"""SummaryPanel: post-session export UI. Sits in a ViewStack page next to the live panels and is bound to a session directory after disconnect / readonly load. Lets the user: 1. Set participant count + run whisperx diarization (heavy, threaded). 2. Rename SPEAKER_xx to real names. 3. Curate which captured frames go into the enhanced transcript. 4. Export `_enhanced.txt` (cheap; re-runnable on rename / reselection). """ import logging from pathlib import Path from threading import Thread import gi gi.require_version("Gtk", "4.0") gi.require_version("Adw", "1") gi.require_version("GdkPixbuf", "2.0") from gi.repository import Gtk, Gdk, GLib, Gio, Adw, Pango, GdkPixbuf, GObject from cht.config import DEFAULT_PARTICIPANTS from cht.session import load_frame_index from cht.summary import merger as summary_merger, pipeline as summary_pipeline log = logging.getLogger(__name__) class SummaryPanel(Gtk.Box): """Post-session export controls.""" __gsignals__ = { "status-changed": (GObject.SignalFlags.RUN_FIRST, None, (str,)), } def __init__(self, **kwargs): super().__init__(orientation=Gtk.Orientation.VERTICAL, spacing=8, **kwargs) self.set_margin_top(8) self.set_margin_bottom(8) self.set_margin_start(8) self.set_margin_end(8) self._session_dir: Path | None = None self._diarized: dict | None = None self._speaker_entries: dict[str, Gtk.Entry] = {} self._frame_checks: dict[str, Gtk.CheckButton] = {} self._busy = False self._build_ui() self._set_enabled(False) # -- UI construction -- def _build_ui(self): # Header title = Gtk.Label(label="Session export") title.add_css_class("title-3") title.set_halign(Gtk.Align.START) self.append(title) self._session_label = Gtk.Label(label="No session bound") self._session_label.add_css_class("dim-label") self._session_label.set_halign(Gtk.Align.START) self.append(self._session_label) # --- Diarization box --- diar_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=4) diar_box.add_css_class("card") diar_box.set_margin_top(4) diar_header = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) diar_header.set_margin_top(8) diar_header.set_margin_bottom(4) diar_header.set_margin_start(8) diar_header.set_margin_end(8) diar_header.append(Gtk.Label(label="Participants:")) self._participants_spin = Gtk.SpinButton.new_with_range(1, 10, 1) self._participants_spin.set_value(DEFAULT_PARTICIPANTS) diar_header.append(self._participants_spin) self._run_btn = Gtk.Button(label="Run diarization") self._run_btn.add_css_class("suggested-action") self._run_btn.connect("clicked", self._on_run_diarize) diar_header.append(self._run_btn) diar_box.append(diar_header) self._progress = Gtk.ProgressBar() self._progress.set_show_text(True) self._progress.set_text("") self._progress.set_margin_start(8) self._progress.set_margin_end(8) diar_box.append(self._progress) self._status_label = Gtk.Label(label="") self._status_label.set_halign(Gtk.Align.START) self._status_label.set_wrap(True) self._status_label.set_ellipsize(Pango.EllipsizeMode.END) self._status_label.add_css_class("caption") self._status_label.set_margin_start(8) self._status_label.set_margin_end(8) self._status_label.set_margin_bottom(8) diar_box.append(self._status_label) self.append(diar_box) # --- Speakers + frames in a paned area for vertical scrolling --- self._speakers_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=4) self._speakers_box.add_css_class("card") self._speakers_box.set_margin_top(4) speakers_header = Gtk.Label(label="Speaker names") speakers_header.add_css_class("heading") speakers_header.set_halign(Gtk.Align.START) speakers_header.set_margin_top(8) speakers_header.set_margin_start(8) self._speakers_box.append(speakers_header) self._speakers_list = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=4) self._speakers_list.set_margin_start(8) self._speakers_list.set_margin_end(8) self._speakers_list.set_margin_bottom(8) self._speakers_box.append(self._speakers_list) self.append(self._speakers_box) self._speakers_box.set_visible(False) # --- Frame picker --- frames_card = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=4) frames_card.add_css_class("card") frames_card.set_margin_top(4) frames_card.set_vexpand(True) frames_header = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) frames_header.set_margin_top(8) frames_header.set_margin_start(8) frames_header.set_margin_end(8) frames_title = Gtk.Label(label="Frames") frames_title.add_css_class("heading") frames_header.append(frames_title) self._frames_summary = Gtk.Label(label="") self._frames_summary.add_css_class("dim-label") frames_header.append(self._frames_summary) spacer = Gtk.Box() spacer.set_hexpand(True) frames_header.append(spacer) select_all_btn = Gtk.Button(label="Select all") select_all_btn.add_css_class("flat") select_all_btn.connect("clicked", lambda b: self._toggle_all_frames(True)) frames_header.append(select_all_btn) deselect_all_btn = Gtk.Button(label="Deselect all") deselect_all_btn.add_css_class("flat") deselect_all_btn.connect("clicked", lambda b: self._toggle_all_frames(False)) frames_header.append(deselect_all_btn) frames_card.append(frames_header) scroll = Gtk.ScrolledWindow() scroll.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC) scroll.set_vexpand(True) scroll.set_margin_start(8) scroll.set_margin_end(8) scroll.set_margin_bottom(8) self._frames_flow = Gtk.FlowBox() self._frames_flow.set_selection_mode(Gtk.SelectionMode.NONE) self._frames_flow.set_max_children_per_line(8) self._frames_flow.set_min_children_per_line(2) self._frames_flow.set_homogeneous(True) self._frames_flow.set_row_spacing(4) self._frames_flow.set_column_spacing(4) scroll.set_child(self._frames_flow) frames_card.append(scroll) self.append(frames_card) # --- Export controls --- export_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) export_box.set_margin_top(4) self._export_btn = Gtk.Button(label="Export enhanced transcript") self._export_btn.add_css_class("suggested-action") self._export_btn.connect("clicked", self._on_export) export_box.append(self._export_btn) self._open_btn = Gtk.Button(label="Open output") self._open_btn.connect("clicked", self._on_open_output) self._open_btn.set_sensitive(False) export_box.append(self._open_btn) self._export_status = Gtk.Label(label="") self._export_status.set_halign(Gtk.Align.START) self._export_status.add_css_class("caption") self._export_status.set_ellipsize(Pango.EllipsizeMode.END) self._export_status.set_hexpand(True) export_box.append(self._export_status) self.append(export_box) # -- Public API -- def bind_session(self, session_dir: Path | None): """Attach the panel to a session directory (or None to clear).""" self._session_dir = session_dir self._diarized = None self._speaker_entries.clear() self._frame_checks.clear() self._clear_widget(self._speakers_list) self._clear_widget(self._frames_flow) self._speakers_box.set_visible(False) self._progress.set_fraction(0.0) self._progress.set_text("") self._status_label.set_text("") self._export_status.set_text("") self._open_btn.set_sensitive(False) self._last_output: Path | None = None if session_dir is None: self._session_label.set_text("No session bound") self._frames_summary.set_text("") self._set_enabled(False) return self._session_label.set_text(f"Session: {session_dir.name}") self._set_enabled(True) self._load_frames() # Reuse cached diarization if present. if summary_pipeline.has_diarization(session_dir): try: self._diarized = summary_pipeline.load_diarization(session_dir) self._populate_speakers() self._status_label.set_text("Loaded cached diarization.") self._progress.set_fraction(1.0) self._progress.set_text("cached") except Exception as e: log.warning("Failed to load cached diarization: %s", e) def set_streaming(self, streaming: bool): """Disable the panel while a live session is running.""" self._set_enabled(not streaming and self._session_dir is not None) # -- Diarization -- def _on_run_diarize(self, _btn): if self._busy or not self._session_dir: return num_speakers = int(self._participants_spin.get_value()) self._busy = True self._run_btn.set_sensitive(False) self._export_btn.set_sensitive(False) self._progress.set_fraction(0.05) self._progress.set_text("starting…") self._status_label.set_text("") session_dir = self._session_dir def _worker(): try: def on_progress(line, frac): GLib.idle_add(self._update_progress, line, frac) diarized = summary_pipeline.run_diarization( session_dir, num_speakers=num_speakers, on_progress=on_progress, ) GLib.idle_add(self._on_diarize_done, diarized, None) except Exception as e: log.exception("Diarization failed") GLib.idle_add(self._on_diarize_done, None, str(e)) Thread(target=_worker, daemon=True, name="diarize").start() def _update_progress(self, line: str | None, frac: float | None): if frac is not None: self._progress.set_fraction(min(1.0, max(0.0, frac))) else: # Pulse mode if no fraction hint. self._progress.pulse() if line: self._progress.set_text(_short(line, 40)) self._status_label.set_text(line) return False def _on_diarize_done(self, diarized: dict | None, err: str | None): self._busy = False self._run_btn.set_sensitive(True) self._export_btn.set_sensitive(True) if err: self._progress.set_fraction(0.0) self._progress.set_text("failed") self._status_label.set_text(f"Error: {err}") return False self._progress.set_fraction(1.0) self._progress.set_text("done") self._diarized = diarized speakers = summary_merger.collect_speakers(diarized) if diarized else [] self._status_label.set_text( f"Diarization complete. Detected speakers: {', '.join(speakers) or '(none)'}" ) self._populate_speakers() return False def _populate_speakers(self): self._clear_widget(self._speakers_list) self._speaker_entries.clear() if not self._diarized: self._speakers_box.set_visible(False) return speakers = summary_merger.collect_speakers(self._diarized) if not speakers: self._speakers_box.set_visible(False) return for sp in speakers: row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) label = Gtk.Label(label=sp) label.set_width_chars(14) label.set_xalign(0) row.append(label) entry = Gtk.Entry() entry.set_placeholder_text(sp) entry.set_hexpand(True) row.append(entry) self._speakers_list.append(row) self._speaker_entries[sp] = entry self._speakers_box.set_visible(True) # -- Frame picker -- def _load_frames(self): self._clear_widget(self._frames_flow) self._frame_checks.clear() if not self._session_dir: return frames = load_frame_index(self._session_dir / "frames") for f in frames: self._add_frame_thumb(f["id"], f["path"], f["timestamp"]) self._update_frames_summary() def _add_frame_thumb(self, frame_id: str, path: Path, timestamp: float): box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2) try: pixbuf = GdkPixbuf.Pixbuf.new_from_file_at_scale( str(path), 192, 108, True, ) texture = Gdk.Texture.new_for_pixbuf(pixbuf) pic = Gtk.Picture.new_for_paintable(texture) pic.set_content_fit(Gtk.ContentFit.CONTAIN) pic.set_size_request(192, 108) box.append(pic) except Exception as e: log.debug("thumbnail load failed for %s: %s", path, e) placeholder = Gtk.Label(label=frame_id) placeholder.set_size_request(192, 108) box.append(placeholder) m, s = divmod(int(timestamp), 60) check = Gtk.CheckButton(label=f"{frame_id} [{m:02d}:{s:02d}]") check.set_active(True) check.connect("toggled", lambda _b: self._update_frames_summary()) box.append(check) self._frames_flow.append(box) self._frame_checks[frame_id] = check def _toggle_all_frames(self, value: bool): for check in self._frame_checks.values(): check.set_active(value) def _update_frames_summary(self): total = len(self._frame_checks) selected = sum(1 for c in self._frame_checks.values() if c.get_active()) self._frames_summary.set_text(f"{selected}/{total} selected") # -- Export -- def _on_export(self, _btn): if not self._session_dir: return if not summary_pipeline.has_diarization(self._session_dir): self._export_status.set_text("Run diarization first.") return selected_ids = {fid for fid, c in self._frame_checks.items() if c.get_active()} name_map = { sp: entry.get_text().strip() for sp, entry in self._speaker_entries.items() if entry.get_text().strip() } try: out = summary_pipeline.export( self._session_dir, selected_frame_ids=selected_ids, name_map=name_map, ) except Exception as e: log.exception("Export failed") self._export_status.set_text(f"Error: {e}") return self._last_output = out self._open_btn.set_sensitive(True) self._export_status.set_text(f"Wrote {out}") def _on_open_output(self, _btn): if not getattr(self, "_last_output", None): return uri = Gio.File.new_for_path(str(self._last_output)).get_uri() Gio.AppInfo.launch_default_for_uri(uri, None) # -- helpers -- def _set_enabled(self, enabled: bool): for w in (self._participants_spin, self._run_btn, self._export_btn): w.set_sensitive(enabled and not self._busy) def _clear_widget(self, container: Gtk.Box | Gtk.FlowBox): while child := container.get_first_child(): container.remove(child) def _short(s: str, n: int) -> str: s = s.strip() return s if len(s) <= n else s[: n - 1] + "…"