Check for experimental codecs

auto_editor/analyze.py

@@ -27,6 +27,16 @@
     from auto_editor.utils.log import Log
 
 
+__all__ = (
+    "LevelError",
+    "Levels",
+    "check_audio",
+    "check_video",
+    "iter_audio",
+    "iter_motion",
+)
+
+
 class LevelError(Exception):
     pass
 
@@ -69,45 +79,58 @@ def mut_remove_large(
             active = False
 
 
-def iter_audio(src, tb: Fraction, stream: int = 0) -> Iterator[np.float32]:
-    fifo = AudioFifo()
-    try:
-        container = av.open(src.path, "r")
-        audio_stream = container.streams.audio[stream]
-        sample_rate = audio_stream.rate
+def check_audio(src: FileInfo, stream: int, log: Log) -> av.AudioStream:
+    container = av.open(src.path, "r")
+    audio_stream = container.streams.audio[stream]
 
-        exact_size = (1 / tb) * sample_rate
-        accumulated_error = 0
+    codec = audio_stream.codec
+    if codec.experimental:
+        log.error(f"`{codec.name}` cannot be used because it is experimental")
 
-        # Resample so that audio data is between [-1, 1]
-        resampler = av.AudioResampler(
-            av.AudioFormat("flt"), audio_stream.layout, sample_rate
-        )
+    return audio_stream
 
-        for frame in container.decode(audio=stream):
-            frame.pts = None  # Skip time checks
 
-            for reframe in resampler.resample(frame):
-                fifo.write(reframe)
+def check_video(src: FileInfo, stream: int, log: Log) -> av.VideoStream:
+    container = av.open(src.path, "r")
+    video_stream = container.streams.video[stream]
 
-            while fifo.samples >= ceil(exact_size):
-                size_with_error = exact_size + accumulated_error
-                current_size = round(size_with_error)
-                accumulated_error = size_with_error - current_size
+    codec = video_stream.codec
+    if codec.experimental:
+        log.error(f"`{codec.name}` cannot be used because it is experimental")
 
-                audio_chunk = fifo.read(current_size)
-                assert audio_chunk is not None
-                arr = audio_chunk.to_ndarray().flatten()
-                yield np.max(np.abs(arr))
+    return video_stream
 
-    finally:
-        container.close()
 
+def iter_audio(audio_stream: av.AudioStream, tb: Fraction) -> Iterator[np.float32]:
+    fifo = AudioFifo()
+    sr = audio_stream.rate
+
+    exact_size = (1 / tb) * sr
+    accumulated_error = Fraction(0)
 
-def iter_motion(src, tb, stream: int, blur: int, width: int) -> Iterator[np.float32]:
-    container = av.open(src.path, "r")
+    # Resample so that audio data is between [-1, 1]
+    resampler = av.AudioResampler(av.AudioFormat("flt"), audio_stream.layout, sr)
 
-    video = container.streams.video[stream]
+    for frame in audio_stream.decode():
+        frame.pts = None  # Skip time checks
+
+        for reframe in resampler.resample(frame):
+            fifo.write(reframe)
+
+        while fifo.samples >= ceil(exact_size):
+            size_with_error = exact_size + accumulated_error
+            current_size = round(size_with_error)
+            accumulated_error = size_with_error - current_size
+
+            audio_chunk = fifo.read(current_size)
+            assert audio_chunk is not None
+            arr = audio_chunk.to_ndarray().flatten()
+            yield np.max(np.abs(arr))
+
+
+def iter_motion(
+    video: av.VideoStream, tb: Fraction, blur: int, width: int
+) -> Iterator[np.float32]:
     video.thread_type = "AUTO"
 
     prev_frame = None
@@ -125,7 +148,7 @@ def iter_motion(src, tb, stream: int, blur: int, width: int) -> Iterator[np.floa
         graph.add("buffersink"),
     ).configure()
 
-    for unframe in container.decode(video):
+    for unframe in video.decode():
         if unframe.pts is None:
             continue
 
@@ -151,8 +174,6 @@ def iter_motion(src, tb, stream: int, blur: int, width: int) -> Iterator[np.floa
         prev_frame = current_frame
         prev_index = index
 
-    container.close()
-
 
 def obj_tag(path: Path, kind: str, tb: Fraction, obj: Sequence[object]) -> str:
     mod_time = int(path.stat().st_mtime)
@@ -175,7 +196,8 @@ def media_length(self) -> int:
             if (arr := self.read_cache("audio", (0,))) is not None:
                 return len(arr)
 
-            result = sum(1 for _ in iter_audio(self.src, self.tb, 0))
+            audio_stream = check_audio(self.src, 0, self.log)
+            result = sum(1 for _ in iter_audio(audio_stream, self.tb))
             self.log.debug(f"Audio Length: {result}")
             return result
 
@@ -253,7 +275,9 @@ def audio(self, stream: int) -> NDArray[np.float32]:
 
         result = np.zeros((inaccurate_dur), dtype=np.float32)
         index = 0
-        for value in iter_audio(self.src, self.tb, stream):
+
+        audio_stream = check_audio(self.src, stream, self.log)
+        for value in iter_audio(audio_stream, self.tb):
             if index > len(result) - 1:
                 result = np.concatenate(
                     (result, np.zeros((len(result)), dtype=np.float32))
@@ -286,7 +310,9 @@ def motion(self, stream: int, blur: int, width: int) -> NDArray[np.float32]:
 
         result = np.zeros((inaccurate_dur), dtype=np.float32)
         index = 0
-        for value in iter_motion(self.src, self.tb, stream, blur, width):
+
+        video = check_video(self.src, stream, self.log)
+        for value in iter_motion(video, self.tb, blur, width):
             if index > len(result) - 1:
                 result = np.concatenate(
                     (result, np.zeros((len(result)), dtype=np.float32))

auto_editor/subcommands/levels.py

@@ -7,7 +7,7 @@
 
 import numpy as np
 
-from auto_editor.analyze import LevelError, Levels, iter_audio, iter_motion
+from auto_editor.analyze import *
 from auto_editor.ffwrapper import initFileInfo
 from auto_editor.lang.palet import env
 from auto_editor.lib.contracts import is_bool, is_nat, is_nat1, is_str, is_void, orc
@@ -130,9 +130,11 @@ def main(sys_args: list[str] = sys.argv[1:]) -> None:
         levels = Levels(src, tb, bar, False, log, strict=True)
         try:
             if method == "audio":
-                print_arr_gen(iter_audio(src, tb, **obj))
+                audio_stream = check_audio(src, obj["stream"], log)
+                print_arr_gen(iter_audio(audio_stream, tb))
             elif method == "motion":
-                print_arr_gen(iter_motion(src, tb, **obj))
+                video_stream = check_video(src, obj["stream"], log)
+                print_arr_gen(iter_motion(video_stream, tb, obj["blur"], obj["width"]))
             elif method == "subtitle":
                 print_arr(levels.subtitle(**obj))
             elif method == "none":

pyproject.toml

@@ -10,7 +10,7 @@ authors = [{ name = "WyattBlue", email = "wyattblue@auto-editor.com" }]
 requires-python = ">=3.10,<3.14"
 dependencies = [
     "numpy>=1.24,<3.0",
-    "pyav==14.0.0rc4",
+    "pyav==14.*",
 ]
 keywords = [
     "video", "audio", "media", "editor", "editing",