Skip to content

progress_tap

anomaly_match_ui.utils.progress_tap

Loguru-sink helper for turning log stream into structured UI progress events.

The training and prediction subprocesses already emit per-batch progress via tqdm; :class:Session._launch_prediction_subprocess relays every subprocess stdout line through logger.info("[subprocess] {}", line) so the lines flow into the main kernel's loguru stream (and into the screen's Output widget).

This module lets a UI screen tap that stream read-only — without touching the background thread that produced the line — to extract 71/93 style progress counts and any cutana module messages that are useful as spinner captions. The tap is a loguru sink, so installing and removing it has no effect on the background work; it simply forks a copy of the log messages to the supplied callbacks.

LogProgressTap(*, on_tqdm=None, on_cutana_hint=None)

A loguru sink that forks log messages to progress-aware callbacks.

Install once when a long-running operation starts, uninstall on completion. The sink runs in whichever thread loguru dispatches from (usually the thread that produced the message), so callbacks must be cheap and widget updates must be safe to perform from a non-UI thread — which is the case for ipywidgets attribute writes.

Parameters:

Name Type Description Default
on_tqdm Callable[[dict], None] | None

Optional callback invoked with the parsed dict from :func:parse_tqdm_line on every matching line.

 None
on_cutana_hint Callable[[str], None] | None

Optional callback invoked with the raw message text when the line matches :func:looks_like_cutana_status.

 None
Source code in anomaly_match_ui/utils/progress_tap.py 
156
157
158
159
160
161
162
163
164
def __init__(
    self,
    *,
    on_tqdm: Callable[[dict], None] | None = None,
    on_cutana_hint: Callable[[str], None] | None = None,
) -> None:
    self._on_tqdm = on_tqdm
    self._on_cutana_hint = on_cutana_hint
    self._sink_id: int | None = None

install()

Attach the sink to the shared loguru logger.

Calling this twice without an intervening :meth:uninstall is a no-op so callers don't need to track state.

Source code in anomaly_match_ui/utils/progress_tap.py 
166
167
168
169
170
171
172
173
174
def install(self) -> None:
    """Attach the sink to the shared loguru logger.

    Calling this twice without an intervening :meth:`uninstall` is a
    no-op so callers don't need to track state.
    """
    if self._sink_id is not None:
        return
    self._sink_id = logger.add(self._sink, level="INFO", format="{message}")

uninstall()

Detach the sink. Safe to call multiple times.

Source code in anomaly_match_ui/utils/progress_tap.py 
176
177
178
179
180
181
182
183
184
185
def uninstall(self) -> None:
    """Detach the sink.  Safe to call multiple times."""
    if self._sink_id is None:
        return
    try:
        logger.remove(self._sink_id)
    except ValueError:
        # Sink already removed (e.g. via logger.remove() elsewhere).
        pass
    self._sink_id = None

clean_hint_text(text)

Return text with any relayed subprocess loguru header removed.

Parameters:

Name Type Description Default
text str

A log line matched by :func:looks_like_cutana_status, which for subprocess-relayed lines still carries the child's time | LEVEL | location - prefix.

 required

Returns:

Type Description
str

Just the human-readable message, suitable for a status label.
Source code in anomaly_match_ui/utils/progress_tap.py 
 99
100
101
102
103
104
105
106
107
108
109
110
def clean_hint_text(text: str) -> str:
    """Return *text* with any relayed subprocess loguru header removed.

    Args:
        text: A log line matched by :func:`looks_like_cutana_status`, which
            for subprocess-relayed lines still carries the child's
            ``time | LEVEL | location - `` prefix.

    Returns:
        Just the human-readable message, suitable for a status label.
    """
    return _LOG_HEADER_RE.sub("", text).strip()

parse_tqdm_line(text)

Extract a tqdm progress event from a single log line.

Parameters:

Name Type Description Default
text str

Raw log message text (without loguru formatting).

 required

Returns:

Type Description
dict | None

Dict with desc, percent, current, total, timing
dict | None

keys when text looks like a tqdm status line; None otherwise.
Source code in anomaly_match_ui/utils/progress_tap.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def parse_tqdm_line(text: str) -> dict | None:
    """Extract a tqdm progress event from a single log line.

    Args:
        text: Raw log message text (without loguru formatting).

    Returns:
        Dict with ``desc``, ``percent``, ``current``, ``total``, ``timing``
        keys when *text* looks like a tqdm status line; ``None`` otherwise.
    """
    m = _TQDM_RE.search(text)
    if m is None:
        return None
    return {
        "desc": m.group("desc").strip(),
        "percent": int(m.group("pct")),
        "current": int(m.group("current")),
        "total": int(m.group("total")),
        "timing": m.group("timing").strip(),
    }

looks_like_cutana_status(text)

Return True when text matches one of the known Cutana status hints.

Source code in anomaly_match_ui/utils/progress_tap.py 
135
136
137
def looks_like_cutana_status(text: str) -> bool:
    """Return True when *text* matches one of the known Cutana status hints."""
    return _CUTANA_HINT_RE.search(text) is not None