molop.utils.progressbar¶
Progress bars and parallel execution helpers. The native-reconstruction boundary and loky executor lifecycle are owned by a separate execution layer; this module keeps the legacy guard imports for existing callers.
A directly runnable demonstration of every capture mode is available as
scripts/progress_capture_demo.py:
MolOP renders progress with tqdm, but every bar
also broadcasts machine-readable
ProgressEvent notifications.
Third-party tools can capture MolOP progress without parsing terminal output:
- In-process — register a listener with
register_progress_listener(or use theprogress_listenercontext manager) and receive every event synchronously. - In-process, live state — use the thread-safe
ProgressRecorderto keep an up-to-date snapshot of every running bar, readable from any thread at any time (even while the bars are still running). - Cross-process — use
progress_jsonl_sinkto append JSON Lines to a file that other tools can tail or read.
MolOP closes only an idle loky executor that it created and owns. An idle executor already owned by a host application is left alone; pending host work makes native reconstruction fail closed until that result stream is consumed or closed.
Events are emitted even when the terminal bar is suppressed
(molopconfig.show_progress_bar = False), because event emission is
independent of rendering. A single bar emits start → (update /
description)* → close events carrying a stable task_id, the description,
done/total counters and a timestamp.
import molop.utils.progressbar as pb
def on_progress(event: pb.ProgressEvent) -> None:
print(f"{event.description}: {event.done}/{event.total}")
with pb.progress_listener(on_progress):
list(pb.AdaptiveProgress(range(5), desc="parse", total=5, disable=True))
# Cross-process capture: tail -f progress.jsonl from another tool
with pb.progress_jsonl_sink("progress.jsonl"):
list(pb.AdaptiveProgress(range(5), desc="parse", total=5, disable=True))
Keeping a live, thread-safe snapshot (ProgressRecorder)¶
Every event is a snapshot: it carries the full done / total /
description state of its bar, never a delta. A consumer therefore only needs
to overwrite its record with each event — no accumulation, no drift, and a
missed event self-heals on the next one.
ProgressRecorder is the official
implementation of that pattern. It is a listener itself, so it works as a
context manager; from any other thread you can query
snapshot() — for example
a GUI poller or a dashboard — and it reflects the bars’ current state even
while they are still running.
from molop.utils.progressbar import ProgressRecorder, parallel_map
recorder = ProgressRecorder()
with recorder: # registers + unregisters automatically
results = parallel_map(
lambda value: value * 2,
range(5),
n_jobs=1,
total=5,
disable=True,
return_results=True,
)
# After the run: full event timeline
print(recorder.history())
# Mid-run, from another thread (e.g. a GUI update loop):
# for task_id, event in recorder.snapshot().items():
# widget.set_value(event.done / event.total if event.total else 0)
# percent = recorder.percent(task_id) # 0.0-100.0, or None if unknown
The same recorder works as a plain listener without the context manager:
from molop.utils.progressbar import parallel_map, register_progress_listener
recorder = ProgressRecorder()
unregister = register_progress_listener(recorder)
try:
parallel_map(lambda value: value * 2, range(5), n_jobs=1, disable=True)
finally:
unregister()
Key points for third-party consumers:
- Key the recorded state by
task_id— concurrent bars never collide. closedoes not imply 100%: an interrupted run closes withdone < total. Always use the event’s owndone/totalvalues.- The listener is invoked synchronously from the thread driving the bar. A
serial bar runs on the caller’s thread;
parallel_mapmay emit from joblib’s completion callback thread. Marshal UI updates back to your UI/event-loop thread, or simply pollsnapshot(). - Events may arrive faster than the terminal redraws (tqdm throttles
rendering, not emission). Compute percentages from
done/total, not from event counts, and throttle your own rendering if needed.
Adaptive progress bars and parallel execution helpers.
Progress rendering is delegated to tqdm (with an auto-selected backend), but
every bar also broadcasts machine-readable :class:ProgressEvent notifications.
Third-party tools can capture MolOP progress without parsing terminal output:
- In-process — register a listener with
:func:
register_progress_listener(or the :func:progress_listenercontext manager) and receive every event synchronously, or use the thread-safe :class:ProgressRecorderto keep a live snapshot of every running bar. - Cross-process — use :func:
progress_jsonl_sinkto append JSON Lines to a file that other tools can tail or read.
JsonLinesProgressListener
¶
Progress listener that appends each event as one JSON line to a file.
Suitable for cross-process capture: an external tool can tail the file
(tail -f, a reader process, a dashboard) and parse JSON Lines
(https://jsonlines.org/). The file is flushed after every event.
源代码位于: src/molop/utils/progressbar.py
class JsonLinesProgressListener:
"""Progress listener that appends each event as one JSON line to a file.
Suitable for cross-process capture: an external tool can tail the file
(``tail -f``, a reader process, a dashboard) and parse JSON Lines
(https://jsonlines.org/). The file is flushed after every event.
"""
def __init__(self, path: os.PathLike[str] | str) -> None:
self.path = Path(path)
self.path.parent.mkdir(parents=True, exist_ok=True)
self._handle = self.path.open("a", encoding="utf-8")
self._lock = threading.Lock()
self._closed = False
def __call__(self, event: ProgressEvent) -> None:
with self._lock:
if self._closed:
return
self._handle.write(json.dumps(asdict(event)) + "\n")
self._handle.flush()
def close(self) -> None:
with self._lock:
if not self._closed:
self._handle.close()
self._closed = True
ProgressEvent
dataclass
¶
A machine-readable progress notification emitted by MolOP progress bars.
属性:
| 名称 | 类型 | 描述 |
|---|---|---|
task_id |
int
|
Stable identifier of the bar instance that emitted the event. |
event |
Literal['start', 'update', 'description', 'close']
|
Kind of event: |
description |
str
|
Current bar description (the |
done |
int | float
|
Completed units so far. |
total |
int | float | None
|
Total units, or None while unknown (e.g. streamed inputs). |
timestamp |
float
|
Wall-clock Unix timestamp of the event. |
源代码位于: src/molop/utils/progressbar.py
@dataclass(frozen=True, slots=True)
class ProgressEvent:
"""A machine-readable progress notification emitted by MolOP progress bars.
Attributes:
task_id: Stable identifier of the bar instance that emitted the event.
event: Kind of event: ``"start"``, ``"update"``, ``"description"`` or
``"close"``. Events are emitted regardless of whether the bar is
rendered (``disable=True`` only suppresses terminal output).
description: Current bar description (the ``desc`` argument), possibly empty.
done: Completed units so far.
total: Total units, or None while unknown (e.g. streamed inputs).
timestamp: Wall-clock Unix timestamp of the event.
"""
task_id: int
event: Literal["start", "update", "description", "close"]
description: str
done: int | float
total: int | float | None
timestamp: float
ProgressRecorder
¶
Thread-safe, snapshot-based tracker of every live MolOP progress bar.
The recorder is itself a progress listener. Register it with
:func:register_progress_listener or simply use it as a context manager,
and it maintains an up-to-date view of every running progress bar keyed by
task_id. Because events are state snapshots (each carries the full
done/total/description), the recorded state is always in sync
with the bars — it can be read from any thread at any time, including
while the bars are still running.
Example¶
.. code-block:: python
from molop.utils.progressbar import ProgressRecorder, parallel_map
rec = ProgressRecorder()
with rec: # registers and unregisters the listener automatically
results = parallel_map(
lambda value: value * 2,
range(3),
n_jobs=1,
total=3,
disable=True,
return_results=True,
)
print(rec.history()) # full event timeline
# Mid-run, from another thread (e.g. a GUI poller or a dashboard):
# for task_id, event in rec.snapshot().items():
# widget.set(event.done / event.total)
源代码位于: src/molop/utils/progressbar.py
class ProgressRecorder:
"""Thread-safe, snapshot-based tracker of every live MolOP progress bar.
The recorder is itself a progress listener. Register it with
:func:`register_progress_listener` or simply use it as a context manager,
and it maintains an up-to-date view of every running progress bar keyed by
``task_id``. Because events are state snapshots (each carries the full
``done``/``total``/``description``), the recorded state is always in sync
with the bars — it can be read from any thread at any time, including
while the bars are still running.
Example
-------
.. code-block:: python
from molop.utils.progressbar import ProgressRecorder, parallel_map
rec = ProgressRecorder()
with rec: # registers and unregisters the listener automatically
results = parallel_map(
lambda value: value * 2,
range(3),
n_jobs=1,
total=3,
disable=True,
return_results=True,
)
print(rec.history()) # full event timeline
# Mid-run, from another thread (e.g. a GUI poller or a dashboard):
# for task_id, event in rec.snapshot().items():
# widget.set(event.done / event.total)
"""
def __init__(self, *, keep_history: bool = True) -> None:
self._lock = threading.Lock()
self._active: dict[int, ProgressEvent] = {}
self._history: list[ProgressEvent] = []
self._keep_history = keep_history
self._unregisters: list[Callable[[], None]] = []
def __enter__(self) -> "ProgressRecorder":
self._unregisters.append(register_progress_listener(self))
return self
def __exit__(
self,
exc_type: type[BaseException] | None,
exc_value: BaseException | None,
traceback: TracebackType | None,
) -> None:
if self._unregisters:
self._unregisters.pop()()
def __call__(self, event: ProgressEvent) -> None:
with self._lock:
if event.event == "close":
self._active.pop(event.task_id, None)
else:
self._active[event.task_id] = event
if self._keep_history:
self._history.append(event)
def snapshot(self) -> dict[int, ProgressEvent]:
"""Latest state of every live progress bar, keyed by ``task_id``."""
with self._lock:
return dict(self._active)
def active_tasks(self) -> tuple[int, ...]:
"""Task ids of the currently live progress bars."""
with self._lock:
return tuple(self._active)
def latest(self, task_id: int) -> ProgressEvent | None:
"""Latest event of ``task_id``, or None if that bar is not live."""
with self._lock:
return self._active.get(task_id)
def percent(self, task_id: int) -> float | None:
"""Completion percentage (0-100) of ``task_id``, or None if unknown."""
with self._lock:
event = self._active.get(task_id)
if event is None or not event.total:
return None
return event.done / event.total * 100.0
def history(self) -> tuple[ProgressEvent, ...]:
"""Full recorded event timeline (empty when ``keep_history=False``)."""
with self._lock:
return tuple(self._history)
def clear(self) -> None:
"""Drop all recorded state (active bars and history)."""
with self._lock:
self._active.clear()
self._history.clear()
is_loky_worker()
¶
loky_parallel_guard()
¶
native_reconstruction_guard()
¶
Compatibility wrapper for the execution-layer native guard.
parallel_map(func, iterable, n_jobs=-1, total=None, desc='Processing', disable=False, return_as=None, tqdm_kwargs=None, progress_callback=None, *, return_results=None, **joblib_kwargs)
¶
parallel_map(
func: Callable[[T], R],
iterable: Iterable[T],
n_jobs: int = -1,
total: int | None = None,
desc: str = "Processing",
disable: bool = False,
return_as: str | None = None,
tqdm_kwargs: dict[str, Any] | None = None,
progress_callback: ProgressListener | None = None,
*,
return_results: Literal[False],
**joblib_kwargs: Any,
) -> None
parallel_map(
func: Callable[[T], R],
iterable: Iterable[T],
n_jobs: int = -1,
total: int | None = None,
desc: str = "Processing",
disable: bool = False,
return_as: str | None = None,
tqdm_kwargs: dict[str, Any] | None = None,
progress_callback: ProgressListener | None = None,
*,
return_results: Literal[True],
**joblib_kwargs: Any,
) -> Iterable[R]
parallel_map(
func: Callable[[T], R],
iterable: Iterable[T],
n_jobs: int = -1,
total: int | None = None,
desc: str = "Processing",
disable: bool = False,
return_as: str | None = None,
tqdm_kwargs: dict[str, Any] | None = None,
progress_callback: ProgressListener | None = None,
*,
return_results: None = None,
**joblib_kwargs: Any,
) -> Iterable[R] | None
A parallel wrapper with type hints and progress bar
Parameters¶
func : Callable[[T], R]
The function to apply to each item in the iterable.
iterable : Iterable[T]
The iterable to apply the function to.
n_jobs : int, optional
The number of parallel jobs to run. Default is -1, which means using all available cores.
total : Optional[int], optional
The total number of items in the iterable. Default is None, which means inferring the total from the iterable.
desc : str, optional
The description to display in the progress bar. Default is “Processing”.
disable : bool, optional
Whether to disable the progress bar. Default is False.
tqdm_kwargs : Optional[dict[str, Any]], optional
Additional keyword arguments to pass to tqdm. Default is None.
progress_callback : Optional[Callable[[ProgressEvent], None]], optional
Optional callback invoked with a :class:ProgressEvent for every
progress update of the wrapped bar, in addition to the global listener
registry (see :func:register_progress_listener). Default is None.
return_results : Optional[bool], optional
When False, exhaust execution and return None. When None, list results
containing only implicit None values are returned as None.
**joblib_kwargs : Any
Additional keyword arguments to pass to joblib.Parallel.
Returns¶
Iterable[R] | None The iterable of results after applying the function to each item, or None for side-effect-only calls.
源代码位于: src/molop/utils/progressbar.py
def parallel_map(
func: Callable[[T], R],
iterable: Iterable[T],
n_jobs: int = -1,
total: int | None = None,
desc: str = "Processing",
disable: bool = False,
return_as: str | None = None,
tqdm_kwargs: dict[str, Any] | None = None,
progress_callback: ProgressListener | None = None,
*,
return_results: bool | None = None,
**joblib_kwargs: Any,
) -> Iterable[R] | None:
"""
A parallel wrapper with type hints and progress bar
Parameters
----------
func : Callable[[T], R]
The function to apply to each item in the iterable.
iterable : Iterable[T]
The iterable to apply the function to.
n_jobs : int, optional
The number of parallel jobs to run. Default is -1, which means using all available cores.
total : Optional[int], optional
The total number of items in the iterable. Default is None, which means inferring the total from the iterable.
desc : str, optional
The description to display in the progress bar. Default is "Processing".
disable : bool, optional
Whether to disable the progress bar. Default is False.
tqdm_kwargs : Optional[dict[str, Any]], optional
Additional keyword arguments to pass to tqdm. Default is None.
progress_callback : Optional[Callable[[ProgressEvent], None]], optional
Optional callback invoked with a :class:`ProgressEvent` for every
progress update of the wrapped bar, in addition to the global listener
registry (see :func:`register_progress_listener`). Default is None.
return_results : Optional[bool], optional
When False, exhaust execution and return None. When None, list results
containing only implicit None values are returned as None.
**joblib_kwargs : Any
Additional keyword arguments to pass to joblib.Parallel.
Returns
-------
Iterable[R] | None
The iterable of results after applying the function to each item, or None
for side-effect-only calls.
"""
original_joblib_kwargs = joblib_kwargs
joblib_kwargs = _with_default_joblib_backend(dict(joblib_kwargs))
effective_return_as = return_as
if effective_return_as is None:
# joblib's legacy multiprocessing backend cannot expose a result
# generator. Keep the streaming default for loky and other backends,
# but select a compatible eager result for this explicit backend.
effective_return_as = (
("list" if joblib_kwargs.get("backend") == "multiprocessing" else "generator")
if original_joblib_kwargs
else "list"
)
tqdm_kwargs = dict(tqdm_kwargs or {})
# Accept progress_callback either as a first-class argument or as a
# pass-through key in tqdm_kwargs without double-binding it.
tqdm_callback = tqdm_kwargs.pop("progress_callback", None)
if progress_callback is None:
progress_callback = tqdm_callback
progress_total = total
if progress_total is None:
try:
progress_total = len(iterable) # type: ignore[arg-type]
except (AttributeError, TypeError):
progress_total = None
def new_progress() -> Any:
return AdaptiveProgress(
None,
desc=desc,
total=progress_total,
disable=disable,
progress_callback=progress_callback,
**tqdm_kwargs,
)
if n_jobs == 1:
if effective_return_as in {"generator", "generator_unordered"}:
def serial_generator() -> Iterable[R]:
progress = new_progress()
try:
for item in iterable:
result = func(item)
progress.update(1)
yield result
finally:
_close_progress_bar(progress)
return _finalize_parallel_results(serial_generator(), return_results)
progress = new_progress()
try:
source = iter(iterable)
except Exception:
_close_progress_bar(progress)
raise
try:
results: list[R] = []
for item in source:
results.append(func(item))
progress.update(1)
return _finalize_parallel_results(results, return_results)
finally:
_close_progress_bar(progress)
parallel = Parallel(n_jobs=n_jobs, return_as=effective_return_as, **joblib_kwargs)
enter = getattr(parallel, "__enter__", None)
if effective_return_as in {"generator", "generator_unordered"}:
# Do not leave loky's reusable executor alive while a later native
# MolGR prewarm runs in this process. The returned generator owns the
# coordination guard until its results are fully consumed or abandoned.
def process_generator() -> Iterable[R]:
progress = new_progress()
observe_parallel, has_parallel_hook = _install_parallel_progress_hook(
parallel, progress
)
try:
with loky_parallel_guard():
if callable(enter):
with parallel as executor:
source = iter(iterable)
results = executor(delayed(func)(item) for item in source)
if results is None:
raise ValueError("The parallel map returned None.")
for result in cast(Iterable[R], results):
if not has_parallel_hook:
progress.update(1)
yield result
else:
source = iter(iterable)
results = parallel(delayed(func)(item) for item in source)
if results is None:
raise ValueError("The parallel map returned None.")
for result in cast(Iterable[R], results):
if not has_parallel_hook:
progress.update(1)
yield result
finally:
observe_parallel()
_close_progress_bar(progress)
return _finalize_parallel_results(process_generator(), return_results)
progress = new_progress()
try:
source = iter(iterable)
except Exception:
_close_progress_bar(progress)
raise
observe_parallel, has_parallel_hook = _install_parallel_progress_hook(parallel, progress)
try:
with loky_parallel_guard():
raw_results: Any
if callable(enter):
with parallel as executor:
raw_results = executor(delayed(func)(item) for item in source)
else:
raw_results = parallel(delayed(func)(item) for item in source)
if raw_results is None:
raise ValueError("The parallel map returned None.")
observe_parallel()
completed_results: list[R]
if isinstance(raw_results, list):
completed_results = raw_results
else:
completed_results = list(cast(Iterable[R], raw_results))
if not has_parallel_hook:
for _ in completed_results:
progress.update(1)
return _finalize_parallel_results(completed_results, return_results)
finally:
observe_parallel()
_close_progress_bar(progress)
progress_jsonl_sink(path)
¶
Capture MolOP progress as JSON Lines appended to path.
Every progress event produced inside the with block is appended to
path, one JSON object per line, e.g.::
{
"task_id": 1,
"event": "start",
"description": "MolOP parsing with 4 processes",
"done": 0,
"total": 12,
"timestamp": 1700000000.0,
}
The listener is unregistered and the file handle closed on block exit.
源代码位于: src/molop/utils/progressbar.py
@contextmanager
def progress_jsonl_sink(
path: os.PathLike[str] | str,
) -> Generator[JsonLinesProgressListener, None, None]:
"""Capture MolOP progress as JSON Lines appended to ``path``.
Every progress event produced inside the ``with`` block is appended to
``path``, one JSON object per line, e.g.::
{
"task_id": 1,
"event": "start",
"description": "MolOP parsing with 4 processes",
"done": 0,
"total": 12,
"timestamp": 1700000000.0,
}
The listener is unregistered and the file handle closed on block exit.
"""
listener = JsonLinesProgressListener(path)
unregister = register_progress_listener(listener)
try:
yield listener
finally:
unregister()
listener.close()
progress_listener(listener)
¶
Register listener for the duration of the with block.
源代码位于: src/molop/utils/progressbar.py
register_progress_listener(listener)
¶
Register listener to receive every MolOP progress event.
Listeners are invoked synchronously from the thread driving the progress bar, once per event, and must be cheap; exceptions raised by a listener are logged and never interrupt progress. Returns an unregister callable.
源代码位于: src/molop/utils/progressbar.py
def register_progress_listener(listener: ProgressListener) -> Callable[[], None]:
"""Register ``listener`` to receive every MolOP progress event.
Listeners are invoked synchronously from the thread driving the progress
bar, once per event, and must be cheap; exceptions raised by a listener are
logged and never interrupt progress. Returns an unregister callable.
"""
registration = _ProgressListenerRegistration(listener)
with _progress_listeners_lock:
_progress_listeners.append(registration)
def unregister() -> None:
with _progress_listeners_lock, suppress(ValueError):
_progress_listeners.remove(registration)
return unregister
unregister_progress_listener(listener)
¶
Remove a previously registered progress listener (no-op if absent).
源代码位于: src/molop/utils/progressbar.py
def unregister_progress_listener(listener: ProgressListener) -> None:
"""Remove a previously registered progress listener (no-op if absent)."""
with _progress_listeners_lock:
_progress_listeners[:] = [
registration
for registration in _progress_listeners
if not _same_listener(registration.listener, listener)
]