add cut_into_multiple_parts_v2 with friendly usage.

Author: fujiawei-dev

ffmpeg/tools/avtools.py

@@ -5,11 +5,11 @@
 '''
 import tempfile
 from pathlib import Path
-from typing import Dict, List, Optional, Union
+from typing import Dict, List, NamedTuple, Optional, Union
 
 from .. import constants
 from .._ffmpeg import input, merge_outputs, output
-from .._utils import seconds_to_string
+from .._utils import seconds_to_string, string_to_seconds
 
 __all__ = [
     "adjust_tempo",
@@ -64,11 +64,16 @@ def modify_metadata(src: Union[str, Path], dst: Union[str, Path], *,
 
 
 def separate_video_stream(src: Union[str, Path], dst: Union[str, Path]):
-    input(src).output(dst, an=True, vcodec="copy").run()
+    input(src, enable_cuda=False).output(dst, an=True, enable_cuda=False, vcodec="copy").run()
 
 
-def separate_audio_stream(src: Union[str, Path], dst: Union[str, Path]):
-    input(src).output(dst, vn=True, acodec="copy").run()
+def separate_audio_stream(src: Union[str, Path], dst: Union[str, Path], pcm_format=False):
+    if pcm_format:
+        kwargs = dict(format=constants.S16LE, acodec=constants.PCM_S16LE, ac=1, ar="16k")
+    else:
+        kwargs = dict(acodec=constants.COPY)
+
+    input(src, enable_cuda=False).output(dst, vn=True, enable_cuda=False, **kwargs).run()
 
 
 def convert_format(src: Union[str, Path], dst: Union[str, Path], *,
@@ -77,8 +82,8 @@ def convert_format(src: Union[str, Path], dst: Union[str, Path], *,
 
 
 def cut_into_multiple_parts(src: Union[str, Path], dst: Union[str, Path],
-                            *, durations: List[float], vcodec="libx264",
-                            enable_cuda=True, overwrite=True):
+                            *, durations: List[Union[float, int, str]], vcodec="libx264",
+                            enable_cuda=True, overwrite=True, accumulative=False):
     """Cut the video or audio into multiple parts.
 
     Example:
@@ -98,11 +103,17 @@ def cut_into_multiple_parts(src: Union[str, Path], dst: Union[str, Path],
 
     for order, duration in enumerate(durations):
         # skip negative value
-        if duration is not None and duration < 0:
+        if isinstance(duration, (int, float)) and duration < 0:
             start_position -= duration
             continue
 
-        outs.append(raw.output(f"{dst / path.stem}_part{order}{path.suffix}",
+        if isinstance(duration, str):
+            duration = string_to_seconds(duration)
+
+        if isinstance(duration, (int, float)) and accumulative:
+            duration -= start_position
+
+        outs.append(raw.output(f"{dst / path.stem}_{order}{path.suffix}",
                                acodec="copy", vcodec=vcodec, enable_cuda=enable_cuda,
                                start_position=seconds_to_string(start_position), duration=duration))
 
@@ -112,6 +123,35 @@ def cut_into_multiple_parts(src: Union[str, Path], dst: Union[str, Path],
     merge_outputs(*outs).run(overwrite=overwrite)
 
 
+class TrimPair(NamedTuple):
+    Start: Union[str, int, float]
+    End: Union[str, int, float]
+    IsDuration: bool = False
+
+
+def cut_into_multiple_parts_v2(src: Union[str, Path], dst: Union[str, Path],
+                               *, start_duration_pairs: List[TrimPair], vcodec="libx264",
+                               enable_cuda=True, overwrite=True):
+    outs = []
+    path = Path(src)
+    raw = input(src, enable_cuda=enable_cuda)
+
+    for order, pair in enumerate(start_duration_pairs):
+        start_position = string_to_seconds(pair.Start)
+        end_position = string_to_seconds(pair.End)
+
+        if not pair.IsDuration and end_position > start_position:
+            duration = end_position - start_position
+        else:
+            duration = end_position
+
+        outs.append(raw.output(f"{dst / path.stem}_{order}{path.suffix}",
+                               acodec="copy", vcodec=vcodec, enable_cuda=enable_cuda,
+                               start_position=start_position, duration=duration))
+
+    merge_outputs(*outs).run(overwrite=overwrite)
+
+
 def cut_one_part(src: Union[str, Path], dst: Union[str, Path], *, vcodec="libx264",
                  enable_cuda=True, overwrite=True, start: Union[str, int, float] = None,
                  end: Union[str, int, float] = None, duration: Union[int, float] = None,