18643271357

Committed 20 Oct 2025 05:38AM UTC coverage: 69.746% (-1.4%) from 71.158%

Build # 18643271357

Build Type

Pull #119

github

Committed by

web-flow

Commit Message

Merge 123afec07 into 661879c59

Pull Request Pull Request #119: Add Windows taskbar progress integration

Run Details

83 of 196 new or added lines in 8 files covered. (42.35%)

1279 existing lines in 23 files now uncovered.

5542 of 7946 relevant lines covered (69.75%)

0.7 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.28

/talks_reducer/cli.py

"""Command line interface for the talks reducer package."""

from __future__ import annotations

import argparse
import os
import shutil
import subprocess
import sys
import time
from importlib import import_module
from pathlib import Path
from typing import Callable, Dict, List, Optional, Sequence, Tuple

from . import audio
from .ffmpeg import FFmpegNotFoundError
from .logging_config import configure_logging_from_env
from .models import ProcessingOptions, default_temp_folder
from .pipeline import speed_up_video
from .progress import TqdmProgressReporter
from .version_utils import resolve_version


def _build_parser() -> argparse.ArgumentParser:
    """Create the argument parser used by the command line interface."""

    parser = argparse.ArgumentParser(
        description="Modifies a video file to play at different speeds when there is sound vs. silence.",
    )

    # Add version argument
    pkg_version = resolve_version()

    parser.add_argument(
        "--version",
        action="version",
        version=f"talks-reducer {pkg_version}",
    )

    parser.add_argument(
        "input_file",
        type=str,
        nargs="+",
        help="The video file(s) you want modified. Can be one or more directories and / or single files.",
    )
    parser.add_argument(
        "-o",
        "--output_file",
        type=str,
        dest="output_file",
        help="The output file. Only usable if a single file is given. If not included, it'll append _ALTERED to the name.",
    )
    parser.add_argument(
        "--temp_folder",
        type=str,
        default=str(default_temp_folder()),
        help="The file path of the temporary working folder.",
    )
    parser.add_argument(
        "-t",
        "--silent_threshold",
        type=float,
        dest="silent_threshold",
        help="The volume amount that frames' audio needs to surpass to be considered sounded. Defaults to 0.05.",
    )
    parser.add_argument(
        "-S",
        "--sounded_speed",
        type=float,
        dest="sounded_speed",
        help="The speed that sounded (spoken) frames should be played at. Defaults to 1.",
    )
    parser.add_argument(
        "-s",
        "--silent_speed",
        type=float,
        dest="silent_speed",
        help="The speed that silent frames should be played at. Defaults to 4.",
    )
    parser.add_argument(
        "-fm",
        "--frame_margin",
        type=float,
        dest="frame_spreadage",
        help="Some silent frames adjacent to sounded frames are included to provide context. Defaults to 2.",
    )
    parser.add_argument(
        "-sr",
        "--sample_rate",
        type=float,
        dest="sample_rate",
        help="Sample rate of the input and output videos. Usually extracted automatically by FFmpeg.",
    )
    parser.add_argument(
        "--keyframe-interval",
        type=float,
        dest="keyframe_interval_seconds",
        help="Override the keyframe spacing in seconds when using --small. Defaults to 30.",
    )
    parser.add_argument(
        "--video-codec",
        choices=["h264", "hevc", "av1"],
        default="hevc",
        help=(
            "Select the video encoder used for the final render (default: hevc — "
            "h.265 for roughly 25% smaller files). Pick h264 (about 10% faster) "
            "when speed matters or av1 (no advantages) for experimental runs."
        ),
    )
    parser.add_argument(
        "--prefer-global-ffmpeg",
        action="store_true",
        help="Use an FFmpeg binary from PATH before falling back to the bundled static build.",
    )
    parser.add_argument(
        "--small",
        action="store_true",
        help="Apply small file optimizations: resize video to 720p (or 480p with --480), audio to 128k bitrate, best compression (uses CUDA if available).",
    )
    parser.add_argument(
        "--480",
        dest="small_480",
        action="store_true",
        help="Use with --small to scale video to 480p instead of 720p.",
    )
    parser.add_argument(
        "--url",
        dest="server_url",
        default=None,
        help="Process videos via a Talks Reducer server at the provided base URL (for example, http://localhost:9005).",
    )
    parser.add_argument(
        "--host",
        dest="host",
        default=None,
        help="Shortcut for --url when targeting a Talks Reducer server on port 9005 (for example, localhost).",
    )
    parser.add_argument(
        "--server-stream",
        action="store_true",
        help="Stream remote progress updates when using --url.",
    )
    return parser


def gather_input_files(paths: List[str]) -> List[str]:
    """Expand provided paths into a flat list of files that contain audio streams."""

    files: List[str] = []
    for input_path in paths:
        if os.path.isfile(input_path) and audio.is_valid_input_file(input_path):
            files.append(os.path.abspath(input_path))
        elif os.path.isdir(input_path):
            for file in os.listdir(input_path):
                candidate = os.path.join(input_path, file)
                if audio.is_valid_input_file(candidate):
                    files.append(candidate)
    return files


def _print_total_time(start_time: float) -> None:
    """Print the elapsed processing time since *start_time*."""

    end_time = time.time()
    total_time = end_time - start_time
    hours, remainder = divmod(total_time, 3600)
    minutes, seconds = divmod(remainder, 60)
    print(f"\nTime: {int(hours)}h {int(minutes)}m {seconds:.2f}s")


class CliApplication:
    """Coordinator for CLI processing with dependency injection support."""

    def __init__(
        self,
        *,
        gather_files: Callable[[List[str]], List[str]],
        send_video: Optional[Callable[..., Tuple[Path, str, str]]],
        speed_up: Callable[[ProcessingOptions, object], object],
        reporter_factory: Callable[[], object],
        remote_error_message: Optional[str] = None,
    ) -> None:
        self._gather_files = gather_files
        self._send_video = send_video
        self._speed_up = speed_up
        self._reporter_factory = reporter_factory
        self._remote_error_message = remote_error_message

    def run(self, parsed_args: argparse.Namespace) -> Tuple[int, List[str]]:
        """Execute the CLI pipeline for *parsed_args*."""

        start_time = time.time()
        files = self._gather_files(parsed_args.input_file)

        args: Dict[str, object] = {
            key: value for key, value in vars(parsed_args).items() if value is not None
        }
        del args["input_file"]

        if "host" in args:
            del args["host"]

        if len(files) > 1 and "output_file" in args:
            del args["output_file"]

        if getattr(parsed_args, "small_480", False) and not getattr(
            parsed_args, "small", False
        ):
            print(
                "Warning: --480 has no effect unless --small is also provided.",
                file=sys.stderr,
            )

        error_messages: List[str] = []
        reporter_logs: List[str] = []

        if getattr(parsed_args, "server_url", None):
            remote_success, remote_errors, fallback_logs = self._process_via_server(
                files, parsed_args, start_time
            )
            error_messages.extend(remote_errors)
            reporter_logs.extend(fallback_logs)
            if remote_success:
                return 0, error_messages

        reporter = self._reporter_factory()
        for message in reporter_logs:
            reporter.log(message)

        for index, file in enumerate(files):
            print(
                f"Processing file {index + 1}/{len(files)} '{os.path.basename(file)}'"
            )
            local_options = dict(args)

            option_kwargs: Dict[str, object] = {"input_file": Path(file)}

            if "output_file" in local_options:
                option_kwargs["output_file"] = Path(local_options["output_file"])
            if "temp_folder" in local_options:
                option_kwargs["temp_folder"] = Path(local_options["temp_folder"])
            if "silent_threshold" in local_options:
                option_kwargs["silent_threshold"] = float(
                    local_options["silent_threshold"]
                )
            if "silent_speed" in local_options:
                option_kwargs["silent_speed"] = float(local_options["silent_speed"])
            if "sounded_speed" in local_options:
                option_kwargs["sounded_speed"] = float(local_options["sounded_speed"])
            if "frame_spreadage" in local_options:
                option_kwargs["frame_spreadage"] = int(local_options["frame_spreadage"])
            if "sample_rate" in local_options:
                option_kwargs["sample_rate"] = int(local_options["sample_rate"])
            if "keyframe_interval_seconds" in local_options:
                option_kwargs["keyframe_interval_seconds"] = float(
                    local_options["keyframe_interval_seconds"]
                )
            if "video_codec" in local_options:
                option_kwargs["video_codec"] = str(local_options["video_codec"])
            if "small" in local_options:
                option_kwargs["small"] = bool(local_options["small"])
            if local_options.get("small_480"):
                option_kwargs["small_target_height"] = 480
            if "prefer_global_ffmpeg" in local_options:
                option_kwargs["prefer_global_ffmpeg"] = bool(
                    local_options["prefer_global_ffmpeg"]
                )
            options = ProcessingOptions(**option_kwargs)

            try:
                result = self._speed_up(options, reporter=reporter)
            except FFmpegNotFoundError as exc:
                message = str(exc)
                return 1, [*error_messages, message]

            reporter.log(f"Completed: {result.output_file}")
            summary_parts: List[str] = []
            time_ratio = getattr(result, "time_ratio", None)
            size_ratio = getattr(result, "size_ratio", None)
            if time_ratio is not None:
                summary_parts.append(f"{time_ratio * 100:.0f}% time")
            if size_ratio is not None:
                summary_parts.append(f"{size_ratio * 100:.0f}% size")
            if summary_parts:
                reporter.log("Result: " + ", ".join(summary_parts))

        _print_total_time(start_time)
        return 0, error_messages

    def _process_via_server(
        self,
        files: Sequence[str],
        parsed_args: argparse.Namespace,
        start_time: float,
    ) -> Tuple[bool, List[str], List[str]]:
        """Upload *files* to the configured server and download the results."""

        if not self._send_video:
            message = self._remote_error_message or "Server processing is unavailable."
            fallback_notice = "Falling back to local processing pipeline."
            return False, [message, fallback_notice], [message, fallback_notice]

        server_url = parsed_args.server_url
        if not server_url:
            message = "Server URL was not provided."
            fallback_notice = "Falling back to local processing pipeline."
            return False, [message, fallback_notice], [message, fallback_notice]

        output_override: Optional[Path] = None
        if parsed_args.output_file and len(files) == 1:
            output_override = Path(parsed_args.output_file).expanduser()
        elif parsed_args.output_file and len(files) > 1:
            print(
                "Warning: --output is ignored when processing multiple files via the server.",
                file=sys.stderr,
            )

        remote_option_values: Dict[str, object] = {}
        if parsed_args.silent_threshold is not None:
            remote_option_values["silent_threshold"] = float(
                parsed_args.silent_threshold
            )
        if parsed_args.silent_speed is not None:
            remote_option_values["silent_speed"] = float(parsed_args.silent_speed)
        if parsed_args.sounded_speed is not None:
            remote_option_values["sounded_speed"] = float(parsed_args.sounded_speed)
        if getattr(parsed_args, "video_codec", None):
            remote_option_values["video_codec"] = str(parsed_args.video_codec)
        if getattr(parsed_args, "prefer_global_ffmpeg", False):
            remote_option_values["prefer_global_ffmpeg"] = True

        unsupported_options: List[str] = []
        for name in (
            "frame_spreadage",
            "sample_rate",
            "temp_folder",
            "keyframe_interval_seconds",
        ):
            if getattr(parsed_args, name) is not None:
                unsupported_options.append(f"--{name.replace('_', '-')}")

        if unsupported_options:
            print(
                "Warning: the following options are ignored when using --url: "
                + ", ".join(sorted(unsupported_options)),
                file=sys.stderr,
            )

        small_480_mode = bool(getattr(parsed_args, "small_480", False)) and bool(
            getattr(parsed_args, "small", False)
        )
        if small_480_mode:
            remote_option_values["small_480"] = True

        for index, file in enumerate(files, start=1):
            basename = os.path.basename(file)
            print(
                f"Processing file {index}/{len(files)} '{basename}' via server {server_url}"
            )
            printed_log_header = False
            progress_state: dict[str, tuple[Optional[int], Optional[int], str]] = {}
            stream_updates = bool(getattr(parsed_args, "server_stream", False))

            def _stream_server_log(line: str) -> None:
                nonlocal printed_log_header
                if not printed_log_header:
                    print("\nServer log:", flush=True)
                    printed_log_header = True
                print(line, flush=True)

            def _stream_progress(
                desc: str, current: Optional[int], total: Optional[int], unit: str
            ) -> None:
                key = desc or "Processing"
                state = (current, total, unit)
                if progress_state.get(key) == state:
                    return
                progress_state[key] = state

                parts: List[str] = []
                if current is not None and total and total > 0:
                    percent = (current / total) * 100
                    parts.append(f"{current}/{total}")
                    parts.append(f"{percent:.1f}%")
                elif current is not None:
                    parts.append(str(current))
                if unit:
                    parts.append(unit)
                message = " ".join(parts).strip()
                print(f"{key}: {message or 'update'}", flush=True)

            try:
                destination, summary, log_text = self._send_video(
                    input_path=Path(file),
                    output_path=output_override,
                    server_url=server_url,
                    small=bool(parsed_args.small),
                    small_480=small_480_mode,
                    **remote_option_values,
                    log_callback=_stream_server_log,
                    stream_updates=stream_updates,
                    progress_callback=_stream_progress if stream_updates else None,
                )
            except Exception as exc:  # pragma: no cover - network failure safeguard
                message = f"Failed to process {basename} via server: {exc}"
                fallback_notice = "Falling back to local processing pipeline."
                return False, [message, fallback_notice], [message, fallback_notice]

            print(summary)
            print(f"Saved processed video to {destination}")
            if log_text.strip() and not printed_log_header:
                print("\nServer log:\n" + log_text)

        _print_total_time(start_time)
        return True, [], []


def _launch_gui(argv: Sequence[str]) -> bool:
    """Attempt to launch the GUI with the provided arguments."""

    try:
        gui_module = import_module(".gui", __package__)
    except ImportError:
        return False

    gui_main = getattr(gui_module, "main", None)
    if gui_main is None:
        return False

    return bool(gui_main(list(argv)))


def _launch_server(argv: Sequence[str]) -> bool:
    """Attempt to launch the Gradio server with the provided arguments."""

    try:
        server_module = import_module(".server", __package__)
    except ImportError:
        return False

    server_main = getattr(server_module, "main", None)
    if server_main is None:
        return False

    server_main(list(argv))
    return True


def _find_server_tray_binary() -> Optional[Path]:
    """Return the best available path to the server tray executable."""

    binary_name = "talks-reducer-server-tray"
    candidates: List[Path] = []

    which_path = shutil.which(binary_name)
    if which_path:
        candidates.append(Path(which_path))

    try:
        launcher_dir = Path(sys.argv[0]).resolve().parent
    except Exception:
        launcher_dir = None

    potential_names = [binary_name]
    if sys.platform == "win32":
        potential_names = [f"{binary_name}.exe", binary_name]

    if launcher_dir is not None:
        for name in potential_names:
            candidates.append(launcher_dir / name)

    for candidate in candidates:
        if candidate and candidate.exists() and os.access(candidate, os.X_OK):
            return candidate

    return None


def _should_hide_subprocess_console() -> bool:
    """Return ``True` ` when a detached Windows launch should hide the console."""

    if sys.platform != "win32":
        return False

    try:
        import ctypes
    except Exception:  # pragma: no cover - optional runtime dependency
        return False

    try:
        get_console_window = ctypes.windll.kernel32.GetConsoleWindow  # type: ignore[attr-defined]
    except Exception:  # pragma: no cover - platform specific guard
        return False

    try:
        handle = get_console_window()
    except Exception:  # pragma: no cover - defensive fallback
        return False

    return handle == 0


def _launch_server_tray_binary(argv: Sequence[str]) -> bool:
    """Launch the packaged server tray executable when available."""

    command = _find_server_tray_binary()
    if command is None:
        return False

    tray_args = [str(command), *list(argv)]

    run_kwargs: Dict[str, object] = {"check": False}

    if sys.platform == "win32":
        no_window_flag = getattr(subprocess, "CREATE_NO_WINDOW", 0)
        if no_window_flag and _should_hide_subprocess_console():
            run_kwargs["creationflags"] = no_window_flag

    try:
        result = subprocess.run(tray_args, **run_kwargs)
    except OSError:
        return False

    return result.returncode == 0


def _launch_server_tray(argv: Sequence[str]) -> bool:
    """Attempt to launch the server tray helper with the provided arguments."""

    if _launch_server_tray_binary(argv):
        return True

    try:
        tray_module = import_module(".server_tray", __package__)
    except ImportError:
        return False

    tray_main = getattr(tray_module, "main", None)
    if tray_main is None:
        return False

    tray_main(list(argv))
    return True


def main(argv: Optional[Sequence[str]] = None) -> None:
    """Entry point for the command line interface.

    Launch the GUI when run without arguments, otherwise defer to the CLI.
    """

    configure_logging_from_env()

    if argv is None:
        argv_list = sys.argv[1:]
    else:
        argv_list = list(argv)

    if "--server" in argv_list:
        index = argv_list.index("--server")
        tray_args = argv_list[index + 1 :]
        if not _launch_server_tray(tray_args):
            print("Server tray mode is unavailable.", file=sys.stderr)
            sys.exit(1)
        return

    if argv_list and argv_list[0] in {"server", "serve"}:
        if not _launch_server(argv_list[1:]):
            print("Gradio server mode is unavailable.", file=sys.stderr)
            sys.exit(1)
        return

    if not argv_list:
        if _launch_gui(argv_list):
            return

        parser = _build_parser()
        parser.print_help()
        return

    parser = _build_parser()
    parsed_args = parser.parse_args(argv_list)

    host_value = getattr(parsed_args, "host", None)
    if host_value:
        parsed_args.server_url = f"http://{host_value}:9005"

    send_video = None
    remote_error_message: Optional[str] = None
    try:  # pragma: no cover - optional dependency guard
        from . import service_client
    except ImportError as exc:
        remote_error_message = (
            "Server mode requires the gradio_client dependency. " f"({exc})"
        )
    else:
        send_video = service_client.send_video

    application = CliApplication(
        gather_files=gather_input_files,
        send_video=send_video,
        speed_up=speed_up_video,
        reporter_factory=TqdmProgressReporter,
        remote_error_message=remote_error_message,
    )

    exit_code, error_messages = application.run(parsed_args)
    for message in error_messages:
        print(message, file=sys.stderr)
    if exit_code:
        sys.exit(exit_code)


if __name__ == "__main__":
    main()

1	"""Command line interface for the talks reducer package."""
2
3	from __future__ import annotations	1✔
4
5	import argparse	1✔
6	import os	1✔
7	import shutil	1✔
8	import subprocess	1✔
9	import sys	1✔
10	import time	1✔
11	from importlib import import_module	1✔
12	from pathlib import Path	1✔
13	from typing import Callable, Dict, List, Optional, Sequence, Tuple	1✔
14
15	from . import audio	1✔
16	from .ffmpeg import FFmpegNotFoundError	1✔
17	from .logging_config import configure_logging_from_env	1✔
18	from .models import ProcessingOptions, default_temp_folder	1✔
19	from .pipeline import speed_up_video	1✔
20	from .progress import TqdmProgressReporter	1✔
21	from .version_utils import resolve_version	1✔
22
23
24	def _build_parser() -> argparse.ArgumentParser:	1✔
25	"""Create the argument parser used by the command line interface."""
26
27	parser = argparse.ArgumentParser(	1✔
28	description="Modifies a video file to play at different speeds when there is sound vs. silence.",
29	)
30
31	# Add version argument
32	pkg_version = resolve_version()	1✔
33
34	parser.add_argument(	1✔
35	"--version",
36	action="version",
37	version=f"talks-reducer {pkg_version}",
38	)
39
40	parser.add_argument(	1✔
41	"input_file",
42	type=str,
43	nargs="+",
44	help="The video file(s) you want modified. Can be one or more directories and / or single files.",
45	)
46	parser.add_argument(	1✔
47	"-o",
48	"--output_file",
49	type=str,
50	dest="output_file",
51	help="The output file. Only usable if a single file is given. If not included, it'll append _ALTERED to the name.",
52	)
53	parser.add_argument(	1✔
54	"--temp_folder",
55	type=str,
56	default=str(default_temp_folder()),
57	help="The file path of the temporary working folder.",
58	)
59	parser.add_argument(	1✔
60	"-t",
61	"--silent_threshold",
62	type=float,
63	dest="silent_threshold",
64	help="The volume amount that frames' audio needs to surpass to be considered sounded. Defaults to 0.05.",
65	)
66	parser.add_argument(	1✔
67	"-S",
68	"--sounded_speed",
69	type=float,
70	dest="sounded_speed",
71	help="The speed that sounded (spoken) frames should be played at. Defaults to 1.",
72	)
73	parser.add_argument(	1✔
74	"-s",
75	"--silent_speed",
76	type=float,
77	dest="silent_speed",
78	help="The speed that silent frames should be played at. Defaults to 4.",
79	)
80	parser.add_argument(	1✔
81	"-fm",
82	"--frame_margin",
83	type=float,
84	dest="frame_spreadage",
85	help="Some silent frames adjacent to sounded frames are included to provide context. Defaults to 2.",
86	)
87	parser.add_argument(	1✔
88	"-sr",
89	"--sample_rate",
90	type=float,
91	dest="sample_rate",
92	help="Sample rate of the input and output videos. Usually extracted automatically by FFmpeg.",
93	)
94	parser.add_argument(	1✔
95	"--keyframe-interval",
96	type=float,
97	dest="keyframe_interval_seconds",
98	help="Override the keyframe spacing in seconds when using --small. Defaults to 30.",
99	)
100	parser.add_argument(	1✔
101	"--video-codec",
102	choices=["h264", "hevc", "av1"],
103	default="hevc",
104	help=(
105	"Select the video encoder used for the final render (default: hevc — "
106	"h.265 for roughly 25% smaller files). Pick h264 (about 10% faster) "
107	"when speed matters or av1 (no advantages) for experimental runs."
108	),
109	)
110	parser.add_argument(	1✔
111	"--prefer-global-ffmpeg",
112	action="store_true",
113	help="Use an FFmpeg binary from PATH before falling back to the bundled static build.",
114	)
115	parser.add_argument(	1✔
116	"--small",
117	action="store_true",
118	help="Apply small file optimizations: resize video to 720p (or 480p with --480), audio to 128k bitrate, best compression (uses CUDA if available).",
119	)
120	parser.add_argument(	1✔
121	"--480",
122	dest="small_480",
123	action="store_true",
124	help="Use with --small to scale video to 480p instead of 720p.",
125	)
126	parser.add_argument(	1✔
127	"--url",
128	dest="server_url",
129	default=None,
130	help="Process videos via a Talks Reducer server at the provided base URL (for example, http://localhost:9005).",
131	)
132	parser.add_argument(	1✔
133	"--host",
134	dest="host",
135	default=None,
136	help="Shortcut for --url when targeting a Talks Reducer server on port 9005 (for example, localhost).",
137	)
138	parser.add_argument(	1✔
139	"--server-stream",
140	action="store_true",
141	help="Stream remote progress updates when using --url.",
142	)
143	return parser	1✔
144
145
146	def gather_input_files(paths: List[str]) -> List[str]:	1✔
147	"""Expand provided paths into a flat list of files that contain audio streams."""
148
149	files: List[str] = []	1✔
150	for input_path in paths:	1✔
151	if os.path.isfile(input_path) and audio.is_valid_input_file(input_path):	1✔
152	files.append(os.path.abspath(input_path))	1✔
153	elif os.path.isdir(input_path):	1✔
154	for file in os.listdir(input_path):	1✔
155	candidate = os.path.join(input_path, file)	1✔
156	if audio.is_valid_input_file(candidate):	1✔
157	files.append(candidate)	1✔
158	return files	1✔
159
160
161	def _print_total_time(start_time: float) -> None:	1✔
162	"""Print the elapsed processing time since start_time."""
163
164	end_time = time.time()	1✔
165	total_time = end_time - start_time	1✔
166	hours, remainder = divmod(total_time, 3600)	1✔
167	minutes, seconds = divmod(remainder, 60)	1✔
168	print(f"\nTime: {int(hours)}h {int(minutes)}m {seconds:.2f}s")	1✔
169
170
171	class CliApplication:	1✔
172	"""Coordinator for CLI processing with dependency injection support."""
173
174	def __init__(	1✔
175	self,
176	*,
177	gather_files: Callable[[List[str]], List[str]],
178	send_video: Optional[Callable[..., Tuple[Path, str, str]]],
179	speed_up: Callable[[ProcessingOptions, object], object],
180	reporter_factory: Callable[[], object],
181	remote_error_message: Optional[str] = None,
182	) -> None:
183	self._gather_files = gather_files	1✔
184	self._send_video = send_video	1✔
185	self._speed_up = speed_up	1✔
186	self._reporter_factory = reporter_factory	1✔
187	self._remote_error_message = remote_error_message	1✔
188
189	def run(self, parsed_args: argparse.Namespace) -> Tuple[int, List[str]]:	1✔
190	"""Execute the CLI pipeline for parsed_args."""
191
192	start_time = time.time()	1✔
193	files = self._gather_files(parsed_args.input_file)	1✔
194
195	args: Dict[str, object] = {	1✔
196	key: value for key, value in vars(parsed_args).items() if value is not None
197	}
198	del args["input_file"]	1✔
199
200	if "host" in args:	1✔
UNCOV 201	del args["host"]	×
202
203	if len(files) > 1 and "output_file" in args:	1✔
UNCOV 204	del args["output_file"]	×
205
206	if getattr(parsed_args, "small_480", False) and not getattr(	1✔
207	parsed_args, "small", False
208	):
UNCOV 209	print(	×
210	"Warning: --480 has no effect unless --small is also provided.",
211	file=sys.stderr,
212	)
213
214	error_messages: List[str] = []	1✔
215	reporter_logs: List[str] = []	1✔
216
217	if getattr(parsed_args, "server_url", None):	1✔
218	remote_success, remote_errors, fallback_logs = self._process_via_server(	1✔
219	files, parsed_args, start_time
220	)
221	error_messages.extend(remote_errors)	1✔
222	reporter_logs.extend(fallback_logs)	1✔
223	if remote_success:	1✔
224	return 0, error_messages	1✔
225
226	reporter = self._reporter_factory()	1✔
227	for message in reporter_logs:	1✔
228	reporter.log(message)	1✔
229
230	for index, file in enumerate(files):	1✔
231	print(	1✔
232	f"Processing file {index + 1}/{len(files)} '{os.path.basename(file)}'"
233	)
234	local_options = dict(args)	1✔
235
236	option_kwargs: Dict[str, object] = {"input_file": Path(file)}	1✔
237
238	if "output_file" in local_options:	1✔
239	option_kwargs["output_file"] = Path(local_options["output_file"])	1✔
240	if "temp_folder" in local_options:	1✔
241	option_kwargs["temp_folder"] = Path(local_options["temp_folder"])	1✔
242	if "silent_threshold" in local_options:	1✔
243	option_kwargs["silent_threshold"] = float(	1✔
244	local_options["silent_threshold"]
245	)
246	if "silent_speed" in local_options:	1✔
247	option_kwargs["silent_speed"] = float(local_options["silent_speed"])	1✔
248	if "sounded_speed" in local_options:	1✔
249	option_kwargs["sounded_speed"] = float(local_options["sounded_speed"])	1✔
250	if "frame_spreadage" in local_options:	1✔
251	option_kwargs["frame_spreadage"] = int(local_options["frame_spreadage"])	1✔
252	if "sample_rate" in local_options:	1✔
253	option_kwargs["sample_rate"] = int(local_options["sample_rate"])	1✔
254	if "keyframe_interval_seconds" in local_options:	1✔
255	option_kwargs["keyframe_interval_seconds"] = float(	1✔
256	local_options["keyframe_interval_seconds"]
257	)
258	if "video_codec" in local_options:	1✔
259	option_kwargs["video_codec"] = str(local_options["video_codec"])	1✔
260	if "small" in local_options:	1✔
261	option_kwargs["small"] = bool(local_options["small"])	1✔
262	if local_options.get("small_480"):	1✔
UNCOV 263	option_kwargs["small_target_height"] = 480	×
264	if "prefer_global_ffmpeg" in local_options:	1✔
265	option_kwargs["prefer_global_ffmpeg"] = bool(	1✔
266	local_options["prefer_global_ffmpeg"]
267	)
268	options = ProcessingOptions(**option_kwargs)	1✔
269
270	try:	1✔
271	result = self._speed_up(options, reporter=reporter)	1✔
UNCOV 272	except FFmpegNotFoundError as exc:	×
UNCOV 273	message = str(exc)	×
UNCOV 274	return 1, [*error_messages, message]	×
275
276	reporter.log(f"Completed: {result.output_file}")	1✔
277	summary_parts: List[str] = []	1✔
278	time_ratio = getattr(result, "time_ratio", None)	1✔
279	size_ratio = getattr(result, "size_ratio", None)	1✔
280	if time_ratio is not None:	1✔
281	summary_parts.append(f"{time_ratio * 100:.0f}% time")	1✔
282	if size_ratio is not None:	1✔
283	summary_parts.append(f"{size_ratio * 100:.0f}% size")	1✔
284	if summary_parts:	1✔
285	reporter.log("Result: " + ", ".join(summary_parts))	1✔
286
287	_print_total_time(start_time)	1✔
288	return 0, error_messages	1✔
289
290	def _process_via_server(	1✔
291	self,
292	files: Sequence[str],
293	parsed_args: argparse.Namespace,
294	start_time: float,
295	) -> Tuple[bool, List[str], List[str]]:
296	"""Upload files to the configured server and download the results."""
297
298	if not self._send_video:	1✔
299	message = self._remote_error_message or "Server processing is unavailable."	1✔
300	fallback_notice = "Falling back to local processing pipeline."	1✔
301	return False, [message, fallback_notice], [message, fallback_notice]	1✔
302
303	server_url = parsed_args.server_url	1✔
304	if not server_url:	1✔
UNCOV 305	message = "Server URL was not provided."	×
UNCOV 306	fallback_notice = "Falling back to local processing pipeline."	×
UNCOV 307	return False, [message, fallback_notice], [message, fallback_notice]	×
308
309	output_override: Optional[Path] = None	1✔
310	if parsed_args.output_file and len(files) == 1:	1✔
UNCOV 311	output_override = Path(parsed_args.output_file).expanduser()	×
312	elif parsed_args.output_file and len(files) > 1:	1✔
313	print(	1✔
314	"Warning: --output is ignored when processing multiple files via the server.",
315	file=sys.stderr,
316	)
317
318	remote_option_values: Dict[str, object] = {}	1✔
319	if parsed_args.silent_threshold is not None:	1✔
320	remote_option_values["silent_threshold"] = float(	1✔
321	parsed_args.silent_threshold
322	)
323	if parsed_args.silent_speed is not None:	1✔
324	remote_option_values["silent_speed"] = float(parsed_args.silent_speed)	1✔
325	if parsed_args.sounded_speed is not None:	1✔
326	remote_option_values["sounded_speed"] = float(parsed_args.sounded_speed)	1✔
327	if getattr(parsed_args, "video_codec", None):	1✔
328	remote_option_values["video_codec"] = str(parsed_args.video_codec)	1✔
329	if getattr(parsed_args, "prefer_global_ffmpeg", False):	1✔
330	remote_option_values["prefer_global_ffmpeg"] = True	1✔
331
332	unsupported_options: List[str] = []	1✔
333	for name in (	1✔
334	"frame_spreadage",
335	"sample_rate",
336	"temp_folder",
337	"keyframe_interval_seconds",
338	):
339	if getattr(parsed_args, name) is not None:	1✔
340	unsupported_options.append(f"--{name.replace('_', '-')}")	1✔
341
342	if unsupported_options:	1✔
343	print(	1✔
344	"Warning: the following options are ignored when using --url: "
345	+ ", ".join(sorted(unsupported_options)),
346	file=sys.stderr,
347	)
348
349	small_480_mode = bool(getattr(parsed_args, "small_480", False)) and bool(	1✔
350	getattr(parsed_args, "small", False)
351	)
352	if small_480_mode:	1✔
UNCOV 353	remote_option_values["small_480"] = True	×
354
355	for index, file in enumerate(files, start=1):	1✔
356	basename = os.path.basename(file)	1✔
357	print(	1✔
358	f"Processing file {index}/{len(files)} '{basename}' via server {server_url}"
359	)
360	printed_log_header = False	1✔
361	progress_state: dict[str, tuple[Optional[int], Optional[int], str]] = {}	1✔
362	stream_updates = bool(getattr(parsed_args, "server_stream", False))	1✔
363
364	def _stream_server_log(line: str) -> None:	1✔
365	nonlocal printed_log_header
366	if not printed_log_header:	1✔
367	print("\nServer log:", flush=True)	1✔
368	printed_log_header = True	1✔
369	print(line, flush=True)	1✔
370
371	def _stream_progress(	1✔
372	desc: str, current: Optional[int], total: Optional[int], unit: str
373	) -> None:
374	key = desc or "Processing"	1✔
375	state = (current, total, unit)	1✔
376	if progress_state.get(key) == state:	1✔
377	return	1✔
378	progress_state[key] = state	1✔
379
380	parts: List[str] = []	1✔
381	if current is not None and total and total > 0:	1✔
382	percent = (current / total) * 100	1✔
383	parts.append(f"{current}/{total}")	1✔
384	parts.append(f"{percent:.1f}%")	1✔
385	elif current is not None:	1✔
UNCOV 386	parts.append(str(current))	×
387	if unit:	1✔
388	parts.append(unit)	1✔
389	message = " ".join(parts).strip()	1✔
390	print(f"{key}: {message or 'update'}", flush=True)	1✔
391
392	try:	1✔
393	destination, summary, log_text = self._send_video(	1✔
394	input_path=Path(file),
395	output_path=output_override,
396	server_url=server_url,
397	small=bool(parsed_args.small),
398	small_480=small_480_mode,
399	**remote_option_values,
400	log_callback=_stream_server_log,
401	stream_updates=stream_updates,
402	progress_callback=_stream_progress if stream_updates else None,
403	)
404	except Exception as exc: # pragma: no cover - network failure safeguard
405	message = f"Failed to process {basename} via server: {exc}"
406	fallback_notice = "Falling back to local processing pipeline."
407	return False, [message, fallback_notice], [message, fallback_notice]
408
409	print(summary)	1✔
410	print(f"Saved processed video to {destination}")	1✔
411	if log_text.strip() and not printed_log_header:	1✔
412	print("\nServer log:\n" + log_text)	1✔
413
414	_print_total_time(start_time)	1✔
415	return True, [], []	1✔
416
417
418	def _launch_gui(argv: Sequence[str]) -> bool:	1✔
419	"""Attempt to launch the GUI with the provided arguments."""
420
UNCOV 421	try:	×
UNCOV 422	gui_module = import_module(".gui", __package__)	×
UNCOV 423	except ImportError:	×
UNCOV 424	return False	×
425
UNCOV 426	gui_main = getattr(gui_module, "main", None)	×
UNCOV 427	if gui_main is None:	×
UNCOV 428	return False	×
429
UNCOV 430	return bool(gui_main(list(argv)))	×
431
432
433	def _launch_server(argv: Sequence[str]) -> bool:	1✔
434	"""Attempt to launch the Gradio server with the provided arguments."""
435
UNCOV 436	try:	×
UNCOV 437	server_module = import_module(".server", __package__)	×
UNCOV 438	except ImportError:	×
UNCOV 439	return False	×
440
UNCOV 441	server_main = getattr(server_module, "main", None)	×
UNCOV 442	if server_main is None:	×
UNCOV 443	return False	×
444
UNCOV 445	server_main(list(argv))	×
UNCOV 446	return True	×
447
448
449	def _find_server_tray_binary() -> Optional[Path]:	1✔
450	"""Return the best available path to the server tray executable."""
451
452	binary_name = "talks-reducer-server-tray"	1✔
453	candidates: List[Path] = []	1✔
454
455	which_path = shutil.which(binary_name)	1✔
456	if which_path:	1✔
457	candidates.append(Path(which_path))	1✔
458
459	try:	1✔
460	launcher_dir = Path(sys.argv[0]).resolve().parent	1✔
UNCOV 461	except Exception:	×
UNCOV 462	launcher_dir = None	×
463
464	potential_names = [binary_name]	1✔
465	if sys.platform == "win32":	1✔
UNCOV 466	potential_names = [f"{binary_name}.exe", binary_name]	×
467
468	if launcher_dir is not None:	1✔
469	for name in potential_names:	1✔
470	candidates.append(launcher_dir / name)	1✔
471
472	for candidate in candidates:	1✔
473	if candidate and candidate.exists() and os.access(candidate, os.X_OK):	1✔
474	return candidate	1✔
475
UNCOV 476	return None	×
477
478
479	def _should_hide_subprocess_console() -> bool:	1✔
480	"""Return ``True` ` when a detached Windows launch should hide the console."""
481
482	if sys.platform != "win32":	1✔
483	return False	1✔
484
485	try:	1✔
486	import ctypes	1✔
487	except Exception: # pragma: no cover - optional runtime dependency
488	return False
489
490	try:	1✔
491	get_console_window = ctypes.windll.kernel32.GetConsoleWindow # type: ignore[attr-defined]	1✔
492	except Exception: # pragma: no cover - platform specific guard
493	return False
494
495	try:	1✔
496	handle = get_console_window()	1✔
497	except Exception: # pragma: no cover - defensive fallback
498	return False
499
500	return handle == 0	1✔
501
502
503	def _launch_server_tray_binary(argv: Sequence[str]) -> bool:	1✔
504	"""Launch the packaged server tray executable when available."""
505
506	command = _find_server_tray_binary()	1✔
507	if command is None:	1✔
508	return False	1✔
509
510	tray_args = [str(command), *list(argv)]	1✔
511
512	run_kwargs: Dict[str, object] = {"check": False}	1✔
513
514	if sys.platform == "win32":	1✔
515	no_window_flag = getattr(subprocess, "CREATE_NO_WINDOW", 0)	1✔
516	if no_window_flag and _should_hide_subprocess_console():	1✔
517	run_kwargs["creationflags"] = no_window_flag	1✔
518
519	try:	1✔
520	result = subprocess.run(tray_args, **run_kwargs)	1✔
UNCOV 521	except OSError:	×
UNCOV 522	return False	×
523
524	return result.returncode == 0	1✔
525
526
527	def _launch_server_tray(argv: Sequence[str]) -> bool:	1✔
528	"""Attempt to launch the server tray helper with the provided arguments."""
529
530	if _launch_server_tray_binary(argv):	1✔
UNCOV 531	return True	×
532
533	try:	1✔
534	tray_module = import_module(".server_tray", __package__)	1✔
UNCOV 535	except ImportError:	×
UNCOV 536	return False	×
537
538	tray_main = getattr(tray_module, "main", None)	1✔
539	if tray_main is None:	1✔
UNCOV 540	return False	×
541
542	tray_main(list(argv))	1✔
543	return True	1✔
544
545
546	def main(argv: Optional[Sequence[str]] = None) -> None:	1✔
547	"""Entry point for the command line interface.
548
549	Launch the GUI when run without arguments, otherwise defer to the CLI.
550	"""
551
552	configure_logging_from_env()	1✔
553
554	if argv is None:	1✔
UNCOV 555	argv_list = sys.argv[1:]	×
556	else:
557	argv_list = list(argv)	1✔
558
559	if "--server" in argv_list:	1✔
560	index = argv_list.index("--server")	1✔
561	tray_args = argv_list[index + 1 :]	1✔
562	if not _launch_server_tray(tray_args):	1✔
563	print("Server tray mode is unavailable.", file=sys.stderr)	1✔
564	sys.exit(1)	1✔
565	return	1✔
566
567	if argv_list and argv_list[0] in {"server", "serve"}:	1✔
568	if not _launch_server(argv_list[1:]):	1✔
569	print("Gradio server mode is unavailable.", file=sys.stderr)	1✔
570	sys.exit(1)	1✔
571	return	1✔
572
573	if not argv_list:	1✔
574	if _launch_gui(argv_list):	1✔
575	return	1✔
576
UNCOV 577	parser = _build_parser()	×
UNCOV 578	parser.print_help()	×
UNCOV 579	return	×
580
581	parser = _build_parser()	1✔
582	parsed_args = parser.parse_args(argv_list)	1✔
583
584	host_value = getattr(parsed_args, "host", None)	1✔
585	if host_value:	1✔
UNCOV 586	parsed_args.server_url = f"http://{host_value}:9005"	×
587
588	send_video = None	1✔
589	remote_error_message: Optional[str] = None	1✔
590	try: # pragma: no cover - optional dependency guard
591	from . import service_client
UNCOV 592	except ImportError as exc:	×
UNCOV 593	remote_error_message = (	×
594	"Server mode requires the gradio_client dependency. " f"({exc})"
595	)
596	else:
597	send_video = service_client.send_video	1✔
598
599	application = CliApplication(	1✔
600	gather_files=gather_input_files,
601	send_video=send_video,
602	speed_up=speed_up_video,
603	reporter_factory=TqdmProgressReporter,
604	remote_error_message=remote_error_message,
605	)
606
607	exit_code, error_messages = application.run(parsed_args)	1✔
608	for message in error_messages:	1✔
UNCOV 609	print(message, file=sys.stderr)	×
610	if exit_code:	1✔
UNCOV 611	sys.exit(exit_code)	×
612
613
614	if __name__ == "__main__":	1✔
UNCOV 615	main()	×

popstas / talks-reducer / 18643271357

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous