GH-26: Handle all possible inputs and outputs (GH-27)

* Fix the names to correspond to the functionality
* Group CLI-related stuff together
* Convert inputs into one format
* Implement output directory and metadata file path calculators
* Handle file existence and implement skip/overwrite logic

Author: ArtyomVancyan

thumbnails/__init__.py

@@ -1,111 +1,18 @@
-import json
-import os
-import shutil
-from datetime import timedelta
-from distutils.dir_util import copy_tree
-
-from PIL import Image
-
-from .formatter import FileFormatter
-from .formatter import FormatterFactory
-from .thumbnails import Thumbnails
-
-DEFAULT_AS = "vtt"
-DEFAULT_COMPRESS = 1.
-DEFAULT_INTERVAL = 1.
-DEFAULT_BASEPATH = ""
-
-
-def register_formatter(typename):
-    """Register a new thumbnail formatter to the factory."""
-
-    def _register_factory(cls):
-        if not issubclass(cls, FileFormatter):
-            raise ValueError("The formatter must implement the FileFormatter interface.")
-
-        cls.extension = typename
-        FormatterFactory.thumbnails[typename] = cls
-        return cls
-
-    return _register_factory
-
-
-@register_formatter("vtt")
-class VTTFormatter(FileFormatter):
-    """Implements the methods for generating thumbnails in the WebVTT format."""
-
-    def __init__(self, video):
-        super().__init__(video)
-        self._master_name = self.filename + ".png"
-
-    def prepare_thumbnails(self):
-        _thumbnails = self.thumbnails(True)
-        master = Image.new(mode="RGBA", size=next(_thumbnails))
-
-        for frame, start, end, x, y in self.thumbnails():
-            with Image.open(frame) as image:
-                image = image.resize((self.width, self.height), Image.ANTIALIAS)
-                master.paste(image, (x, y))
-
-        master.save(self._master_name)
-        self.tempdir.cleanup()
-
-    def generate(self):
-        def _format_time(secs):
-            delta = timedelta(seconds=secs)
-            return ("0%s.000" % delta)[:12]
-
-        _lines = ["WEBVTT\n\n"]
-        _img_src = self.basepath + self._master_name
-
-        for frame, start, end, x, y in self.thumbnails():
-            _thumbnail = "%s --> %s\n%s#xywh=%d,%d,%d,%d\n\n" % (
-                _format_time(start), _format_time(end),
-                _img_src, x, y, self.width, self.height
-            )
-            _lines.append(_thumbnail)
-
-        with open(self.thumbnail_file, "w") as fp:
-            fp.writelines(_lines)
-
-
-@register_formatter("json")
-class JSONFormatter(FileFormatter):
-    """Implements the methods for generating thumbnails in the JSON format."""
-
-    def __init__(self, video):
-        super().__init__(video)
-        self._outdir = "outdir"  # temp dirname
-
-    def prepare_thumbnails(self):
-        if os.path.isdir(self._outdir):
-            shutil.rmtree(self._outdir)
-        copy_tree(self.tempdir.name, self._outdir)
-        self.tempdir.cleanup()
-
-    def generate(self):
-        _content = {}
-
-        for frame, start, end, x, y in self.thumbnails():
-            frame = self._outdir + os.sep + os.path.split(frame)[1]
-            with Image.open(frame) as image:
-                image.resize((self.width, self.height), Image.ANTIALIAS).save(frame)
-                _thumbnail = {
-                    "src": self.basepath + frame,
-                    "width": "%spx" % self.width,
-                }
-                _content[int(start)] = _thumbnail
-
-        with open(self.thumbnail_file, "w") as fp:
-            json.dump(_content, fp, indent=2)
-
+from .thumbnail import Thumbnail
+from .thumbnail import ThumbnailExistsError
+from .thumbnail import ThumbnailFactory
+from .thumbnail import ThumbnailJSON
+from .thumbnail import ThumbnailVTT
+from .thumbnail import register_thumbnail
+from .video import Video
 
 __version__ = "v1.0"
 __all__ = (
-    "register_formatter",
-    "FormatterFactory",
-    "FileFormatter",
-    "JSONFormatter",
-    "VTTFormatter",
-    "Thumbnails",
+    "Thumbnail",
+    "ThumbnailExistsError",
+    "ThumbnailFactory",
+    "ThumbnailJSON",
+    "ThumbnailVTT",
+    "register_thumbnail",
+    "Video",
 )

thumbnails/__main__.py

@@ -1,69 +1,72 @@
 import concurrent.futures
 import functools
+import itertools
 import os
 
 import click
 
-from . import DEFAULT_AS
-from . import DEFAULT_BASEPATH
-from . import DEFAULT_COMPRESS
-from . import DEFAULT_INTERVAL
-from . import FormatterFactory
-from . import Thumbnails
-from . import __version__
+from . import Thumbnail
+from . import ThumbnailExistsError
+from . import ThumbnailFactory
+from . import Video
+from .cli import cli
 
 
-def worker(video, as_):
+def worker(video, fmt, base, skip, output):
     """Generate thumbnails for a single video."""
-    video.extract_frames()
-    formatter = FormatterFactory.create_formatter(as_, video)
-    formatter.prepare_thumbnails()
-    formatter.generate()
+    try:
+        thumbnail = ThumbnailFactory.create_thumbnail(fmt, video, base, skip, output)
+    except ThumbnailExistsError:
+        return print("Skipping '%s'" % video.filepath)
+    thumbnail.prepare_frames()
+    thumbnail.generate()
 
 
-class _ThumbnailsCLI(click.Command):
-    """This class overrides the usages section of the help message."""
-
-    def format_usage(self, ctx, formatter):
-        usages = (
-            "[OPTIONS] INPUT_DIR OUTPUT_DIR",
-            "[OPTIONS] INPUT_FILE OUTPUT_FILE",
-            "[OPTIONS] INPUT_FILES... OUTPUT_DIR",
-        )
-        formatter.write_usage(ctx.command_path, "\n\t\t\t".join(usages), prefix="Usages: ")
+@cli
+def main(compress=None, interval=None, base=None, inputs=None, output=None, skip=None, **kwargs):
+    """TODO: This section will be completed after fixing the issue #26."""
 
+    def listdir(directory):
+        """Lists all files in the given directory with absolute paths."""
+        for basedir, _, files in os.walk(directory):
+            for file in filter(os.path.isfile, files):
+                yield os.path.abspath(os.path.join(basedir, file))
 
-# This defines a set of supported values for the particular option of the CLI.
-_type = click.Choice(FormatterFactory.thumbnails.keys(), case_sensitive=False)
+    if all(map(os.path.isfile, inputs)):
+        inputs = set(map(os.path.abspath, inputs))
+    elif all(map(os.path.isdir, inputs)):
+        inputs = set(itertools.chain(*map(listdir, inputs)))
+    else:
+        exit("Inputs must be all files or all directories.")
 
+    fmt = kwargs.pop("format")
+    inputs = dict(zip(map(lambda i: Thumbnail.metadata_path(i, output, fmt), inputs), inputs))
 
-@click.command(cls=_ThumbnailsCLI)
-@click.option("--as", "-F", default=DEFAULT_AS, type=_type, help="Output format. Default is %s." % DEFAULT_AS)
-@click.option("--compress", "-C", default=DEFAULT_COMPRESS, help="The image scale coefficient. A number from 0 to 1.")
-@click.option("--interval", "-I", default=DEFAULT_INTERVAL, help="The interval between neighbor thumbnails in seconds.")
-@click.option("--basepath", "-B", default=DEFAULT_BASEPATH, help="The prefix of the thumbnails path can be customized.")
-@click.argument("inputs", required=True, type=click.Path(), nargs=-1)
-@click.argument("output", required=True, type=click.Path(), nargs=1)
-@click.version_option(__version__)
-def thumbnails_cli(compress, interval, basepath, inputs, output, **kwargs):
-    """TODO: This section will be completed after fixing the issue #26."""
-    as_ = kwargs.pop("as")
-    output_is_directory = all((len(inputs) > 1, *map(os.path.isfile, inputs))) or os.path.isdir(inputs[0])
+    if not skip and any(map(os.path.exists, inputs.keys())):
+        skip = not click.confirm("Do you want to overwrite already existing files?")
 
     with concurrent.futures.ThreadPoolExecutor() as executor:
         videos = executor.map(
             functools.partial(
-                Thumbnails,
+                Video,
                 compress=compress,
                 interval=interval,
-                basepath=basepath
             ),
-            inputs,
+            inputs.values(),
         )
 
     with concurrent.futures.ProcessPoolExecutor() as executor:
-        executor.map(functools.partial(worker, as_=as_), videos)
+        executor.map(
+            functools.partial(
+                worker,
+                fmt=fmt,
+                base=base,
+                skip=skip,
+                output=output,
+            ),
+            videos,
+        )
 
 
 if __name__ == "__main__":
-    thumbnails_cli()
+    main()

thumbnails/cli.py

@@ -0,0 +1,43 @@
+import functools
+
+import click
+
+from . import ThumbnailFactory
+from . import __version__
+
+# Default values of the particular option of the CLI.
+DEFAULT_BASE = ""
+DEFAULT_SKIP = False
+DEFAULT_OUTPUT = None
+DEFAULT_FORMAT = "vtt"
+DEFAULT_COMPRESS = 1.0
+DEFAULT_INTERVAL = 1.0
+
+# Help messages of the particular option of the CLI.
+HELP_BASE = "The prefix of the thumbnails path can be customized."
+HELP_SKIP = "Skip the existing thumbnails. Default is not set."
+HELP_OUTPUT = "The output directory. Default is the current directory."
+HELP_FORMAT = "Output format. Default is %s." % DEFAULT_FORMAT
+HELP_COMPRESS = "The image scale coefficient. A number from 0 to 1."
+HELP_INTERVAL = "The interval between neighbor thumbnails in seconds."
+
+# This defines a choice of supported values for the '--format' option of the CLI.
+format_choice = click.Choice(ThumbnailFactory.thumbnails.keys(), case_sensitive=False)
+
+
+def cli(func):
+    @click.command()
+    @click.option("--compress", "-C", default=DEFAULT_COMPRESS, help=HELP_COMPRESS)
+    @click.option("--interval", "-I", default=DEFAULT_INTERVAL, help=HELP_INTERVAL)
+    @click.option("--base", "-B", default=DEFAULT_BASE, help=HELP_BASE)
+    @click.option("--skip", "-S", default=DEFAULT_SKIP, help=HELP_SKIP, is_flag=True)
+    @click.option("--output", "-O", default=DEFAULT_OUTPUT, type=click.Path(), help=HELP_OUTPUT)
+    @click.option("--format", "-F", default=DEFAULT_FORMAT, type=format_choice, help=HELP_FORMAT)
+    @click.argument("inputs", required=True, type=click.Path(), nargs=-1)
+    @click.version_option(__version__, "-V", "--version")
+    @click.help_option("-h", "--help")
+    @functools.wraps(func)
+    def wrapper(*args, **kwargs):
+        return func(*args, **kwargs)
+
+    return wrapper

thumbnails/ffmpeg.py

@@ -11,22 +11,10 @@
 class _FFMpeg:
     """This class is used to parse the metadata of a video file."""
 
-    def __init__(self, filename):
-        duration, self.size = self._parse_metadata(filename)
+    def __init__(self, filepath):
+        duration, self.size = self._parse_metadata(filepath)
         self.duration = int(duration + 1)
 
-    @staticmethod
-    def _cross_platform_popen_params(bufsize=100000):
-        popen_params = {
-            "bufsize": bufsize,
-            "stdout": subprocess.PIPE,
-            "stderr": subprocess.PIPE,
-            "stdin": subprocess.DEVNULL,
-        }
-        if os.name == "nt":
-            popen_params["creationflags"] = 0x08000000
-        return popen_params
-
     @staticmethod
     def _parse_duration(stdout):
         """Parse the duration of a video from stdout."""
@@ -42,21 +30,25 @@ def _parse_size(stdout):
         match_size = re.search(size_regex, stdout, re.M)
         return tuple(map(int, match_size.groups()))
 
-    def _parse_metadata(self, filename):
+    def _parse_metadata(self, filepath):
         """Parse the metadata of a video file."""
-        meta = immeta(filename)
+        meta = immeta(filepath)
         duration, size = meta.get("duration"), meta.get("size")
 
         if not all((duration, size)):
             # Parse the metadata of the video formats
             # that are not supported by imageio.
 
-            cmd = (ffmpeg_bin, "-hide_banner", "-i", filename)
-
-            popen_params = self._cross_platform_popen_params()
-            process = subprocess.Popen(cmd, **popen_params)
+            process = subprocess.Popen(
+                (ffmpeg_bin, "-hide_banner", "-i", filepath),
+                bufsize=100000,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                stdin=subprocess.DEVNULL,
+                creationflags=0x08000000 if os.name == "nt" else 0,
+            )
             _, stderr = process.communicate()
-            stdout = stderr.decode("utf8", errors="ignore")
+            stdout = (stderr or b"").decode("utf8", errors="ignore")
 
             duration = self._parse_duration(stdout)
             size = self._parse_size(stdout)