Simplify making v3 timeline

auto_editor/__main__.py

@@ -68,20 +68,6 @@ def main_options(parser: ArgumentParser) -> ArgumentParser:
         metavar="[START,STOP ...]",
         help="The range of media that will be added in, opposite of --cut-out",
     )
-    parser.add_argument(
-        "--mark-as-loud",
-        type=time_range,
-        nargs="*",
-        metavar="[START,STOP ...]",
-        help='The range that will be marked as "loud"',
-    )
-    parser.add_argument(
-        "--mark-as-silent",
-        type=time_range,
-        nargs="*",
-        metavar="[START,STOP ...]",
-        help='The range that will be marked as "silent"',
-    )
     parser.add_argument(
         "--set-speed-for-range",
         "--set-speed",

auto_editor/analyze.py

@@ -34,6 +34,7 @@
     from fractions import Fraction
     from typing import Any
 
+    from av.filter import FilterContext
     from numpy.typing import NDArray
 
     from auto_editor.ffwrapper import FileInfo
@@ -93,7 +94,7 @@ class LevelError(Exception):
     pass
 
 
-def link_nodes(*nodes: Any) -> None:
+def link_nodes(*nodes: FilterContext) -> None:
     for c, n in zip(nodes, nodes[1:]):
         c.link_to(n)
 
@@ -403,6 +404,7 @@ def motion(self, s: int, blur: int, width: int) -> NDArray[np.float_]:
 
             prev_image = image
 
+            assert isinstance(frame.time, float)
             index = int(frame.time * self.tb)
             self.bar.tick(index)
 

auto_editor/make_layers.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from fractions import Fraction
-from typing import TYPE_CHECKING, Any, NamedTuple
+from typing import TYPE_CHECKING, NamedTuple
 
 import numpy as np
 
@@ -11,7 +11,6 @@
 from auto_editor.lib.data_structs import print_str
 from auto_editor.lib.err import MyError
 from auto_editor.timeline import ASpace, TlAudio, TlVideo, VSpace, v1, v3
-from auto_editor.utils.chunks import Chunks, chunkify, chunks_len, merge_chunks
 from auto_editor.utils.func import mut_margin
 from auto_editor.utils.types import Args, CoerceError, time
 
@@ -20,6 +19,7 @@
 
     from auto_editor.output import Ensure
     from auto_editor.utils.bar import Bar
+    from auto_editor.utils.chunks import Chunks
     from auto_editor.utils.log import Log
 
     BoolList = NDArray[np.bool_]
@@ -128,59 +128,33 @@ def make_timeline(
         tb = inp.get_fps() if args.frame_rate is None else args.frame_rate
         res = inp.get_res() if args.resolution is None else args.resolution
 
-    chunks, vclips, aclips = make_layers(
-        sources,
-        ensure,
-        tb,
-        args.edit_based_on,
-        args.margin,
-        args.cut_out,
-        args.add_in,
-        args.mark_as_silent,
-        args.mark_as_loud,
-        args.set_speed_for_range,
-        args.silent_speed,
-        args.video_speed,
-        bar,
-        temp,
-        log,
-    )
-
-    v1_compatiable = None if inp is None else v1(inp, chunks)
-    return v3(inp, tb, sr, res, args.background, vclips, aclips, v1_compatiable)
-
-
-def make_layers(
-    sources: list[FileInfo],
-    ensure: Ensure,
-    tb: Fraction,
-    method: str,
-    margin: tuple[str, str],
-    cut_out: list[list[str]],
-    add_in: list[list[str]],
-    mark_silent: list[list[str]],
-    mark_loud: list[list[str]],
-    speed_ranges: list[tuple[float, str, str]],
-    silent_speed: float,
-    loud_speed: float,
-    bar: Bar,
-    temp: str,
-    log: Log,
-) -> tuple[Chunks, VSpace, ASpace]:
-    start = 0
-    all_clips: list[list[Clip]] = []
-    all_chunks: list[Chunks] = []
-
     try:
-        start_margin = time(margin[0], tb)
-        end_margin = time(margin[1], tb)
+        start_margin = time(args.margin[0], tb)
+        end_margin = time(args.margin[1], tb)
     except CoerceError as e:
         log.error(e)
 
-    speed_map = [silent_speed, loud_speed]
+    method = args.edit_based_on
+
+    has_loud = np.array([], dtype=np.bool_)
+    src_index = np.array([], dtype=np.int32)
+    concat = np.concatenate
+
+    for i, src in enumerate(sources):
+        filesetup = FileSetup(src, ensure, len(sources) < 2, tb, bar, temp, log)
+
+        edit_result = run_interpreter_for_edit_option(method, filesetup)
+        mut_margin(edit_result, start_margin, end_margin)
+
+        has_loud = concat((has_loud, edit_result))
+        src_index = concat((src_index, np.full(len(edit_result), i, dtype=np.int32)))
+
+    # Setup for handling custom speeds
+    speed_index = has_loud.astype(np.uint)
+    speed_map = [args.silent_speed, args.video_speed]
     speed_hash = {
-        0: silent_speed,
-        1: loud_speed,
+        0: args.silent_speed,
+        1: args.video_speed,
     }
 
     def get_speed_index(speed: float) -> int:
@@ -201,63 +175,98 @@ def parse_time(val: str, arr: NDArray) -> int:
         except CoerceError as e:
             log.error(e)
 
-    def mut_set_range(arr: NDArray, _ranges: list[list[str]], index: Any) -> None:
+    def mut_set_range(arr: NDArray, _ranges: list[list[str]], index: float) -> None:
         for _range in _ranges:
             assert len(_range) == 2
             pair = [parse_time(val, arr) for val in _range]
             arr[pair[0] : pair[1]] = index
 
-    for src in sources:
-        filesetup = FileSetup(src, ensure, len(sources) < 2, tb, bar, temp, log)
-        has_loud = run_interpreter_for_edit_option(method, filesetup)
+    try:
+        if len(args.cut_out) > 0:
+            # always cut out even if 'silent_speed' is not 99,999
+            mut_set_range(speed_index, args.cut_out, get_speed_index(99_999))
+
+        if len(args.add_in) > 0:
+            # set to 'video_speed' index
+            mut_set_range(speed_index, args.add_in, 1.0)
+
+        for speed_range in args.set_speed_for_range:
+            speed = speed_range[0]
+            _range = list(speed_range[1:])
+            mut_set_range(speed_index, [_range], get_speed_index(speed))
+    except CoerceError as e:
+        log.error(e)
 
-        if len(mark_loud) > 0:
-            mut_set_range(has_loud, mark_loud, loud_speed)
+    def echunk(
+        arr: NDArray, src_index: NDArray[np.int32]
+    ) -> list[tuple[FileInfo, int, int, float]]:
+        arr_length = len(has_loud)
 
-        if len(mark_silent) > 0:
-            mut_set_range(has_loud, mark_silent, silent_speed)
+        chunks = []
+        start = 0
+        doi = 0
+        for j in range(1, arr_length):
+            if (arr[j] != arr[j - 1]) or (src_index[j] != src_index[j - 1]):
+                src = sources[src_index[j - 1]]
+                chunks.append((src, start, j - doi, speed_map[arr[j - 1]]))
+                start = j - doi
 
-        mut_margin(has_loud, start_margin, end_margin)
+                if src_index[j] != src_index[j - 1]:
+                    start = 0
+                    doi = j
 
-        # Setup for handling custom speeds
-        has_loud = has_loud.astype(np.uint)
+        src = sources[src_index[j]]
+        chunks.append((src, start, arr_length, speed_map[arr[j]]))
+        return chunks
 
-        try:
-            if len(cut_out) > 0:
-                # always cut out even if 'silent_speed' is not 99,999
-                mut_set_range(has_loud, cut_out, get_speed_index(99_999))
-
-            if len(add_in) > 0:
-                # set to 'video_speed' index
-                mut_set_range(has_loud, add_in, 1)
-
-            for speed_range in speed_ranges:
-                speed = speed_range[0]
-                _range = list(speed_range[1:])
-                mut_set_range(has_loud, [_range], get_speed_index(speed))
-        except CoerceError as e:
-            log.error(e)
+    clips: list[Clip] = []
+    i = 0
+    start = 0
+    for chunk in echunk(speed_index, src_index):
+        if chunk[3] != 99999:
+            dur = round((chunk[2] - chunk[1]) / chunk[3])
+            if dur == 0:
+                continue
 
-        chunks = chunkify(has_loud, speed_hash)
+            offset = chunk[1]
 
-        all_chunks.append(chunks)
-        all_clips.append(clipify(chunks, src, start))
-        start += round(chunks_len(chunks))
+            if not (clips and clips[-1].start == round(start)):
+                clips.append(Clip(start, dur, offset, chunk[3], chunk[0]))
+            start += dur
+            i += 1
 
     vtl: VSpace = []
     atl: ASpace = []
+    for c in clips:
+        if c.src.videos:
+            if len(vtl) == 0:
+                vtl.append([])
+            vtl[0].append(TlVideo(c.start, c.dur, c.src, c.offset, c.speed, 0))
+
+    for c in clips:
+        for a in range(len(c.src.audios)):
+            if a >= len(atl):
+                atl.append([])
+            atl[a].append(TlAudio(c.start, c.dur, c.src, c.offset, c.speed, 1, a))
+
+    # Turn long silent/loud array to formatted chunk list.
+    # Example: [1, 1, 1, 2, 2], {1: 1.0, 2: 1.5} => [(0, 3, 1.0), (3, 5, 1.5)]
+    def chunkify(arr: NDArray, smap: dict[int, float]) -> Chunks:
+        arr_length = len(arr)
+
+        chunks = []
+        start = 0
+        for j in range(1, arr_length):
+            if arr[j] != arr[j - 1]:
+                chunks.append((start, j, smap[arr[j - 1]]))
+                start = j
+        chunks.append((start, arr_length, smap[arr[j]]))
+        return chunks
+
+    if len(sources) == 1 and inp is not None:
+        chunks = chunkify(speed_index, speed_hash)
+        v1_compatiable = v1(inp, chunks)
+    else:
+        v1_compatiable = None
 
-    for clips in all_clips:
-        for c in clips:
-            if c.src.videos:
-                if len(vtl) == 0:
-                    vtl.append([])
-                vtl[0].append(TlVideo(c.start, c.dur, c.src, c.offset, c.speed, 0))
-
-        for c in clips:
-            for a in range(len(c.src.audios)):
-                if a >= len(atl):
-                    atl.append([])
-                atl[a].append(TlAudio(c.start, c.dur, c.src, c.offset, c.speed, 1, a))
-
-    return merge_chunks(all_chunks), vtl, atl
+    return v3(inp, tb, sr, res, args.background, vtl, atl, v1_compatiable)

auto_editor/subcommands/test.py

@@ -248,7 +248,6 @@ def high_speed_test():
 
     # Issue #184
     def units():
-        run.main(["example.mp4"], ["--mark_as_loud", "20s,22sec", "25secs,26.5seconds"])
         run.main(["example.mp4"], ["--edit", "all/e", "--set-speed", "125%,-30,end"])
         return run.main(["example.mp4"], ["--edit", "audio:threshold=4%"])
 
@@ -412,7 +411,7 @@ def multi_track_edit():
         return out
 
     def concat():
-        out = run.main(["example.mp4"], ["--mark_as_silent", "0,171"], "hmm.mp4")
+        out = run.main(["example.mp4"], ["--cut-out", "0,171"], "hmm.mp4")
         out2 = run.main(["example.mp4", "hmm.mp4"], ["--debug"])
         return out, out2
 

auto_editor/utils/chunks.py

@@ -1,46 +1,2 @@
-from __future__ import annotations
-
-from fractions import Fraction
-from typing import TYPE_CHECKING
-
-if TYPE_CHECKING:
-    from numpy.typing import NDArray
-
 Chunk = tuple[int, int, float]
 Chunks = list[Chunk]
-
-
-# Turn long silent/loud array to formatted chunk list.
-# Example: [1, 1, 1, 2, 2], {1: 1.0, 2: 1.5} => [(0, 3, 1.0), (3, 5, 1.5)]
-def chunkify(arr: NDArray, smap: dict[int, float]) -> Chunks:
-    arr_length = len(arr)
-
-    chunks = []
-    start = 0
-    for j in range(1, arr_length):
-        if arr[j] != arr[j - 1]:
-            chunks.append((start, j, smap[arr[j - 1]]))
-            start = j
-    chunks.append((start, arr_length, smap[arr[j]]))
-    return chunks
-
-
-def chunks_len(chunks: Chunks) -> Fraction:
-    _len = Fraction(0)
-    for chunk in chunks:
-        if chunk[2] != 99999:
-            speed = Fraction(chunk[2])
-            _len += Fraction(chunk[1] - chunk[0], speed)
-    return _len
-
-
-def merge_chunks(all_chunks: list[Chunks]) -> Chunks:
-    chunks = []
-    start = 0
-    for _chunks in all_chunks:
-        for chunk in _chunks:
-            chunks.append((chunk[0] + start, chunk[1] + start, chunk[2]))
-        if _chunks:
-            start += _chunks[-1][1]
-
-    return chunks

auto_editor/utils/types.py

@@ -232,8 +232,6 @@ class Args:
     no_seek: bool = False
     cut_out: list[list[str]] = field(default_factory=list)
     add_in: list[list[str]] = field(default_factory=list)
-    mark_as_loud: list[list[str]] = field(default_factory=list)
-    mark_as_silent: list[list[str]] = field(default_factory=list)
     set_speed_for_range: list[tuple[float, str, str]] = field(default_factory=list)
     frame_rate: Fraction | None = None
     sample_rate: int | None = None