diff --git a/scenedetect/output/__init__.py b/scenedetect/output/__init__.py index 7e5fac31..19e19418 100644 --- a/scenedetect/output/__init__.py +++ b/scenedetect/output/__init__.py @@ -16,6 +16,7 @@ """ import csv +import functools import json import logging import math @@ -68,8 +69,39 @@ logger = logging.getLogger("pyscenedetect") +_F = ty.TypeVar("_F", bound=ty.Callable[..., ty.Any]) + + +def _open_output_file(parameter_name: str, *, newline: str | None = None): + """Allows a writer's first argument to be an open text file or an output path. + + File handles remain open after the writer returns. Paths are opened in write mode and closed + when the writer returns. + """ + + def decorator(func: _F) -> _F: + @functools.wraps(func) + def wrapper(*args: ty.Any, **kwargs: ty.Any): + output_file = args[0] if args else kwargs[parameter_name] + + if not isinstance(output_file, (str, Path)): + return func(*args, **kwargs) + + with open(output_file, "w", newline=newline) as file_handle: + if args: + return func(file_handle, *args[1:], **kwargs) + + kwargs[parameter_name] = file_handle + return func(**kwargs) + + return ty.cast(_F, wrapper) + + return decorator + + +@_open_output_file(parameter_name="output_csv_file", newline="") def write_scene_list( - output_csv_file: ty.TextIO | str | Path, + output_csv_file: str | Path | ty.TextIO, scene_list: SceneList, include_cut_list: bool = True, cut_list: CutList | None = None, @@ -93,17 +125,7 @@ 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 + output_csv_file = ty.cast(ty.TextIO, output_csv_file) csv_writer = csv.writer(output_csv_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: @@ -304,8 +326,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("output_path") def write_scene_list_edl( - output_path: str | Path, + output_path: str | Path | ty.TextIO, scene_list: SceneList, title: str = "PySceneDetect", reel: str = "AX", @@ -314,7 +337,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 +346,7 @@ 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) + output_file = ty.cast(ty.TextIO, 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 +357,16 @@ 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 + logger.info( + "Writing scenes in EDL format to %s", + getattr(output_file, "name", ""), + ) + # `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 +385,9 @@ def _frame_timecode_seconds(tc: FrameTimecode) -> Fraction: return Fraction(tc.pts) * tc.time_base +@_open_output_file(parameter_name="output_path") def write_scene_list_fcpx( - output_path: str | Path, + output_path: str | Path | ty.TextIO, scene_list: SceneList, video_path: str | Path, frame_rate: Fraction, @@ -374,7 +401,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 +411,7 @@ def write_scene_list_fcpx( of `video_path`. """ assert scene_list - output_path = Path(output_path) + output_file = ty.cast(ty.TextIO, output_path) video_path = Path(video_path) if video_name is None: video_name = video_path.stem @@ -453,13 +481,16 @@ 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) + logger.info( + "Writing scenes in FCPX format to %s", + getattr(output_file, "name", ""), + ) + output_file.write(pretty_xml) +@_open_output_file(parameter_name="output_path") def write_scene_list_fcp7( - output_path: str | Path, + output_path: str | Path | ty.TextIO, scene_list: SceneList, video_path: str | Path, frame_rate: Fraction, @@ -474,7 +505,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 +518,7 @@ 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) + output_file = ty.cast(ty.TextIO, output_path) video_path = Path(video_path) if video_name is None: video_name = video_path.stem @@ -570,16 +602,19 @@ 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) + logger.info( + "Writing scenes in FCP format to %s", + getattr(output_file, "name", ""), + ) + 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(parameter_name="output_path") def write_scene_list_otio( - output_path: str | Path, + output_path: str | Path | ty.TextIO, scene_list: SceneList, video_path: str | Path, frame_rate: Fraction, @@ -591,7 +626,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 +635,7 @@ 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) + output_file = ty.cast(ty.TextIO, output_path) video_path = Path(video_path) if name is None: name = video_path.stem @@ -680,7 +716,9 @@ 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") + logger.info( + "Writing scenes in OTIO format to %s", + getattr(output_file, "name", ""), + ) + json.dump(otio, output_file, indent=4) + output_file.write("\n") diff --git a/tests/test_output.py b/tests/test_output.py index 1f434873..a81eae36 100644 --- a/tests/test_output.py +++ b/tests/test_output.py @@ -291,12 +291,31 @@ 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_file_handle(): + """EDL output accepts an open text file.""" + scenes = _fake_scenes(_FPS_CFR, [(0, 30)]) + buf = StringIO() + write_scene_list_edl(buf, scenes) + text = buf.getvalue() + assert "TITLE:" in text + + def test_write_scene_list_edl_accepts_str_path(tmp_path: Path): - """`output_path` must accept both Path and str.""" + """EDL output accepts a string 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() + assert "TITLE:" in output_path.read_text() + + +def test_write_scene_list_edl_accepts_path(tmp_path: Path): + """EDL output accepts a `Path`.""" + scenes = _fake_scenes(_FPS_CFR, [(0, 30)]) + output_path = tmp_path / "scenes.edl" + write_scene_list_edl(output_path, scenes) + assert output_path.exists() + assert "TITLE:" in output_path.read_text() def test_write_scene_list_edl_with_start_timecode_smpte(tmp_path: Path): @@ -428,6 +447,56 @@ 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" +def test_write_scene_list_fcpx_file_handle(): + """FCPXML output accepts an open text file.""" + scenes = _fake_scenes(_FPS_NTSC, [(48, 96), (96, 144)]) + buf = StringIO() + # `video_path` need not exist; only `.absolute().as_uri()` is called on it. + write_scene_list_fcpx( + output_path=buf, + scene_list=scenes, + video_path=Path("fake_video.mp4"), + frame_rate=_FPS_NTSC, + frame_size=(1280, 544), + ) + + assert buf.getvalue().startswith(' reference.""" scenes = _fake_scenes(_FPS_NTSC, [(0, 48), (48, 96)]) @@ -478,6 +547,51 @@ def test_write_scene_list_fcp7_cfr_sets_ntsc_false(tmp_path: Path): assert ntsc is not None and ntsc.text == "False" +def test_write_scene_list_fcp7_file_handle(): + """FCP7 XML output accepts an open text file.""" + scenes = _fake_scenes(_FPS_CFR, [(0, 30)]) + buf = StringIO() + write_scene_list_fcp7( + buf, + scene_list=scenes, + video_path=Path("source.mp4"), + frame_rate=_FPS_CFR, + frame_size=(640, 360), + ) + text = buf.getvalue() + assert text.startswith('