From 7b6780bd86a1a60d1a26491c29fc2409e80d506e Mon Sep 17 00:00:00 2001 From: Dylan De Faoite Date: Wed, 29 Apr 2026 20:15:35 +0100 Subject: [PATCH] refactor(clip): extract media functions out of core This is in preperation for testing --- rewind/core.py | 119 +++--------------------------------------------- rewind/media.py | 109 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 116 insertions(+), 112 deletions(-) create mode 100644 rewind/media.py diff --git a/rewind/core.py b/rewind/core.py index d3571ab..8cf38a7 100644 --- a/rewind/core.py +++ b/rewind/core.py @@ -1,12 +1,10 @@ #!/usr/bin/env python3 import os import datetime -import subprocess -import tempfile -from rewind.state import load_state, add_marker_to_state, remove_marker_from_state from rewind.config import Config -from tqdm import tqdm +from rewind.media import get_ts_files, concat_ts_files +from rewind.state import load_state, add_marker_to_state, remove_marker_from_state config = Config() @@ -24,12 +22,12 @@ def clip(seconds_from_end: float) -> None: end_timestamp = datetime.datetime.now().timestamp() length = end_timestamp - start_timestamp - files, start_offset, end_offset = _get_ts_files( + files, start_offset, end_offset = get_ts_files( start_timestamp, end_timestamp ) - _concat_ts_files(files, start_offset, end_offset, length, output_path) + concat_ts_files(files, start_offset, end_offset, length, output_path) print(f"Created clip: {output_path}") def save(first_marker: str, second_marker: str): @@ -45,12 +43,12 @@ def save(first_marker: str, second_marker: str): output_file_name = f"{datetime.datetime.fromtimestamp(first_timestamp).strftime('%Y-%m-%d_%H:%M:%S')}-[{first_marker}-{second_marker}].mp4" output_path = os.path.join(vod_dir, output_file_name) - files, start_offset, end_offset = _get_ts_files( + files, start_offset, end_offset = get_ts_files( first_timestamp, second_timestamp ) - _concat_ts_files(files, start_offset, end_offset, second_timestamp - first_timestamp, output_path) + concat_ts_files(files, start_offset, end_offset, second_timestamp - first_timestamp, output_path) print(f"Created video file: {output_path}") def mark(name: str) -> None: @@ -90,107 +88,4 @@ def marker_exists(name: str) -> bool: if marker["name"] == name: return True - return False - -""" -Retrieves .ts files recorded between the specified timestamps. -Returns a list of file paths and extra start and end offsets if needed. -get_duration() is used as little as possible since it is slow. -end_timestamp of a file is the start time of the next file. -""" -def _get_ts_files(start_timestamp: float, end_timestamp: float) -> tuple[list[str], float, float]: - ts_files = [f for f in load_state()["files"] if os.path.exists(f["path"])] - selected_files = [] - start_offset = 0.0 - end_offset = 0.0 - - for i, file_info in enumerate(ts_files): - file_start = file_info["timestamp"] - file_end = ts_files[i + 1]["timestamp"] if i + 1 < len(ts_files) else get_duration(file_info["path"]) + file_start - - if file_end <= start_timestamp: - continue - if file_start >= end_timestamp: - break - - selected_files.append(file_info["path"]) - - if file_start <= start_timestamp < file_end: - start_offset = start_timestamp - file_start - if file_start < end_timestamp <= file_end: - end_offset = file_end - end_timestamp - - return selected_files, start_offset, end_offset - -def _concat_ts_files(file_list: list[str], start_offset: float, end_offset: float, length: float, output_file: str) -> None: - tmp_file = tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) - try: - for file_path in file_list: - tmp_file.write(f"file '{file_path}'\n") - tmp_file.close() - - cmd = ["ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-nostats", "-progress", "pipe:1"] - if start_offset > 0: - cmd += ["-ss", str(start_offset)] - if end_offset > 0: - cmd += ["-t", str(length)] - cmd += ["-f", "concat", "-safe", "0", "-i", tmp_file.name, "-c", "copy"] - cmd.append(output_file) - - process = subprocess.Popen( - cmd, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - bufsize=1, - ) - - init_ms_val = 0 - init_ms_val_set = False - - with tqdm( - total=length, - unit="s", - unit_scale=True, - unit_divisor=60, - desc="Processing", - leave=True, - ) as pbar: - for line in process.stdout: # type: ignore[union-attr] - line = line.strip() - - if line.startswith("out_time_ms="): - out_time_ms = int(line.split("=")[1]) - - if not init_ms_val_set: - init_ms_val = out_time_ms - init_ms_val_set = True - out_time_ms -= init_ms_val - - seconds = abs(out_time_ms / 1_000_000) - pbar.n = min(seconds, length) - pbar.refresh() - - elif line == "progress=end": - break - - ret = process.wait() - if ret != 0: - raise RuntimeError("ffmpeg failed") - finally: - os.unlink(tmp_file.name) - -def get_duration(file_path: str) -> float: - result = subprocess.run( - ["ffprobe", "-v", "error", "-show_entries", - "format=duration", "-of", - "default=noprint_wrappers=1:nokey=1", file_path], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - ) - - if result.returncode != 0: - raise RuntimeError(f"ffprobe failed for file {file_path}") - - return float(result.stdout.strip()) \ No newline at end of file + return False \ No newline at end of file diff --git a/rewind/media.py b/rewind/media.py new file mode 100644 index 0000000..760b3f5 --- /dev/null +++ b/rewind/media.py @@ -0,0 +1,109 @@ +import os +import subprocess +import tempfile + +from rewind.state import load_state +from tqdm import tqdm + +""" +Retrieves .ts files recorded between the specified timestamps. +Returns a list of file paths and extra start and end offsets if needed. +get_duration() is used as little as possible since it is slow. +end_timestamp of a file is the start time of the next file. +""" +def get_ts_files(start_timestamp: float, end_timestamp: float) -> tuple[list[str], float, float]: + ts_files = [f for f in load_state()["files"] if os.path.exists(f["path"])] + selected_files = [] + start_offset = 0.0 + end_offset = 0.0 + + for i, file_info in enumerate(ts_files): + file_start = file_info["timestamp"] + file_end = ts_files[i + 1]["timestamp"] if i + 1 < len(ts_files) else get_duration(file_info["path"]) + file_start + + if file_end <= start_timestamp: + continue + if file_start >= end_timestamp: + break + + selected_files.append(file_info["path"]) + + if file_start <= start_timestamp < file_end: + start_offset = start_timestamp - file_start + if file_start < end_timestamp <= file_end: + end_offset = file_end - end_timestamp + + return selected_files, start_offset, end_offset + +def concat_ts_files(file_list: list[str], start_offset: float, end_offset: float, length: float, output_file: str) -> None: + tmp_file = tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) + try: + for file_path in file_list: + tmp_file.write(f"file '{file_path}'\n") + tmp_file.close() + + cmd = ["ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-nostats", "-progress", "pipe:1"] + if start_offset > 0: + cmd += ["-ss", str(start_offset)] + if end_offset > 0: + cmd += ["-t", str(length)] + cmd += ["-f", "concat", "-safe", "0", "-i", tmp_file.name, "-c", "copy"] + cmd.append(output_file) + + process = subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + bufsize=1, + ) + + init_ms_val = 0 + init_ms_val_set = False + + with tqdm( + total=length, + unit="s", + unit_scale=True, + unit_divisor=60, + desc="Processing", + leave=True, + ) as pbar: + for line in process.stdout: # type: ignore[union-attr] + line = line.strip() + + if line.startswith("out_time_ms="): + out_time_ms = int(line.split("=")[1]) + + if not init_ms_val_set: + init_ms_val = out_time_ms + init_ms_val_set = True + out_time_ms -= init_ms_val + + seconds = abs(out_time_ms / 1_000_000) + pbar.n = min(seconds, length) + pbar.refresh() + + elif line == "progress=end": + break + + ret = process.wait() + if ret != 0: + raise RuntimeError("ffmpeg failed") + finally: + os.unlink(tmp_file.name) + +def get_duration(file_path: str) -> float: + result = subprocess.run( + ["ffprobe", "-v", "error", "-show_entries", + "format=duration", "-of", + "default=noprint_wrappers=1:nokey=1", file_path], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + + if result.returncode != 0: + raise RuntimeError(f"ffprobe failed for file {file_path}") + + return float(result.stdout.strip()) \ No newline at end of file