Add ffmagick

1 files changed, 210 insertions, 0 deletions

diff --git a/ffmagick/ffmagick.py b/ffmagick/ffmagick.py
new file mode 100644
index 0000000..9bc01d3
--- /dev/null
+++ b/ffmagick/ffmagick.py
@@ -0,0 +1,210 @@
+#!/usr/bin/python
+
+import sys, os
+import argparse
+import subprocess
+from enum import Enum
+import tomllib as toml
+
+# Simple progress display, in a similar interface to the Progress library.
+class ProgressBar:
+    def __init__(self, text, max):
+        self.max = max
+        self.text = text
+        self.current = 0
+
+    def next(self):
+        self.current += 1
+        print("\r{} {}/{}".format(self.text, self.current, self.max), flush=True,
+              end="")
+
+    def finish(self):
+        print(flush=True)
+
+try:
+    from progress.bar import Bar as Progress
+    HAS_PROGRESS = True
+except:
+    HAS_PROGRESS = False
+    Progress = ProgressBar
+
+class Processor(Enum):
+    FFMPEG = 1
+    IMAGEMAGICK = 2
+
+PROFILES = [
+    {
+        "name": "half_small",
+        "desc": "Scale image to half, and compress to low-quality AVIF",
+        "processor": "IMAGEMAGICK",
+        "command": "-scale 50% -quality 70",
+        "output_ext": "avif",
+    }, {
+        "name": "intermediate",
+        "desc": "Convert video to ProRes",
+        "processor": "FFMPEG",
+        "command": "-c:v prores_ks -profile:v 3 -qscale:v 9 -vendor apl0 -pix_fmt yuv422p10le -c:a copy",
+        "output_ext": "mov",
+    }]
+
+class Profile:
+    def __init__(self, profile_dict):
+        self.name = profile_dict.get("name", "")
+        self.desc = profile_dict.get("desc", "")
+        self.processor = Processor[profile_dict.get("processor", "IMAGEMAGICK")]
+        self.command = profile_dict.get("command", "")
+        self.output_ext = profile_dict.get("output_ext", "")
+
+    @classmethod
+    def get(cls, name):
+        for p in PROFILES:
+            if p["name"] == name:
+                return cls(p)
+
+    def __str__(self):
+        return "{} ({}): {}\n{}".format(self.name, self.processor.name,
+                                        self.desc, self.command)
+
+def getConfigFiles():
+    files = []
+    f = os.path.join(os.path.dirname(os.path.realpath(__file__)),
+                     "ffmagick.toml")
+    files.append((f, os.path.exists(f)))
+    f = "/etc/ffmagick.toml"
+    files.append((f, os.path.exists(f)))
+    if "HOME" in os.environ:
+        home = os.environ["HOME"]
+        if os.name == "nt" and home.endswith(':') and home.find('/') == -1 \
+           and home.find('\\') == -1:
+            # $HOME is a drive letter (e.g. “D:”) on Windows.
+            f = os.path.join(home, os.sep, ".config", "ffmagick.toml")
+        else:
+            f = os.path.join(home, ".config", "ffmagick.toml")
+        files.append((f, os.path.exists(f)))
+    return files
+
+class Configuration:
+    def __init__(self):
+        self._imagemagick_bin = None
+        self._ffmpeg_bin = None
+        self.profiles = []
+
+    # Get configuration from files in the system.
+    @classmethod
+    def get(cls):
+        conf = cls()
+        for f in getConfigFiles():
+            if f[1]:
+                conf.merge(Configuration.readFromFile(f[0]))
+        return conf
+
+    @property
+    def imagemagick_bin(self):
+        if self._imagemagick_bin is None:
+            return "magick"
+        else:
+            return self._imagemagick_bin
+
+    @property
+    def ffmpeg_bin(self):
+        if self._ffmpeg_bin is None:
+            return "ffmpeg"
+        else:
+            return self._ffmpeg_bin
+
+    @classmethod
+    def readFromFile(cls, filename):
+        with open(filename, 'rb') as f:
+            data = toml.load(f)
+
+        result = cls()
+        for key in data:
+            item = data[key]
+            if isinstance(item, dict):
+                p = Profile(item)
+                p.name = key
+                result.profiles.append(p)
+            else:
+                if key == "imagemagick_bin":
+                    result._imagemagick_bin = item
+                elif key == "ffmpeg_bin":
+                    result._ffmpeg_bin = item
+                else:
+                    print("ERROR: invalid configuration key:", key)
+        return result
+
+    def merge(self, conf):
+        if conf._imagemagick_bin is not None:
+            self._imagemagick_bin = conf._imagemagick_bin
+        if conf._ffmpeg_bin is not None:
+            self._ffmpeg_bin = conf._ffmpeg_bin
+        self.profiles = conf.profiles + self.profiles
+
+    def getProfile(self, name):
+        for p in self.profiles:
+            if p.name == name:
+                return p
+
+def runOnce(config, profile, input_file):
+    output_file = os.path.splitext(input_file)[0] + "." + profile.output_ext
+    if profile.processor == Processor.FFMPEG:
+        cmd = "\"{}\" -y -i \"{}\" {} \"{}\"".format(
+            config.ffmpeg_bin, input_file, profile.command, output_file)
+    elif profile.processor == Processor.IMAGEMAGICK:
+        cmd = "\"{}\" \"{}\" {} \"{}\"".format(
+            config.imagemagick_bin, input_file, profile.command, output_file)
+    return subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+                          text=True)
+
+def runAll(config, profile, input_files):
+    bar = Progress("Processing", max=len(input_files))
+    for f in input_files:
+        bar.next()
+        result = runOnce(config, profile, f)
+        if result.returncode != 0:
+            bar.finish()
+            print("ERROR: failed to process {}. Log:\n")
+            print(result.stdout)
+            return False
+    bar.finish()
+    return True
+
+def main():
+    import argparse
+    parser = argparse.ArgumentParser(
+        prog='FFmagick', description='What the program does')
+    parser.add_argument('files', metavar="FILE", nargs='*',
+                        help="File to process")
+    parser.add_argument('-p', "--profile", metavar="PROFILE", dest="profile",
+                        help="Specify the profile to process the files")
+    parser.add_argument("-l", "--list-profiles", action="store_true",
+                        dest="should_list_profiles",
+                        help="List all profiles")
+    parser.add_argument("--list-config-files", action="store_true",
+                        dest="should_list_conf_files",
+                        help="List all config files found")
+    args = parser.parse_args()
+
+    config = Configuration.get()
+    if args.should_list_profiles:
+        for p in config.profiles:
+            print(p)
+            print()
+        return 0
+
+    if args.should_list_conf_files:
+        for f in getConfigFiles():
+            print("✅" if f[1] else "❌", f[0])
+        return 0
+
+    if "profile" not in args:
+        print("Profile needed")
+        return 1
+
+    if not runAll(config, config.getProfile(args.profile), args.files):
+        return 1
+
+    return 0
+
+if __name__ == "__main__":
+    sys.exit(main())