diff --git a/scenedetect/__main__.py b/scenedetect/__main__.py index 7c9ec1b9..5a8f7e4c 100755 --- a/scenedetect/__main__.py +++ b/scenedetect/__main__.py @@ -22,10 +22,10 @@ def main(): """PySceneDetect command-line interface (CLI) entry point.""" - cli_ctx = CliContext() + context = CliContext() try: # Process command line arguments and subcommands to initialize the context. - scenedetect.main(obj=cli_ctx) # Parse CLI arguments with registered callbacks. + scenedetect.main(obj=context) # Parse CLI arguments with registered callbacks. except SystemExit as exit: help_command = any(arg in sys.argv for arg in ["-h", "--help"]) if help_command or exit.code != 0: @@ -38,12 +38,12 @@ def main(): # no progress bars get created, we instead create a fake context manager. This is done here # to avoid needing a separate context manager at each point a progress bar is created. log_redirect = ( - FakeTqdmLoggingRedirect() if cli_ctx.quiet_mode else logging_redirect_tqdm(loggers=[logger]) + FakeTqdmLoggingRedirect() if context.quiet_mode else logging_redirect_tqdm(loggers=[logger]) ) with log_redirect: try: - run_scenedetect(cli_ctx) + run_scenedetect(context) except KeyboardInterrupt: logger.info("Stopped.") if __debug__: @@ -52,7 +52,7 @@ def main(): if __debug__: raise else: - logger.critical("Unhandled exception:", exc_info=ex) + logger.critical("ERROR: Unhandled exception:", exc_info=ex) raise SystemExit(1) from None diff --git a/scenedetect/_cli/__init__.py b/scenedetect/_cli/__init__.py index 18047181..208e4d1d 100644 --- a/scenedetect/_cli/__init__.py +++ b/scenedetect/_cli/__init__.py @@ -20,13 +20,23 @@ import inspect import logging +import os +import os.path import typing as ty import click import scenedetect -from scenedetect._cli.config import CHOICE_MAP, CONFIG_FILE_PATH, CONFIG_MAP -from scenedetect._cli.context import USER_CONFIG, CliContext +import scenedetect._cli.commands as cli_commands +from scenedetect._cli.config import ( + CHOICE_MAP, + CONFIG_FILE_PATH, + CONFIG_MAP, + DEFAULT_JPG_QUALITY, + DEFAULT_WEBP_QUALITY, + TimecodeFormat, +) +from scenedetect._cli.context import USER_CONFIG, CliContext, check_split_video_requirements from scenedetect.backends import AVAILABLE_BACKENDS from scenedetect.detectors import ( AdaptiveDetector, @@ -35,7 +45,8 @@ HistogramDetector, ThresholdDetector, ) -from scenedetect.platform import get_system_version_info +from scenedetect.platform import get_cv2_imwrite_params, get_system_version_info +from scenedetect.scene_manager import Interpolation _PROGRAM_VERSION = scenedetect.__version__ """Used to avoid name conflict with named `scenedetect` command below.""" @@ -305,8 +316,10 @@ def scenedetect( Global options (e.g. -i/--input, -c/--config) must be specified before any commands and their options. The order of commands is not strict, but each command must only be specified once. """ - assert isinstance(ctx.obj, CliContext) - ctx.obj.handle_options( + ctx = ctx.obj + assert isinstance(ctx, CliContext) + + ctx.handle_options( input_path=input, output=output, framerate=framerate, @@ -334,7 +347,6 @@ def scenedetect( @click.pass_context def help_command(ctx: click.Context, command_name: str): """Print help for command (`help [command]`).""" - assert isinstance(ctx.obj, CliContext) assert isinstance(ctx.parent.command, click.MultiCommand) parent_command = ctx.parent.command all_commands = set(parent_command.list_commands(ctx)) @@ -358,7 +370,6 @@ def help_command(ctx: click.Context, command_name: str): @click.pass_context def about_command(ctx: click.Context): """Print license/copyright info.""" - assert isinstance(ctx.obj, CliContext) click.echo("") click.echo(click.style(_LINE_SEPARATOR, fg="cyan")) click.echo(click.style(" About PySceneDetect %s" % _PROGRAM_VERSION, fg="yellow")) @@ -371,7 +382,6 @@ def about_command(ctx: click.Context): @click.pass_context def version_command(ctx: click.Context): """Print PySceneDetect version.""" - assert isinstance(ctx.obj, CliContext) click.echo("") click.echo(get_system_version_info()) ctx.exit() @@ -421,12 +431,23 @@ def time_command( {scenedetect_with_video} time --start 0 --end 1000 """ - assert isinstance(ctx.obj, CliContext) - ctx.obj.handle_time( - start=start, - duration=duration, - end=end, - ) + ctx = ctx.obj + assert isinstance(ctx, CliContext) + + if duration is not None and end is not None: + raise click.BadParameter( + "Only one of --duration/-d or --end/-e can be specified, not both.", + param_hint="time", + ) + logger.debug("Setting video time:\n start: %s, duration: %s, end: %s", start, duration, end) + # *NOTE*: The Python API uses 0-based frame indices, but the CLI uses 1-based indices to + # match the default start number used by `ffmpeg` when saving frames as images. As such, + # we must correct start time if set as frames. See the test_cli_time* tests for for details. + ctx.start_time = ctx.parse_timecode(start, correct_pts=True) + ctx.end_time = ctx.parse_timecode(end) + ctx.duration = ctx.parse_timecode(duration) + if ctx.start_time and ctx.end_time and (ctx.start_time + 1) > ctx.end_time: + raise click.BadParameter("-e/--end time must be greater than -s/--start") @click.command("detect-content", cls=_Command) @@ -525,8 +546,9 @@ def detect_content_command( {scenedetect_with_video} detect-content --threshold 27.5 """ - assert isinstance(ctx.obj, CliContext) - detector_args = ctx.obj.get_detect_content_params( + ctx = ctx.obj + assert isinstance(ctx, CliContext) + detector_args = ctx.get_detect_content_params( threshold=threshold, luma_only=luma_only, min_scene_len=min_scene_len, @@ -534,8 +556,7 @@ def detect_content_command( kernel_size=kernel_size, filter_mode=filter_mode, ) - logger.debug("Adding detector: ContentDetector(%s)", detector_args) - ctx.obj.add_detector(ContentDetector(**detector_args)) + ctx.add_detector(ContentDetector, detector_args) @click.command("detect-adaptive", cls=_Command) @@ -636,8 +657,9 @@ def detect_adaptive_command( {scenedetect_with_video} detect-adaptive --threshold 3.2 """ - assert isinstance(ctx.obj, CliContext) - detector_args = ctx.obj.get_detect_adaptive_params( + ctx = ctx.obj + assert isinstance(ctx, CliContext) + detector_args = ctx.get_detect_adaptive_params( threshold=threshold, min_content_val=min_content_val, min_delta_hsv=min_delta_hsv, @@ -647,8 +669,7 @@ def detect_adaptive_command( weights=weights, kernel_size=kernel_size, ) - logger.debug("Adding detector: AdaptiveDetector(%s)", detector_args) - ctx.obj.add_detector(AdaptiveDetector(**detector_args)) + ctx.add_detector(AdaptiveDetector, detector_args) @click.command("detect-threshold", cls=_Command) @@ -715,15 +736,15 @@ def detect_threshold_command( {scenedetect_with_video} detect-threshold --threshold 15 """ - assert isinstance(ctx.obj, CliContext) - detector_args = ctx.obj.get_detect_threshold_params( + ctx = ctx.obj + assert isinstance(ctx, CliContext) + detector_args = ctx.get_detect_threshold_params( threshold=threshold, fade_bias=fade_bias, add_last_scene=add_last_scene, min_scene_len=min_scene_len, ) - logger.debug("Adding detector: ThresholdDetector(%s)", detector_args) - ctx.obj.add_detector(ThresholdDetector(**detector_args)) + ctx.add_detector(ThresholdDetector, detector_args) @click.command("detect-hist", cls=_Command) @@ -785,14 +806,12 @@ def detect_hist_command( {scenedetect_with_video} detect-hist --threshold 0.1 --bins 240 """ - assert isinstance(ctx.obj, CliContext) - - assert isinstance(ctx.obj, CliContext) - detector_args = ctx.obj.get_detect_hist_params( + ctx = ctx.obj + assert isinstance(ctx, CliContext) + detector_args = ctx.get_detect_hist_params( threshold=threshold, bins=bins, min_scene_len=min_scene_len ) - logger.debug("Adding detector: HistogramDetector(%s)", detector_args) - ctx.obj.add_detector(HistogramDetector(**detector_args)) + ctx.add_detector(HistogramDetector, detector_args) @click.command("detect-hash", cls=_Command) @@ -870,14 +889,12 @@ def detect_hash_command( {scenedetect_with_video} detect-hash --size 32 --lowpass 3 """ - assert isinstance(ctx.obj, CliContext) - - assert isinstance(ctx.obj, CliContext) - detector_args = ctx.obj.get_detect_hash_params( + ctx = ctx.obj + assert isinstance(ctx, CliContext) + detector_args = ctx.get_detect_hash_params( threshold=threshold, size=size, lowpass=lowpass, min_scene_len=min_scene_len ) - logger.debug("Adding detector: HashDetector(%s)", detector_args) - ctx.obj.add_detector(HashDetector(**detector_args)) + ctx.add_detector(HashDetector, detector_args) @click.command("load-scenes", cls=_Command) @@ -911,9 +928,23 @@ def load_scenes_command( {scenedetect_with_video} load-scenes -i scenes.csv --start-col-name "Start Timecode" """ - assert isinstance(ctx.obj, CliContext) - logger.debug("Loading scenes from %s (start_col_name = %s)", input, start_col_name) - ctx.obj.handle_load_scenes(input=input, start_col_name=start_col_name) + ctx = ctx.obj + assert isinstance(ctx, CliContext) + + logger.debug("Will load scenes from %s (start_col_name = %s)", input, start_col_name) + if ctx.scene_manager.get_num_detectors() > 0: + raise click.ClickException("The load-scenes command cannot be used with detectors.") + if ctx.load_scenes_input: + raise click.ClickException("The load-scenes command must only be specified once.") + input = os.path.abspath(input) + if not os.path.exists(input): + raise click.BadParameter( + f"Could not load scenes, file does not exist: {input}", param_hint="-i/--input" + ) + ctx.load_scenes_input = input + ctx.load_scenes_column_name = ctx.config.get_value( + "load-scenes", "start-col-name", start_col_name + ) @click.command("export-html", cls=_Command) @@ -958,13 +989,20 @@ def export_html_command( image_height: ty.Optional[int], ): """Export scene list to HTML file. Requires save-images unless --no-images is specified.""" - assert isinstance(ctx.obj, CliContext) - ctx.obj.handle_export_html( - filename=filename, - no_images=no_images, - image_width=image_width, - image_height=image_height, - ) + ctx = ctx.obj + assert isinstance(ctx, CliContext) + + no_images = no_images or ctx.config.get_value("export-html", "no-images") + if not ctx.save_images and not no_images: + raise click.BadArgumentUsage( + "export-html requires that save-images precedes it or --no-images is specified." + ) + export_html_args = { + "html_name_format": ctx.config.get_value("export-html", "filename", filename), + "image_width": ctx.config.get_value("export-html", "image-width", image_width), + "image_height": ctx.config.get_value("export-html", "image-height", image_height), + } + ctx.add_command(cli_commands.export_html, export_html_args) @click.command("list-scenes", cls=_Command) @@ -1018,14 +1056,23 @@ def list_scenes_command( skip_cuts: bool, ): """Create scene list CSV file (will be named $VIDEO_NAME-Scenes.csv by default).""" - assert isinstance(ctx.obj, CliContext) - ctx.obj.handle_list_scenes( - output=output, - filename=filename, - no_output_file=no_output_file, - quiet=quiet, - skip_cuts=skip_cuts, - ) + ctx = ctx.obj + assert isinstance(ctx, CliContext) + + no_output_file = no_output_file or ctx.config.get_value("list-scenes", "no-output-file") + scene_list_dir = ctx.config.get_value("list-scenes", "output", output) + scene_list_name_format = ctx.config.get_value("list-scenes", "filename", filename) + list_scenes_args = { + "cut_format": TimecodeFormat[ctx.config.get_value("list-scenes", "cut-format").upper()], + "display_scenes": ctx.config.get_value("list-scenes", "display-scenes"), + "display_cuts": ctx.config.get_value("list-scenes", "display-cuts"), + "scene_list_output": not no_output_file, + "scene_list_name_format": scene_list_name_format, + "skip_cuts": skip_cuts or ctx.config.get_value("list-scenes", "skip-cuts"), + "output_dir": scene_list_dir, + "quiet": quiet or ctx.config.get_value("list-scenes", "quiet") or ctx.quiet_mode, + } + ctx.add_command(cli_commands.list_scenes, list_scenes_args) @click.command("split-video", cls=_Command) @@ -1134,18 +1181,73 @@ def split_video_command( {scenedetect_with_video} split-video --filename \\$VIDEO_NAME-Clip-\\$SCENE_NUMBER """ - assert isinstance(ctx.obj, CliContext) - ctx.obj.handle_split_video( - output=output, - filename=filename, - quiet=quiet, - copy=copy, - high_quality=high_quality, - rate_factor=rate_factor, - preset=preset, - args=args, - mkvmerge=mkvmerge, - ) + ctx = ctx.obj + assert isinstance(ctx, CliContext) + + check_split_video_requirements(use_mkvmerge=mkvmerge) + if "%" in ctx.video_stream.path or "://" in ctx.video_stream.path: + error = "The split-video command is incompatible with image sequences/URLs." + raise click.BadParameter(error, param_hint="split-video") + + # We only load the config values for these flags/options if none of the other + # encoder flags/options were set via the CLI to avoid any conflicting options + # (e.g. if the config file sets `high-quality = yes` but `--copy` is specified). + if not (mkvmerge or copy or high_quality or args or rate_factor or preset): + mkvmerge = ctx.config.get_value("split-video", "mkvmerge") + copy = ctx.config.get_value("split-video", "copy") + high_quality = ctx.config.get_value("split-video", "high-quality") + rate_factor = ctx.config.get_value("split-video", "rate-factor") + preset = ctx.config.get_value("split-video", "preset") + args = ctx.config.get_value("split-video", "args") + + # Disallow certain combinations of options. + if mkvmerge or copy: + command = "mkvmerge (-m)" if mkvmerge else "copy (-c)" + if high_quality: + raise click.BadParameter( + "high-quality (-hq) cannot be used with %s" % (command), + param_hint="split-video", + ) + if args: + raise click.BadParameter( + "args (-a) cannot be used with %s" % (command), param_hint="split-video" + ) + if rate_factor: + raise click.BadParameter( + "rate-factor (crf) cannot be used with %s" % (command), param_hint="split-video" + ) + if preset: + raise click.BadParameter( + "preset (-p) cannot be used with %s" % (command), param_hint="split-video" + ) + + # mkvmerge-Specific Options + if mkvmerge and copy: + logger.warning("copy mode (-c) ignored due to mkvmerge mode (-m).") + + # ffmpeg-Specific Options + if copy: + args = "-map 0:v:0 -map 0:a? -map 0:s? -c:v copy -c:a copy" + elif not args: + if rate_factor is None: + rate_factor = 22 if not high_quality else 17 + if preset is None: + preset = "veryfast" if not high_quality else "slow" + args = ( + "-map 0:v:0 -map 0:a? -map 0:s? " + f"-c:v libx264 -preset {preset} -crf {rate_factor} -c:a aac" + ) + if filename: + logger.info("Output file name format: %s", filename) + + split_video_args = { + "name_format": ctx.config.get_value("split-video", "filename", filename), + "use_mkvmerge": mkvmerge, + "output_dir": ctx.config.get_value("split-video", "output", output), + "show_output": not quiet, + "ffmpeg_args": args, + } + ctx.add_command(cli_commands.split_video, split_video_args) @click.command("save-images", cls=_Command) @@ -1279,21 +1381,65 @@ def save_images_command( {scenedetect_with_video} save-images --filename \\$SCENE_NUMBER-img\\$IMAGE_NUMBER """ - assert isinstance(ctx.obj, CliContext) - ctx.obj.handle_save_images( - num_images=num_images, - output=output, - filename=filename, - jpeg=jpeg, - webp=webp, - quality=quality, - png=png, - compression=compression, - frame_margin=frame_margin, - scale=scale, - height=height, - width=width, + ctx = ctx.obj + assert isinstance(ctx, CliContext) + + if "://" in ctx.video_stream.path: + error_str = "\nThe save-images command is incompatible with URLs." + logger.error(error_str) + raise click.BadParameter(error_str, param_hint="save-images") + num_flags = sum([1 if flag else 0 for flag in [jpeg, webp, png]]) + if num_flags > 1: + logger.error(".") + raise click.BadParameter("Only one image type can be specified.", param_hint="save-images") + elif num_flags == 0: + image_format = ctx.config.get_value("save-images", "format").lower() + jpeg = image_format == "jpeg" + webp = image_format == "webp" + png = image_format == "png" + + if not any((scale, height, width)): + scale = ctx.config.get_value("save-images", "scale") + height = ctx.config.get_value("save-images", "height") + width = ctx.config.get_value("save-images", "width") + scale_method = Interpolation[ctx.config.get_value("save-images", "scale-method").upper()] + quality = ( + (DEFAULT_WEBP_QUALITY if webp else DEFAULT_JPG_QUALITY) + if ctx.config.is_default("save-images", "quality") + else ctx.config.get_value("save-images", "quality") ) + compression = ctx.config.get_value("save-images", "compression", compression) + image_extension = "jpg" if jpeg else "png" if png else "webp" + valid_params = get_cv2_imwrite_params() + if image_extension not in valid_params or valid_params[image_extension] is None: + error_strs = [ + "Image encoder type `%s` not supported." % image_extension.upper(), + "The specified encoder type could not be found in the current OpenCV module.", + "To enable this output format, please update the installed version of OpenCV.", + "If you build OpenCV, ensure the the proper dependencies are enabled. ", + ] + logger.debug("\n".join(error_strs)) + raise click.BadParameter("\n".join(error_strs), param_hint="save-images") + output = ctx.config.get_value("save-images", "output", output) + + save_images_args = { + "encoder_param": compression if png else quality, + "frame_margin": ctx.config.get_value("save-images", "frame-margin", frame_margin), + "height": height, + "image_extension": image_extension, + "image_name_template": ctx.config.get_value("save-images", "filename", filename), + "interpolation": scale_method, + "num_images": ctx.config.get_value("save-images", "num-images", num_images), + "output_dir": output, + "scale": scale, + "show_progress": ctx.quiet_mode, + "width": width, + } + ctx.add_command(cli_commands.save_images, save_images_args) + + # Record that we added a save-images command to the pipeline so we can allow export-html + # to run afterwards (it is dependent on the output). + ctx.save_images = True # ---------------------------------------------------------------------- diff --git a/scenedetect/_cli/commands.py b/scenedetect/_cli/commands.py new file mode 100644 index 00000000..83a23bf2 --- /dev/null +++ b/scenedetect/_cli/commands.py @@ -0,0 +1,213 @@ +# +# PySceneDetect: Python-Based Video Scene Detector +# ------------------------------------------------------------------- +# [ Site: https://scenedetect.com ] +# [ Docs: https://scenedetect.com/docs/ ] +# [ Github: https://github.com/Breakthrough/PySceneDetect/ ] +# +# Copyright (C) 2014-2024 Brandon Castellano . +# PySceneDetect is licensed under the BSD 3-Clause License; see the +# included LICENSE file, or visit one of the above pages for details. +# +"""Logic for PySceneDetect commands that operate on the result of the processing pipeline. + +In addition to the the arguments registered with the command, commands will be called with the +current command-line context, as well as the processing result (scenes and cuts). +""" + +import logging +import typing as ty +from string import Template + +from scenedetect._cli.context import CliContext +from scenedetect.platform import get_and_create_path +from scenedetect.scene_manager import ( + CutList, + Interpolation, + SceneList, + write_scene_list, + write_scene_list_html, +) +from scenedetect.scene_manager import ( + save_images as save_images_impl, +) +from scenedetect.video_splitter import split_video_ffmpeg, split_video_mkvmerge + +logger = logging.getLogger("pyscenedetect") + + +def export_html( + context: CliContext, + scenes: SceneList, + cuts: CutList, + image_width: int, + image_height: int, + html_name_format: str, +): + """Handles the `export-html` command.""" + (image_filenames, output_dir) = ( + context.save_images_result + if context.save_images_result is not None + else (None, context.output_dir) + ) + html_filename = Template(html_name_format).safe_substitute(VIDEO_NAME=context.video_stream.name) + if not html_filename.lower().endswith(".html"): + html_filename += ".html" + html_path = get_and_create_path(html_filename, output_dir) + write_scene_list_html( + output_html_filename=html_path, + scene_list=scenes, + cut_list=cuts, + image_filenames=image_filenames, + image_width=image_width, + image_height=image_height, + ) + + +def list_scenes( + context: CliContext, + scenes: SceneList, + cuts: CutList, + scene_list_output: bool, + scene_list_name_format: str, + output_dir: str, + skip_cuts: bool, + quiet: bool, + display_scenes: bool, + display_cuts: bool, + cut_format: str, +): + """Handles the `list-scenes` command.""" + # Write scene list CSV to if required. + if scene_list_output: + scene_list_filename = Template(scene_list_name_format).safe_substitute( + VIDEO_NAME=context.video_stream.name + ) + if not scene_list_filename.lower().endswith(".csv"): + scene_list_filename += ".csv" + scene_list_path = get_and_create_path( + scene_list_filename, + output_dir, + ) + logger.info("Writing scene list to CSV file:\n %s", scene_list_path) + with open(scene_list_path, "w") as scene_list_file: + write_scene_list( + output_csv_file=scene_list_file, + scene_list=scenes, + include_cut_list=not skip_cuts, + cut_list=cuts, + ) + # Suppress output if requested. + if quiet: + return + # Print scene list. + if display_scenes: + logger.info( + """Scene List: +----------------------------------------------------------------------- + | Scene # | Start Frame | Start Time | End Frame | End Time | +----------------------------------------------------------------------- +%s +-----------------------------------------------------------------------""", + "\n".join( + [ + " | %5d | %11d | %s | %11d | %s |" + % ( + i + 1, + start_time.get_frames() + 1, + start_time.get_timecode(), + end_time.get_frames(), + end_time.get_timecode(), + ) + for i, (start_time, end_time) in enumerate(scenes) + ] + ), + ) + # Print cut list. + if cuts and display_cuts: + logger.info( + "Comma-separated timecode list:\n %s", + ",".join([cut_format.format(cut) for cut in cuts]), + ) + + +def save_images( + context: CliContext, + scenes: SceneList, + cuts: CutList, + num_images: int, + frame_margin: int, + image_extension: str, + encoder_param: int, + image_name_template: str, + output_dir: ty.Optional[str], + show_progress: bool, + scale: int, + height: int, + width: int, + interpolation: Interpolation, +): + """Handles the `save-images` command.""" + del cuts # save-images only uses scenes. + + images = save_images_impl( + scene_list=scenes, + video=context.video_stream, + num_images=num_images, + frame_margin=frame_margin, + image_extension=image_extension, + encoder_param=encoder_param, + image_name_template=image_name_template, + output_dir=output_dir, + show_progress=show_progress, + scale=scale, + height=height, + width=width, + interpolation=interpolation, + ) + # Save the result for use by `export-html` if required. + context.save_images_result = (images, output_dir) + + +def split_video( + context: CliContext, + scenes: SceneList, + cuts: CutList, + name_format: str, + use_mkvmerge: bool, + output_dir: str, + show_output: bool, + ffmpeg_args: str, +): + """Handles the `split-video` command.""" + del cuts # split-video only uses scenes. + + # Add proper extension to filename template if required. + dot_pos = name_format.rfind(".") + extension_length = 0 if dot_pos < 0 else len(name_format) - (dot_pos + 1) + # If using mkvmerge, force extension to .mkv. + if use_mkvmerge and not name_format.endswith(".mkv"): + name_format += ".mkv" + # Otherwise, if using ffmpeg, only add an extension if one doesn't exist. + elif not 2 <= extension_length <= 4: + name_format += ".mp4" + if use_mkvmerge: + split_video_mkvmerge( + input_video_path=context.video_stream.path, + scene_list=scenes, + output_dir=output_dir, + output_file_template=name_format, + show_output=show_output, + ) + else: + split_video_ffmpeg( + input_video_path=context.video_stream.path, + scene_list=scenes, + output_dir=output_dir, + output_file_template=name_format, + arg_override=ffmpeg_args, + show_progress=not context.quiet_mode, + show_output=show_output, + ) + if scenes: + logger.info("Video splitting completed, scenes written to disk.") diff --git a/scenedetect/_cli/config.py b/scenedetect/_cli/config.py index 929587ad..3ea5babe 100644 --- a/scenedetect/_cli/config.py +++ b/scenedetect/_cli/config.py @@ -30,7 +30,7 @@ from scenedetect.scene_manager import Interpolation from scenedetect.video_splitter import DEFAULT_FFMPEG_ARGS -VALID_PYAV_THREAD_MODES = ["NONE", "SLICE", "FRAME", "AUTO"] +PYAV_THREADING_MODES = ["NONE", "SLICE", "FRAME", "AUTO"] class OptionParseFailure(Exception): @@ -306,7 +306,7 @@ def format(self, timecode: FrameTimecode) -> str: "display-cuts": True, "display-scenes": True, "filename": "$VIDEO_NAME-Scenes.csv", - "output": "", + "output": None, "no-output-file": False, "quiet": False, "skip-cuts": False, @@ -320,7 +320,7 @@ def format(self, timecode: FrameTimecode) -> str: "frame-skip": 0, "merge-last-scene": False, "min-scene-len": TimecodeValue("0.6s"), - "output": "", + "output": None, "verbosity": "info", }, "save-images": { @@ -330,7 +330,7 @@ def format(self, timecode: FrameTimecode) -> str: "frame-margin": 1, "height": 0, "num-images": 3, - "output": "", + "output": None, "quality": RangeValue(_PLACEHOLDER, min_val=0, max_val=100), "scale": 1.0, "scale-method": "linear", @@ -342,7 +342,7 @@ def format(self, timecode: FrameTimecode) -> str: "filename": "$VIDEO_NAME-Scene-$SCENE_NUMBER", "high-quality": False, "mkvmerge": False, - "output": "", + "output": None, "preset": "veryfast", "quiet": False, "rate-factor": RangeValue(22, min_val=0, max_val=100), @@ -354,7 +354,7 @@ def format(self, timecode: FrameTimecode) -> str: CHOICE_MAP: Dict[str, Dict[str, List[str]]] = { "backend-pyav": { - "threading_mode": [mode.lower() for mode in VALID_PYAV_THREAD_MODES], + "threading_mode": [mode.lower() for mode in PYAV_THREADING_MODES], }, "detect-content": { "filter-mode": [mode.name.lower() for mode in FlashFilter.Mode], @@ -580,7 +580,6 @@ def get_value( command: str, option: str, override: Optional[ConfigValue] = None, - ignore_default: bool = False, ) -> ConfigValue: """Get the current setting or default value of the specified command option.""" assert command in CONFIG_MAP and option in CONFIG_MAP[command] @@ -590,8 +589,6 @@ def get_value( value = self._config[command][option] else: value = CONFIG_MAP[command][option] - if ignore_default: - return None if issubclass(type(value), ValidatedValue): return value.value return value diff --git a/scenedetect/_cli/context.py b/scenedetect/_cli/context.py index de0e95a0..1062cc71 100644 --- a/scenedetect/_cli/context.py +++ b/scenedetect/_cli/context.py @@ -12,7 +12,6 @@ """Context of which command-line options and config settings the user provided.""" import logging -import os import typing as ty import click @@ -21,11 +20,8 @@ from scenedetect import AVAILABLE_BACKENDS, open_video from scenedetect._cli.config import ( CHOICE_MAP, - DEFAULT_JPG_QUALITY, - DEFAULT_WEBP_QUALITY, ConfigLoadFailure, ConfigRegistry, - TimecodeFormat, ) from scenedetect.detectors import ( AdaptiveDetector, @@ -35,7 +31,7 @@ ThresholdDetector, ) from scenedetect.frame_timecode import MAX_FPS_DELTA, FrameTimecode -from scenedetect.platform import get_cv2_imwrite_params, init_logger +from scenedetect.platform import init_logger from scenedetect.scene_detector import FlashFilter, SceneDetector from scenedetect.scene_manager import Interpolation, SceneManager from scenedetect.stats_manager import StatsManager @@ -45,35 +41,7 @@ logger = logging.getLogger("pyscenedetect") USER_CONFIG = ConfigRegistry(throw_exception=False) - - -def parse_timecode( - value: ty.Optional[str], frame_rate: float, correct_pts: bool = False -) -> FrameTimecode: - """Parses a user input string into a FrameTimecode assuming the given framerate. - - If value is None, None will be returned instead of processing the value. - - Raises: - click.BadParameter - """ - if value is None: - return None - try: - if correct_pts and value.isdigit(): - value = int(value) - if value >= 1: - value -= 1 - return FrameTimecode(timecode=value, fps=frame_rate) - except ValueError as ex: - raise click.BadParameter( - "timecode must be in seconds (100.0), frames (100), or HH:MM:SS" - ) from ex - - -def contains_sequence_or_url(video_path: str) -> bool: - """Checks if the video path is a URL or image sequence.""" - return "%" in video_path or "://" in video_path +"""The user config, which can be overriden by command-line. If not found, will be default config.""" def check_split_video_requirements(use_mkvmerge: bool) -> None: @@ -103,88 +71,85 @@ def check_split_video_requirements(use_mkvmerge: bool) -> None: class CliContext: - """Context of the command-line interface and config file parameters passed between sub-commands. - - Handles validation of options taken in from the CLI *and* configuration files. - - After processing the main program options via `handle_options`, the CLI will then call - the respective `handle_*` method for each command. Once all commands have been - processed, the main program actions are executed by passing this object to the - `run_scenedetect` function in `scenedetect.cli.controller`. + """The state of the application representing what video will be processed, how, and what to do + with the result. This includes handling all input options via command line and config file. + Once the CLI creates a context, it is executed by passing it to the + `scenedetect._cli.controller.run_scenedetect` function. """ def __init__(self): - self.config = USER_CONFIG - self.video_stream: VideoStream = None + # State: + self.config: ConfigRegistry = USER_CONFIG + self.quiet_mode: bool = None self.scene_manager: SceneManager = None self.stats_manager: StatsManager = None - self.added_detector: bool = False - - # Global `scenedetect` Options - self.output_dir: str = None # -o/--output - self.quiet_mode: bool = None # -q/--quiet or -v/--verbosity quiet - self.stats_file_path: str = None # -s/--stats - self.drop_short_scenes: bool = None # --drop-short-scenes - self.merge_last_scene: bool = None # --merge-last-scene - self.min_scene_len: FrameTimecode = None # -m/--min-scene-len - self.frame_skip: int = None # -fs/--frame-skip - self.default_detector: ty.Tuple[ty.Type[SceneDetector], ty.Dict[str, ty.Any]] = ( - None # [global] default-detector - ) + self.save_images: bool = False # True if the save-images command was specified + self.save_images_result: ty.Any = (None, None) # Result of save-images used by export-html - # `time` Command Options - self.time: bool = False + # Input: + self.video_stream: VideoStream = None + self.load_scenes_input: str = None # load-scenes -i/--input + self.load_scenes_column_name: str = None # load-scenes -c/--start-col-name self.start_time: FrameTimecode = None # time -s/--start self.end_time: FrameTimecode = None # time -e/--end self.duration: FrameTimecode = None # time -d/--duration + self.frame_skip: int = None + + # Options: + self.drop_short_scenes: bool = None + self.merge_last_scene: bool = None + self.min_scene_len: FrameTimecode = None + self.default_detector: ty.Tuple[ty.Type[SceneDetector], ty.Dict[str, ty.Any]] = None + self.output_dir: str = None + self.stats_file_path: str = None + + # Output Commands (e.g. split-video, save-images): + # Commands to run after the detection pipeline. Stored as (callback, args) and invoked with + # the results of the detection pipeline by the controller. + self.commands: ty.List[ty.Tuple[ty.Callable, ty.Dict[str, ty.Any]]] = [] + + def add_command(self, command: ty.Callable, command_args: ty.Dict[str, ty.Any]): + """Add `command` to the processing pipeline. Will be called after processing the input.""" + if "output_dir" in command_args and command_args["output_dir"] is None: + command_args["output_dir"] = self.output_dir + logger.debug("Adding command: %s(%s)", command.__name__, command_args) + self.commands.append((command, command_args)) + + def add_detector(self, detector: ty.Type[SceneDetector], detector_args: ty.Dict[str, ty.Any]): + """Instantiate and add `detector` to the processing pipeline.""" + if self.load_scenes_input: + raise click.ClickException("The load-scenes command cannot be used with detectors.") + logger.debug("Adding detector: %s(%s)", detector.__name__, detector_args) + self.scene_manager.add_detector(detector(**detector_args)) - # `save-images` Command Options - self.save_images: bool = False - self.image_extension: str = None # save-images -j/--jpeg, -w/--webp, -p/--png - self.image_dir: str = None # save-images -o/--output - self.image_param: int = None # save-images -q/--quality if -j/-w, - # otherwise -c/--compression if -p - self.image_name_format: str = None # save-images -f/--name-format - self.num_images: int = None # save-images -n/--num-images - self.frame_margin: int = 1 # save-images -m/--frame-margin - self.scale: float = None # save-images -s/--scale - self.height: int = None # save-images -h/--height - self.width: int = None # save-images -w/--width - self.scale_method: Interpolation = None # [save-images] scale-method - - # `split-video` Command Options - self.split_video: bool = False - self.split_mkvmerge: bool = None # split-video -m/--mkvmerge - self.split_args: str = None # split-video -a/--args, -c/--copy - self.split_dir: str = None # split-video -o/--output - self.split_name_format: str = None # split-video -f/--filename - self.split_quiet: bool = None # split-video -q/--quiet - - # `list-scenes` Command Options - self.list_scenes: bool = False - self.list_scenes_quiet: bool = None # list-scenes -q/--quiet - self.scene_list_dir: str = None # list-scenes -o/--output - self.scene_list_name_format: str = None # list-scenes -f/--filename - self.scene_list_output: bool = None # list-scenes -n/--no-output-file - self.skip_cuts: bool = None # list-scenes -s/--skip-cuts - self.display_cuts: bool = True # [list-scenes] display-cuts - self.display_scenes: bool = True # [list-scenes] display-scenes - self.cut_format: TimecodeFormat = TimecodeFormat.TIMECODE # [list-scenes] cut-format - - # `export-html` Command Options - self.export_html: bool = False - self.html_name_format: str = None # export-html -f/--filename - self.html_include_images: bool = None # export-html --no-images - self.image_width: int = None # export-html -w/--image-width - self.image_height: int = None # export-html -h/--image-height - - # `load-scenes` Command Options - self.load_scenes_input: str = None # load-scenes -i/--input - self.load_scenes_column_name: str = None # load-scenes -c/--start-col-name + def ensure_detector(self): + """Ensures at least one detector has been instantiated, otherwise adds a default one.""" + if self.scene_manager.get_num_detectors() == 0: + logger.debug("No detector specified, adding default detector.") + (detector_type, detector_args) = self.default_detector + self.add_detector(detector_type, detector_args) - # - # Command Handlers - # + def parse_timecode(self, value: ty.Optional[str], correct_pts: bool = False) -> FrameTimecode: + """Parses a user input string into a FrameTimecode assuming the given framerate. If `value` + is None it will be passed through without processing. + + Raises: + click.BadParameter, click.ClickException + """ + if value is None: + return None + try: + if self.video_stream is None: + raise click.ClickException("No input video (-i/--input) was specified.") + if correct_pts and value.isdigit(): + value = int(value) + if value >= 1: + value -= 1 + return FrameTimecode(timecode=value, fps=self.video_stream.frame_rate) + except ValueError as ex: + raise click.BadParameter( + "timecode must be in seconds (100.0), frames (100), or HH:MM:SS" + ) from ex def handle_options( self, @@ -216,11 +181,13 @@ def handle_options( # TODO(v1.0): Make the stats value optional (e.g. allow -s only), and allow use of # $VIDEO_NAME macro in the name. Default to $VIDEO_NAME.csv. + # The `scenedetect` command was just started, let's initialize logging and try to load any + # config files that were specified. try: init_failure = not self.config.initialized init_log = self.config.get_init_log() quiet = not init_failure and quiet - self._initialize_logging(quiet=quiet, verbosity=verbosity, logfile=logfile) + self._initialize_logging(quiet, verbosity, logfile) # Configuration file was specified via CLI argument -c/--config. if config and not init_failure: @@ -262,26 +229,21 @@ def handle_options( param_hint="frame skip + stats file", ) - # Handle the case where -i/--input was not specified (e.g. for the `help` command). + # Handle case where -i/--input was not specified (e.g. for the `help` command). if input_path is None: return - # Have to load the input video to obtain a time base before parsing timecodes. - self._open_video_stream( - input_path=input_path, - framerate=framerate, - backend=self.config.get_value("global", "backend", backend, ignore_default=True), - ) + # Load the input video to obtain a time base for parsing timecodes. + self._open_video_stream(input_path, framerate, backend) - self.output_dir = output if output else self.config.get_value("global", "output") + self.output_dir = self.config.get_value("global", "output", output) if self.output_dir: - logger.info("Output directory set:\n %s", self.output_dir) + logger.debug("Output directory set:\n %s", self.output_dir) - self.min_scene_len = parse_timecode( + self.min_scene_len = self.parse_timecode( min_scene_len if min_scene_len is not None else self.config.get_value("global", "min-scene-len"), - self.video_stream.frame_rate, ) self.drop_short_scenes = drop_short_scenes or self.config.get_value( "global", "drop-short-scenes" @@ -329,6 +291,10 @@ def handle_options( ] self.scene_manager = scene_manager + # + # Detector Parameters + # + def get_detect_content_params( self, threshold: ty.Optional[float] = None, @@ -338,9 +304,7 @@ def get_detect_content_params( kernel_size: ty.Optional[int] = None, filter_mode: ty.Optional[str] = None, ) -> ty.Dict[str, ty.Any]: - """Handle detect-content command options and return args to construct one with.""" - self._ensure_input_open() - + """Get a dict containing user options to construct a ContentDetector with.""" if self.drop_short_scenes: min_scene_len = 0 else: @@ -349,7 +313,7 @@ def get_detect_content_params( min_scene_len = self.min_scene_len.frame_num else: min_scene_len = self.config.get_value("detect-content", "min-scene-len") - min_scene_len = parse_timecode(min_scene_len, self.video_stream.frame_rate).frame_num + min_scene_len = self.parse_timecode(min_scene_len).frame_num if weights is not None: try: @@ -381,7 +345,6 @@ def get_detect_adaptive_params( min_delta_hsv: ty.Optional[float] = None, ) -> ty.Dict[str, ty.Any]: """Handle detect-adaptive command options and return args to construct one with.""" - self._ensure_input_open() # TODO(v0.7): Remove these branches when removing -d/--min-delta-hsv. if min_delta_hsv is not None: @@ -407,7 +370,7 @@ def get_detect_adaptive_params( min_scene_len = self.min_scene_len.frame_num else: min_scene_len = self.config.get_value("detect-adaptive", "min-scene-len") - min_scene_len = parse_timecode(min_scene_len, self.video_stream.frame_rate).frame_num + min_scene_len = self.parse_timecode(min_scene_len).frame_num if weights is not None: try: @@ -435,7 +398,6 @@ def get_detect_threshold_params( min_scene_len: ty.Optional[str] = None, ) -> ty.Dict[str, ty.Any]: """Handle detect-threshold command options and return args to construct one with.""" - self._ensure_input_open() if self.drop_short_scenes: min_scene_len = 0 @@ -445,7 +407,7 @@ def get_detect_threshold_params( min_scene_len = self.min_scene_len.frame_num else: min_scene_len = self.config.get_value("detect-threshold", "min-scene-len") - min_scene_len = parse_timecode(min_scene_len, self.video_stream.frame_rate).frame_num + min_scene_len = self.parse_timecode(min_scene_len).frame_num # TODO(v1.0): add_last_scene cannot be disabled right now. return { "add_final_scene": add_last_scene @@ -455,23 +417,6 @@ def get_detect_threshold_params( "threshold": self.config.get_value("detect-threshold", "threshold", threshold), } - def handle_load_scenes(self, input: ty.AnyStr, start_col_name: ty.Optional[str]): - """Handle `load-scenes` command options.""" - self._ensure_input_open() - if self.added_detector: - raise click.ClickException("The load-scenes command cannot be used with detectors.") - if self.load_scenes_input: - raise click.ClickException("The load-scenes command must only be specified once.") - input = os.path.abspath(input) - if not os.path.exists(input): - raise click.BadParameter( - f"Could not load scenes, file does not exist: {input}", param_hint="-i/--input" - ) - self.load_scenes_input = input - self.load_scenes_column_name = self.config.get_value( - "load-scenes", "start-col-name", start_col_name - ) - def get_detect_hist_params( self, threshold: ty.Optional[float] = None, @@ -479,7 +424,7 @@ def get_detect_hist_params( min_scene_len: ty.Optional[str] = None, ) -> ty.Dict[str, ty.Any]: """Handle detect-hist command options and return args to construct one with.""" - self._ensure_input_open() + if self.drop_short_scenes: min_scene_len = 0 else: @@ -488,7 +433,7 @@ def get_detect_hist_params( min_scene_len = self.min_scene_len.frame_num else: min_scene_len = self.config.get_value("detect-hist", "min-scene-len") - min_scene_len = parse_timecode(min_scene_len, self.video_stream.frame_rate).frame_num + min_scene_len = self.parse_timecode(min_scene_len).frame_num return { "bins": self.config.get_value("detect-hist", "bins", bins), "min_scene_len": min_scene_len, @@ -503,7 +448,7 @@ def get_detect_hash_params( min_scene_len: ty.Optional[str] = None, ) -> ty.Dict[str, ty.Any]: """Handle detect-hash command options and return args to construct one with.""" - self._ensure_input_open() + if self.drop_short_scenes: min_scene_len = 0 else: @@ -512,7 +457,7 @@ def get_detect_hash_params( min_scene_len = self.min_scene_len.frame_num else: min_scene_len = self.config.get_value("detect-hash", "min-scene-len") - min_scene_len = parse_timecode(min_scene_len, self.video_stream.frame_rate).frame_num + min_scene_len = self.parse_timecode(min_scene_len).frame_num return { "lowpass": self.config.get_value("detect-hash", "lowpass", lowpass), "min_scene_len": min_scene_len, @@ -520,274 +465,6 @@ def get_detect_hash_params( "threshold": self.config.get_value("detect-hash", "threshold", threshold), } - def handle_export_html( - self, - filename: ty.Optional[ty.AnyStr], - no_images: bool, - image_width: ty.Optional[int], - image_height: ty.Optional[int], - ): - """Handle `export-html` command options.""" - self._ensure_input_open() - if self.export_html: - self._on_duplicate_command("export_html") - - no_images = no_images or self.config.get_value("export-html", "no-images") - self.html_include_images = not no_images - - self.html_name_format = self.config.get_value("export-html", "filename", filename) - self.image_width = self.config.get_value("export-html", "image-width", image_width) - self.image_height = self.config.get_value("export-html", "image-height", image_height) - - if not self.save_images and not no_images: - raise click.BadArgumentUsage( - "The export-html command requires that the save-images command\n" - "is specified before it, unless --no-images is specified." - ) - logger.info("HTML file name format:\n %s", filename) - - self.export_html = True - - def handle_list_scenes( - self, - output: ty.Optional[ty.AnyStr], - filename: ty.Optional[ty.AnyStr], - no_output_file: bool, - quiet: bool, - skip_cuts: bool, - ): - """Handle `list-scenes` command options.""" - self._ensure_input_open() - if self.list_scenes: - self._on_duplicate_command("list-scenes") - - self.display_cuts = self.config.get_value("list-scenes", "display-cuts") - self.display_scenes = self.config.get_value("list-scenes", "display-scenes") - self.skip_cuts = skip_cuts or self.config.get_value("list-scenes", "skip-cuts") - self.cut_format = TimecodeFormat[self.config.get_value("list-scenes", "cut-format").upper()] - self.list_scenes_quiet = quiet or self.config.get_value("list-scenes", "quiet") - no_output_file = no_output_file or self.config.get_value("list-scenes", "no-output-file") - - self.scene_list_dir = self.config.get_value( - "list-scenes", "output", output, ignore_default=True - ) - self.scene_list_name_format = self.config.get_value("list-scenes", "filename", filename) - if self.scene_list_name_format is not None and not no_output_file: - logger.info("Scene list filename format:\n %s", self.scene_list_name_format) - self.scene_list_output = not no_output_file - if self.scene_list_dir is not None: - logger.info("Scene list output directory:\n %s", self.scene_list_dir) - - self.list_scenes = True - - def handle_split_video( - self, - output: ty.Optional[ty.AnyStr], - filename: ty.Optional[ty.AnyStr], - quiet: bool, - copy: bool, - high_quality: bool, - rate_factor: ty.Optional[int], - preset: ty.Optional[str], - args: ty.Optional[str], - mkvmerge: bool, - ): - """Handle `split-video` command options.""" - self._ensure_input_open() - if self.split_video: - self._on_duplicate_command("split-video") - - check_split_video_requirements(use_mkvmerge=mkvmerge) - - if contains_sequence_or_url(self.video_stream.path): - error_str = "The split-video command is incompatible with image sequences/URLs." - raise click.BadParameter(error_str, param_hint="split-video") - - ## - ## Common Arguments/Options - ## - - self.split_video = True - self.split_quiet = quiet or self.config.get_value("split-video", "quiet") - self.split_dir = self.config.get_value("split-video", "output", output, ignore_default=True) - if self.split_dir is not None: - logger.info("Video output path set: \n%s", self.split_dir) - self.split_name_format = self.config.get_value("split-video", "filename", filename) - - # We only load the config values for these flags/options if none of the other - # encoder flags/options were set via the CLI to avoid any conflicting options - # (e.g. if the config file sets `high-quality = yes` but `--copy` is specified). - if not (mkvmerge or copy or high_quality or args or rate_factor or preset): - mkvmerge = self.config.get_value("split-video", "mkvmerge") - copy = self.config.get_value("split-video", "copy") - high_quality = self.config.get_value("split-video", "high-quality") - rate_factor = self.config.get_value("split-video", "rate-factor") - preset = self.config.get_value("split-video", "preset") - args = self.config.get_value("split-video", "args") - - # Disallow certain combinations of flags/options. - if mkvmerge or copy: - command = "mkvmerge (-m)" if mkvmerge else "copy (-c)" - if high_quality: - raise click.BadParameter( - "high-quality (-hq) cannot be used with %s" % (command), - param_hint="split-video", - ) - if args: - raise click.BadParameter( - "args (-a) cannot be used with %s" % (command), param_hint="split-video" - ) - if rate_factor: - raise click.BadParameter( - "rate-factor (crf) cannot be used with %s" % (command), param_hint="split-video" - ) - if preset: - raise click.BadParameter( - "preset (-p) cannot be used with %s" % (command), param_hint="split-video" - ) - - ## - ## mkvmerge-Specific Arguments/Options - ## - if mkvmerge: - if copy: - logger.warning("copy mode (-c) ignored due to mkvmerge mode (-m).") - self.split_mkvmerge = True - logger.info("Using mkvmerge for video splitting.") - return - - ## - ## ffmpeg-Specific Arguments/Options - ## - if copy: - args = "-map 0:v:0 -map 0:a? -map 0:s? -c:v copy -c:a copy" - elif not args: - if rate_factor is None: - rate_factor = 22 if not high_quality else 17 - if preset is None: - preset = "veryfast" if not high_quality else "slow" - args = ( - "-map 0:v:0 -map 0:a? -map 0:s? " - f"-c:v libx264 -preset {preset} -crf {rate_factor} -c:a aac" - ) - - logger.info("ffmpeg arguments: %s", args) - self.split_args = args - if filename: - logger.info("Output file name format: %s", filename) - - def handle_save_images( - self, - num_images: ty.Optional[int], - output: ty.Optional[ty.AnyStr], - filename: ty.Optional[ty.AnyStr], - jpeg: bool, - webp: bool, - quality: ty.Optional[int], - png: bool, - compression: ty.Optional[int], - frame_margin: ty.Optional[int], - scale: ty.Optional[float], - height: ty.Optional[int], - width: ty.Optional[int], - ): - """Handle `save-images` command options.""" - self._ensure_input_open() - if self.save_images: - self._on_duplicate_command("save-images") - - if "://" in self.video_stream.path: - error_str = "\nThe save-images command is incompatible with URLs." - logger.error(error_str) - raise click.BadParameter(error_str, param_hint="save-images") - - num_flags = sum([1 if flag else 0 for flag in [jpeg, webp, png]]) - if num_flags > 1: - logger.error("Multiple image type flags set for save-images command.") - raise click.BadParameter( - "Only one image type (JPG/PNG/WEBP) can be specified.", param_hint="save-images" - ) - # Only use config params for image format if one wasn't specified. - elif num_flags == 0: - image_format = self.config.get_value("save-images", "format").lower() - jpeg = image_format == "jpeg" - webp = image_format == "webp" - png = image_format == "png" - - # Only use config params for scale/height/width if none of them are specified explicitly. - if scale is None and height is None and width is None: - self.scale = self.config.get_value("save-images", "scale") - self.height = self.config.get_value("save-images", "height") - self.width = self.config.get_value("save-images", "width") - else: - self.scale = scale - self.height = height - self.width = width - - self.scale_method = Interpolation[ - self.config.get_value("save-images", "scale-method").upper() - ] - - default_quality = DEFAULT_WEBP_QUALITY if webp else DEFAULT_JPG_QUALITY - quality = ( - default_quality - if self.config.is_default("save-images", "quality") - else self.config.get_value("save-images", "quality") - ) - - compression = self.config.get_value("save-images", "compression", compression) - self.image_param = compression if png else quality - - self.image_extension = "jpg" if jpeg else "png" if png else "webp" - valid_params = get_cv2_imwrite_params() - if self.image_extension not in valid_params or valid_params[self.image_extension] is None: - error_strs = [ - "Image encoder type `%s` not supported." % self.image_extension.upper(), - "The specified encoder type could not be found in the current OpenCV module.", - "To enable this output format, please update the installed version of OpenCV.", - "If you build OpenCV, ensure the the proper dependencies are enabled. ", - ] - logger.debug("\n".join(error_strs)) - raise click.BadParameter("\n".join(error_strs), param_hint="save-images") - - self.image_dir = self.config.get_value("save-images", "output", output, ignore_default=True) - - self.image_name_format = self.config.get_value("save-images", "filename", filename) - self.num_images = self.config.get_value("save-images", "num-images", num_images) - self.frame_margin = self.config.get_value("save-images", "frame-margin", frame_margin) - - image_type = ("jpeg" if jpeg else self.image_extension).upper() - image_param_type = "Compression" if png else "Quality" - image_param_type = " [%s: %d]" % (image_param_type, self.image_param) - logger.info("Image output format set: %s%s", image_type, image_param_type) - if self.image_dir is not None: - logger.info("Image output directory set:\n %s", os.path.abspath(self.image_dir)) - - self.save_images = True - - def handle_time(self, start, duration, end): - """Handle `time` command options.""" - self._ensure_input_open() - if self.time: - self._on_duplicate_command("time") - if duration is not None and end is not None: - raise click.BadParameter( - "Only one of --duration/-d or --end/-e can be specified, not both.", - param_hint="time", - ) - logger.debug( - "Setting video time:\n start: %s, duration: %s, end: %s", start, duration, end - ) - # *NOTE*: The Python API uses 0-based frame indices, but the CLI uses 1-based indices to - # match the default start number used by `ffmpeg` when saving frames as images. As such, - # we must correct start time if set as frames. See the test_cli_time* tests for for details. - self.start_time = parse_timecode(start, self.video_stream.frame_rate, correct_pts=True) - self.end_time = parse_timecode(end, self.video_stream.frame_rate) - self.duration = parse_timecode(duration, self.video_stream.frame_rate) - if self.start_time and self.end_time and (self.start_time + 1) > self.end_time: - raise click.BadParameter("-e/--end time must be greater than -s/--start") - self.time = True - # # Private Methods # @@ -825,27 +502,11 @@ def _initialize_logging( # Initialize logger with the set CLI args / user configuration. init_logger(log_level=curr_verbosity, show_stdout=not self.quiet_mode, log_file=logfile) - def add_detector(self, detector): - """Add Detector: Adds a detection algorithm to the CliContext's SceneManager.""" - if self.load_scenes_input: - raise click.ClickException("The load-scenes command cannot be used with detectors.") - self._ensure_input_open() - self.scene_manager.add_detector(detector) - self.added_detector = True - - def _ensure_input_open(self) -> None: - """Ensure self.video_stream was initialized (i.e. -i/--input was specified), - otherwise raises an exception. Should only be used from commands that require an - input video to process the options (e.g. those that require a timecode). - - Raises: - click.BadParameter: self.video_stream was not initialized. - """ - if self.video_stream is None: - raise click.ClickException("No input video (-i/--input) was specified.") - def _open_video_stream( - self, input_path: ty.AnyStr, framerate: ty.Optional[float], backend: ty.Optional[str] + self, + input_path: ty.AnyStr, + framerate: ty.Optional[float], + backend: ty.Optional[str], ): if "%" in input_path and backend != "opencv": raise click.BadParameter( @@ -855,14 +516,13 @@ def _open_video_stream( if framerate is not None and framerate < MAX_FPS_DELTA: raise click.BadParameter("Invalid framerate specified!", param_hint="-f/--framerate") try: - if backend is None: - backend = self.config.get_value("global", "backend") - else: - if backend not in AVAILABLE_BACKENDS: - raise click.BadParameter( - "Specified backend %s is not available on this system!" % backend, - param_hint="-b/--backend", - ) + backend = self.config.get_value("global", "backend", backend) + if backend not in AVAILABLE_BACKENDS: + raise click.BadParameter( + "Specified backend %s is not available on this system!" % backend, + param_hint="-b/--backend", + ) + # Open the video with the specified backend, loading any required config settings. if backend == "pyav": self.video_stream = open_video( @@ -905,22 +565,3 @@ def _open_video_stream( raise click.BadParameter( "Input error:\n\n\t%s\n" % str(ex), param_hint="-i/--input" ) from None - - def _on_duplicate_command(self, command: str) -> None: - """Called when a command is duplicated to stop parsing and raise an error. - - Arguments: - command: Command that was duplicated for error context. - - Raises: - click.BadParameter - """ - error_strs = [] - error_strs.append("Error: Command %s specified multiple times." % command) - error_strs.append("The %s command may appear only one time.") - - logger.error("\n".join(error_strs)) - raise click.BadParameter( - "\n Command %s may only be specified once." % command, - param_hint="%s command" % command, - ) diff --git a/scenedetect/_cli/controller.py b/scenedetect/_cli/controller.py index eae039d4..313997fd 100644 --- a/scenedetect/_cli/controller.py +++ b/scenedetect/_cli/controller.py @@ -16,26 +16,15 @@ import os import time import typing as ty -from string import Template -from scenedetect._cli.context import CliContext, check_split_video_requirements +from scenedetect._cli.context import CliContext from scenedetect.frame_timecode import FrameTimecode from scenedetect.platform import get_and_create_path -from scenedetect.scene_manager import ( - get_scenes_from_cuts, - save_images, - write_scene_list, - write_scene_list_html, -) -from scenedetect.video_splitter import split_video_ffmpeg, split_video_mkvmerge +from scenedetect.scene_manager import CutList, SceneList, get_scenes_from_cuts from scenedetect.video_stream import SeekError logger = logging.getLogger("pyscenedetect") -SceneList = ty.List[ty.Tuple[FrameTimecode, FrameTimecode]] - -CutList = ty.List[FrameTimecode] - def run_scenedetect(context: CliContext): """Perform main CLI application control logic. Run once all command-line options and @@ -55,46 +44,49 @@ def run_scenedetect(context: CliContext): logger.info("Skipping detection, loading scenes from: %s", context.load_scenes_input) if context.stats_file_path: logger.warning("WARNING: -s/--stats will be ignored due to load-scenes.") - scene_list, cut_list = _load_scenes(context) - scene_list = _postprocess_scene_list(context, scene_list) - logger.info("Loaded %d scenes.", len(scene_list)) + scenes, cuts = _load_scenes(context) + scenes = _postprocess_scene_list(context, scenes) + logger.info("Loaded %d scenes.", len(scenes)) else: # Perform scene detection on input. - scene_list, cut_list = _detect(context) - scene_list = _postprocess_scene_list(context, scene_list) + scenes, cuts = _detect(context) + scenes = _postprocess_scene_list(context, scenes) # Handle -s/--stats option. _save_stats(context) - if scene_list: + if scenes: logger.info( "Detected %d scenes, average shot length %.1f seconds.", - len(scene_list), - sum([(end_time - start_time).get_seconds() for start_time, end_time in scene_list]) - / float(len(scene_list)), + len(scenes), + sum([(end_time - start_time).get_seconds() for start_time, end_time in scenes]) + / float(len(scenes)), ) else: logger.info("No scenes detected.") - # Handle list-scenes command. - _list_scenes(context, scene_list, cut_list) + # Handle post-processing commands the user wants to run (see scenedetect._cli.commands). + for handler, kwargs in context.commands: + handler(context=context, scenes=scenes, cuts=cuts, **kwargs) - # Handle save-images command. - image_filenames = _save_images(context, scene_list) - # Handle export-html command. - _export_html(context, scene_list, cut_list, image_filenames) +def _postprocess_scene_list(context: CliContext, scene_list: SceneList) -> SceneList: + # Handle --merge-last-scene. If set, when the last scene is shorter than --min-scene-len, + # it will be merged with the previous one. + if context.merge_last_scene and context.min_scene_len is not None and context.min_scene_len > 0: + if len(scene_list) > 1 and (scene_list[-1][1] - scene_list[-1][0]) < context.min_scene_len: + new_last_scene = (scene_list[-2][0], scene_list[-1][1]) + scene_list = scene_list[:-2] + [new_last_scene] - # Handle split-video command. - _split_video(context, scene_list) + # Handle --drop-short-scenes. + if context.drop_short_scenes and context.min_scene_len > 0: + scene_list = [s for s in scene_list if (s[1] - s[0]) >= context.min_scene_len] + return scene_list -def _detect(context: CliContext) -> ty.Optional[ty.Tuple[SceneList, CutList]]: - # Use default detector if one was not specified. - if context.scene_manager.get_num_detectors() == 0: - detector_type, detector_args = context.default_detector - logger.debug("Using default detector: %s(%s)" % (detector_type.__name__, detector_args)) - context.scene_manager.add_detector(detector_type(**detector_args)) +def _detect(context: CliContext) -> ty.Optional[ty.Tuple[SceneList, CutList]]: perf_start_time = time.time() + + context.ensure_detector() if context.start_time is not None: logger.debug("Seeking to start time...") try: @@ -159,158 +151,6 @@ def _save_stats(context: CliContext) -> None: logger.debug("No frame metrics updated, skipping update of the stats file.") -def _list_scenes(context: CliContext, scene_list: SceneList, cut_list: CutList) -> None: - """Handles the `list-scenes` command.""" - if not context.list_scenes: - return - # Write scene list CSV to if required. - if context.scene_list_output: - scene_list_filename = Template(context.scene_list_name_format).safe_substitute( - VIDEO_NAME=context.video_stream.name - ) - if not scene_list_filename.lower().endswith(".csv"): - scene_list_filename += ".csv" - scene_list_path = get_and_create_path( - scene_list_filename, - context.scene_list_dir if context.scene_list_dir is not None else context.output_dir, - ) - logger.info("Writing scene list to CSV file:\n %s", scene_list_path) - with open(scene_list_path, "w") as scene_list_file: - write_scene_list( - output_csv_file=scene_list_file, - scene_list=scene_list, - include_cut_list=not context.skip_cuts, - cut_list=cut_list, - ) - # Suppress output if requested. - if context.list_scenes_quiet: - return - # Print scene list. - if context.display_scenes: - logger.info( - """Scene List: ------------------------------------------------------------------------ - | Scene # | Start Frame | Start Time | End Frame | End Time | ------------------------------------------------------------------------ -%s ------------------------------------------------------------------------""", - "\n".join( - [ - " | %5d | %11d | %s | %11d | %s |" - % ( - i + 1, - start_time.get_frames() + 1, - start_time.get_timecode(), - end_time.get_frames(), - end_time.get_timecode(), - ) - for i, (start_time, end_time) in enumerate(scene_list) - ] - ), - ) - # Print cut list. - if cut_list and context.display_cuts: - logger.info( - "Comma-separated timecode list:\n %s", - ",".join([context.cut_format.format(cut) for cut in cut_list]), - ) - - -def _save_images( - context: CliContext, scene_list: SceneList -) -> ty.Optional[ty.Dict[int, ty.List[str]]]: - """Handles the `save-images` command.""" - if not context.save_images: - return None - # Command can override global output directory setting. - output_dir = context.output_dir if context.image_dir is None else context.image_dir - return save_images( - scene_list=scene_list, - video=context.video_stream, - num_images=context.num_images, - frame_margin=context.frame_margin, - image_extension=context.image_extension, - encoder_param=context.image_param, - image_name_template=context.image_name_format, - output_dir=output_dir, - show_progress=not context.quiet_mode, - scale=context.scale, - height=context.height, - width=context.width, - interpolation=context.scale_method, - ) - - -def _export_html( - context: CliContext, - scene_list: SceneList, - cut_list: CutList, - image_filenames: ty.Optional[ty.Dict[int, ty.List[str]]], -) -> None: - """Handles the `export-html` command.""" - if not context.export_html: - return - # Command can override global output directory setting. - output_dir = context.output_dir if context.image_dir is None else context.image_dir - html_filename = Template(context.html_name_format).safe_substitute( - VIDEO_NAME=context.video_stream.name - ) - if not html_filename.lower().endswith(".html"): - html_filename += ".html" - html_path = get_and_create_path(html_filename, output_dir) - logger.info("Exporting to html file:\n %s:", html_path) - if not context.html_include_images: - image_filenames = None - write_scene_list_html( - html_path, - scene_list, - cut_list, - image_filenames=image_filenames, - image_width=context.image_width, - image_height=context.image_height, - ) - - -def _split_video(context: CliContext, scene_list: SceneList) -> None: - """Handles the `split-video` command.""" - if not context.split_video: - return - output_path_template = context.split_name_format - # Add proper extension to filename template if required. - dot_pos = output_path_template.rfind(".") - extension_length = 0 if dot_pos < 0 else len(output_path_template) - (dot_pos + 1) - # If using mkvmerge, force extension to .mkv. - if context.split_mkvmerge and not output_path_template.endswith(".mkv"): - output_path_template += ".mkv" - # Otherwise, if using ffmpeg, only add an extension if one doesn't exist. - elif not 2 <= extension_length <= 4: - output_path_template += ".mp4" - # Ensure the appropriate tool is available before handling split-video. - check_split_video_requirements(context.split_mkvmerge) - # Command can override global output directory setting. - output_dir = context.output_dir if context.split_dir is None else context.split_dir - if context.split_mkvmerge: - split_video_mkvmerge( - input_video_path=context.video_stream.path, - scene_list=scene_list, - output_dir=output_dir, - output_file_template=output_path_template, - show_output=not (context.quiet_mode or context.split_quiet), - ) - else: - split_video_ffmpeg( - input_video_path=context.video_stream.path, - scene_list=scene_list, - output_dir=output_dir, - output_file_template=output_path_template, - arg_override=context.split_args, - show_progress=not context.quiet_mode, - show_output=not (context.quiet_mode or context.split_quiet), - ) - if scene_list: - logger.info("Video splitting completed, scenes written to disk.") - - def _load_scenes(context: CliContext) -> ty.Tuple[SceneList, CutList]: assert context.load_scenes_input assert os.path.exists(context.load_scenes_input) @@ -353,18 +193,3 @@ def _load_scenes(context: CliContext) -> ty.Tuple[SceneList, CutList]: return get_scenes_from_cuts( cut_list=cut_list, start_pos=start_time, end_pos=end_time ), cut_list - - -def _postprocess_scene_list(context: CliContext, scene_list: SceneList) -> SceneList: - # Handle --merge-last-scene. If set, when the last scene is shorter than --min-scene-len, - # it will be merged with the previous one. - if context.merge_last_scene and context.min_scene_len is not None and context.min_scene_len > 0: - if len(scene_list) > 1 and (scene_list[-1][1] - scene_list[-1][0]) < context.min_scene_len: - new_last_scene = (scene_list[-2][0], scene_list[-1][1]) - scene_list = scene_list[:-2] + [new_last_scene] - - # Handle --drop-short-scenes. - if context.drop_short_scenes and context.min_scene_len > 0: - scene_list = [s for s in scene_list if (s[1] - s[0]) >= context.min_scene_len] - - return scene_list diff --git a/scenedetect/scene_manager.py b/scenedetect/scene_manager.py index dc3bba04..f844ba57 100644 --- a/scenedetect/scene_manager.py +++ b/scenedetect/scene_manager.py @@ -106,6 +106,12 @@ def on_new_scene(frame_img: numpy.ndarray, frame_num: int): logger = logging.getLogger("pyscenedetect") +SceneList = List[Tuple[FrameTimecode, FrameTimecode]] +"""Type hint for a list of scenes in the form (start time, end time).""" + +CutList = List[FrameTimecode] +"""Type hint for a list of cuts, where each timecode represents the first frame of a new shot.""" + # TODO: This value can and should be tuned for performance improvements as much as possible, # until accuracy falls, on a large enough dataset. This has yet to be done, but the current # value doesn't seem to have caused any issues at least. @@ -158,11 +164,11 @@ def compute_downscale_factor(frame_width: int, effective_width: int = DEFAULT_MI def get_scenes_from_cuts( - cut_list: Iterable[FrameTimecode], + cut_list: CutList, start_pos: Union[int, FrameTimecode], end_pos: Union[int, FrameTimecode], base_timecode: Optional[FrameTimecode] = None, -) -> List[Tuple[FrameTimecode, FrameTimecode]]: +) -> SceneList: """Returns a list of tuples of start/end FrameTimecodes for each scene based on a list of detected scene cuts/breaks. @@ -207,9 +213,9 @@ def get_scenes_from_cuts( def write_scene_list( output_csv_file: TextIO, - scene_list: Iterable[Tuple[FrameTimecode, FrameTimecode]], + scene_list: SceneList, include_cut_list: bool = True, - cut_list: Optional[Iterable[FrameTimecode]] = None, + cut_list: Optional[CutList] = None, ) -> None: """Writes the given list of scenes to an output file handle in CSV format. @@ -263,14 +269,14 @@ def write_scene_list( def write_scene_list_html( - output_html_filename, - scene_list, - cut_list=None, - css=None, - css_class="mytable", - image_filenames=None, - image_width=None, - image_height=None, + output_html_filename: str, + scene_list: SceneList, + cut_list: Optional[CutList] = None, + css: str = None, + css_class: str = "mytable", + image_filenames: Optional[Dict[int, List[str]]] = None, + image_width: Optional[int] = None, + image_height: Optional[int] = None, ): """Writes the given list of scenes to an output file handle in html format. @@ -287,6 +293,7 @@ def write_scene_list_html( image_width: Optional desired width of images in table in pixels image_height: Optional desired height of images in table in pixels """ + logger.info("Exporting scenes to html:\n %s:", output_html_filename) if not css: css = """ table.mytable { @@ -386,11 +393,9 @@ def write_scene_list_html( # -# TODO(v1.0): Refactor to take a SceneList object; consider moving this and save scene list -# to a better spot, or just move them to scene_list.py. -# +# TODO(v1.0): Consider moving all post-processing functionality into a separate submodule. def save_images( - scene_list: List[Tuple[FrameTimecode, FrameTimecode]], + scene_list: SceneList, video: VideoStream, num_images: int = 3, frame_margin: int = 1, @@ -474,7 +479,7 @@ def save_images( # Setup flags and init progress bar if available. completed = True - logger.info("Generating output images (%d per scene)...", num_images) + logger.info(f"Saving {num_images} images per scene to {output_dir}, format {image_extension}") progress_bar = None if show_progress: progress_bar = tqdm(total=len(scene_list) * num_images, unit="images", dynamic_ncols=True) @@ -537,6 +542,7 @@ def save_images( video.seek(image_timecode) frame_im = video.read() if frame_im is not None: + # TODO: Add extension to template. # TODO: Allow NUM to be a valid suffix in addition to NUMBER. file_path = "%s.%s" % ( filename_template.safe_substitute( @@ -740,7 +746,7 @@ def clear_detectors(self) -> None: def get_scene_list( self, base_timecode: Optional[FrameTimecode] = None, start_in_scene: bool = False - ) -> List[Tuple[FrameTimecode, FrameTimecode]]: + ) -> SceneList: """Return a list of tuples of start/end FrameTimecodes for each detected scene. Arguments: @@ -779,7 +785,7 @@ def _get_cutting_list(self) -> List[int]: # Ensure all cuts are unique by using a set to remove all duplicates. return [self._base_timecode + cut for cut in sorted(set(self._cutting_list))] - def _get_event_list(self) -> List[Tuple[FrameTimecode, FrameTimecode]]: + def _get_event_list(self) -> SceneList: if not self._event_list: return [] assert self._base_timecode is not None @@ -1065,8 +1071,10 @@ def _decode_thread( # def get_cut_list( - self, base_timecode: Optional[FrameTimecode] = None, show_warning: bool = True - ) -> List[FrameTimecode]: + self, + base_timecode: Optional[FrameTimecode] = None, + show_warning: bool = True, + ) -> CutList: """[DEPRECATED] Return a list of FrameTimecodes of the detected scene changes/cuts. Unlike get_scene_list, the cutting list returns a list of FrameTimecodes representing @@ -1092,9 +1100,7 @@ def get_cut_list( logger.error("`get_cut_list()` is deprecated and will be removed in a future release.") return self._get_cutting_list() - def get_event_list( - self, base_timecode: Optional[FrameTimecode] = None - ) -> List[Tuple[FrameTimecode, FrameTimecode]]: + def get_event_list(self, base_timecode: Optional[FrameTimecode] = None) -> SceneList: """[DEPRECATED] DO NOT USE. Get a list of start/end timecodes of sparse detection events. diff --git a/scenedetect/video_splitter.py b/scenedetect/video_splitter.py index 8b41834d..0fbb9720 100644 --- a/scenedetect/video_splitter.py +++ b/scenedetect/video_splitter.py @@ -193,9 +193,9 @@ def split_video_mkvmerge( if not scene_list: return 0 - logger.info( - "Splitting input video using mkvmerge, output path template:\n %s", output_file_template - ) + logger.info("Splitting video with mkvmerge, output path template:\n %s", output_file_template) + if output_dir: + logger.info("Output folder:\n %s", output_file_template) if video_name is None: video_name = Path(input_video_path).stem @@ -301,9 +301,9 @@ def split_video_ffmpeg( if not scene_list: return 0 - logger.info( - "Splitting input video using ffmpeg, output path template:\n %s", output_file_template - ) + logger.info("Splitting video with ffmpeg, output path template:\n %s", output_file_template) + if output_dir: + logger.info("Output folder:\n %s", output_file_template) if video_name is None: video_name = Path(input_video_path).stem