molop.utils.progressbar¶

进度条与并行执行辅助函数。native 重建边界和 loky 执行器生命周期由独立执行层管理；本模块保留旧的 guard 导入路径以兼容已有调用方。

所有捕获方式的可直接运行演示位于 scripts/progress_capture_demo.py：

Bash

uv run --frozen python scripts/progress_capture_demo.py

MolOP 使用 tqdm 渲染进度条，但每个进度条同时还会广播机器可读的 ProgressEvent 事件。第三方工具无需解析终端输出即可捕获 MolOP 的处理进度：

进程内 — 通过 register_progress_listener 注册监听器（或使用 progress_listener 上下文管理器），同步接收每一个事件。
进程内、实时状态 — 使用线程安全的 ProgressRecorder 维护所有运行中进度条的实时快照，任何线程、任何时刻（包括进度条尚未跑完时）都可以查询。
跨进程 — 使用 progress_jsonl_sink 将 JSON Lines 追加写入文件，其他工具可以 tail 或读取该文件。

MolOP 只会关闭自己创建且已经空闲的 loky 执行器。宿主程序已有的空闲执行器不会被关闭；如果宿主执行器仍有待处理任务，native 重建会 fail-closed 并要求先消费或关闭该结果流。

即使终端进度条被关闭（molopconfig.show_progress_bar = False），事件仍然会发出，因为事件发送与渲染相互独立。每个进度条发出 start → (update / description)* → close 事件，携带稳定的 task_id、描述、done/total 计数和时间戳。

Python

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

# 跨进程捕获：其他工具执行 tail -f progress.jsonl
with pb.progress_jsonl_sink("progress.jsonl"):
    list(pb.AdaptiveProgress(range(5), desc="parse", total=5, disable=True))

回调输出示例

Text Only

parse: 0/5
parse: 1/5
parse: 2/5
parse: 3/5
parse: 4/5
parse: 5/5

实时、线程安全的快照（`ProgressRecorder`）¶

每个事件都是快照：它携带该进度条完整的 done / total / description 状态，而不是增量。因此消费者只需要用事件覆盖自己的记录即可——无需累加、不会漂移，即使漏掉某个事件，下一个事件到来时也会自动重新对齐。

ProgressRecorder 是该模式的官方实现。它本身就是一个监听器，可以直接用作上下文管理器；任何其他线程都可以通过 snapshot() 查询——例如 GUI 轮询或仪表盘——即使在进度条尚未跑完时，也能拿到与条完全一致的实时状态。

Python

from molop.utils.progressbar import ProgressRecorder, parallel_map

recorder = ProgressRecorder()

with recorder:                      # 自动注册 + 注销监听器
    results = parallel_map(
        lambda value: value * 2,
        range(5),
        n_jobs=1,
        total=5,
        disable=True,
        return_results=True,
    )

# 运行结束后：完整事件时间线
print(recorder.history())

# 运行中，从其他线程（例如 GUI 更新循环）：
#   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；未知时为 None

快照输出示例

Text Only

(ProgressEvent(task_id=1, event='start', done=0, total=5), ...)

同一个 recorder 也可以不用上下文管理器，作为普通监听器使用：

Python

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

第三方消费者需要注意的要点：

按 task_id 区分记录——并发的多个进度条不会互相覆盖。
close 并不代表 100%：被中断的运行会以 done < total 关闭，请始终使用事件自带的 done / total 数值。
监听器在驱动进度条的线程里同步调用。串行进度条使用调用方线程；parallel_map 可能在 Joblib 的完成回调线程中发出事件。如果消费者需要更新 UI，请把更新 marshal 回 UI/事件循环线程，或者直接轮询 snapshot()。
事件频率可能高于终端重绘频率（tqdm 只节流渲染、不节流事件）。计算百分比请用 done / total，不要数事件条数；必要时自己节流渲染。

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_listener context manager) and receive every event synchronously, or use the thread-safe :class:ProgressRecorder to keep a live snapshot of every running bar.
Cross-process — use :func:progress_jsonl_sink to 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.

Source code in src/molop/utils/progressbar.py

Python

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.

Attributes:

Name	Type	Description
`task_id`	`int`	Stable identifier of the bar instance that emitted the event.
`event`	`Literal['start', 'update', 'description', 'close']`	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`	`str`	Current bar description (the `desc` argument), possibly empty.
`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.

Source code in src/molop/utils/progressbar.py

Python

@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

Text Only

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)

Source code in src/molop/utils/progressbar.py

Python

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

`active_tasks()` ¶

Task ids of the currently live progress bars.

Source code in src/molop/utils/progressbar.py

Python

def active_tasks(self) -> tuple[int, ...]:
    """Task ids of the currently live progress bars."""

    with self._lock:
        return tuple(self._active)

`clear()` ¶

Drop all recorded state (active bars and history).

Source code in src/molop/utils/progressbar.py

Python

def clear(self) -> None:
    """Drop all recorded state (active bars and history)."""

    with self._lock:
        self._active.clear()
        self._history.clear()

`history()` ¶

Full recorded event timeline (empty when keep_history=False).

Source code in src/molop/utils/progressbar.py

Python

def history(self) -> tuple[ProgressEvent, ...]:
    """Full recorded event timeline (empty when ``keep_history=False``)."""

    with self._lock:
        return tuple(self._history)

`latest(task_id)` ¶

Latest event of task_id, or None if that bar is not live.

Source code in src/molop/utils/progressbar.py

Python

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)

`percent(task_id)` ¶

Completion percentage (0-100) of task_id, or None if unknown.

Source code in src/molop/utils/progressbar.py

Python

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

`snapshot()` ¶

Latest state of every live progress bar, keyed by task_id.

Source code in src/molop/utils/progressbar.py

Python

def snapshot(self) -> dict[int, ProgressEvent]:
    """Latest state of every live progress bar, keyed by ``task_id``."""

    with self._lock:
        return dict(self._active)

`is_loky_worker()` ¶

Compatibility wrapper for :func:molop.utils.execution.is_loky_worker.

Source code in src/molop/utils/progressbar.py

Python

def is_loky_worker() -> bool:
    """Compatibility wrapper for :func:`molop.utils.execution.is_loky_worker`."""

    return _execution.is_loky_worker()

`loky_parallel_guard()` ¶

Compatibility wrapper for the execution-layer parallel guard.

Source code in src/molop/utils/progressbar.py

Python

@contextmanager
def loky_parallel_guard() -> Generator[None, None, None]:
    """Compatibility wrapper for the execution-layer parallel guard."""

    with _execution.loky_parallel_guard():
        yield

`native_reconstruction_guard()` ¶

Compatibility wrapper for the execution-layer native guard.

Source code in src/molop/utils/progressbar.py

Python

@contextmanager
def native_reconstruction_guard() -> Generator[None, None, None]:
    """Compatibility wrapper for the execution-layer native guard."""

    with _execution.native_reconstruction_guard(process_validator=ensure_current_process):
        yield

`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)` ¶

Python

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

Python

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]

Python

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.

Source code in src/molop/utils/progressbar.py

Python

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.::

Text Only

{
    "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.

Source code in src/molop/utils/progressbar.py

Python

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

Source code in src/molop/utils/progressbar.py

Python

@contextmanager
def progress_listener(listener: ProgressListener) -> Generator[ProgressListener, None, None]:
    """Register ``listener`` for the duration of the ``with`` block."""

    unregister = register_progress_listener(listener)
    try:
        yield listener
    finally:
        unregister()

`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.

Source code in src/molop/utils/progressbar.py

Python

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

Source code in src/molop/utils/progressbar.py

Python

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

molop.utils.progressbar¶

实时、线程安全的快照（ProgressRecorder）¶

JsonLinesProgressListener ¶

ProgressEvent dataclass ¶

ProgressRecorder ¶

Example¶

active_tasks() ¶

clear() ¶

history() ¶

latest(task_id) ¶

percent(task_id) ¶

snapshot() ¶

is_loky_worker() ¶

loky_parallel_guard() ¶

native_reconstruction_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) ¶

Parameters¶

Returns¶

progress_jsonl_sink(path) ¶

progress_listener(listener) ¶

register_progress_listener(listener) ¶

unregister_progress_listener(listener) ¶

实时、线程安全的快照（`ProgressRecorder`）¶

`JsonLinesProgressListener` ¶

`ProgressEvent` `dataclass` ¶

`ProgressRecorder` ¶

`active_tasks()` ¶

`clear()` ¶

`history()` ¶

`latest(task_id)` ¶

`percent(task_id)` ¶

`snapshot()` ¶

`is_loky_worker()` ¶

`loky_parallel_guard()` ¶

`native_reconstruction_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)` ¶

`progress_jsonl_sink(path)` ¶

`progress_listener(listener)` ¶

`register_progress_listener(listener)` ¶

`unregister_progress_listener(listener)` ¶