Skip to content

prediction_screen

anomaly_match_ui.screens.prediction_screen

Prediction screen — live gallery with polling, histogram, and profiler stats.

PredictionScreen(app)

Bases: GalleryScreenBase

Prediction screen with live gallery, histogram, and profiler stats.

Polls the AnomalyScoreDB while the subprocess runs, updating a paginated thumbnail gallery, score histogram, and statistics panel in real time.

Parameters:

Name Type Description Default
app AnomalyMatchApp

The parent application instance used for navigation.

required
Source code in anomaly_match_ui/screens/prediction_screen.py
def __init__(self, app: AnomalyMatchApp) -> None:
    super().__init__(app)
    self._running = False
    self._stop_requested = False
    self._prediction_thread: threading.Thread | None = None
    self._current_chunk: int = 0
    self._total_chunks: int = 0
    self._log_sink = ScreenLogSink("PredictionScreen")

build()

Build and return the prediction screen layout.

Returns:

Type Description
Widget

The root VBox widget for the prediction screen.

Source code in anomaly_match_ui/screens/prediction_screen.py
def build(self) -> widgets.Widget:
    """Build and return the prediction screen layout.

    Returns:
        The root VBox widget for the prediction screen.
    """
    # Terminal output (styled like training screen)
    self.out = widgets.Output(
        layout=widgets.Layout(
            border="1px solid white",
            height="200px",
            background_color=BG_COLOR,
            overflow="auto",
        ),
        style={"color": "white"},
    )

    self._header = HeaderWidget(
        self.app, self.out, show_back_button=True, back_to="prediction_setup"
    )

    # ── Info/control row ─────────────────────────────────────────
    self._config_html = widgets.HTML(
        value="",
        layout=widgets.Layout(flex="1 1 auto", padding="0 8px"),
    )
    self._loading_html = widgets.HTML(
        value="",
        layout=widgets.Layout(width="20px"),
    )
    self._countdown_html = widgets.HTML(
        value="",
        layout=widgets.Layout(width="auto", padding="0 4px"),
    )
    self._interval_dropdown = widgets.Dropdown(
        options=_INTERVAL_OPTIONS,
        value=_DEFAULT_POLL_INTERVAL,
        layout=widgets.Layout(width="60px", height="24px"),
        style={"description_width": "0px"},
    )
    info_row = HBox(
        [self._config_html, self._loading_html, self._countdown_html, self._interval_dropdown],
        layout=widgets.Layout(
            align_items="center",
            padding="2px 12px",
            background_color=ESA_BLUE_DEEP,
            gap="4px",
        ),
    )

    # ── Progress row (status + bar + buttons) ────────────────────
    self.status_html = widgets.HTML(
        value=self._status_text("Ready"),
        layout=widgets.Layout(
            width="auto",
            min_width="140px",
            flex="0 0 auto",
            padding="0 8px 0 0",
        ),
    )

    self.progress = ProgressPanel()
    self.progress.progress_bar.layout.flex = "1 1 auto"
    self.progress.progress_bar.layout.height = "28px"

    self.start_btn = widgets.Button(
        description="Start Prediction",
        button_style="primary",
        layout=widgets.Layout(width="150px", height="28px"),
    )
    self.start_btn.on_click(lambda _: self._on_start())

    self.stop_btn = widgets.Button(
        description="Stop",
        button_style="danger",
        disabled=True,
        layout=widgets.Layout(width="70px", height="28px"),
        style={"button_color": ESA_RED},
    )
    self.stop_btn.on_click(lambda _: self._on_stop())

    progress_row = HBox(
        [
            self.status_html,
            self.progress.spinner,
            self.progress.progress_bar,
            self.progress.status_label,
            self.start_btn,
            self.stop_btn,
        ],
        layout=widgets.Layout(
            gap="8px",
            align_items="center",
            padding="4px 12px",
        ),
    )

    # ── Gallery (left panel) ─────────────────────────────────────
    self.gallery = GalleryWidget(
        on_magnify=self._on_magnify,
        on_star=self._on_star,
    )
    self.gallery.set_on_sort_change(self._on_sort_change)
    self.gallery.set_on_page_change(self._on_page_change)
    self.gallery.set_on_scale_change(self._on_scale_change)

    left_panel = VBox(
        [self.gallery.widget],
        layout=widgets.Layout(
            flex="1 1 auto",
            min_width="0",
            background_color=BG_COLOR,
        ),
    )

    # ── Right panel — stats + histogram ──────────────────────────
    self.stats_html = widgets.HTML(
        value=self._render_stats(0, 0.0, 0.0),
        layout=widgets.Layout(padding="4px 8px"),
    )

    self._histogram_image = widgets.Image(
        format="png",
        layout=widgets.Layout(
            width="300px",
            height="200px",
            object_fit="contain",
            background_color=BG_COLOR,
            display="none",
        ),
    )

    self.profiler_html = widgets.HTML(
        value="",
        layout=widgets.Layout(padding="4px 8px"),
    )

    right_panel = VBox(
        [
            widgets.HTML(
                value=(
                    f'<div style="color:{ESA_BLUE_BRIGHT}; font-size:14px;'
                    f' font-weight:bold; padding:4px 8px;">Statistics</div>'
                ),
            ),
            self.stats_html,
            self._histogram_image,
            self.profiler_html,
        ],
        layout=widgets.Layout(
            width="320px",
            flex="0 0 auto",
            background_color=BG_COLOR,
            border_left=f"1px solid {ESA_BLUE_DEEP}",
            padding="0 0 0 8px",
        ),
    )

    main_content = HBox(
        [left_panel, right_panel],
        layout=widgets.Layout(
            background_color=BG_COLOR,
            padding="8px 0",
            gap="0px",
        ),
    )

    # Collect action buttons for busy-state toggling
    self._action_buttons = [self.start_btn]

    # Assemble layout
    root = VBox(
        [
            self._header.widget,
            info_row,
            progress_row,
            main_content,
            self.out,
        ],
        layout=widgets.Layout(background_color=BG_COLOR),
    )
    return root

on_enter()

Set up terminal output and refresh config summary.

Source code in anomaly_match_ui/screens/prediction_screen.py
def on_enter(self) -> None:
    """Set up terminal output and refresh config summary."""
    BackendInterface.set_terminal_output(self.out)
    self._log_sink.attach(self.out)
    self._update_config_summary()

    # Reopen DB monitor if returning from detail screen or on resume
    # TODO(#324): add public accessor
    if self._db_path and BackendInterface._score_db is None:
        search_dir = self.cfg.prediction_search_dir or ""
        BackendInterface.open_prediction_monitor(self._db_path, search_dir)

    # Auto-detect DB on first entry (resume scenario)
    if not self._db_path and self.cfg.output_dir:
        candidate = BackendInterface.get_prediction_db_path()
        if os.path.exists(candidate):
            self._db_path = candidate
            search_dir = self.cfg.prediction_search_dir or ""
            BackendInterface.open_prediction_monitor(self._db_path, search_dir)

            # Load total source count for progress display
            if self._total_source_count == 0:
                try:
                    self._file_type, self._total_source_count = (
                        BackendInterface.detect_and_count_sources(
                            self.cfg.prediction_search_dir
                        )
                    )
                except Exception as exc:
                    logger.debug("Failed to detect total sources on resume: %s", exc)

    # Restart polling if prediction is still running
    if self._running and not self._polling:
        self._start_polling()

    # Refresh gallery, stats, histogram, and profiler when we have results
    if self._db_path:
        count = BackendInterface.get_prediction_count()
        if count > 0:
            self._last_db_count = count
            self._refresh_gallery()

            elapsed = time.time() - self._start_time if self._start_time > 0 else 0
            speed = count / elapsed if elapsed > 0 else 0
            self._update_stats_panel(count, speed, elapsed)
            if count >= 10:
                self._update_histogram()

        # Reload profiler stats if prediction finished
        if not self._running:
            self._load_profiler_stats()

on_leave()

Stop polling when navigating away. Keep DB open for detail screen.

Source code in anomaly_match_ui/screens/prediction_screen.py
def on_leave(self) -> None:
    """Stop polling when navigating away. Keep DB open for detail screen."""
    self._stop_polling()
    self._log_sink.detach()