Use VideoTimestamps

muxtools/audio/extractors.py

@@ -4,6 +4,7 @@
 from fractions import Fraction
 from typing import Sequence
 from pathlib import Path
+from video_timestamps import TimeType
 import os
 import re
 
@@ -17,7 +18,7 @@
 from ..utils.types import Trim, PathLike, TrackType
 from ..utils.env import get_temp_workdir, run_commandline, communicate_stdout
 from ..utils.subprogress import run_cmd_pb, ProgressBarConfig
-from ..utils.convert import frame_to_timedelta, format_timedelta, frame_to_ms
+from ..utils.convert import frame_to_ms, format_timedelta
 
 __all__ = ["Eac3to", "Sox", "FFMpeg", "MkvExtract"]
 
@@ -172,7 +173,7 @@ class Trimmer(Trimmer):
 
         :param trim:                Can be a single trim or a sequence of trims.
         :param trim_use_ms:         Will use milliseconds instead of frame numbers
-        :param fps:                 Fps fraction that will be used for the conversion. Also accepts a timecode (v2) file.
+        :param fps:                 Fps fraction that will be used for the conversion. Also accepts a timecode (v2, v4) file.
         :param preserve_delay:      Will preserve existing container delay
         :param num_frames:          Total number of frames used for calculations
         :param output:              Custom output. Can be a dir or a file.
@@ -208,13 +209,13 @@ def _targs(self, trim: Trim) -> str:
                 if self.trim_use_ms:
                     arg += f" -ss {format_timedelta(timedelta(milliseconds=trim[0]))}"
                 else:
-                    arg += f" -ss {format_timedelta(frame_to_timedelta(trim[0], self.fps))}"
+                    arg += f" -ss {format_timedelta(timedelta(milliseconds=frame_to_ms(trim[0], TimeType.EXACT, self.fps, False)))}"
             if trim[1] is not None and trim[1] != 0:
                 end_frame = self.num_frames + trim[1] if trim[1] < 0 else trim[1]
                 if self.trim_use_ms:
                     arg += f" -to {format_timedelta(timedelta(milliseconds=trim[1]))}"
                 else:
-                    arg += f" -to {format_timedelta(frame_to_timedelta(end_frame, self.fps))}"
+                    arg += f" -to {format_timedelta(timedelta(milliseconds=frame_to_ms(end_frame, TimeType.EXACT, self.fps, False)))}"
             return arg
 
         def trim_audio(self, input: AudioFile, quiet: bool = True) -> AudioFile:
@@ -248,7 +249,7 @@ def trim_audio(self, input: AudioFile, quiet: bool = True) -> AudioFile:
                 args.append(str(out.resolve()))
                 if not run_commandline(args, quiet):
                     if tr[0] and lossy:
-                        ms = tr[0] if self.trim_use_ms else frame_to_ms(tr[0], self.fps)
+                        ms = tr[0] if self.trim_use_ms else frame_to_ms(tr[0], TimeType.EXACT, self.fps, False)
                         cont_delay = self._calc_delay(ms, ainfo.num_samples(), getattr(minfo, "sampling_rate", 48000))
                         debug(f"Additional delay of {cont_delay} ms will be applied to fix remaining sync", self)
                         if self.preserve_delay:
@@ -366,7 +367,7 @@ class Sox(Trimmer):
     :param trim:                List of Trims or a single Trim, which is a Tuple of two frame numbers or milliseconds
     :param preserve_delay:      Keeps existing container delay if True
     :param trim_use_ms:         Will use milliseconds instead of frame numbers
-    :param fps:                 The fps fraction used for the calculations. Also accepts a timecode (v2) file.
+    :param fps:                 The fps fraction used for the calculations. Also accepts a timecode (v2, v4) file.
     :param num_frames:          Total number of frames used for calculations
     :param output:              Custom output. Can be a dir or a file.
                                 Do not specify an extension unless you know what you're doing.
@@ -386,7 +387,7 @@ def _conv(self, val: int | None):
         if self.trim_use_ms:
             return abs(val) / 1000
         else:
-            return frame_to_timedelta(abs(val), self.fps).total_seconds()
+            return frame_to_ms(abs(val), TimeType.EXACT, self.fps, False) / 1000
 
     def trim_audio(self, input: AudioFile, quiet: bool = True) -> AudioFile:
         import sox

muxtools/functions.py

@@ -33,7 +33,7 @@ def do_audio(
     :param fileIn:          Input file
     :param track:           Audio track number
     :param trims:           Frame ranges to trim and/or combine, e.g. (24, -24) or [(24, 500), (700, 900)]
-    :param fps:             FPS Fraction used for the conversion to time. Also accepts a timecode (v2) file.
+    :param fps:             FPS Fraction used for the conversion to time. Also accepts a timecode (v2, v4) file.
     :param num_frames:      Total number of frames, used for negative numbers in trims
     :param extractor:       Tool used to extract the audio
     :param trimmer:         Tool used to trim the audio

muxtools/misc/chapters.py

@@ -4,6 +4,7 @@
 from fractions import Fraction
 from pathlib import Path
 from typing import TypeVar
+from video_timestamps import TimeType
 import os
 import re
 
@@ -15,7 +16,7 @@
 from ..utils.parsing import parse_ogm, parse_xml
 from ..utils.files import clean_temp_files, ensure_path_exists, ensure_path
 from ..utils.env import get_temp_workdir, get_workdir, run_commandline
-from ..utils.convert import format_timedelta, frame_to_timedelta, timedelta_to_frame
+from ..utils.convert import format_timedelta, frame_to_ms, ms_to_frame
 
 __all__ = ["Chapters"]
 
@@ -31,7 +32,7 @@ def __init__(
         Convenience class for chapters
 
         :param chapter_source:      Input either txt with ogm chapters, xml or (a list of) self defined chapters.
-        :param fps:                 Needed for timestamp convertion. Assumes 24000/1001 by default. Also accepts a timecode (v2) file.
+        :param fps:                 Needed for timestamp convertion. Assumes 24000/1001 by default. Also accepts a timecode (v2, v4) file.
         :param _print:              Prints chapters after parsing and after trimming.
         """
         self.fps = fps
@@ -54,7 +55,7 @@ def __init__(
         for ch in self.chapters:
             if isinstance(ch[0], int):
                 current = list(ch)
-                current[0] = frame_to_timedelta(current[0], self.fps)
+                current[0] = timedelta(milliseconds=frame_to_ms(current[0], TimeType.EXACT, self.fps, False))
                 chapters.append(tuple(current))
             else:
                 chapters.append(ch)
@@ -67,15 +68,15 @@ def trim(self: ChaptersSelf, trim_start: int = 0, trim_end: int = 0, num_frames:
         if trim_start > 0:
             chapters: list[Chapter] = []
             for chapter in self.chapters:
-                if timedelta_to_frame(chapter[0]) == 0:
+                if ms_to_frame(int(chapter[0].total_seconds() * 1000), TimeType.START, self.fps) == 0:
                     chapters.append(chapter)
                     continue
-                if timedelta_to_frame(chapter[0]) - trim_start < 0:
+                if ms_to_frame(int(chapter[0].total_seconds() * 1000), TimeType.START, self.fps) - trim_start < 0:
                     continue
                 current = list(chapter)
-                current[0] = current[0] - frame_to_timedelta(trim_start, self.fps)
+                current[0] = current[0] - timedelta(milliseconds=frame_to_ms(trim_start, TimeType.EXACT, self.fps, False))
                 if num_frames:
-                    if current[0] > frame_to_timedelta(num_frames - 1, self.fps):
+                    if current[0] > timedelta(milliseconds=frame_to_ms(num_frames - 1, TimeType.EXACT, self.fps, False)):
                         continue
                 chapters.append(tuple(current))
 
@@ -84,7 +85,7 @@ def trim(self: ChaptersSelf, trim_start: int = 0, trim_end: int = 0, num_frames:
             if trim_end > 0:
                 chapters: list[Chapter] = []
                 for chapter in self.chapters:
-                    if timedelta_to_frame(chapter[0], self.fps) < trim_end:
+                    if ms_to_frame(int(chapter[0].total_seconds() * 1000), TimeType.START, self.fps) < trim_end:
                         chapters.append(chapter)
                 self.chapters = chapters
 
@@ -125,7 +126,7 @@ def add(self: ChaptersSelf, chapters: Chapter | list[Chapter], index: int = 0) -
         for ch in chapters:
             if isinstance(ch[0], int):
                 current = list(ch)
-                current[0] = frame_to_timedelta(current[0], self.fps)
+                current[0] = timedelta(milliseconds=frame_to_ms(current[0], TimeType.EXACT, self.fps, False))
                 converted.append(tuple(current))
             else:
                 converted.append(ch)
@@ -143,14 +144,9 @@ def shift_chapter(self: ChaptersSelf, chapter: int = 0, shift_amount: int = 0) -
         :param shift_amount:    Frames to shift by
         """
         ch = list(self.chapters[chapter])
-        shift_delta = frame_to_timedelta(abs(shift_amount), self.fps)
-        if shift_amount < 0:
-            shifted_frame = ch[0] - shift_delta
-        else:
-            shifted_frame = ch[0] + shift_delta
-
-        if shifted_frame.total_seconds() > 0:
-            ch[0] = shifted_frame
+        ch_frame = ms_to_frame(int(ch[0].total_seconds() * 1000), TimeType.START, self.fps) + abs(shift_amount)
+        if ch_frame >= 0:
+            ch[0] = timedelta(milliseconds=frame_to_ms(ch_frame, TimeType.EXACT, self.fps, False))
         else:
             ch[0] = timedelta(seconds=0)
         self.chapters[chapter] = tuple(ch)
@@ -170,7 +166,7 @@ def print(self: ChaptersSelf) -> ChaptersSelf:
         """
         info("Chapters:")
         for time, name in self.chapters:
-            print(f"{name}: {format_timedelta(time)} | {timedelta_to_frame(time, self.fps)}")
+            print(f"{name}: {format_timedelta(time)} | {ms_to_frame(int(time.total_seconds() * 1000), TimeType.START, self.fps)}")
         print("", end="\n")
         return self
 
@@ -209,7 +205,7 @@ def from_sub(
         Extract chapters from an ass file or a SubFile.
 
         :param file:            Input ass file or SubFile
-        :param fps:             FPS passed to the chapter class for further operations. Also accepts a timecode (v2) file.
+        :param fps:             FPS passed to the chapter class for further operations. Also accepts a timecode (v2, v4) file.
         :param use_actor_field: Uses the actor field instead of the effect field for identification.
         :param markers:         Markers to check for.
         :param _print:          Prints the chapters after parsing
@@ -255,7 +251,7 @@ def from_mkv(file: PathLike, fps: Fraction | PathLike = Fraction(24000, 1001), _
         Extract chapters from mkv.
 
         :param file:            Input mkv file
-        :param fps:             FPS passed to the chapter class for further operations. Also accepts a timecode (v2) file.
+        :param fps:             FPS passed to the chapter class for further operations. Also accepts a timecode (v2, v4) file.
         :param _print:          Prints the chapters after parsing
         """
         caller = "Chapters.from_mkv"

muxtools/subtitle/sub.py

@@ -6,6 +6,7 @@
 from datetime import timedelta
 from fractions import Fraction
 from pathlib import Path
+from video_timestamps import TimeType
 import shutil
 import json
 import re
@@ -17,7 +18,7 @@
 from ..utils.download import get_executable
 from ..utils.types import PathLike, TrackType
 from ..utils.log import debug, error, info, warn
-from ..utils.convert import frame_to_timedelta, timedelta_to_frame
+from ..utils.convert import frame_to_ms, ms_to_frame
 from ..utils.env import get_temp_workdir, get_workdir, run_commandline
 from ..utils.files import ensure_path_exists, get_absolute_track, make_output, clean_temp_files, uniquify_path
 from ..muxing.muxfiles import MuxingFile
@@ -298,15 +299,15 @@ def shift_0(
 
         This does not currently exactly reproduce the aegisub behaviour but it should have the same effect.
 
-        :param fps:             The fps fraction used for conversions. Also accepts a timecode (v2) file.
+        :param fps:             The fps fraction used for conversions. Also accepts a timecode (v2, v4) file.
         :param allowed_styles:  A list of style names this will run on. Will run on every line if None.
         """
 
         def _func(lines: LINES):
             for line in lines:
                 if not allowed_styles or line.style.lower() in allowed_styles:
-                    line.start = frame_to_timedelta(timedelta_to_frame(line.start, fps, exclude_boundary=True), fps, True)
-                    line.end = frame_to_timedelta(timedelta_to_frame(line.end, fps, exclude_boundary=True), fps, True)
+                    line.start = timedelta(milliseconds=frame_to_ms(ms_to_frame(int(line.start.total_seconds() * 1000), TimeType.START, fps), TimeType.START, fps))
+                    line.end = timedelta(milliseconds=frame_to_ms(ms_to_frame(int(line.end.total_seconds() * 1000), TimeType.END, fps), TimeType.END, fps))
 
         return self.manipulate_lines(_func)
 
@@ -327,7 +328,7 @@ def merge(
         :param sync:            Can be None to not adjust timing at all, an int for a frame number or a string for a syncpoint name.
         :param sync2:           The syncpoint you want to use for the second file.
                                 This is needed if you specified a frame for sync and still want to use a specific syncpoint.
-        :param fps:             The fps used for time calculations. Also accepts a timecode (v2) file.
+        :param fps:             The fps used for time calculations. Also accepts a timecode (v2, v4) file.
         :param use_actor_field: Checks the actor field instead of effect for the names if True.
         :param no_error:        Don't error and warn instead if syncpoint not found.
         :param sort_lines:      Sort the lines by the starting timestamp.
@@ -349,7 +350,7 @@ def merge(
             if target is None and isinstance(sync, str):
                 field = line.name if use_actor_field else line.effect
                 if field.lower().strip() == sync.lower().strip() or line.text.lower().strip() == sync.lower().strip():
-                    target = timedelta_to_frame(line.start, fps, exclude_boundary=True) + 1
+                    target = ms_to_frame(int(line.start.total_seconds() * 1000), TimeType.START, fps) + 1
 
         if target is None and isinstance(sync, str):
             msg = f"Syncpoint '{sync}' was not found."
@@ -367,7 +368,7 @@ def merge(
                 sync2 = sync2 or sync
             field = line.name if use_actor_field else line.effect
             if field.lower().strip() == sync2.lower().strip() or line.text.lower().strip() == sync2.lower().strip():
-                second_sync = timedelta_to_frame(line.start, fps, exclude_boundary=True) + 1
+                second_sync = ms_to_frame(int(line.start.total_seconds() * 1000), TimeType.START, fps) + 1
                 mergedoc.events.remove(line)
                 break
 
@@ -376,7 +377,7 @@ def merge(
         # Assume the first line to be the second syncpoint if none was found
         if second_sync is None:
             for line in filter(lambda event: event.TYPE != "Comment", sorted_lines):
-                second_sync = timedelta_to_frame(line.start, fps, exclude_boundary=True) + 1
+                second_sync = ms_to_frame(int(line.start.total_seconds() * 1000), TimeType.START, fps) + 1
                 break
 
         # Merge lines from file
@@ -388,8 +389,8 @@ def merge(
 
             # Apply frame offset
             offset = (target or -1) - second_sync
-            line.start = frame_to_timedelta(timedelta_to_frame(line.start, fps, exclude_boundary=True) + offset, fps, True)
-            line.end = frame_to_timedelta(timedelta_to_frame(line.end, fps, exclude_boundary=True) + offset, fps, True)
+            line.start = timedelta(milliseconds=frame_to_ms(ms_to_frame(int(line.start.total_seconds() * 1000), TimeType.START, fps) + offset, TimeType.START, fps))
+            line.end = timedelta(milliseconds=frame_to_ms(ms_to_frame(int(line.end.total_seconds() * 1000), TimeType.END, fps) + offset, TimeType.END, fps))
             tomerge.append(line)
 
         if tomerge:
@@ -648,23 +649,23 @@ def shift(self: SubFileSelf, frames: int, fps: Fraction | PathLike = Fraction(24
         Shifts all lines by any frame number.
 
         :param frames:              Number of frames to shift by
-        :param fps:                 FPS needed for the timing calculations. Also accepts a timecode (v2) file.
+        :param fps:                 FPS needed for the timing calculations. Also accepts a timecode (v2, v4) file.
         :param delete_before_zero:  Delete lines that would be before 0 after shifting.
         """
 
         def shift_lines(lines: LINES):
             new_list = list[_Line]()
             for line in lines:
-                start = timedelta_to_frame(line.start, fps, exclude_boundary=True) + frames
+                start = ms_to_frame(int(line.start.total_seconds() * 1000), TimeType.START, fps) + frames
                 if start < 0:
                     if delete_before_zero:
                         continue
                     start = 0
-                start = frame_to_timedelta(start, fps, compensate=True)
-                end = timedelta_to_frame(line.end, fps, exclude_boundary=True) + frames
+                start = timedelta(milliseconds=frame_to_ms(start, TimeType.START, fps))
+                end = ms_to_frame(int(line.end.total_seconds() * 1000), TimeType.END, fps) + frames
                 if end < 0:
                     continue
-                end = frame_to_timedelta(end, fps, compensate=True)
+                end = timedelta(milliseconds=frame_to_ms(end, TimeType.END, fps))
                 line.start = start
                 line.end = end
                 new_list.append(line)
@@ -703,19 +704,19 @@ def from_srt(
         :param file:            Input srt file
         :param an8_all_caps:    Automatically an8 every full caps line with over 7 characters because they're usually signs.
         :param style_all_caps:  Also set the style of these lines to "Sign" wether it exists or not.
-        :param fps:             FPS needed for the time conversion. Also accepts a timecode (v2) file.
+        :param fps:             FPS needed for the time conversion. Also accepts a timecode (v2, v4) file.
         :param encoding:        Encoding used to read the file. Defaults to UTF8.
         """
         caller = "SubFile.from_srt"
         file = ensure_path_exists(file, caller)
 
         compiled = re.compile(SRT_REGEX, re.MULTILINE)
 
-        def srt_timedelta(timestamp: str) -> timedelta:
+        def srt_timedelta(timestamp: str, time_type: TimeType) -> timedelta:
             args = timestamp.split(",")[0].split(":")
             parsed = timedelta(hours=int(args[0]), minutes=int(args[1]), seconds=int(args[2]), milliseconds=int(timestamp.split(",")[1]))
-            cope = timedelta_to_frame(parsed, fps, exclude_boundary=True)
-            cope = frame_to_timedelta(cope, fps, compensate=True)
+            cope = ms_to_frame(int(parsed.total_seconds() * 1000), time_type, fps)
+            cope = timedelta(milliseconds=frame_to_ms(cope, time_type, fps))
             return cope
 
         def convert_tags(text: str) -> tuple[str, bool]:
@@ -737,8 +738,8 @@ def convert_tags(text: str) -> tuple[str, bool]:
         with open(file, "r", encoding=encoding) as reader:
             content = reader.read() + "\n"
             for match in compiled.finditer(content):
-                start = srt_timedelta(match["start"])
-                end = srt_timedelta(match["end"])
+                start = srt_timedelta(match["start"], TimeType.START)
+                end = srt_timedelta(match["end"], TimeType.END)
                 text, sign = convert_tags(match["text"])
                 doc.events.append(Dialogue(layer=99, start=start, end=end, text=text, style="Sign" if sign and style_all_caps else "Default"))