19548235201

Committed 20 Nov 2025 07:02PM UTC coverage: 68.202% (+0.1%) from 68.072%

Build # 19548235201

Build Type

push

github

Committed by

web-flow

Commit Message

refactor: centralize progress callbacks (#134)

Run Details

59 of 61 new or added lines in 7 files covered. (96.72%)

77 existing lines in 7 files now uncovered.

5924 of 8686 relevant lines covered (68.2%)

0.68 hits per line

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

98.21

/talks_reducer/progress.py

"""Progress reporting utilities shared by the CLI and GUI layers."""

from __future__ import annotations

import sys
from contextlib import AbstractContextManager
from dataclasses import dataclass
from typing import Callable, Optional, Protocol, runtime_checkable

from tqdm import tqdm


@runtime_checkable
class ProgressHandle(Protocol):
    """Represents a single progress task that can be updated incrementally."""

    @property
    def current(self) -> int:
        """Return the number of processed units."""

    def ensure_total(self, total: int) -> None:
        """Increase the total units when FFmpeg reports a larger frame count."""

    def advance(self, amount: int) -> None:
        """Advance the progress cursor by ``amount`` units."""

    def finish(self) -> None:
        """Mark the task as finished, filling in any remaining progress."""


@runtime_checkable
class ProgressReporter(Protocol):
    """Interface used by the pipeline to stream progress information."""

    def log(self, message: str) -> None:
        """Emit an informational log message to the user interface."""

    def task(
        self, *, desc: str = "", total: Optional[int] = None, unit: str = ""
    ) -> AbstractContextManager[ProgressHandle]:
        """Return a context manager managing a :class:`ProgressHandle`."""


@dataclass
class _NullProgressHandle:
    """No-op implementation for environments that do not need progress."""

    total: Optional[int] = None
    current: int = 0

    def ensure_total(self, total: int) -> None:
        self.total = max(self.total or 0, total)

    def advance(self, amount: int) -> None:
        self.current += amount

    def finish(self) -> None:
        if self.total is not None:
            self.current = self.total


class NullProgressReporter(ProgressReporter):
    """Progress reporter that ignores all output."""

    def log(self, message: str) -> None:  # pragma: no cover - intentional no-op
        del message

    def task(
        self, *, desc: str = "", total: Optional[int] = None, unit: str = ""
    ) -> AbstractContextManager[ProgressHandle]:
        del desc, unit

        class _Context(AbstractContextManager[ProgressHandle]):
            def __init__(self, handle: _NullProgressHandle) -> None:
                self._handle = handle

            def __enter__(self) -> ProgressHandle:
                return self._handle

            def __exit__(self, exc_type, exc, tb) -> bool:
                return False

        return _Context(_NullProgressHandle(total=total))


class CallbackProgressHandle(AbstractContextManager[ProgressHandle]):
    """Progress handle that triggers callbacks when progress changes."""

    def __init__(
        self,
        *,
        desc: str = "",
        total: Optional[int] = None,
        on_start: Optional[Callable[[str, Optional[int]], None]] = None,
        on_update: Optional[Callable[[int, Optional[int], str], None]] = None,
        on_finish: Optional[Callable[[int, Optional[int], str], None]] = None,
        infer_total_on_finish: bool = False,
    ) -> None:
        self._desc = desc
        self._total = total
        self._current = 0
        self._on_update = on_update
        self._on_finish = on_finish
        self._infer_total_on_finish = infer_total_on_finish
        self._on_start = on_start

        if self._on_start is not None:
            self._on_start(self._desc, self._total)

    @property
    def current(self) -> int:
        return self._current

    def ensure_total(self, total: int) -> None:
        if total > 0 and (self._total is None or total > self._total):
            self._total = total
            if self._on_update is not None:
                self._on_update(self._current, self._total, self._desc)

    def advance(self, amount: int) -> None:
        if amount <= 0:
            return
        self._current += amount
        if self._on_update is not None:
            self._on_update(self._current, self._total, self._desc)

    def finish(self) -> None:
        if self._total is not None and self._current < self._total:
            self._current = self._total
        elif self._total is None and self._infer_total_on_finish:
            self._total = self._current if self._current > 0 else 1

        if self._on_update is not None:
            self._on_update(self._current, self._total, self._desc)
        if self._on_finish is not None:
            self._on_finish(self._current, self._total, self._desc)

    def __enter__(self) -> ProgressHandle:
        return self

    def __exit__(self, exc_type, exc, tb) -> bool:
        if exc_type is None:
            self.finish()
        return False


@dataclass
class _TqdmProgressHandle(AbstractContextManager[ProgressHandle]):
    """Wraps a :class:`tqdm.tqdm` instance to match :class:`ProgressHandle`."""

    bar: tqdm

    @property
    def current(self) -> int:
        return int(self.bar.n)

    def ensure_total(self, total: int) -> None:
        if self.bar.total is None or total > self.bar.total:
            self.bar.total = total

    def advance(self, amount: int) -> None:
        if amount > 0:
            self.bar.update(amount)

    def finish(self) -> None:
        if self.bar.total is not None and self.bar.n < self.bar.total:
            self.bar.update(self.bar.total - self.bar.n)

    def __enter__(self) -> ProgressHandle:
        return self

    def __exit__(self, exc_type, exc, tb) -> bool:
        if exc_type is None:
            self.finish()
        self.bar.close()
        return False


class TqdmProgressReporter(ProgressReporter):
    """Adapter that renders pipeline progress using :mod:`tqdm`."""

    def __init__(self) -> None:
        self._bar_format = (
            "{desc:<20} {percentage:3.0f}%"
            "|{bar:10}|"
            " {n_fmt:>6}/{total_fmt:>6} [{elapsed:^5}<{remaining:^5}, {rate_fmt}{postfix}]"
        )

    def log(self, message: str) -> None:
        tqdm.write(message)

    def task(
        self, *, desc: str = "", total: Optional[int] = None, unit: str = ""
    ) -> AbstractContextManager[ProgressHandle]:
        bar = tqdm(
            total=total,
            desc=desc,
            unit=unit,
            bar_format=self._bar_format,
            file=sys.stderr,
        )
        return _TqdmProgressHandle(bar)


class SignalProgressReporter(NullProgressReporter):
    """Placeholder implementation for GUI integrations.

    UI front-ends can subclass this type and emit framework-specific signals when
    progress updates arrive.
    """

    pass


__all__ = [
    "CallbackProgressHandle",
    "ProgressHandle",
    "ProgressReporter",
    "NullProgressReporter",
    "TqdmProgressReporter",
    "SignalProgressReporter",
]

1	"""Progress reporting utilities shared by the CLI and GUI layers."""
2
3	from __future__ import annotations	1✔
4
5	import sys	1✔
6	from contextlib import AbstractContextManager	1✔
7	from dataclasses import dataclass	1✔
8	from typing import Callable, Optional, Protocol, runtime_checkable	1✔
9
10	from tqdm import tqdm	1✔
11
12
13	@runtime_checkable	1✔
14	class ProgressHandle(Protocol):	1✔
15	"""Represents a single progress task that can be updated incrementally."""
16
17	@property	1✔
18	def current(self) -> int:	1✔
19	"""Return the number of processed units."""
20
21	def ensure_total(self, total: int) -> None:	1✔
22	"""Increase the total units when FFmpeg reports a larger frame count."""
23
24	def advance(self, amount: int) -> None:	1✔
25	"""Advance the progress cursor by ``amount`` units."""
26
27	def finish(self) -> None:	1✔
28	"""Mark the task as finished, filling in any remaining progress."""
29
30
31	@runtime_checkable	1✔
32	class ProgressReporter(Protocol):	1✔
33	"""Interface used by the pipeline to stream progress information."""
34
35	def log(self, message: str) -> None:	1✔
36	"""Emit an informational log message to the user interface."""
37
38	def task(	1✔
39	self, *, desc: str = "", total: Optional[int] = None, unit: str = ""
40	) -> AbstractContextManager[ProgressHandle]:
41	"""Return a context manager managing a :class:`ProgressHandle`."""
42
43
44	@dataclass	1✔
45	class _NullProgressHandle:	1✔
46	"""No-op implementation for environments that do not need progress."""
47
48	total: Optional[int] = None	1✔
49	current: int = 0	1✔
50
51	def ensure_total(self, total: int) -> None:	1✔
52	self.total = max(self.total or 0, total)	1✔
53
54	def advance(self, amount: int) -> None:	1✔
55	self.current += amount	1✔
56
57	def finish(self) -> None:	1✔
58	if self.total is not None:	1✔
59	self.current = self.total	1✔
60
61
62	class NullProgressReporter(ProgressReporter):	1✔
63	"""Progress reporter that ignores all output."""
64
65	def log(self, message: str) -> None: # pragma: no cover - intentional no-op
66	del message
67
68	def task(	1✔
69	self, *, desc: str = "", total: Optional[int] = None, unit: str = ""
70	) -> AbstractContextManager[ProgressHandle]:
71	del desc, unit	1✔
72
73	class _Context(AbstractContextManager[ProgressHandle]):	1✔
74	def __init__(self, handle: _NullProgressHandle) -> None:	1✔
75	self._handle = handle	1✔
76
77	def __enter__(self) -> ProgressHandle:	1✔
78	return self._handle	1✔
79
80	def __exit__(self, exc_type, exc, tb) -> bool:	1✔
81	return False	1✔
82
83	return _Context(_NullProgressHandle(total=total))	1✔
84
85
86	class CallbackProgressHandle(AbstractContextManager[ProgressHandle]):	1✔
87	"""Progress handle that triggers callbacks when progress changes."""
88
89	def __init__(	1✔
90	self,
91	*,
92	desc: str = "",
93	total: Optional[int] = None,
94	on_start: Optional[Callable[[str, Optional[int]], None]] = None,
95	on_update: Optional[Callable[[int, Optional[int], str], None]] = None,
96	on_finish: Optional[Callable[[int, Optional[int], str], None]] = None,
97	infer_total_on_finish: bool = False,
98	) -> None:
99	self._desc = desc	1✔
100	self._total = total	1✔
101	self._current = 0	1✔
102	self._on_update = on_update	1✔
103	self._on_finish = on_finish	1✔
104	self._infer_total_on_finish = infer_total_on_finish	1✔
105	self._on_start = on_start	1✔
106
107	if self._on_start is not None:	1✔
108	self._on_start(self._desc, self._total)	1✔
109
110	@property	1✔
111	def current(self) -> int:	1✔
112	return self._current	1✔
113
114	def ensure_total(self, total: int) -> None:	1✔
115	if total > 0 and (self._total is None or total > self._total):	1✔
116	self._total = total	1✔
117	if self._on_update is not None:	1✔
118	self._on_update(self._current, self._total, self._desc)	1✔
119
120	def advance(self, amount: int) -> None:	1✔
121	if amount <= 0:	1✔
NEW 122	return	×
123	self._current += amount	1✔
124	if self._on_update is not None:	1✔
125	self._on_update(self._current, self._total, self._desc)	1✔
126
127	def finish(self) -> None:	1✔
128	if self._total is not None and self._current < self._total:	1✔
129	self._current = self._total	1✔
130	elif self._total is None and self._infer_total_on_finish:	1✔
NEW 131	self._total = self._current if self._current > 0 else 1	×
132
133	if self._on_update is not None:	1✔
134	self._on_update(self._current, self._total, self._desc)	1✔
135	if self._on_finish is not None:	1✔
136	self._on_finish(self._current, self._total, self._desc)	1✔
137
138	def __enter__(self) -> ProgressHandle:	1✔
139	return self	1✔
140
141	def __exit__(self, exc_type, exc, tb) -> bool:	1✔
142	if exc_type is None:	1✔
143	self.finish()	1✔
144	return False	1✔
145
146
147	@dataclass	1✔
148	class _TqdmProgressHandle(AbstractContextManager[ProgressHandle]):	1✔
149	"""Wraps a :class:`tqdm.tqdm` instance to match :class:`ProgressHandle`."""
150
151	bar: tqdm	1✔
152
153	@property	1✔
154	def current(self) -> int:	1✔
155	return int(self.bar.n)	1✔
156
157	def ensure_total(self, total: int) -> None:	1✔
158	if self.bar.total is None or total > self.bar.total:	1✔
159	self.bar.total = total	1✔
160
161	def advance(self, amount: int) -> None:	1✔
162	if amount > 0:	1✔
163	self.bar.update(amount)	1✔
164
165	def finish(self) -> None:	1✔
166	if self.bar.total is not None and self.bar.n < self.bar.total:	1✔
167	self.bar.update(self.bar.total - self.bar.n)	1✔
168
169	def __enter__(self) -> ProgressHandle:	1✔
170	return self	1✔
171
172	def __exit__(self, exc_type, exc, tb) -> bool:	1✔
173	if exc_type is None:	1✔
174	self.finish()	1✔
175	self.bar.close()	1✔
176	return False	1✔
177
178
179	class TqdmProgressReporter(ProgressReporter):	1✔
180	"""Adapter that renders pipeline progress using :mod:`tqdm`."""
181
182	def __init__(self) -> None:	1✔
183	self._bar_format = (	1✔
184	"{desc:<20} {percentage:3.0f}%"
185	"\|{bar:10}\|"
186	" {n_fmt:>6}/{total_fmt:>6} [{elapsed:^5}<{remaining:^5}, {rate_fmt}{postfix}]"
187	)
188
189	def log(self, message: str) -> None:	1✔
190	tqdm.write(message)	1✔
191
192	def task(	1✔
193	self, *, desc: str = "", total: Optional[int] = None, unit: str = ""
194	) -> AbstractContextManager[ProgressHandle]:
195	bar = tqdm(	1✔
196	total=total,
197	desc=desc,
198	unit=unit,
199	bar_format=self._bar_format,
200	file=sys.stderr,
201	)
202	return _TqdmProgressHandle(bar)	1✔
203
204
205	class SignalProgressReporter(NullProgressReporter):	1✔
206	"""Placeholder implementation for GUI integrations.
207
208	UI front-ends can subclass this type and emit framework-specific signals when
209	progress updates arrive.
210	"""
211
212	pass	1✔
213
214
215	__all__ = [	1✔
216	"CallbackProgressHandle",
217	"ProgressHandle",
218	"ProgressReporter",
219	"NullProgressReporter",
220	"TqdmProgressReporter",
221	"SignalProgressReporter",
222	]

popstas / talks-reducer / 19548235201

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