GH-5: Refactor the functions' positions (GH-9)

Author: ArtyomVancyan

ffmpeg/__init__.py

@@ -1,61 +1 @@
-# import asyncio
-
-import imageio
-from PIL import Image
-
-
-class Ffmpeg:
-
-    def __init__(self, filename):
-        self.filename = filename
-        self.reader = imageio.get_reader(filename)
-        self.meta = self.reader.get_meta_data()
-        self.duration = self.meta.get("duration")
-        self.fps = self.meta.get("fps")
-
-    def get_frame(self, t):
-        pos = int(self.fps * t + 0.00001) + 1
-        return self.reader.get_data(pos)
-
-    def get_frame_buffer(self, t=0):
-        image = self.get_frame(t)
-        image = image.astype("uint8")
-        return Image.fromarray(image)
-
-    def close(self):
-        self.reader.close()
-
-    def __del__(self):
-        self.close()
-
-# class Ffmpeg:
-#
-#     def __init__(self, filename):
-#         self.filename = filename
-#
-#     def get_frame(self, t):
-#         pos = int(self.fps * t + 0.00001) + 1
-#         return self.reader.get_data(pos)
-#
-#     def get_frame_buffer(self, t=0):
-#         # image = await asyncio.to_thread(self.get_frame, t)
-#         image = self.get_frame(t)
-#         image = image.astype("uint8")
-#         return Image.fromarray(image)
-#
-#     async def __aenter__(self):
-#         self.reader = await asyncio.to_thread(
-#             imageio.get_reader, self.filename)
-#         self.meta = self.reader.get_meta_data()
-#         self.duration = self.meta.get("duration")
-#         self.fps = self.meta.get("fps")
-#         return self
-#
-#     async def __aexit__(self, *_):
-#         self.close()
-#
-#     def close(self):
-#         self.reader.close()
-#
-#     def __del__(self):
-#         self.close()
+from .ffmpeg import FFMpeg

ffmpeg/ffmpeg.py

@@ -11,39 +11,72 @@
 FFMPEG_BINARY = imageio_ffmpeg.get_ffmpeg_exe()
 
 
-def cross_platform_popen_params(popen_params):
-    if os.name == "nt":
-        popen_params["creationflags"] = 0x08000000
-    return popen_params
-
-
-class Ffmpeg:
-
+class FFMpeg:
     def __init__(self, filename):
-
-        self.filename = filename
-
-        duration, size = image_meta(filename)
+        duration, size = self.parse_metadata(filename)
         self.area = reduce(int.__mul__, size)
         self.bytes = self.area * 4
+        self.filename = filename
         self.duration = duration
         self.size = size
 
+    @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):
+        duration_regex = r"duration[^\n]+([0-9][0-9]:[0-9][0-9]:[0-9][0-9].[0-9][0-9])"
+        time = re.search(duration_regex, stdout, re.M | re.I).group(1)
+        time = [float(part.replace(",", ".")) for part in time.split(":")]
+        return sum(mult * part for mult, part in zip((3600, 60, 1), time))
+
+    @staticmethod
+    def _parse_size(stdout):
+        size_regex = r"\s(\d+)x(\d+)[,\s]"
+        match_size = re.search(size_regex, stdout, re.M)
+        return tuple(map(int, match_size.groups()))
+
+    def parse_metadata(self, filename):
+        meta = immeta(filename)
+        duration, size = meta.get("duration"), meta.get("size")
+
+        if not all([duration, size]):
+            cmd = [FFMPEG_BINARY, "-hide_banner", "-i", filename]
+
+            popen_params = self.cross_platform_popen_params()
+            process = subprocess.Popen(cmd, **popen_params)
+            _, stderr = process.communicate()
+            stdout = stderr.decode("utf8", errors="ignore")
+
+            process.terminate()
+            del process
+
+            duration = self._parse_duration(stdout)
+            size = self._parse_size(stdout)
+
+        return duration, size
+
     @staticmethod
     def frame_to_buffer(image):
         image = image.astype("uint8")
         return Image.fromarray(image)
 
-    def get_frame(self, start_time=0):
+    def get_frame(self, start_time):
         if start_time != 0:
             offset = min(1, start_time)
             i_arg = [
-                "-ss",
-                "%.06f" % (start_time - offset),
-                "-i",
-                self.filename,
-                "-ss",
-                "%.06f" % offset,
+                "-ss", "%.06f" % (start_time - offset),
+                "-i", self.filename,
+                "-ss", "%.06f" % offset,
             ]
         else:
             i_arg = ["-i", self.filename]
@@ -52,32 +85,17 @@ def get_frame(self, start_time=0):
                 [FFMPEG_BINARY]
                 + i_arg
                 + [
-                    "-loglevel",
-                    "error",
-                    "-f",
-                    "image2pipe",
-                    "-vf",
-                    "scale=%d:%d" % tuple(self.size),
-                    "-sws_flags",
-                    "bicubic",
-                    "-pix_fmt",
-                    "rgba",
-                    "-vcodec",
-                    "rawvideo",
-                    "-",
+                    "-loglevel", "error",
+                    "-f", "image2pipe",
+                    "-vf", "scale=%d:%d" % tuple(self.size),
+                    "-sws_flags", "bicubic",
+                    "-pix_fmt", "rgba",
+                    "-vcodec", "rawvideo", "-",
                 ]
         )
-        popen_params = cross_platform_popen_params(
-            {
-                "bufsize": self.bytes + 100,
-                "stdout": subprocess.PIPE,
-                "stderr": subprocess.PIPE,
-                "stdin": subprocess.DEVNULL,
-            }
-        )
 
+        popen_params = self.cross_platform_popen_params(self.bytes + 100)
         process = subprocess.Popen(cmd, **popen_params)
-
         buffer = process.stdout.read(self.bytes)
 
         process.terminate()
@@ -91,49 +109,3 @@ def get_frame(self, start_time=0):
         result.shape = (*self.size[::-1], len(buffer) // self.area)
 
         return result
-
-
-def convert_to_seconds(time):
-    time = [float(part.replace(",", ".")) for part in time.split(":")]
-    return sum(mult * part for mult, part in zip((3600, 60, 1), time))
-
-
-def parse_duration(stdout):
-    duration_regex = r"duration[^\n]+([0-9][0-9]:[0-9][0-9]:[0-9][0-9].[0-9][0-9])"
-    match_duration = re.search(duration_regex, stdout, re.M | re.I)
-    return convert_to_seconds(match_duration.group(1))
-
-
-def parse_size(stdout):
-    size_regex = r"\s(\d+)x(\d+)[,\s]"
-    match_size = re.search(size_regex, stdout, re.M)
-    return tuple(map(int, match_size.groups()))
-
-
-def image_meta(filename):
-    meta = immeta(filename)
-    duration, size = meta.get("duration"), meta.get("size")
-
-    if not all([duration, size]):
-        cmd = [FFMPEG_BINARY, "-hide_banner", "-i", filename]
-
-        popen_params = cross_platform_popen_params(
-            {
-                "bufsize": 10 ** 5,
-                "stdout": subprocess.PIPE,
-                "stderr": subprocess.PIPE,
-                "stdin": subprocess.DEVNULL,
-            }
-        )
-
-        process = subprocess.Popen(cmd, **popen_params)
-        _, stderr = process.communicate()
-        stdout = stderr.decode("utf8", errors="ignore")
-
-        process.terminate()
-        del process
-
-        duration = parse_duration(stdout)
-        size = parse_size(stdout)
-
-    return duration, size

thumbnails.py

@@ -4,7 +4,7 @@
 
 from PIL import Image
 
-from ffmpeg.ffmpeg import Ffmpeg
+from ffmpeg import FFMpeg
 
 width, height = 300, 200
 interval = 20
@@ -44,7 +44,7 @@ def worker(video):
 
 
 async def main():
-    fps = [Ffmpeg(fp) for fp in files]
+    fps = [FFMpeg(fp) for fp in files]
 
     with concurrent.futures.ProcessPoolExecutor() as executor:
         executor.map(worker, fps)