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

53.85

/talks_reducer/progress.py

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

from __future__ import annotations

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

from tqdm import tqdm

from .windows_taskbar import TaskbarProgressState

if TYPE_CHECKING:  # pragma: no cover - import for typing only
    from .windows_taskbar import TaskbarProgress


logger = logging.getLogger(__name__)


@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))


@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


class _TaskbarTaskContext(AbstractContextManager[ProgressHandle]):
    """Context manager that wraps another progress task with taskbar updates."""

    def __init__(
        self,
        reporter: "TaskbarProgressReporter",
        context: AbstractContextManager[ProgressHandle],
        total: Optional[int],
    ) -> None:
        self._reporter = reporter
        self._context = context
        self._total = total

    def __enter__(self) -> ProgressHandle:
        handle = self._context.__enter__()
        self._reporter._start(total=self._total, current=handle.current)
        return _TaskbarProgressHandle(handle, self._reporter)

    def __exit__(self, exc_type, exc, tb) -> bool:
        try:
            return self._context.__exit__(exc_type, exc, tb)
        finally:
            self._reporter._finalize()


@dataclass
class _TaskbarProgressHandle:
    """Progress handle proxy that mirrors events to the Windows taskbar."""

    _delegate: ProgressHandle
    _reporter: "TaskbarProgressReporter"

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

    def ensure_total(self, total: int) -> None:
        self._delegate.ensure_total(total)
        self._reporter._on_total(total=total, current=self.current)

    def advance(self, amount: int) -> None:
        previous = self.current
        self._delegate.advance(amount)
        if amount > 0 and self.current != previous:
            self._reporter._on_advance(current=self.current)

    def finish(self) -> None:
        self._delegate.finish()
        self._reporter._on_finish(current=self.current)


class TaskbarProgressReporter(ProgressReporter):
    """Progress reporter that mirrors updates to the Windows taskbar."""

    def __init__(self, delegate: ProgressReporter, taskbar: "TaskbarProgress") -> None:
        self._delegate = delegate
        self._taskbar = taskbar
        self._enabled = True
        self._total: Optional[int] = None
        self._current: int = 0
        self._finalized = False
        logger.debug(
            "TaskbarProgressReporter initialised with delegate=%s taskbar=%s",
            type(delegate).__name__,
            taskbar,
        )

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

    def task(
        self, *, desc: str = "", total: Optional[int] = None, unit: str = ""
    ) -> AbstractContextManager[ProgressHandle]:
        logger.debug(
            "Creating taskbar-mirrored task desc=%r total=%s unit=%r", desc, total, unit
        )
        context = self._delegate.task(desc=desc, total=total, unit=unit)
        self._finalized = False
        return _TaskbarTaskContext(self, context, total)

    # Internal hooks -----------------------------------------------------
    def _start(self, total: Optional[int], current: int) -> None:
        self._total = total
        self._current = current
        if not self._enabled:
            logger.debug(
                "Skipping taskbar start because reporter is disabled (total=%s current=%s)",
                total,
                current,
            )
            return
        try:
            if not total or total <= 0:
                logger.debug(
                    "Setting taskbar progress state to indeterminate (current=%s)",
                    current,
                )
                self._taskbar.set_progress_state(TaskbarProgressState.INDETERMINATE)
            else:
                logger.debug(
                    "Initialising taskbar progress (current=%s total=%s)",
                    current,
                    total,
                )
                self._taskbar.set_progress_state(TaskbarProgressState.NORMAL)
                self._taskbar.set_progress_value(current, total)
        except Exception as exc:  # pragma: no cover - Windows-specific logging
            self._disable(exc)

    def _on_total(self, total: int, current: int) -> None:
        if total <= 0:
            return
        self._total = max(self._total or 0, total)
        if not self._enabled or not self._total:
            logger.debug(
                "Skipping taskbar total update (enabled=%s total=%s current=%s)",
                self._enabled,
                self._total,
                current,
            )
            return
        try:
            logger.debug(
                "Updating taskbar total/current (total=%s current=%s)",
                self._total,
                current,
            )
            self._taskbar.set_progress_state(TaskbarProgressState.NORMAL)
            self._taskbar.set_progress_value(current, self._total)
        except Exception as exc:  # pragma: no cover - Windows-specific logging
            self._disable(exc)

    def _on_advance(self, current: int) -> None:
        self._current = current
        if not self._enabled or not self._total:
            logger.debug(
                "Skipping taskbar advance (enabled=%s total=%s current=%s)",
                self._enabled,
                self._total,
                current,
            )
            return
        try:
            logger.debug(
                "Advancing taskbar progress (current=%s total=%s)",
                current,
                self._total,
            )
            self._taskbar.set_progress_value(current, self._total)
        except Exception as exc:  # pragma: no cover - Windows-specific logging
            self._disable(exc)

    def _on_finish(self, current: int) -> None:
        self._current = current
        if not self._enabled:
            logger.debug(
                "Skipping taskbar finish because reporter is disabled (current=%s)",
                current,
            )
            return
        try:
            if self._total and self._total > 0:
                logger.debug(
                    "Finishing taskbar progress (current=%s total=%s)",
                    current,
                    self._total,
                )
                self._taskbar.set_progress_value(self._total, self._total)
            self._taskbar.clear()
            logger.debug("Cleared taskbar progress after finish")
        except Exception as exc:  # pragma: no cover - Windows-specific logging
            self._disable(exc)

    def _finalize(self) -> None:
        if self._finalized:
            logger.debug("Taskbar reporter already finalised; skipping cleanup")
            return
        self._finalized = True
        try:
            if self._enabled:
                self._taskbar.clear()
                logger.debug("Cleared taskbar progress during finalise")
        except Exception:  # pragma: no cover - best-effort cleanup
            pass
        finally:
            try:
                self._taskbar.close()
                logger.debug("Closed taskbar progress helper")
            except Exception:  # pragma: no cover - best-effort cleanup
                pass

    def _disable(self, exc: Exception) -> None:
        if not self._enabled:
            return
        self._enabled = False
        logger.debug(
            "Disabling Windows taskbar progress updates: %s", exc, exc_info=True
        )


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

1	"""Progress reporting utilities shared by the CLI and GUI layers."""
2
3	from __future__ import annotations	1✔
4
5	import logging	1✔
6	import sys	1✔
7	from contextlib import AbstractContextManager	1✔
8	from dataclasses import dataclass	1✔
9	from typing import TYPE_CHECKING, Optional, Protocol, runtime_checkable	1✔
10
11	from tqdm import tqdm	1✔
12
13	from .windows_taskbar import TaskbarProgressState	1✔
14
15	if TYPE_CHECKING: # pragma: no cover - import for typing only
16	from .windows_taskbar import TaskbarProgress
17
18
19	logger = logging.getLogger(__name__)	1✔
20
21
22	@runtime_checkable	1✔
23	class ProgressHandle(Protocol):	1✔
24	"""Represents a single progress task that can be updated incrementally."""
25
26	@property	1✔
27	def current(self) -> int:	1✔
28	"""Return the number of processed units."""
29
30	def ensure_total(self, total: int) -> None:	1✔
31	"""Increase the total units when FFmpeg reports a larger frame count."""
32
33	def advance(self, amount: int) -> None:	1✔
34	"""Advance the progress cursor by ``amount`` units."""
35
36	def finish(self) -> None:	1✔
37	"""Mark the task as finished, filling in any remaining progress."""
38
39
40	@runtime_checkable	1✔
41	class ProgressReporter(Protocol):	1✔
42	"""Interface used by the pipeline to stream progress information."""
43
44	def log(self, message: str) -> None:	1✔
45	"""Emit an informational log message to the user interface."""
46
47	def task(	1✔
48	self, *, desc: str = "", total: Optional[int] = None, unit: str = ""
49	) -> AbstractContextManager[ProgressHandle]:
50	"""Return a context manager managing a :class:`ProgressHandle`."""
51
52
53	@dataclass	1✔
54	class _NullProgressHandle:	1✔
55	"""No-op implementation for environments that do not need progress."""
56
57	total: Optional[int] = None	1✔
58	current: int = 0	1✔
59
60	def ensure_total(self, total: int) -> None:	1✔
61	self.total = max(self.total or 0, total)	1✔
62
63	def advance(self, amount: int) -> None:	1✔
64	self.current += amount	1✔
65
66	def finish(self) -> None:	1✔
67	if self.total is not None:	1✔
68	self.current = self.total	1✔
69
70
71	class NullProgressReporter(ProgressReporter):	1✔
72	"""Progress reporter that ignores all output."""
73
74	def log(self, message: str) -> None: # pragma: no cover - intentional no-op
75	del message
76
77	def task(	1✔
78	self, *, desc: str = "", total: Optional[int] = None, unit: str = ""
79	) -> AbstractContextManager[ProgressHandle]:
80	del desc, unit	1✔
81
82	class _Context(AbstractContextManager[ProgressHandle]):	1✔
83	def __init__(self, handle: _NullProgressHandle) -> None:	1✔
84	self._handle = handle	1✔
85
86	def __enter__(self) -> ProgressHandle:	1✔
87	return self._handle	1✔
88
89	def __exit__(self, exc_type, exc, tb) -> bool:	1✔
90	return False	1✔
91
92	return _Context(_NullProgressHandle(total=total))	1✔
93
94
95	@dataclass	1✔
96	class _TqdmProgressHandle(AbstractContextManager[ProgressHandle]):	1✔
97	"""Wraps a :class:`tqdm.tqdm` instance to match :class:`ProgressHandle`."""
98
99	bar: tqdm	1✔
100
101	@property	1✔
102	def current(self) -> int:	1✔
103	return int(self.bar.n)	1✔
104
105	def ensure_total(self, total: int) -> None:	1✔
106	if self.bar.total is None or total > self.bar.total:	1✔
107	self.bar.total = total	1✔
108
109	def advance(self, amount: int) -> None:	1✔
110	if amount > 0:	1✔
111	self.bar.update(amount)	1✔
112
113	def finish(self) -> None:	1✔
114	if self.bar.total is not None and self.bar.n < self.bar.total:	1✔
115	self.bar.update(self.bar.total - self.bar.n)	1✔
116
117	def __enter__(self) -> ProgressHandle:	1✔
118	return self	1✔
119
120	def __exit__(self, exc_type, exc, tb) -> bool:	1✔
121	if exc_type is None:	1✔
122	self.finish()	1✔
123	self.bar.close()	1✔
124	return False	1✔
125
126
127	class TqdmProgressReporter(ProgressReporter):	1✔
128	"""Adapter that renders pipeline progress using :mod:`tqdm`."""
129
130	def __init__(self) -> None:	1✔
131	self._bar_format = (	1✔
132	"{desc:<20} {percentage:3.0f}%"
133	"\|{bar:10}\|"
134	" {n_fmt:>6}/{total_fmt:>6} [{elapsed:^5}<{remaining:^5}, {rate_fmt}{postfix}]"
135	)
136
137	def log(self, message: str) -> None:	1✔
138	tqdm.write(message)	1✔
139
140	def task(	1✔
141	self, *, desc: str = "", total: Optional[int] = None, unit: str = ""
142	) -> AbstractContextManager[ProgressHandle]:
143	bar = tqdm(	1✔
144	total=total,
145	desc=desc,
146	unit=unit,
147	bar_format=self._bar_format,
148	file=sys.stderr,
149	)
150	return _TqdmProgressHandle(bar)	1✔
151
152
153	class SignalProgressReporter(NullProgressReporter):	1✔
154	"""Placeholder implementation for GUI integrations.
155
156	UI front-ends can subclass this type and emit framework-specific signals when
157	progress updates arrive.
158	"""
159
160	pass	1✔
161
162
163	class _TaskbarTaskContext(AbstractContextManager[ProgressHandle]):	1✔
164	"""Context manager that wraps another progress task with taskbar updates."""
165
166	def __init__(	1✔
167	self,
168	reporter: "TaskbarProgressReporter",
169	context: AbstractContextManager[ProgressHandle],
170	total: Optional[int],
171	) -> None:
NEW 172	self._reporter = reporter	×
NEW 173	self._context = context	×
NEW 174	self._total = total	×
175
176	def __enter__(self) -> ProgressHandle:	1✔
NEW 177	handle = self._context.__enter__()	×
NEW 178	self._reporter._start(total=self._total, current=handle.current)	×
NEW 179	return _TaskbarProgressHandle(handle, self._reporter)	×
180
181	def __exit__(self, exc_type, exc, tb) -> bool:	1✔
NEW 182	try:	×
NEW 183	return self._context.__exit__(exc_type, exc, tb)	×
184	finally:
NEW 185	self._reporter._finalize()	×
186
187
188	@dataclass	1✔
189	class _TaskbarProgressHandle:	1✔
190	"""Progress handle proxy that mirrors events to the Windows taskbar."""
191
192	_delegate: ProgressHandle	1✔
193	_reporter: "TaskbarProgressReporter"	1✔
194
195	@property	1✔
196	def current(self) -> int:	1✔
NEW 197	return self._delegate.current	×
198
199	def ensure_total(self, total: int) -> None:	1✔
NEW 200	self._delegate.ensure_total(total)	×
NEW 201	self._reporter._on_total(total=total, current=self.current)	×
202
203	def advance(self, amount: int) -> None:	1✔
NEW 204	previous = self.current	×
NEW 205	self._delegate.advance(amount)	×
NEW 206	if amount > 0 and self.current != previous:	×
NEW 207	self._reporter._on_advance(current=self.current)	×
208
209	def finish(self) -> None:	1✔
NEW 210	self._delegate.finish()	×
NEW 211	self._reporter._on_finish(current=self.current)	×
212
213
214	class TaskbarProgressReporter(ProgressReporter):	1✔
215	"""Progress reporter that mirrors updates to the Windows taskbar."""
216
217	def __init__(self, delegate: ProgressReporter, taskbar: "TaskbarProgress") -> None:	1✔
NEW 218	self._delegate = delegate	×
NEW 219	self._taskbar = taskbar	×
NEW 220	self._enabled = True	×
NEW 221	self._total: Optional[int] = None	×
NEW 222	self._current: int = 0	×
NEW 223	self._finalized = False	×
NEW 224	logger.debug(	×
225	"TaskbarProgressReporter initialised with delegate=%s taskbar=%s",
226	type(delegate).__name__,
227	taskbar,
228	)
229
230	def log(self, message: str) -> None:	1✔
NEW 231	self._delegate.log(message)	×
232
233	def task(	1✔
234	self, *, desc: str = "", total: Optional[int] = None, unit: str = ""
235	) -> AbstractContextManager[ProgressHandle]:
NEW 236	logger.debug(	×
237	"Creating taskbar-mirrored task desc=%r total=%s unit=%r", desc, total, unit
238	)
NEW 239	context = self._delegate.task(desc=desc, total=total, unit=unit)	×
NEW 240	self._finalized = False	×
NEW 241	return _TaskbarTaskContext(self, context, total)	×
242
243	# Internal hooks -----------------------------------------------------
244	def _start(self, total: Optional[int], current: int) -> None:	1✔
NEW 245	self._total = total	×
NEW 246	self._current = current	×
NEW 247	if not self._enabled:	×
NEW 248	logger.debug(	×
249	"Skipping taskbar start because reporter is disabled (total=%s current=%s)",
250	total,
251	current,
252	)
NEW 253	return	×
NEW 254	try:	×
NEW 255	if not total or total <= 0:	×
NEW 256	logger.debug(	×
257	"Setting taskbar progress state to indeterminate (current=%s)",
258	current,
259	)
NEW 260	self._taskbar.set_progress_state(TaskbarProgressState.INDETERMINATE)	×
261	else:
NEW 262	logger.debug(	×
263	"Initialising taskbar progress (current=%s total=%s)",
264	current,
265	total,
266	)
NEW 267	self._taskbar.set_progress_state(TaskbarProgressState.NORMAL)	×
NEW 268	self._taskbar.set_progress_value(current, total)	×
269	except Exception as exc: # pragma: no cover - Windows-specific logging
270	self._disable(exc)
271
272	def _on_total(self, total: int, current: int) -> None:	1✔
NEW 273	if total <= 0:	×
NEW 274	return	×
NEW 275	self._total = max(self._total or 0, total)	×
NEW 276	if not self._enabled or not self._total:	×
NEW 277	logger.debug(	×
278	"Skipping taskbar total update (enabled=%s total=%s current=%s)",
279	self._enabled,
280	self._total,
281	current,
282	)
NEW 283	return	×
NEW 284	try:	×
NEW 285	logger.debug(	×
286	"Updating taskbar total/current (total=%s current=%s)",
287	self._total,
288	current,
289	)
NEW 290	self._taskbar.set_progress_state(TaskbarProgressState.NORMAL)	×
NEW 291	self._taskbar.set_progress_value(current, self._total)	×
292	except Exception as exc: # pragma: no cover - Windows-specific logging
293	self._disable(exc)
294
295	def _on_advance(self, current: int) -> None:	1✔
NEW 296	self._current = current	×
NEW 297	if not self._enabled or not self._total:	×
NEW 298	logger.debug(	×
299	"Skipping taskbar advance (enabled=%s total=%s current=%s)",
300	self._enabled,
301	self._total,
302	current,
303	)
NEW 304	return	×
NEW 305	try:	×
NEW 306	logger.debug(	×
307	"Advancing taskbar progress (current=%s total=%s)",
308	current,
309	self._total,
310	)
NEW 311	self._taskbar.set_progress_value(current, self._total)	×
312	except Exception as exc: # pragma: no cover - Windows-specific logging
313	self._disable(exc)
314
315	def _on_finish(self, current: int) -> None:	1✔
NEW 316	self._current = current	×
NEW 317	if not self._enabled:	×
NEW 318	logger.debug(	×
319	"Skipping taskbar finish because reporter is disabled (current=%s)",
320	current,
321	)
NEW 322	return	×
NEW 323	try:	×
NEW 324	if self._total and self._total > 0:	×
NEW 325	logger.debug(	×
326	"Finishing taskbar progress (current=%s total=%s)",
327	current,
328	self._total,
329	)
NEW 330	self._taskbar.set_progress_value(self._total, self._total)	×
NEW 331	self._taskbar.clear()	×
NEW 332	logger.debug("Cleared taskbar progress after finish")	×
333	except Exception as exc: # pragma: no cover - Windows-specific logging
334	self._disable(exc)
335
336	def _finalize(self) -> None:	1✔
NEW 337	if self._finalized:	×
NEW 338	logger.debug("Taskbar reporter already finalised; skipping cleanup")	×
NEW 339	return	×
NEW 340	self._finalized = True	×
NEW 341	try:	×
NEW 342	if self._enabled:	×
NEW 343	self._taskbar.clear()	×
NEW 344	logger.debug("Cleared taskbar progress during finalise")	×
345	except Exception: # pragma: no cover - best-effort cleanup
346	pass
347	finally:
NEW 348	try:	×
NEW 349	self._taskbar.close()	×
NEW 350	logger.debug("Closed taskbar progress helper")	×
351	except Exception: # pragma: no cover - best-effort cleanup
352	pass
353
354	def _disable(self, exc: Exception) -> None:	1✔
NEW 355	if not self._enabled:	×
NEW 356	return	×
NEW 357	self._enabled = False	×
NEW 358	logger.debug(	×
359	"Disabling Windows taskbar progress updates: %s", exc, exc_info=True
360	)
361
362
363	__all__ = [	1✔
364	"ProgressHandle",
365	"ProgressReporter",
366	"NullProgressReporter",
367	"TqdmProgressReporter",
368	"SignalProgressReporter",
369	"TaskbarProgressReporter",
370	]

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