diff --git a/CHANGELOG.md b/CHANGELOG.md index 7d9ca5c..34ec736 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -64,3 +64,6 @@ - added an executable output-definition layer so a grounded concept can deterministically project into one or more CDM rows, with richer profile shapes to describe them - added deterministic handling for row values that depend on a separate source field, or that should suppress the row entirely, always recorded explicitly rather than silently dropped - fully additive; existing grounding and profile behavior is unchanged + +# 0.4.0 +- added Mermaid-based visualization for output-definition catalogues and projected bundles, and an optional interactive terminal explorer built on them \ No newline at end of file diff --git a/docs/usage/tui.md b/docs/usage/tui.md new file mode 100644 index 0000000..915fd6e --- /dev/null +++ b/docs/usage/tui.md @@ -0,0 +1,58 @@ +# TUI Explorer + +Use the Textual-based TUI when you want to browse a compiled output-definition +catalogue, edit JSON input, run a projection, and inspect the result in one +terminal session. + +## Install + +The TUI dependencies are optional: + +```bash +uv sync --extra tui +``` + +For contributor workflows, `uv sync --extra dev` also includes the TUI +dependencies alongside the test and lint tooling. + +## Launch + +```bash +omop-semantics tui +``` + +This launches the explorer against an empty runtime. The reusable app lives in +`omop_semantics.runtime.tui` and is mainly intended to be embedded by a caller +that already has concrete `OutputDefinition` instances to browse. + +Programmatic use looks like: + +```python +from omop_semantics.runtime import OmopSemanticEngine +from omop_semantics.runtime.tui import run_tui + +engine = OmopSemanticEngine.from_yaml_paths(registry_paths=[], profile_paths=[]) +runtime = engine.build_output_definition_runtime(my_definitions) +run_tui(runtime) +``` + +## What it shows + +- A catalogue tree built from `describe_definition()` +- An editable JSON context pane +- A result tree using the same three-state language as the visualization HTML: + green for projected rows, red for suppressed rows, amber for unresolved rows + or links + +For the meaning of those states, see [Visualization](../visualization.md). + +## Controls + +- `Ctrl+R` runs the selected definition against the JSON payload in the context pane +- `Q` quits the app + +Selecting a definition in the tree loads a starter JSON payload into the +context pane. Child rows and annotations are browsable without changing the +current context. When the app is embedded by an external caller, that payload +can use the flat `SemanticProjectionRequest` shape so it can be pasted +directly into a worker-backed flow. diff --git a/docs/visualization.md b/docs/visualization.md index cf56bc8..6e736d0 100644 --- a/docs/visualization.md +++ b/docs/visualization.md @@ -144,3 +144,8 @@ inputs, and suppression annotations without requiring a concrete input context. Inline notebook display via `IPython.display.HTML(html.raw)` is convenient, but it is best-effort only. Browser viewing of the saved `.html` file is the guaranteed path across notebook front ends and local environments. + +## Interactive exploration + +For a terminal-based browse/edit/run loop, use the Textual explorer documented +in [TUI Explorer](usage/tui.md). diff --git a/mkdocs.yaml b/mkdocs.yaml index b310baa..7a158b5 100644 --- a/mkdocs.yaml +++ b/mkdocs.yaml @@ -50,6 +50,7 @@ plugins: nav: - Home: index.md - Usage: usage.md + - TUI Explorer: usage/tui.md - Visualization: visualization.md - Data Model: data-model.md - Schema & Instances: schema-and-instances.md diff --git a/pyproject.toml b/pyproject.toml index bc5185b..b9be215 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "omop-semantics" -version = "0.3.0" +version = "0.4.0" description = "Define, validate, and use schema-backed semantic conventions for OMOP CDM" readme = "README.md" authors = [ @@ -28,9 +28,13 @@ dev = [ "mypy>=1.8", "ruff>=0.4", "rich>=13.0", + "textual>=0.60", "tornado>=6.5.7", "pygments>=2.20.0" ] +tui = [ + "textual>=0.60", +] docs = [ "mkdocs-material>=9.7.6", "mkdocstrings-python>=2.0.1", diff --git a/src/omop_semantics/__init__.py b/src/omop_semantics/__init__.py index cdb35cc..d318c36 100644 --- a/src/omop_semantics/__init__.py +++ b/src/omop_semantics/__init__.py @@ -33,6 +33,7 @@ def main(argv: "list[str] | None" = None) -> int: Subcommands: gen-models (re)generate the committed pydantic models from the LinkML schema. + tui launch the interactive output-definition explorer. """ import sys @@ -41,6 +42,10 @@ def main(argv: "list[str] | None" = None) -> int: from omop_semantics.schema.codegen import cli as gen_models_cli return gen_models_cli(argv[1:]) + if argv and argv[0] == "tui": + from omop_semantics.runtime.tui.app import cli as tui_cli - print("usage: omop-semantics gen-models [--check] [--out DIR]", file=sys.stderr) + return tui_cli(argv[1:]) + + print("usage: omop-semantics {gen-models|tui} ...", file=sys.stderr) return 2 diff --git a/src/omop_semantics/runtime/tui/__init__.py b/src/omop_semantics/runtime/tui/__init__.py new file mode 100644 index 0000000..4400ceb --- /dev/null +++ b/src/omop_semantics/runtime/tui/__init__.py @@ -0,0 +1,5 @@ +from __future__ import annotations + +from omop_semantics.runtime.tui.app import OutputDefinitionExplorer, run_tui + +__all__ = ["OutputDefinitionExplorer", "run_tui"] diff --git a/src/omop_semantics/runtime/tui/app.py b/src/omop_semantics/runtime/tui/app.py new file mode 100644 index 0000000..80457bb --- /dev/null +++ b/src/omop_semantics/runtime/tui/app.py @@ -0,0 +1,318 @@ +from __future__ import annotations + +import json +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Callable, Mapping + +from rich.text import Text +from textual.app import App, ComposeResult +from textual.binding import Binding +from textual.containers import Horizontal, Vertical +from textual.widgets import Button, Footer, Header, Static, TextArea, Tree + +from omop_semantics.runtime.output_definitions import OutputDefinitionRuntime +from omop_semantics.runtime.viz import DefinitionOutline, derive_status, describe_definition + +from .widgets import ProjectionViewData, populate_result_tree, render_summary + +ProjectCallable = Callable[[str, Mapping[str, Any]], Any] + +_DOMAIN_BY_TABLE = { + "condition_occurrence": "Condition", + "observation": "Observation", + "procedure_occurrence": "Procedure", + "measurement": "Measurement", + "drug_exposure": "Drug", + "device_exposure": "Device", + "death": "Death", + "specimen": "Specimen", + "episode": "Episode", + "episode_event": "Episode", + "visit_occurrence": "Visit", +} + + +@dataclass(frozen=True) +class CatalogueNodeData: + definition_name: str + kind: str + + +class OutputDefinitionExplorer(App[None]): + CSS = """ + Screen { + layout: vertical; + } + + #body { + layout: vertical; + height: 1fr; + } + + #top { + height: 2fr; + } + + #catalogue-panel, #context-panel, #result-panel { + border: round $accent; + padding: 0 1; + } + + #catalogue-panel, #context-panel { + width: 1fr; + } + + #result-panel { + height: 1fr; + } + + .panel-title { + margin-bottom: 1; + text-style: bold; + } + + Tree, TextArea { + height: 1fr; + } + + #run { + margin-top: 1; + width: 100%; + } + + #result-summary { + height: auto; + margin-bottom: 1; + } + """ + + BINDINGS = [ + Binding("ctrl+r", "run_projection", "Run", priority=True), + Binding("q", "quit", "Quit"), + ] + + def __init__( + self, + runtime: OutputDefinitionRuntime, + project_fn: ProjectCallable | None = None, + ) -> None: + super().__init__() + self._runtime = runtime + self._service_mode = project_fn is not None + self._project_fn: ProjectCallable = project_fn or runtime.project + self._selected_definition: str | None = None + + def compose(self) -> ComposeResult: + yield Header() + with Vertical(id="body"): + with Horizontal(id="top"): + with Vertical(id="catalogue-panel"): + yield Static("Catalogue", classes="panel-title") + yield Tree("Output definitions", id="catalogue") + with Vertical(id="context-panel"): + yield Static("Context (editable JSON)", classes="panel-title") + yield TextArea(id="context") + yield Button("Run (ctrl+r)", id="run", variant="primary") + with Vertical(id="result-panel"): + yield Static("Result", classes="panel-title") + yield Static("Select a definition to begin.", id="result-summary") + yield Tree("Projection result", id="result-tree") + yield Footer() + + def on_mount(self) -> None: + catalogue = self.query_one("#catalogue", Tree) + catalogue.show_root = False + catalogue.root.expand() + self._populate_catalogue(catalogue) + + result_tree = self.query_one("#result-tree", Tree) + result_tree.show_root = False + result_tree.root.expand() + + context = self.query_one("#context", TextArea) + if not self._runtime.names(): + context.load_text("{}") + self.query_one("#result-summary", Static).update("No definitions loaded.") + return + + first = catalogue.root.children[0] + catalogue.select_node(first) + self._select_definition(first.data.definition_name if isinstance(first.data, CatalogueNodeData) else None) + + def on_tree_node_selected(self, event: Tree.NodeSelected[CatalogueNodeData]) -> None: + if event.control.id != "catalogue": + return + data = event.node.data + if not isinstance(data, CatalogueNodeData) or data.kind != "definition": + return + if data.definition_name == self._selected_definition: + return + self._select_definition(data.definition_name) + + def on_button_pressed(self, event: Button.Pressed) -> None: + if event.button.id == "run": + self.action_run_projection() + + def action_run_projection(self) -> None: + if self._selected_definition is None: + self._show_result( + ProjectionViewData( + status="error", + message="Select a definition before running a projection.", + ) + ) + return + + context_area = self.query_one("#context", TextArea) + try: + payload = json.loads(context_area.text) + except json.JSONDecodeError as exc: + self._show_result( + ProjectionViewData( + status="error", + message=f"Invalid JSON: {exc.msg} at line {exc.lineno}, column {exc.colno}", + ) + ) + return + + try: + raw_result = self._project_fn(self._selected_definition, payload) + except Exception as exc: + self._show_result( + ProjectionViewData( + status="error", + definition_name=self._selected_definition, + message=str(exc), + ) + ) + return + + if self._service_mode: + view = _extract_service_view(raw_result) + else: + view = _extract_runtime_view(raw_result) + self._show_result(view) + + def _populate_catalogue(self, tree: Tree[CatalogueNodeData]) -> None: + tree.clear() + tree.root.expand() + for definition_name in self._runtime.names(): + outline = describe_definition(self._runtime.get(definition_name)) + definition_node = tree.root.add( + Text(outline.name), + data=CatalogueNodeData(definition_name=outline.name, kind="definition"), + expand=True, + ) + self._add_outline_rows(definition_node, outline) + + def _add_outline_rows(self, definition_node: Any, outline: DefinitionOutline) -> None: + for row in outline.rows: + row_node = definition_node.add( + Text(f"{row.row_id} ({row.cdm_table} / {row.profile_name})"), + data=CatalogueNodeData(definition_name=outline.name, kind="row"), + expand=True, + ) + for target_slot, source_path in row.derived_slots: + row_node.add_leaf( + Text(f"=> {target_slot} from {source_path}", style="yellow3"), + data=CatalogueNodeData(definition_name=outline.name, kind="derived"), + ) + if row.suppressible: + row_node.add_leaf( + Text("! suppresses on match", style="red"), + data=CatalogueNodeData(definition_name=outline.name, kind="suppression"), + ) + + def _select_definition(self, definition_name: str | None) -> None: + self._selected_definition = definition_name + if definition_name is None: + return + context = self.query_one("#context", TextArea) + context.load_text(json.dumps(self._context_template(definition_name), indent=2)) + + def _context_template(self, definition_name: str) -> dict[str, Any]: + compiled = self._runtime.get(definition_name) + table = compiled.row_projections[0].profile.cdm_table if compiled.row_projections else "" + domain = _DOMAIN_BY_TABLE.get(table, "") + if self._service_mode: + return { + "grounded_concept_id": 0, + "grounded_domain": domain, + "definition_hint": definition_name, + "context": {}, + } + grounded: dict[str, Any] = {"concept_id": 0} + if domain: + grounded["domain"] = domain + return {"grounded": grounded, "source": {}} + + def _show_result(self, view: ProjectionViewData) -> None: + self.query_one("#result-summary", Static).update(render_summary(view)) + populate_result_tree(self.query_one("#result-tree", Tree), view) + + +def run_tui( + runtime: OutputDefinitionRuntime, + project_fn: ProjectCallable | None = None, +) -> None: + OutputDefinitionExplorer(runtime, project_fn=project_fn).run() + + +def cli(argv: list[str] | None = None) -> int: + import argparse + + from omop_semantics.runtime import OmopSemanticEngine + + parser = argparse.ArgumentParser(prog="omop-semantics tui") + parser.add_argument("--registry", action="append", default=[], type=Path) + parser.add_argument("--profiles", action="append", default=[], type=Path) + args = parser.parse_args(argv) + + engine = OmopSemanticEngine.from_yaml_paths( + registry_paths=args.registry, + profile_paths=args.profiles, + ) + runtime = engine.build_output_definition_runtime([]) + run_tui(runtime) + return 0 + + +def _extract_runtime_view(bundle: Any) -> ProjectionViewData: + return ProjectionViewData( + status=derive_status( + has_rows=bool(bundle.rows), + has_unresolved=bool(bundle.unresolved_fields), + has_suppressed=bool(bundle.suppressed_rows), + ), + rows=[ + (row.row_id, row.profile.cdm_table, dict(row.fields)) + for row in bundle.rows + ], + suppressed=[ + (row.row_id, row.reason, row.source_field, row.source_code) + for row in bundle.suppressed_rows + ], + unresolved=list(bundle.unresolved_fields), + audit_notes=list(bundle.audit_notes), + definition_name=bundle.definition_name, + role=bundle.role, + ) + + +def _extract_service_view(result: Any) -> ProjectionViewData: + return ProjectionViewData( + status=result.status, + rows=[ + (row.row_id, row.table, dict(row.fields)) + for row in result.rows + ], + suppressed=[ + (row.row_id, row.reason, row.source_field, row.source_code) + for row in result.suppressed_rows + ], + unresolved=list(result.unresolved_fields), + audit_notes=list(result.audit_notes), + definition_name=result.definition_name, + role=result.role, + ) diff --git a/src/omop_semantics/runtime/tui/widgets.py b/src/omop_semantics/runtime/tui/widgets.py new file mode 100644 index 0000000..17d71f4 --- /dev/null +++ b/src/omop_semantics/runtime/tui/widgets.py @@ -0,0 +1,109 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Sequence + +from rich.text import Text +from textual.widgets import Tree + +_STATUS_STYLES = { + "ok": "green", + "partial": "yellow3", + "suppressed": "red", + "no_match": "grey70", + "error": "red", +} + + +@dataclass(frozen=True) +class ProjectionViewData: + status: str + rows: Sequence[tuple[str, str, dict[str, Any]]] = () + suppressed: Sequence[tuple[str, str, str, str]] = () + unresolved: Sequence[dict[str, Any]] = () + audit_notes: Sequence[str] = () + definition_name: str | None = None + role: str | None = None + message: str | None = None + + +def render_summary(data: ProjectionViewData) -> Text: + status_style = _STATUS_STYLES.get(data.status, "white") + summary = Text() + summary.append("Status: ", style="bold") + summary.append(data.status, style=f"bold {status_style}") + if data.definition_name: + summary.append(" | Definition: ", style="bold") + summary.append(data.definition_name) + if data.role: + summary.append(" | Role: ", style="bold") + summary.append(data.role) + if data.message: + summary.append(" | ") + summary.append(data.message, style=status_style) + return summary + + +def populate_result_tree(tree: Tree[object], data: ProjectionViewData) -> None: + tree.clear() + tree.root.expand() + + if data.message and not data.rows and not data.suppressed and not data.unresolved: + tree.root.add_leaf(_label(data.message, data.status)) + return + + if data.rows: + rows_node = tree.root.add(Text("Projected rows", style="bold"), expand=True) + for row_id, table, fields in data.rows: + row_node = rows_node.add(_label(f"[ok] {row_id} ({table})", "ok"), expand=True) + for field, value in sorted(fields.items()): + row_node.add_leaf(Text(f"{field} = {value}")) + + if data.suppressed: + suppressed_node = tree.root.add(Text("Suppressed rows", style="bold"), expand=True) + for row_id, reason, source_field, source_code in data.suppressed: + row_node = suppressed_node.add( + _label(f"[suppressed] {row_id}", "suppressed"), + expand=True, + ) + row_node.add_leaf(Text(f"reason = {reason}", style="red")) + row_node.add_leaf(Text(f"source_field = {source_field}")) + row_node.add_leaf(Text(f"source_code = {source_code}")) + + if data.unresolved: + unresolved_node = tree.root.add(Text("Unresolved", style="bold"), expand=True) + for entry in data.unresolved: + row_id = entry.get("row_id") + if row_id is not None: + node = unresolved_node.add( + _label(f"[unresolved] {row_id}", "partial"), + expand=True, + ) + for field in entry.get("missing_fields", ()): + node.add_leaf(Text(f"missing_field = {field}", style="yellow3")) + continue + + link = entry.get("link") + if link is not None: + label = ( + f"[unresolved] link {link['source_row']} -> {link['target_row']} " + f"({link['relationship_type']})" + ) + node = unresolved_node.add(_label(label, "partial"), expand=True) + for row_name in entry.get("missing_rows", ()): + node.add_leaf(Text(f"missing_row = {row_name}", style="yellow3")) + continue + + unresolved_node.add_leaf(_label(str(entry), "partial")) + + if data.audit_notes: + notes_node = tree.root.add(Text("Audit notes", style="bold"), expand=True) + for note in data.audit_notes: + notes_node.add_leaf(Text(note)) + + if not tree.root.children: + tree.root.add_leaf(_label("No projected rows.", data.status)) + + +def _label(text: str, status: str) -> Text: + return Text(text, style=_STATUS_STYLES.get(status, "white")) diff --git a/tests/test_tui.py b/tests/test_tui.py new file mode 100644 index 0000000..34a5b55 --- /dev/null +++ b/tests/test_tui.py @@ -0,0 +1,378 @@ +from __future__ import annotations + +import asyncio +import json +from types import SimpleNamespace +from typing import Any + +from textual.widgets import TextArea, Tree + +from omop_semantics import INSTANCE_DIR +from omop_semantics.runtime import ( + ContextFieldRef, + DerivationRule, + OmopSemanticEngine, + OutputDefinition, + OutputRowProjection, + SpecialValuePolicy, + derive_status, +) +from omop_semantics.runtime.tui.app import OutputDefinitionExplorer + + +def _engine() -> OmopSemanticEngine: + return OmopSemanticEngine.from_yaml_paths( + registry_paths=[INSTANCE_DIR / "demographic.yaml"], + profile_paths=[], + ) + + +def _runtime(): + return _engine().build_output_definition_runtime( + [_condition_with_status_definition(), _criteria_gate_definition()] + ) + + +def _runtime_with_placeholder(): + return _engine().build_output_definition_runtime( + [_placeholder_definition(), _condition_with_status_definition()] + ) + + +def _placeholder_definition() -> OutputDefinition: + return OutputDefinition( + name="placeholder_definition", + role="condition_modifier", + ) + + +def _condition_with_status_definition() -> OutputDefinition: + return OutputDefinition( + name="condition_with_status_from_secondary_field", + role="condition_modifier", + row_projections=( + OutputRowProjection( + row_id="condition", + profile_name="condition_with_status", + field_bindings={ + "condition_concept_id": ContextFieldRef("grounded.concept_id"), + }, + ), + ), + derivation_rules=( + DerivationRule( + target_row="condition", + target_slot="condition_status_concept_id", + source_field=ContextFieldRef("source.role_field"), + code_map={"1": 32902, "2": 32908}, + suppress_codes=frozenset({"3"}), + ), + ), + ) + + +def _criteria_gate_definition() -> OutputDefinition: + return OutputDefinition( + name="criteria_gate_condition", + role="condition_modifier", + row_projections=( + OutputRowProjection( + row_id="condition", + profile_name="condition_simple", + field_bindings={ + "condition_concept_id": ContextFieldRef("grounded.concept_id"), + }, + special_value_policy=SpecialValuePolicy( + source_field=ContextFieldRef("source.raw_value"), + allowed_special_values=frozenset({"0"}), + suppression_mode="drop", + ), + ), + ), + ) + + +def _service_mode_project(runtime): + def project(definition_name: str, raw: dict[str, Any]): + grounded: dict[str, Any] = { + "concept_id": raw["grounded_concept_id"], + } + if raw.get("grounded_domain"): + grounded["domain"] = raw["grounded_domain"] + bundle = runtime.project( + definition_name, + { + "grounded": grounded, + "source": dict(raw.get("context", {})), + }, + ) + return SimpleNamespace( + status=derive_status( + has_rows=bool(bundle.rows), + has_unresolved=bool(bundle.unresolved_fields), + has_suppressed=bool(bundle.suppressed_rows), + ), + rows=[ + SimpleNamespace(row_id=row.row_id, table=row.profile.cdm_table, fields=dict(row.fields)) + for row in bundle.rows + ], + suppressed_rows=[ + SimpleNamespace( + row_id=row.row_id, + reason=row.reason, + source_field=row.source_field, + source_code=row.source_code, + ) + for row in bundle.suppressed_rows + ], + unresolved_fields=list(bundle.unresolved_fields), + audit_notes=list(bundle.audit_notes), + definition_name=bundle.definition_name, + role=bundle.role, + ) + + return project + + +def _labels(node) -> list[str]: + label = getattr(node.label, "plain", str(node.label)) + labels = [label] + for child in node.children: + labels.extend(_labels(child)) + return labels + + +def _child_with_label(node, label: str): + for child in node.children: + if child.label.plain == label: + return child + raise AssertionError(f"Could not find child node with label {label!r}") + + +def test_tui_catalogue_has_two_top_level_nodes() -> None: + async def scenario() -> None: + runtime = _runtime() + app = OutputDefinitionExplorer(runtime, project_fn=_service_mode_project(runtime)) + async with app.run_test() as pilot: + await pilot.pause() + catalogue = app.query_one("#catalogue", Tree) + labels = [node.label.plain for node in catalogue.root.children] + assert labels == [ + "condition_with_status_from_secondary_field", + "criteria_gate_condition", + ] + + asyncio.run(scenario()) + + +def test_tui_selection_populates_definition_hint_in_context() -> None: + async def scenario() -> None: + runtime = _runtime() + app = OutputDefinitionExplorer(runtime, project_fn=_service_mode_project(runtime)) + async with app.run_test() as pilot: + await pilot.pause() + catalogue = app.query_one("#catalogue", Tree) + catalogue.select_node(catalogue.root.children[1]) + await pilot.pause() + payload = json.loads(app.query_one("#context", TextArea).text) + assert payload["definition_hint"] == "criteria_gate_condition" + + asyncio.run(scenario()) + + +def test_tui_placeholder_definition_loads_without_row_projections() -> None: + async def scenario() -> None: + runtime = _runtime_with_placeholder() + app = OutputDefinitionExplorer(runtime, project_fn=_service_mode_project(runtime)) + async with app.run_test() as pilot: + await pilot.pause() + catalogue = app.query_one("#catalogue", Tree) + placeholder_node = _child_with_label(catalogue.root, "placeholder_definition") + catalogue.select_node(placeholder_node) + await pilot.pause() + payload = json.loads(app.query_one("#context", TextArea).text) + assert payload["definition_hint"] == "placeholder_definition" + assert payload["grounded_domain"] == "" + + asyncio.run(scenario()) + + +def test_tui_selecting_row_node_does_not_reset_context() -> None: + async def scenario() -> None: + runtime = _runtime() + app = OutputDefinitionExplorer(runtime, project_fn=_service_mode_project(runtime)) + async with app.run_test() as pilot: + await pilot.pause() + catalogue = app.query_one("#catalogue", Tree) + context = app.query_one("#context", TextArea) + row_node = catalogue.root.children[0].children[0] + context.load_text('{"edited": "row"}') + catalogue.select_node(row_node) + await pilot.pause() + assert context.text == '{"edited": "row"}' + + asyncio.run(scenario()) + + +def test_tui_selecting_derived_leaf_does_not_reset_context() -> None: + async def scenario() -> None: + runtime = _runtime() + app = OutputDefinitionExplorer(runtime, project_fn=_service_mode_project(runtime)) + async with app.run_test() as pilot: + await pilot.pause() + catalogue = app.query_one("#catalogue", Tree) + context = app.query_one("#context", TextArea) + row_node = catalogue.root.children[0].children[0] + derived_node = _child_with_label( + row_node, + "=> condition_status_concept_id from source.role_field", + ) + context.load_text('{"edited": "derived"}') + catalogue.select_node(derived_node) + await pilot.pause() + assert context.text == '{"edited": "derived"}' + + asyncio.run(scenario()) + + +def test_tui_selecting_suppression_leaf_does_not_reset_context() -> None: + async def scenario() -> None: + runtime = _runtime() + app = OutputDefinitionExplorer(runtime, project_fn=_service_mode_project(runtime)) + async with app.run_test() as pilot: + await pilot.pause() + catalogue = app.query_one("#catalogue", Tree) + context = app.query_one("#context", TextArea) + row_node = catalogue.root.children[1].children[0] + suppression_node = _child_with_label(row_node, "! suppresses on match") + context.load_text('{"edited": "suppression"}') + catalogue.select_node(suppression_node) + await pilot.pause() + assert context.text == '{"edited": "suppression"}' + + asyncio.run(scenario()) + + +def test_tui_selecting_different_definition_still_resets_context() -> None: + async def scenario() -> None: + runtime = _runtime() + app = OutputDefinitionExplorer(runtime, project_fn=_service_mode_project(runtime)) + async with app.run_test() as pilot: + await pilot.pause() + catalogue = app.query_one("#catalogue", Tree) + context = app.query_one("#context", TextArea) + context.load_text('{"edited": "definition"}') + catalogue.select_node(catalogue.root.children[1]) + await pilot.pause() + payload = json.loads(context.text) + assert payload["definition_hint"] == "criteria_gate_condition" + + asyncio.run(scenario()) + + +def test_tui_reselecting_same_definition_does_not_reset_context() -> None: + async def scenario() -> None: + runtime = _runtime() + app = OutputDefinitionExplorer(runtime, project_fn=_service_mode_project(runtime)) + async with app.run_test() as pilot: + await pilot.pause() + catalogue = app.query_one("#catalogue", Tree) + context = app.query_one("#context", TextArea) + definition_node = catalogue.root.children[0] + context.load_text('{"edited": "same-definition"}') + app.on_tree_node_selected(SimpleNamespace(control=catalogue, node=definition_node)) + await pilot.pause() + assert context.text == '{"edited": "same-definition"}' + + asyncio.run(scenario()) + + +def test_tui_run_kept_projection_renders_expected_fields() -> None: + async def scenario() -> None: + runtime = _runtime() + app = OutputDefinitionExplorer(runtime, project_fn=_service_mode_project(runtime)) + async with app.run_test() as pilot: + await pilot.pause() + context = app.query_one("#context", TextArea) + context.load_text( + json.dumps( + { + "grounded_concept_id": 4152280, + "grounded_domain": "Condition", + "definition_hint": "condition_with_status_from_secondary_field", + "context": {"role_field": "1"}, + }, + indent=2, + ) + ) + await pilot.press("ctrl+r") + await pilot.pause() + + labels = _labels(app.query_one("#result-tree", Tree).root) + assert "[ok] condition (condition_occurrence)" in labels + assert "condition_concept_id = 4152280" in labels + assert "condition_status_concept_id = 32902" in labels + + asyncio.run(scenario()) + + +def test_tui_run_suppressed_projection_renders_reason() -> None: + async def scenario() -> None: + runtime = _runtime() + app = OutputDefinitionExplorer(runtime, project_fn=_service_mode_project(runtime)) + async with app.run_test() as pilot: + await pilot.pause() + context = app.query_one("#context", TextArea) + context.load_text( + json.dumps( + { + "grounded_concept_id": 4152280, + "grounded_domain": "Condition", + "definition_hint": "condition_with_status_from_secondary_field", + "context": {"role_field": "3"}, + }, + indent=2, + ) + ) + await pilot.press("ctrl+r") + await pilot.pause() + + labels = _labels(app.query_one("#result-tree", Tree).root) + assert "[suppressed] condition" in labels + assert any("source_code = 3" == label for label in labels) + + asyncio.run(scenario()) + + +def test_tui_invalid_json_shows_inline_error_and_does_not_crash() -> None: + async def scenario() -> None: + runtime = _runtime() + app = OutputDefinitionExplorer(runtime, project_fn=_service_mode_project(runtime)) + async with app.run_test() as pilot: + await pilot.pause() + context = app.query_one("#context", TextArea) + context.load_text("{") + await pilot.press("ctrl+r") + await pilot.pause() + + labels = _labels(app.query_one("#result-tree", Tree).root) + assert any(label.startswith("Invalid JSON:") for label in labels) + + asyncio.run(scenario()) + + +def test_tui_project_exception_shows_inline_error_and_does_not_crash() -> None: + def raising_project(_definition_name: str, _raw: dict[str, Any]): + raise KeyError("grounded_concept_id") + + async def scenario() -> None: + app = OutputDefinitionExplorer(_runtime(), project_fn=raising_project) + async with app.run_test() as pilot: + await pilot.pause() + await pilot.press("ctrl+r") + await pilot.pause() + + labels = _labels(app.query_one("#result-tree", Tree).root) + assert any("grounded_concept_id" in label for label in labels) + + asyncio.run(scenario()) diff --git a/tests/test_viz.py b/tests/test_viz.py index 267bcc0..5855ede 100644 --- a/tests/test_viz.py +++ b/tests/test_viz.py @@ -16,6 +16,7 @@ bundle_to_mermaid, catalogue_to_html, catalogue_to_mermaid, + derive_status, describe_definition, ) @@ -271,3 +272,11 @@ def test_bundle_to_mermaid_escapes_user_supplied_labels() -> None: assert "|" in mermaid assert "