""" Streamlit progress panel — pure renderer of a ProgressSnapshot dict. The panel reads exactly one object (the snapshot) from session state and renders: 1. Overall status line (from snapshot["phase"]) 2. Current item (from snapshot["current_item"] — e.g. batch transcript) 3. Current module (from snapshot["current_module"]) 4. Progress bar (from snapshot["pct"] + counts) 5. Latest event line (from snapshot["latest_event"]) 6. Last 5-10 log lines (tail of snapshot["recent_logs"]) No state is inferred from logs. No state is stored inside this class. StreamlitProgressCallback bridges the ProgressCallback protocol and on_event to the snapshot stored in st.session_state. When given an ``st.empty()`` render_slot, it re-paints the panel on each event so the bar and module count stay live during a blocking run. """ from __future__ import annotations import datetime from typing import Any, Dict, MutableMapping, Optional import streamlit as st from transcriptx.app.progress import ( ProgressEvent, ProgressSnapshot, update_snapshot_from_event, ) # Session-state key under which the analysis running snapshot is stored SNAPSHOT_KEY = "run_progress_snapshot" # Session-state key for the audio preprocessing snapshot PREPROCESS_SNAPSHOT_KEY = "audio_prep_snapshot" # Session-state key for the audio merge snapshot MERGE_SNAPSHOT_KEY = "audio_merge_snapshot" # Session-state key for integrated transcription TRANSCRIPTION_SNAPSHOT_KEY = "transcription_snapshot" # Number of log lines to render in the panel PANEL_LOG_LINES = 8 # --------------------------------------------------------------------------- # Pure rendering function # --------------------------------------------------------------------------- def render_progress_panel( snapshot: ProgressSnapshot, *, unit_label: str = "modules", current_label: str = "Current module", item_label: str = "Current transcript", ) -> None: """ Render the compact progress panel from a snapshot dict. This function is stateless and has no side effects beyond Streamlit widgets. """ phase: str = snapshot.get("phase", "running") # type: ignore[assignment] status: str = snapshot.get("status", "running") # type: ignore[assignment] current_module: str = snapshot.get("current_module", "") # type: ignore[assignment] current_item: str = snapshot.get("current_item", "") # type: ignore[assignment] completed: int = snapshot.get("completed", 0) # type: ignore[assignment] skipped: int = snapshot.get("skipped", 0) # type: ignore[assignment] failed: int = snapshot.get("failed", 0) # type: ignore[assignment] total: int = snapshot.get("total", 0) # type: ignore[assignment] pct: float = snapshot.get("pct", 0.0) # type: ignore[assignment] latest_event: str = snapshot.get("latest_event", "") # type: ignore[assignment] recent_logs: list = snapshot.get("recent_logs", []) # type: ignore[assignment] error: Optional[str] = snapshot.get("error") # type: ignore[assignment] error_code: Optional[str] = snapshot.get("error_code") # type: ignore[assignment] done = completed + skipped + failed # 1. Overall status line # Prefer "Checking inputs…" over "Validating…" — the latter reads like a # post-run QA step, especially when the panel is frozen for a long spinner. phase_labels: Dict[str, str] = { "validating": "Checking inputs…", "running_pipeline": "Running pipeline…", "finalizing": "Finalizing…", "completed": "Completed", "failed": "Failed", "cancelled": "Cancelled", } phase_label = phase_labels.get(phase, phase.replace("_", " ").title()) if status == "completed": st.success(f"**{phase_label}**") elif status == "failed": st.error(f"**{phase_label}**") if error: st.error(error) elif error_code: st.error(f"[{error_code}]") elif status == "cancelled": st.warning(f"**{phase_label}**") else: st.info(f"**{phase_label}**") # 2. Current item (batch transcript / subject) — shown above module so # nested module events never hide which file is being processed. if current_item: item_prefix = ( f"Last {item_label.lower().replace('current ', '')}:" if status in ("completed", "failed", "cancelled") else f"{item_label}:" ) st.markdown(f"{item_prefix} `{current_item}`") # 3. Current module / file if current_module: prefix = ( f"Last {current_label.lower().replace('current ', '')}:" if status in ("completed", "failed", "cancelled") else f"{current_label}:" ) st.markdown(f"{prefix} `{current_module}`") # 4. Progress bar with x / y units if total > 0: bar_label = f"{done} / {total} {unit_label}" if skipped: bar_label += f" · {skipped} skipped" if failed: bar_label += f" · {failed} failed" st.progress(min(pct / 100.0, 1.0), text=bar_label) else: st.progress(0.0) # 5. Latest event line if latest_event: st.caption(latest_event) # 6. Last N log lines if recent_logs: tail = recent_logs[-PANEL_LOG_LINES:] with st.expander("Recent logs", expanded=False): st.text("\n".join(tail)) # --------------------------------------------------------------------------- # StreamlitProgressCallback — bridges ProgressCallback + on_event to snapshot # --------------------------------------------------------------------------- class StreamlitProgressCallback: """ Implements ProgressCallback protocol and bridges pipeline events into the shared ProgressSnapshot stored in st.session_state[snapshot_key]. The snapshot must be initialised in session state before the run starts. This class only mutates it; it never creates it. Pass a custom snapshot_key to isolate concurrent panels (e.g. audio prep uses PREPROCESS_SNAPSHOT_KEY while analysis uses SNAPSHOT_KEY). When ``render_slot`` is an ``st.empty()`` (or compatible) placeholder, the panel is re-painted after snapshot mutations so the bar and module counts update during a blocking run instead of staying frozen under a spinner. """ def __init__( self, snapshot_key: str = SNAPSHOT_KEY, *, render_slot: Any | None = None, snapshot: Optional[MutableMapping[str, Any]] = None, unit_label: str = "modules", current_label: str = "Current module", item_label: str = "Current transcript", ) -> None: self._snapshot_key = snapshot_key self._render_slot = render_slot self._snapshot_obj = snapshot self._unit_label = unit_label self._current_label = current_label self._item_label = item_label def _snap(self) -> Optional[MutableMapping[str, Any]]: if self._snapshot_obj is not None: return self._snapshot_obj return st.session_state.get(self._snapshot_key) def refresh_panel(self) -> None: """Re-render the progress panel into ``render_slot`` when one is bound.""" if self._render_slot is None: return snap = self._snap() if snap is None: return with self._render_slot.container(): render_progress_panel( snap, # type: ignore[arg-type] unit_label=self._unit_label, current_label=self._current_label, item_label=self._item_label, ) # ------------------------------------------------------------------ # ProgressCallback protocol # ------------------------------------------------------------------ def on_stage_start(self, stage_name: str) -> None: snap = self._snap() if snap is not None: # Nested batch transcripts can finish with status=completed; a new # stage must clear that so the panel does not stay on a success banner. snap["status"] = "running" snap["phase"] = stage_name # Prefer a pre-run phrasing; "Validating…" reads like post-analysis QA. stage_labels = { "validating": "Checking inputs…", } snap["latest_event"] = stage_labels.get( stage_name, stage_name.replace("_", " ").title() + "…" ) self.refresh_panel() def on_stage_progress( self, message: str, pct: Optional[float] = None, *, current_item: Optional[str] = None, ) -> None: snap = self._snap() if snap is not None: snap["latest_event"] = message if current_item is not None: snap["current_item"] = current_item if pct is not None: snap["pct"] = min(100.0, float(pct)) self.refresh_panel() def on_stage_complete(self, stage_name: str) -> None: pass # run-level completion is handled via on_event def on_log(self, message: str, level: str = "info") -> None: """Append a timestamped log line to recent_logs. Never infer state from it.""" snap = self._snap() if snap is None: return ts = datetime.datetime.now().strftime("%H:%M:%S") logs: list = snap.get("recent_logs", []) logs.append(f"[{ts}] {message}") if len(logs) > 100: logs = logs[-100:] snap["recent_logs"] = logs # Logs alone should not thrash the panel; stage/event paths refresh. def on_event(self, event: ProgressEvent) -> None: """Update the snapshot from a structured pipeline event.""" snap = self._snap() if snap is not None: update_snapshot_from_event(snap, event) # type: ignore[arg-type] self.refresh_panel() def get_log_text(self) -> str: """Return all recent logs as a single string (for legacy callers).""" snap = self._snap() if snap is None: return "" return "\n".join(snap.get("recent_logs", []))