diff --git a/scenedetect/output/__init__.py b/scenedetect/output/__init__.py index 7e5fac31..0766d2a8 100644 --- a/scenedetect/output/__init__.py +++ b/scenedetect/output/__init__.py @@ -16,9 +16,13 @@ """ import csv +import functools +import inspect +import io import json import logging import math +import os import typing as ty from fractions import Fraction from pathlib import Path @@ -64,12 +68,67 @@ from scenedetect.output.video import ( split_video_mkvmerge as split_video_mkvmerge, ) +from scenedetect.platform import BytePath, StrPath logger = logging.getLogger("pyscenedetect") +_F = ty.TypeVar("_F", bound=ty.Callable[..., ty.Any]) + + +def _open_output_file(output_format: str) -> ty.Callable[[_F], _F]: + """Allows a writer's first argument to be an open text file or a filesystem path. + + File handles remain open after the writer returns. Path-based output is rendered in memory + before the destination is opened, then written as UTF-8 with no newline translation. + """ + + def decorator(func: _F) -> _F: + parameters = tuple(inspect.signature(func).parameters.values()) + assert parameters and parameters[0].kind in ( + inspect.Parameter.POSITIONAL_ONLY, + inspect.Parameter.POSITIONAL_OR_KEYWORD, + ) + parameter_name = parameters[0].name + + @functools.wraps(func) + def wrapper(*args: ty.Any, **kwargs: ty.Any) -> ty.Any: + output_file = args[0] if args else kwargs[parameter_name] + + if not isinstance(output_file, (str, bytes, os.PathLike)): + if not callable(getattr(output_file, "write", None)): + raise TypeError( + f"{parameter_name} must be a filesystem path or writable text file" + ) + logger.info( + "Writing scenes in %s format to %s", + output_format, + getattr(output_file, "name", ""), + ) + return func(*args, **kwargs) + + with io.StringIO(newline="") as output_buffer: + if args: + result = func(output_buffer, *args[1:], **kwargs) + else: + kwargs[parameter_name] = output_buffer + result = func(**kwargs) + contents = output_buffer.getvalue() + + logger.info("Writing scenes in %s format to %s", output_format, output_file) + with open(output_file, "w", encoding="utf-8", newline="") as file_handle: + file_handle.write(contents) + + return result + + return ty.cast(_F, wrapper) + + return decorator + + +@_open_output_file("CSV") def write_scene_list( - output_csv_file: ty.TextIO | str | Path, + output_csv_file: StrPath | BytePath | ty.TextIO, scene_list: SceneList, include_cut_list: bool = True, cut_list: CutList | None = None, @@ -93,18 +152,9 @@ def write_scene_list( Raises: TypeError: "delimiter" must be a 1-character string """ - if isinstance(output_csv_file, (str, Path)): - with open(output_csv_file, "w", newline="") as file_handle: - write_scene_list( - file_handle, - scene_list, - include_cut_list=include_cut_list, - cut_list=cut_list, - col_separator=col_separator, - row_separator=row_separator, - ) - return - csv_writer = csv.writer(output_csv_file, delimiter=col_separator, lineterminator=row_separator) + assert not isinstance(output_csv_file, (str, bytes, os.PathLike)) + output_file = output_csv_file + csv_writer = csv.writer(output_file, delimiter=col_separator, lineterminator=row_separator) # If required, output the cutting list as the first row (i.e. before the header row). if include_cut_list: csv_writer.writerow( @@ -304,8 +354,9 @@ def _parse_edl_start_timecode(value: str, frame_rate: Fraction | float) -> int: return round((hours * 3600 + minutes * 60 + seconds) * float(frame_rate)) + frames +@_open_output_file("EDL") def write_scene_list_edl( - output_path: str | Path, + output_path: StrPath | BytePath | ty.TextIO, scene_list: SceneList, title: str = "PySceneDetect", reel: str = "AX", @@ -314,7 +365,8 @@ def write_scene_list_edl( """Writes the given list of scenes to `output_path` in CMX 3600 EDL format. Arguments: - output_path: Path to write the EDL file to. Parent directories must exist. + output_path: Open text file or path to write the EDL file to. Parent directories must + exist. When a path is provided, the file is closed after writing. scene_list: List of scenes as pairs of FrameTimecodes denoting each scene's start/end. title: Title header written as ``TITLE:`` in the EDL. reel: Reel name used for each event. Typically 2-8 uppercase characters. @@ -322,7 +374,8 @@ def write_scene_list_edl( every event so the EDL aligns with the source media's on-screen timecode. Applied to both source and record columns. """ - output_path = Path(output_path) + assert not isinstance(output_path, (str, bytes, os.PathLike)) + output_file = output_path offset_frames = 0 if start_timecode is not None and start_timecode.strip() and scene_list: frame_rate = scene_list[0][0].frame_rate @@ -333,14 +386,12 @@ def write_scene_list_edl( in_tc = _edl_timecode(start + offset_frames) out_tc = _edl_timecode(end + offset_frames) lines.append(f"{(i + 1):03d} {reel} V C {in_tc} {out_tc} {in_tc} {out_tc}") - logger.info("Writing scenes in EDL format to %s", output_path) - with open(output_path, "w") as f: - # `scenedetect` is imported lazily to avoid a circular import at module load. - import scenedetect + # `scenedetect` is imported lazily to avoid a circular import at module load. + import scenedetect - f.write(f"* CREATED WITH PYSCENEDETECT {scenedetect.__version__}\n") - f.write("\n".join(lines)) - f.write("\n") + output_file.write(f"* CREATED WITH PYSCENEDETECT {scenedetect.__version__}\n") + output_file.write("\n".join(lines)) + output_file.write("\n") def _rational_seconds(value: Fraction) -> str: @@ -359,8 +410,9 @@ def _frame_timecode_seconds(tc: FrameTimecode) -> Fraction: return Fraction(tc.pts) * tc.time_base +@_open_output_file("FCPX") def write_scene_list_fcpx( - output_path: str | Path, + output_path: StrPath | BytePath | ty.TextIO, scene_list: SceneList, video_path: str | Path, frame_rate: Fraction, @@ -374,7 +426,8 @@ def write_scene_list_fcpx( https://developer.apple.com/documentation/professional-video-applications/fcpxml-reference Arguments: - output_path: Path to write the FCPXML file to. Parent directories must exist. + output_path: Open text file or path to write the FCPXML file to. Parent directories must + exist. When a path is provided, the file is closed after writing. scene_list: List of scenes as pairs of FrameTimecodes. Must not be empty. video_path: Path to the source video file; written into the output as a ``file://`` URI. frame_rate: Source frame rate as a rational `Fraction` (e.g. ``Fraction(24000, 1001)``). @@ -383,7 +436,8 @@ def write_scene_list_fcpx( of `video_path`. """ assert scene_list - output_path = Path(output_path) + assert not isinstance(output_path, (str, bytes, os.PathLike)) + output_file = output_path video_path = Path(video_path) if video_name is None: video_name = video_path.stem @@ -453,13 +507,12 @@ def write_scene_list_fcpx( pretty_xml = minidom.parseString(ElementTree.tostring(root, encoding="unicode")).toprettyxml( indent=" " ) - logger.info("Writing scenes in FCPX format to %s", output_path) - with open(output_path, "w") as f: - f.write(pretty_xml) + output_file.write(pretty_xml) +@_open_output_file("FCP") def write_scene_list_fcp7( - output_path: str | Path, + output_path: StrPath | BytePath | ty.TextIO, scene_list: SceneList, video_path: str | Path, frame_rate: Fraction, @@ -474,7 +527,8 @@ def write_scene_list_fcp7( ``pathurl`` is written as a valid ``file://`` URI per the xmeml spec. Arguments: - output_path: Path to write the xmeml file to. Parent directories must exist. + output_path: Open text file or path to write the xmeml file to. Parent directories must + exist. When a path is provided, the file is closed after writing. scene_list: List of scenes as pairs of FrameTimecodes. Must not be empty. video_path: Path to the source video file; written into the output as a ``file://`` URI. frame_rate: Source frame rate as a rational `Fraction`. @@ -486,7 +540,8 @@ def write_scene_list_fcp7( frozen. If None, falls back to the last scene's end time. """ assert scene_list - output_path = Path(output_path) + assert not isinstance(output_path, (str, bytes, os.PathLike)) + output_file = output_path video_path = Path(video_path) if video_name is None: video_name = video_path.stem @@ -570,16 +625,15 @@ def write_scene_list_fcp7( pretty_xml = minidom.parseString(ElementTree.tostring(root, encoding="unicode")).toprettyxml( indent=" " ) - logger.info("Writing scenes in FCP format to %s", output_path) - with open(output_path, "w") as f: - f.write(pretty_xml) + output_file.write(pretty_xml) # TODO: We have to export framerate as a float for OTIO's current format. When OTIO supports # fractional timecodes, we should export the framerate as a rational number instead. # https://github.com/AcademySoftwareFoundation/OpenTimelineIO/issues/190 +@_open_output_file("OTIO") def write_scene_list_otio( - output_path: str | Path, + output_path: StrPath | BytePath | ty.TextIO, scene_list: SceneList, video_path: str | Path, frame_rate: Fraction, @@ -591,7 +645,8 @@ def write_scene_list_otio( OTIO (OpenTimelineIO) timelines can be imported by many video editors. Arguments: - output_path: Path to write the OTIO file to. Parent directories must exist. + output_path: Open text file or path to write the OTIO file to. Parent directories must + exist. When a path is provided, the file is closed after writing. scene_list: List of scenes as pairs of FrameTimecodes. video_path: Path to the source video file; written into the output as an absolute path. frame_rate: Source frame rate as a rational `Fraction`. Exported as a float, as the @@ -599,7 +654,8 @@ def write_scene_list_otio( name: Timeline name. Defaults to the stem of `video_path`. audio: If True (default), include an audio track alongside the video track. """ - output_path = Path(output_path) + assert not isinstance(output_path, (str, bytes, os.PathLike)) + output_file = output_path video_path = Path(video_path) if name is None: name = video_path.stem @@ -680,7 +736,5 @@ def write_scene_list_otio( }, } - logger.info("Writing scenes in OTIO format to %s", output_path) - with open(output_path, "w") as f: - json.dump(otio, f, indent=4) - f.write("\n") + json.dump(otio, output_file, indent=4) + output_file.write("\n") diff --git a/scenedetect/platform.py b/scenedetect/platform.py index 9a783ea2..65e54436 100644 --- a/scenedetect/platform.py +++ b/scenedetect/platform.py @@ -31,6 +31,10 @@ """Type hint for filesystem paths. Accepts a `str` or any object implementing :class:`os.PathLike` (e.g. :class:`pathlib.Path`).""" +BytePath = bytes | os.PathLike[bytes] +"""Type hint for byte-based filesystem paths. Accepts `bytes` or any object implementing +:class:`os.PathLike` that returns `bytes`.""" + DEBUG_MODE: bool = os.environ.get("SCENEDETECT_DEBUG", "").strip().lower() not in ( "", "0", diff --git a/tests/test_output.py b/tests/test_output.py index 1f434873..b419cbfd 100644 --- a/tests/test_output.py +++ b/tests/test_output.py @@ -12,6 +12,8 @@ """Tests for scenedetect.output module.""" import json +import os +import typing as ty from fractions import Fraction from io import StringIO from pathlib import Path @@ -250,32 +252,43 @@ def _fake_scenes(fps: Fraction, frames): return [(FrameTimecode(start, fps=fps), FrameTimecode(end, fps=fps)) for start, end in frames] -def test_write_scene_list_file_handle(): - """Existing callers that pass an open file handle keep working.""" +_OUTPUT_TARGET_KINDS = ("file_handle", "str", "path", "bytes") + + +def _make_output_target( + tmp_path: Path, target_kind: str, filename: str +) -> tuple[ty.Any, ty.Callable[[], str]]: + """Create one supported output target and a function that reads its text.""" + output_path = tmp_path / filename + if target_kind == "file_handle": + output_file = StringIO(newline="") + return output_file, output_file.getvalue + if target_kind == "str": + output_target = str(output_path) + elif target_kind == "bytes": + output_target = os.fsencode(output_path) + else: + assert target_kind == "path" + output_target = output_path + return output_target, lambda: output_path.read_text(encoding="utf-8") + + +@pytest.mark.parametrize("target_kind", _OUTPUT_TARGET_KINDS) +def test_write_scene_list_output_target(tmp_path: Path, target_kind: str): + """CSV output accepts a file handle, string, Path, or bytes path.""" scenes = _fake_scenes(_FPS_CFR, [(0, 30), (30, 60)]) - buf = StringIO() - write_scene_list(buf, scenes, include_cut_list=False) - text = buf.getvalue() + output_target, read_output = _make_output_target(tmp_path, target_kind, "scenes.csv") + write_scene_list(output_target, scenes, include_cut_list=False) + text = read_output() assert "Scene Number" in text assert "00:00:00.000" in text or "00:00:00:00" in text -def test_write_scene_list_accepts_str_path(tmp_path: Path): - """A filesystem path string must not raise TypeError from csv.writer.""" - scenes = _fake_scenes(_FPS_CFR, [(0, 30)]) - output_path = tmp_path / "scenes.csv" - write_scene_list(str(output_path), scenes, include_cut_list=False) - assert output_path.exists() - assert "Scene Number" in output_path.read_text() - - -def test_write_scene_list_accepts_path(tmp_path: Path): - """pathlib.Path is opened with a context manager and closed after write.""" - scenes = _fake_scenes(_FPS_CFR, [(0, 30)]) - output_path = tmp_path / "scenes.csv" - write_scene_list(output_path, scenes, include_cut_list=False) - assert output_path.exists() - assert "Scene Number" in output_path.read_text() +def test_write_scene_list_rejects_invalid_output_target(): + """The shared output decorator rejects objects that are neither paths nor writable files.""" + scenes = [(FrameTimecode(0, 24.0), FrameTimecode(24, 24.0))] + with pytest.raises(TypeError, match="output_csv_file"): + write_scene_list(ty.cast(ty.Any, object()), scenes) def test_write_scene_list_edl(tmp_path: Path): @@ -291,12 +304,15 @@ def test_write_scene_list_edl(tmp_path: Path): assert "002 AX V C 00:00:01:00 00:00:02:00 00:00:01:00 00:00:02:00" in content -def test_write_scene_list_edl_accepts_str_path(tmp_path: Path): - """`output_path` must accept both Path and str.""" +@pytest.mark.parametrize("target_kind", _OUTPUT_TARGET_KINDS) +def test_write_scene_list_edl_output_target(tmp_path: Path, target_kind: str): + """EDL output accepts a file handle, string, Path, or bytes path.""" scenes = _fake_scenes(_FPS_CFR, [(0, 30)]) - output_path = tmp_path / "scenes.edl" - write_scene_list_edl(str(output_path), scenes) - assert output_path.exists() + output_target, read_output = _make_output_target(tmp_path, target_kind, "scenes.edl") + write_scene_list_edl(output_target, scenes, title="Tést") + text = read_output() + assert "TITLE: Tést" in text + assert "\r\n" not in text def test_write_scene_list_edl_with_start_timecode_smpte(tmp_path: Path): @@ -358,8 +374,11 @@ def test_write_scene_list_edl_default_no_offset(tmp_path: Path): def test_write_scene_list_edl_with_start_timecode_invalid_format(tmp_path: Path, bad_value: str): """Malformed start timecodes raise ValueError before writing.""" scenes = _fake_scenes(_FPS_CFR, [(0, 30)]) + output_path = tmp_path / "scenes.edl" + output_path.write_text("existing output", encoding="utf-8") with pytest.raises(ValueError): - write_scene_list_edl(tmp_path / "scenes.edl", scenes, start_timecode=bad_value) + write_scene_list_edl(output_path, scenes, start_timecode=bad_value) + assert output_path.read_text(encoding="utf-8") == "existing output" @pytest.mark.parametrize( @@ -428,6 +447,22 @@ def test_write_scene_list_fcpx_video_name_defaults_to_path_stem(tmp_path: Path): assert asset is not None and asset.attrib["name"] == "my_clip" +@pytest.mark.parametrize("target_kind", _OUTPUT_TARGET_KINDS) +def test_write_scene_list_fcpx_output_target(tmp_path: Path, target_kind: str): + """FCPXML output accepts a file handle, string, Path, or bytes path.""" + scenes = _fake_scenes(_FPS_NTSC, [(48, 96), (96, 144)]) + output_target, read_output = _make_output_target(tmp_path, target_kind, "scenes.xml") + # `video_path` need not exist; only `.absolute().as_uri()` is called on it. + write_scene_list_fcpx( + output_path=output_target, + scene_list=scenes, + video_path=Path("fake_video.mp4"), + frame_rate=_FPS_NTSC, + frame_size=(1280, 544), + ) + assert read_output().startswith(' reference.""" scenes = _fake_scenes(_FPS_NTSC, [(0, 48), (48, 96)]) @@ -478,6 +513,21 @@ def test_write_scene_list_fcp7_cfr_sets_ntsc_false(tmp_path: Path): assert ntsc is not None and ntsc.text == "False" +@pytest.mark.parametrize("target_kind", _OUTPUT_TARGET_KINDS) +def test_write_scene_list_fcp7_output_target(tmp_path: Path, target_kind: str): + """FCP7 XML output accepts a file handle, string, Path, or bytes path.""" + scenes = _fake_scenes(_FPS_CFR, [(0, 30)]) + output_target, read_output = _make_output_target(tmp_path, target_kind, "scenes.xml") + write_scene_list_fcp7( + output_target, + scene_list=scenes, + video_path=Path("source.mp4"), + frame_rate=_FPS_CFR, + frame_size=(640, 360), + ) + assert read_output().startswith('