Get motion with pyav filters, remove pixeldiff

auto_editor/analyze.py

@@ -58,11 +58,6 @@
     pAttr("blur", 9, is_nat),
     pAttr("width", 400, is_nat1),
 )
-pixeldiff_builder = pAttrs(
-    "pixeldiff",
-    pAttr("threshold", 1, is_nat),
-    pAttr("stream", 0, is_nat),
-)
 subtitle_builder = pAttrs(
     "subtitle",
     pAttr("pattern", Required, is_str),
@@ -74,7 +69,6 @@
 builder_map = {
     "audio": audio_builder,
     "motion": motion_builder,
-    "pixeldiff": pixeldiff_builder,
     "subtitle": subtitle_builder,
 }
 
@@ -353,7 +347,6 @@ def cleanhtml(raw_html: str) -> str:
 
     def motion(self, s: int, blur: int, width: int) -> NDArray[np.float64]:
         import av
-        from PIL import ImageChops, ImageFilter
 
         av.logging.set_level(av.logging.PANIC)
 
@@ -370,28 +363,20 @@ def motion(self, s: int, blur: int, width: int) -> NDArray[np.float64]:
         stream = container.streams.video[s]
         stream.thread_type = "AUTO"
 
-        if (
-            stream.duration is None
-            or stream.time_base is None
-            or stream.average_rate is None
-        ):
-            inaccurate_dur = 1
-        else:
-            inaccurate_dur = int(
-                stream.duration * stream.time_base * stream.average_rate
-            )
-
+        inaccurate_dur = 1 if stream.duration is None else stream.duration
         self.bar.start(inaccurate_dur, "Analyzing motion")
 
-        prev_image = None
-        image = None
+        prev_frame = None
+        current_frame = None
         total_pixels = self.src.videos[0].width * self.src.videos[0].height
         index = 0
 
         graph = av.filter.Graph()
         link_nodes(
             graph.add_buffer(template=stream),
             graph.add("scale", f"{width}:-1"),
+            graph.add("format", "gray"),
+            graph.add("gblur", f"sigma={blur}"),
             graph.add("buffersink"),
         )
         graph.configure()
@@ -402,98 +387,31 @@ def motion(self, s: int, blur: int, width: int) -> NDArray[np.float64]:
             graph.push(unframe)
             frame = graph.pull()
 
-            prev_image = image
-
-            assert isinstance(frame.time, float)
+            # Showing progress ...
+            assert frame.time is not None
             index = int(frame.time * self.tb)
-            self.bar.tick(index)
+            if frame.pts is not None:
+                self.bar.tick(frame.pts)
+
+            current_frame = frame.to_ndarray()
 
             if index > len(threshold_list) - 1:
                 threshold_list = np.concatenate(
                     (threshold_list, np.zeros((len(threshold_list)), dtype=np.float64)),
                     axis=0,
                 )
 
-            image = frame.to_image().convert("L")
-
-            if blur > 0:
-                image = image.filter(ImageFilter.GaussianBlur(radius=blur))
-
-            if prev_image is not None:
-                count = np.count_nonzero(ImageChops.difference(prev_image, image))
-
-                threshold_list[index] = count / total_pixels
-
-        self.bar.end()
-        result = threshold_list[:index]
-        del threshold_list
-
-        return self.cache("motion", mobj, result)
-
-    def pixeldiff(self, s: int) -> NDArray[np.uint64]:
-        import av
-        from PIL import ImageChops
-
-        av.logging.set_level(av.logging.PANIC)
-
-        pobj = {"stream": s}
-
-        if s >= len(self.src.videos):
-            raise LevelError(f"pixeldiff: video stream '{s}' does not exist.")
-
-        if (arr := self.read_cache("pixeldiff", pobj)) is not None:
-            return arr
-
-        container = av.open(f"{self.src.path}", "r")
-
-        stream = container.streams.video[s]
-        stream.thread_type = "AUTO"
-
-        if (
-            stream.duration is None
-            or stream.time_base is None
-            or stream.average_rate is None
-        ):
-            inaccurate_dur = 1
-        else:
-            inaccurate_dur = int(
-                stream.duration * stream.time_base * stream.average_rate
-            )
-
-        self.bar.start(inaccurate_dur, "Analyzing pixel diffs")
-
-        prev_image = None
-        image = None
-        index = 0
-
-        threshold_list = np.zeros((1024), dtype=np.uint64)
-
-        for frame in container.decode(stream):
-            prev_image = image
-
-            assert frame.time is not None
-            index = int(frame.time * self.tb)
-            self.bar.tick(index)
-
-            if index > len(threshold_list) - 1:
-                threshold_list = np.concatenate(
-                    (threshold_list, np.zeros((len(threshold_list)), dtype=np.uint64)),
-                    axis=0,
+            if prev_frame is not None:
+                # Use `int6` to avoid underflow with `uint8` datatype
+                diff = np.abs(
+                    prev_frame.astype(np.int16) - current_frame.astype(np.int16)
                 )
+                threshold_list[index] = np.count_nonzero(diff) / total_pixels
 
-            assert isinstance(frame, av.VideoFrame)
-            image = frame.to_image()
-
-            if prev_image is not None:
-                threshold_list[index] = np.count_nonzero(
-                    ImageChops.difference(prev_image, image)
-                )
+            prev_frame = current_frame
 
         self.bar.end()
-        result = threshold_list[:index]
-        del threshold_list
-
-        return self.cache("pixeldiff", pobj, result)
+        return self.cache("motion", mobj, threshold_list[:index])
 
 
 def edit_method(val: str, filesetup: FileSetup, env: Env) -> NDArray[np.bool_]:
@@ -558,8 +476,6 @@ def edit_method(val: str, filesetup: FileSetup, env: Env) -> NDArray[np.bool_]:
                 levels.motion(obj["stream"], obj["blur"], obj["width"]),
                 obj["threshold"],
             )
-        if method == "pixeldiff":
-            return to_threshold(levels.pixeldiff(obj["stream"]), obj["threshold"])
 
         if method == "subtitle":
             return levels.subtitle(

auto_editor/help.py

@@ -41,19 +41,16 @@
     - blur nat? : 9
     - width nat1? : 400
 
- - none  ;  Do not modify the media in anyway; mark all sections as "loud" (1).
- - all/e  ; Cut out everything out; mark all sections as "silent" (0).
-
- - pixeldiff  ; Detect when a certain amount of pixels have changed between frames.
-    - threshold nat? : 1
-    - stream nat? : 0
-
  - subtitle  ; Detect when subtitle matches pattern as a RegEx string.
     - pattern string?
     - stream nat? : 0
     - ignore-case bool? : #f
     - max-count (or/c nat? void?) : (void)
 
+ - none   ; Do not modify the media in anyway; mark all sections as "loud" (1).
+ - all/e  ; Cut out everything out; mark all sections as "silent" (0).
+
+
 Command-line Examples:
   --edit audio
   --edit audio:threshold=4%

auto_editor/lang/palet.py

@@ -51,7 +51,7 @@ class ClosingError(MyError):
 LPAREN, RPAREN, LBRAC, RBRAC, LCUR, RCUR, EOF = "(", ")", "[", "]", "{", "}", "EOF"
 VAL, QUOTE, SEC, DB, DOT, VLIT = "VAL", "QUOTE", "SEC", "DB", "DOT", "VLIT"
 SEC_UNITS = ("s", "sec", "secs", "second", "seconds")
-METHODS = ("audio:", "motion:", "pixeldiff:", "subtitle:")
+METHODS = ("audio:", "motion:", "subtitle:")
 brac_pairs = {LPAREN: RPAREN, LBRAC: RBRAC, LCUR: RCUR}
 
 str_escape = {

auto_editor/subcommands/levels.py

@@ -114,8 +114,6 @@ def main(sys_args: list[str] = sys.argv[1:]) -> None:
                 print_arr(levels.audio(obj["stream"]))
             elif method == "motion":
                 print_arr(levels.motion(obj["stream"], obj["blur"], obj["width"]))
-            elif method == "pixeldiff":
-                print_arr(levels.pixeldiff(obj["stream"]))
             elif method == "subtitle":
                 print_arr(
                     levels.subtitle(

auto_editor/utils/bar.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import sys
+from fractions import Fraction
 from math import floor
 from shutil import get_terminal_size
 from time import localtime, time
@@ -58,7 +59,7 @@ def pretty_time(my_time: float, ampm: bool) -> str:
             return f"{hours:02}:{minutes:02} {ampm_marker}"
         return f"{hours:02}:{minutes:02}"
 
-    def tick(self, index: float) -> None:
+    def tick(self, index: float | Fraction) -> None:
         if self.hide:
             return
 

pyproject.toml

@@ -10,7 +10,6 @@ authors = [{ name = "WyattBlue", email = "[email protected]" }]
 requires-python = ">=3.10"
 dependencies = [
     "numpy>=1.22.0",
-    "pillow==10.2.0",
     "pyav==12.0.2",
     "ae-ffmpeg==1.1.*",
 ]