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

96.0

/talks_reducer/gui/progress.py

"""Progress helpers that bridge the pipeline with the Tkinter GUI."""

from __future__ import annotations

from typing import Callable, Optional

from ..progress import ProgressHandle, SignalProgressReporter


class _GuiProgressHandle(ProgressHandle):
    """Simple progress handle that records totals but only logs milestones."""

    def __init__(self, log_callback: Callable[[str], None], desc: str) -> None:
        self._log_callback = log_callback
        self._desc = desc
        self._current = 0
        self._total: Optional[int] = None
        if desc:
            self._log_callback(f"{desc} started")

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

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

    def advance(self, amount: int) -> None:
        if amount > 0:
            self._current += amount

    def finish(self) -> None:
        if self._total is not None:
            self._current = self._total
        if self._desc:
            self._log_callback(f"{self._desc} completed")

    def __enter__(self) -> "_GuiProgressHandle":
        return self

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


class _TkProgressReporter(SignalProgressReporter):
    """Progress reporter that forwards updates to the GUI thread."""

    def __init__(
        self,
        log_callback: Callable[[str], None],
        process_callback: Optional[Callable] = None,
        *,
        stop_callback: Optional[Callable[[], bool]] = None,
        taskbar_hwnd: Optional[int] = None,
    ) -> None:
        self._log_callback = log_callback
        self.process_callback = process_callback
        self._stop_callback = stop_callback
        self._taskbar_hwnd = taskbar_hwnd

    def log(self, message: str) -> None:
        self._log_callback(message)
        print(message, flush=True)

    def task(
        self, *, desc: str = "", total: Optional[int] = None, unit: str = ""
    ) -> _GuiProgressHandle:
        del total, unit
        return _GuiProgressHandle(self._log_callback, desc)

    def stop_requested(self) -> bool:
        """Return ``True`` when the GUI has asked to cancel processing."""

        if self._stop_callback is None:
            return False
        return bool(self._stop_callback())

    def taskbar_hwnd(self) -> Optional[int]:
        """Return the HWND associated with the GUI window, if available."""

        return self._taskbar_hwnd


__all__ = ["_GuiProgressHandle", "_TkProgressReporter"]

1	"""Progress helpers that bridge the pipeline with the Tkinter GUI."""
2
3	from __future__ import annotations	1✔
4
5	from typing import Callable, Optional	1✔
6
7	from ..progress import ProgressHandle, SignalProgressReporter	1✔
8
9
10	class _GuiProgressHandle(ProgressHandle):	1✔
11	"""Simple progress handle that records totals but only logs milestones."""
12
13	def __init__(self, log_callback: Callable[[str], None], desc: str) -> None:	1✔
14	self._log_callback = log_callback	1✔
15	self._desc = desc	1✔
16	self._current = 0	1✔
17	self._total: Optional[int] = None	1✔
18	if desc:	1✔
19	self._log_callback(f"{desc} started")	1✔
20
21	@property	1✔
22	def current(self) -> int:	1✔
23	return self._current	1✔
24
25	def ensure_total(self, total: int) -> None:	1✔
26	if self._total is None or total > self._total:	1✔
27	self._total = total	1✔
28
29	def advance(self, amount: int) -> None:	1✔
30	if amount > 0:	1✔
31	self._current += amount	1✔
32
33	def finish(self) -> None:	1✔
34	if self._total is not None:	1✔
35	self._current = self._total	1✔
36	if self._desc:	1✔
37	self._log_callback(f"{self._desc} completed")	1✔
38
39	def __enter__(self) -> "_GuiProgressHandle":	1✔
40	return self	1✔
41
42	def __exit__(self, exc_type, exc, tb) -> bool:	1✔
43	if exc_type is None:	1✔
44	self.finish()	1✔
45	return False	1✔
46
47
48	class _TkProgressReporter(SignalProgressReporter):	1✔
49	"""Progress reporter that forwards updates to the GUI thread."""
50
51	def __init__(	1✔
52	self,
53	log_callback: Callable[[str], None],
54	process_callback: Optional[Callable] = None,
55	*,
56	stop_callback: Optional[Callable[[], bool]] = None,
57	taskbar_hwnd: Optional[int] = None,
58	) -> None:
59	self._log_callback = log_callback	1✔
60	self.process_callback = process_callback	1✔
61	self._stop_callback = stop_callback	1✔
62	self._taskbar_hwnd = taskbar_hwnd	1✔
63
64	def log(self, message: str) -> None:	1✔
65	self._log_callback(message)	1✔
66	print(message, flush=True)	1✔
67
68	def task(	1✔
69	self, *, desc: str = "", total: Optional[int] = None, unit: str = ""
70	) -> _GuiProgressHandle:
71	del total, unit	1✔
72	return _GuiProgressHandle(self._log_callback, desc)	1✔
73
74	def stop_requested(self) -> bool:	1✔
75	"""Return ``True`` when the GUI has asked to cancel processing."""
76
77	if self._stop_callback is None:	1✔
78	return False	×
79	return bool(self._stop_callback())	1✔
80
81	def taskbar_hwnd(self) -> Optional[int]:	1✔
82	"""Return the HWND associated with the GUI window, if available."""
83
NEW 84	return self._taskbar_hwnd	×
85
86
87	__all__ = ["_GuiProgressHandle", "_TkProgressReporter"]	1✔

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