""" Model runner panel. Lets users: - Load a local .py model or register a remote API endpoint - Run scoring models against the worklist or a subset of sequences - Run generative models to create new sequences (added to worklist) - View results in a sortable table """ from __future__ import annotations from typing import TYPE_CHECKING import panel as pn import param if TYPE_CHECKING: from ui.state import AppState class ModelRunnerPanel(param.Parameterized): """Model loading and execution panel.""" def __init__(self, state: "AppState", **params: object) -> None: super().__init__(**params) self._state = state self._status_pane = pn.pane.HTML("") # ── Load section ────────────────────────────────────────────────────────── def _build_load_section(self) -> pn.Column: # Local file self._local_path = pn.widgets.TextInput( name="Local model path (.py)", placeholder="/path/to/my_model.py", width=380, ) load_local_btn = pn.widgets.Button(name="Load", button_type="primary", margin=(8, 4)) load_local_btn.on_click(self._on_load_local) # API endpoint self._api_endpoint = pn.widgets.TextInput( name="API Endpoint URL", placeholder="https://model.example.com/api/v1", width=300, ) self._api_name = pn.widgets.TextInput(name="Model Name", placeholder="my_scorer", width=180) self._api_key_input = pn.widgets.PasswordInput(name="API Key (optional)", width=220) self._api_type = pn.widgets.Select( name="Type", options=["Scoring", "Generative"], value="Scoring", width=130 ) load_api_btn = pn.widgets.Button(name="Register", button_type="primary", margin=(8, 4)) load_api_btn.on_click(self._on_register_api) return pn.Column( pn.pane.HTML( '
' 'Load Model
' ), pn.Tabs( ("Local File", pn.Column(pn.Row(self._local_path, load_local_btn))), ("API Endpoint", pn.Column( pn.Row(self._api_endpoint, self._api_name), pn.Row(self._api_key_input, self._api_type, load_api_btn), )), ), ) # ── Loaded models list ───────────────────────────────────────────────────── @param.depends("_state.model_registry") def _loaded_models_view(self) -> pn.Column: registry = self._state.model_registry all_models = registry.all_models if not all_models: return pn.pane.HTML( '
' 'No models loaded yet.
' ) rows = [] for reg in all_models: type_badge_color = "#0284C7" if reg.model_type == "scoring" else "#7C3AED" source_badge = ( 'LOCAL' if reg.source == "local" else 'API' ) type_badge = ( f'' f'{reg.model_type.upper()}' ) remove_btn = pn.widgets.Button( name="✕", button_type="danger", width=30, margin=(1, 2), ) model_name_captured = reg.model.name def _remove(event: object, mn: str = model_name_captured) -> None: self._state.model_registry.unregister(mn) self._state.param.trigger("model_registry") remove_btn.on_click(_remove) rows.append(pn.Row( pn.pane.HTML( f'
' f'{reg.model.name}
' ), pn.pane.HTML(f'
{type_badge} {source_badge}
'), pn.pane.HTML( f'
' f'{reg.model.description}
' ), remove_btn, )) return pn.Column(*rows) # ── Run section ──────────────────────────────────────────────────────────── def _build_run_section(self) -> pn.Column: registry = self._state.model_registry scoring_names = [r.model.name for r in registry.scoring_models] gen_names = [r.model.name for r in registry.generative_models] self._score_model_select = pn.widgets.Select( name="Scoring Model", options=scoring_names or ["(none loaded)"], width=240, ) run_score_btn = pn.widgets.Button( name="Score Worklist", button_type="success", margin=(8, 4) ) run_score_btn.on_click(self._on_run_scoring) self._gen_model_select = pn.widgets.Select( name="Generative Model", options=gen_names or ["(none loaded)"], width=240, ) self._gen_n = pn.widgets.IntInput(name="# sequences", value=10, width=100) run_gen_btn = pn.widgets.Button( name="Generate Sequences", button_type="primary", margin=(8, 4) ) run_gen_btn.on_click(self._on_run_generation) return pn.Column( pn.pane.HTML( '
' 'Run Models
' ), pn.Tabs( ("Score Sequences", pn.Column( pn.Row(self._score_model_select, run_score_btn), pn.pane.HTML( '
' 'Scores all items currently in the worklist.
' ), )), ("Generate Sequences", pn.Column( pn.Row(self._gen_model_select, self._gen_n, run_gen_btn), pn.pane.HTML( '
' 'New sequences are added to the worklist.
' ), )), ), ) # ── Full panel ───────────────────────────────────────────────────────────── def panel(self) -> pn.Column: return pn.Column( pn.pane.HTML( '
' 'Models
' ), self._build_load_section(), pn.layout.Divider(), pn.pane.HTML( '
' 'Loaded Models
' ), pn.panel(self._loaded_models_view), pn.layout.Divider(), self._build_run_section(), self._status_pane, sizing_mode="stretch_width", styles={"padding": "8px 16px"}, ) # ── Event handlers ───────────────────────────────────────────────────────── def _on_load_local(self, event: object) -> None: path = self._local_path.value.strip() if not path: return try: loaded = self._state.model_registry.load_local(path) self._state.param.trigger("model_registry") names = ", ".join(m.name for m in loaded) self._status_pane.object = ( f'
Loaded: {names}
' ) except Exception as e: self._status_pane.object = ( f'
Error: {e}
' ) def _on_register_api(self, event: object) -> None: endpoint = self._api_endpoint.value.strip() name = self._api_name.value.strip() or "api_model" key = self._api_key_input.value or None model_type = self._api_type.value if not endpoint: return try: if model_type == "Scoring": self._state.model_registry.register_api_scorer(endpoint, name, key) else: self._state.model_registry.register_api_generator(endpoint, name, key) self._state.param.trigger("model_registry") self._status_pane.object = ( f'
Registered API: {name}
' ) except Exception as e: self._status_pane.object = ( f'
Error: {e}
' ) def _on_run_scoring(self, event: object) -> None: model_name = self._score_model_select.value if not model_name or model_name == "(none loaded)": return sequences = self._state.worklist.sequences if not sequences: self._status_pane.object = ( '
Worklist is empty.
' ) return try: df = self._state.model_registry.run_scoring(model_name, sequences) score_map = dict(zip(df["id"], df["score"])) for item in self._state.worklist.items: if item.sequence.id in score_map: item.scores[model_name] = score_map[item.sequence.id] self._state.param.trigger("worklist") self._status_pane.object = ( f'
' f'Scored {len(sequences)} sequences with {model_name}
' ) except Exception as e: self._status_pane.object = ( f'
Scoring failed: {e}
' ) def _on_run_generation(self, event: object) -> None: model_name = self._gen_model_select.value if not model_name or model_name == "(none loaded)": return n = self._gen_n.value or 10 try: new_seqs = self._state.model_registry.run_generation(model_name, constraints={}, n=n) self._state.worklist.add_many(new_seqs, origin="generated") self._state.param.trigger("worklist") self._status_pane.object = ( f'
' f'Generated {len(new_seqs)} sequences — added to worklist
' ) except Exception as e: self._status_pane.object = ( f'
Generation failed: {e}
' )