Upload self-contained KGVis Gradio Space

README.md

@@ -1,14 +1,32 @@
 ---
-title: Frame2KG Vis
-emoji: 🚀
-colorFrom: blue
-colorTo: red
+title: KGVis
 sdk: gradio
-sdk_version: 6.16.0
-python_version: '3.13'
 app_file: app.py
-pinned: false
-short_description: Visualisation Tool for Frame2kg Outputs
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# KGVis
+
+Standalone Hugging Face Space export for the KGVis visualizer.
+
+Push the contents of this directory to the root of a Hugging Face Space
+repository. The Space expects `app.py`, `requirements.txt`, and `kgvis/` at the
+repository root.
+
+This Space contains the minimal local `kgvis` package files needed by the
+Gradio app. The package does not need to be published to PyPI because the
+`kgvis/` directory is included directly in this repository.
+
+## Included
+
+- `app.py`: Gradio upload-only visualizer.
+- `requirements.txt`: external Python dependencies for Spaces.
+- `kgvis/plot.py`: Plotly figure construction.
+- `kgvis/schema/`: graph schema parsing.
+
+## Not Included
+
+- No Dash app files.
+- No storage helpers.
+- No saved data, uploaded images, or local test datasets.
+
+Upload one graph JSON file and, optionally, one matching image file in the UI.

app.py

@@ -0,0 +1,540 @@
+from __future__ import annotations
+
+import base64
+import json
+import mimetypes
+from pathlib import Path
+from typing import Any
+
+import gradio as gr
+from PIL import Image
+
+from kgvis.plot import build_figure, empty_figure
+from kgvis.schema import graph_from_dict, graph_to_dict, parse_graph
+
+
+DEFAULT_TOGGLES = ["nodes", "edges", "node_labels", "edge_labels"]
+
+CSS = """
+.kgvis-shell { max-width: 1600px; margin: 0 auto; }
+.kgvis-status textarea { font-family: ui-monospace, SFMono-Regular, Menlo, monospace; }
+.kgvis-plot { min-height: 640px; }
+"""
+
+
+def _path_from_upload(upload: Any) -> Path | None:
+    if upload is None:
+        return None
+    if isinstance(upload, (str, Path)):
+        return Path(upload)
+    if isinstance(upload, dict):
+        for key in ("path", "name"):
+            if upload.get(key):
+                return Path(upload[key])
+    name = getattr(upload, "name", None)
+    if name:
+        return Path(name)
+    return None
+
+
+def _upload_name(upload: Any, path: Path) -> str:
+    if isinstance(upload, dict):
+        return str(upload.get("orig_name") or upload.get("name") or path.name)
+    return str(getattr(upload, "orig_name", None) or path.name)
+
+
+def _image_state_from_upload(upload: Any) -> dict[str, Any]:
+    path = _path_from_upload(upload)
+    if path is None:
+        raise ValueError("No image file was provided.")
+    if not path.exists():
+        raise FileNotFoundError(f"Uploaded image not found: {path}")
+
+    with Image.open(path) as img:
+        width, height = img.size
+    if height == 0:
+        raise ValueError("Image height is zero.")
+
+    mime, _ = mimetypes.guess_type(path.name)
+    if not mime:
+        mime = "image/png"
+    encoded = base64.b64encode(path.read_bytes()).decode("ascii")
+    return {
+        "name": _upload_name(upload, path),
+        "data_uri": f"data:{mime};base64,{encoded}",
+        "aspect": width / height,
+    }
+
+
+def _load_graph_upload(upload: Any) -> tuple[dict[str, Any], dict[str, Any], str]:
+    path = _path_from_upload(upload)
+    if path is None:
+        raise ValueError("No JSON file was provided.")
+    if not path.exists():
+        raise FileNotFoundError(f"Uploaded JSON not found: {path}")
+
+    raw = json.loads(path.read_text(encoding="utf-8"))
+    graph = parse_graph(raw)
+    return graph_to_dict(graph), raw, _upload_name(upload, path)
+
+
+def _node_choices(graph_payload: dict[str, Any] | None) -> list[tuple[str, str]]:
+    if not graph_payload:
+        return []
+    graph = graph_from_dict(graph_payload)
+    return [
+        (f"{node.label or node.id} ({node.id})", node.id)
+        for node in graph.nodes
+        if node.id
+    ]
+
+
+def _summary(graph_payload: dict[str, Any] | None, graph_name: str | None = None) -> str:
+    if not graph_payload:
+        return "No graph loaded."
+    graph = graph_from_dict(graph_payload)
+    name = f"`{graph_name}`: " if graph_name else ""
+    schema = graph.meta.get("schema", "unknown")
+    return f"{name}{len(graph.nodes)} nodes, {len(graph.edges)} edges (schema: {schema})"
+
+
+def _dangling_markdown(graph_payload: dict[str, Any] | None) -> str:
+    if not graph_payload:
+        return "Dangling edges: 0"
+    graph = graph_from_dict(graph_payload)
+    node_ids = {node.id for node in graph.nodes}
+    dangling = [
+        edge
+        for edge in graph.edges
+        if edge.source not in node_ids or edge.target not in node_ids
+    ]
+    if not dangling:
+        return "Dangling edges: 0"
+    lines = [f"Dangling edges: {len(dangling)}"]
+    lines.extend(
+        f"- `{edge.predicate}`: `{edge.source}` -> `{edge.target}`"
+        for edge in dangling
+    )
+    return "\n".join(lines)
+
+
+def _edge_rows(graph_payload: dict[str, Any] | None) -> list[list[str]]:
+    if not graph_payload:
+        return []
+    graph = graph_from_dict(graph_payload)
+    node_ids = {node.id for node in graph.nodes}
+    return [
+        [
+            edge.source,
+            edge.predicate,
+            edge.target,
+            "yes" if edge.source not in node_ids or edge.target not in node_ids else "no",
+        ]
+        for edge in graph.edges
+    ]
+
+
+def _node_details(
+    graph_payload: dict[str, Any] | None,
+    selected_node_id: str | None,
+) -> dict[str, Any]:
+    if not graph_payload:
+        return {"message": "No graph loaded."}
+    if not selected_node_id:
+        return {"message": "Select a node to see details."}
+
+    graph = graph_from_dict(graph_payload)
+    node = next((candidate for candidate in graph.nodes if candidate.id == selected_node_id), None)
+    if node is None:
+        return {"message": "Selected node not found."}
+
+    node_ids = {candidate.id for candidate in graph.nodes}
+    outgoing = [
+        {
+            "predicate": edge.predicate,
+            "source": edge.source,
+            "target": edge.target,
+        }
+        for edge in graph.edges
+        if edge.source == selected_node_id and edge.target in node_ids
+    ]
+    incoming = [
+        {
+            "predicate": edge.predicate,
+            "source": edge.source,
+            "target": edge.target,
+        }
+        for edge in graph.edges
+        if edge.target == selected_node_id and edge.source in node_ids
+    ]
+    dangling = [
+        {
+            "predicate": edge.predicate,
+            "source": edge.source,
+            "target": edge.target,
+        }
+        for edge in graph.edges
+        if (
+            edge.source == selected_node_id
+            and edge.target not in node_ids
+            or edge.target == selected_node_id
+            and edge.source not in node_ids
+        )
+    ]
+
+    return {
+        "id": node.id,
+        "label": node.label,
+        "attributes": node.attributes,
+        "outgoing_edges": outgoing,
+        "incoming_edges": incoming,
+        "dangling_edges": dangling,
+    }
+
+
+def _render_figure(
+    graph_payload: dict[str, Any] | None,
+    image_payload: dict[str, Any] | None,
+    toggles: list[str] | None,
+    layout_mode: str | None,
+    show_image: bool,
+    background_opacity: float | None,
+    show_bboxes: bool,
+    selected_node_id: str | None,
+):
+    if not graph_payload:
+        return empty_figure("Upload a graph JSON to begin.")
+
+    graph = graph_from_dict(graph_payload)
+    active_toggles = toggles or []
+    background = None
+    background_aspect = None
+    if show_image and image_payload:
+        background = image_payload.get("data_uri")
+        background_aspect = image_payload.get("aspect")
+
+    return build_figure(
+        graph,
+        show_nodes="nodes" in active_toggles,
+        show_edges="edges" in active_toggles,
+        show_node_labels="node_labels" in active_toggles and "nodes" in active_toggles,
+        show_edge_labels="edge_labels" in active_toggles and "edges" in active_toggles,
+        selected_node_id=selected_node_id,
+        layout_mode=layout_mode or "force",
+        background_image=background,
+        background_aspect=background_aspect,
+        background_opacity=background_opacity if background_opacity is not None else 0.35,
+        show_bboxes=show_bboxes,
+        highlight_node_id=selected_node_id,
+    )
+
+
+def _render_visual_outputs(
+    graph_payload: dict[str, Any] | None,
+    image_payload: dict[str, Any] | None,
+    toggles: list[str] | None,
+    layout_mode: str | None,
+    show_image: bool,
+    background_opacity: float | None,
+    show_bboxes: bool,
+    selected_node_id: str | None,
+):
+    return (
+        _render_figure(
+            graph_payload,
+            image_payload,
+            toggles,
+            layout_mode,
+            show_image,
+            background_opacity,
+            show_bboxes,
+            selected_node_id,
+        ),
+        _node_details(graph_payload, selected_node_id),
+    )
+
+
+def handle_graph_upload(
+    graph_file: Any,
+    image_payload: dict[str, Any] | None,
+    toggles: list[str] | None,
+    layout_mode: str | None,
+    show_image: bool,
+    background_opacity: float | None,
+    show_bboxes: bool,
+):
+    if graph_file is None:
+        return (
+            None,
+            None,
+            None,
+            gr.update(choices=[], value=None),
+            "No graph loaded.",
+            "Dangling edges: 0",
+            [],
+            empty_figure("Upload a graph JSON to begin."),
+            {"message": "No graph loaded."},
+            None,
+            "Graph cleared.",
+        )
+
+    try:
+        graph_payload, raw_json, graph_name = _load_graph_upload(graph_file)
+    except Exception as exc:
+        return (
+            None,
+            None,
+            None,
+            gr.update(choices=[], value=None),
+            f"Failed to parse graph: {exc}",
+            "Dangling edges: 0",
+            [],
+            empty_figure("Failed to parse graph JSON."),
+            {"message": "No graph loaded."},
+            None,
+            f"Upload failed: {exc}",
+        )
+
+    return (
+        graph_payload,
+        raw_json,
+        graph_name,
+        gr.update(choices=_node_choices(graph_payload), value=None),
+        _summary(graph_payload, graph_name),
+        _dangling_markdown(graph_payload),
+        _edge_rows(graph_payload),
+        _render_figure(
+            graph_payload,
+            image_payload,
+            toggles,
+            layout_mode,
+            show_image,
+            background_opacity,
+            show_bboxes,
+            None,
+        ),
+        _node_details(graph_payload, None),
+        raw_json,
+        f"Loaded {graph_name}.",
+    )
+
+
+def handle_image_upload(
+    image_file: Any,
+    graph_payload: dict[str, Any] | None,
+    toggles: list[str] | None,
+    layout_mode: str | None,
+    show_image: bool,
+    background_opacity: float | None,
+    show_bboxes: bool,
+    selected_node_id: str | None,
+):
+    if image_file is None:
+        return (
+            None,
+            _render_figure(
+                graph_payload,
+                None,
+                toggles,
+                layout_mode,
+                show_image,
+                background_opacity,
+                show_bboxes,
+                selected_node_id,
+            ),
+            "Image cleared.",
+        )
+
+    try:
+        image_payload = _image_state_from_upload(image_file)
+    except Exception as exc:
+        return (
+            None,
+            _render_figure(
+                graph_payload,
+                None,
+                toggles,
+                layout_mode,
+                show_image,
+                background_opacity,
+                show_bboxes,
+                selected_node_id,
+            ),
+            f"Image upload failed: {exc}",
+        )
+
+    return (
+        image_payload,
+        _render_figure(
+            graph_payload,
+            image_payload,
+            toggles,
+            layout_mode,
+            show_image,
+            background_opacity,
+            show_bboxes,
+            selected_node_id,
+        ),
+        f"Loaded image {image_payload['name']}.",
+    )
+
+
+def create_app() -> gr.Blocks:
+    with gr.Blocks(title="KGVis", fill_width=True) as app:
+        graph_state = gr.State(None)
+        raw_state = gr.State(None)
+        graph_name_state = gr.State(None)
+        image_state = gr.State(None)
+
+        with gr.Column(elem_classes=["kgvis-shell"]):
+            gr.Markdown("# KGVis")
+
+            with gr.Row():
+                with gr.Column(scale=1, min_width=300):
+                    graph_file = gr.File(
+                        label="Upload JSON",
+                        file_count="single",
+                        file_types=[".json"],
+                        type="filepath",
+                    )
+                    image_file = gr.File(
+                        label="Upload Image",
+                        file_count="single",
+                        file_types=[".png", ".jpg", ".jpeg"],
+                        type="filepath",
+                    )
+                    status = gr.Textbox(
+                        label="Status",
+                        value="Upload a graph JSON to begin.",
+                        interactive=False,
+                        lines=2,
+                        elem_classes=["kgvis-status"],
+                    )
+                    toggles = gr.CheckboxGroup(
+                        label="View Options",
+                        choices=[
+                            ("Show nodes", "nodes"),
+                            ("Show edges", "edges"),
+                            ("Show node labels", "node_labels"),
+                            ("Show edge labels", "edge_labels"),
+                        ],
+                        value=DEFAULT_TOGGLES,
+                    )
+                    layout_mode = gr.Radio(
+                        label="Layout",
+                        choices=[
+                            ("Force layout", "force"),
+                            ("Image layout (bbox)", "bbox"),
+                        ],
+                        value="force",
+                    )
+                    show_bboxes = gr.Checkbox(label="Show bounding boxes", value=False)
+                    show_image = gr.Checkbox(label="Show image", value=True)
+                    background_opacity = gr.Slider(
+                        label="Background opacity",
+                        minimum=0.0,
+                        maximum=1.0,
+                        step=0.05,
+                        value=0.35,
+                    )
+                    selected_node = gr.Dropdown(
+                        label="Selected Node",
+                        choices=[],
+                        value=None,
+                        interactive=True,
+                    )
+
+                with gr.Column(scale=3, min_width=640):
+                    summary = gr.Markdown("No graph loaded.")
+                    plot = gr.Plot(
+                        label="Graph",
+                        value=empty_figure("Upload a graph JSON to begin."),
+                        elem_classes=["kgvis-plot"],
+                    )
+                    dangling = gr.Markdown("Dangling edges: 0")
+
+                    with gr.Row():
+                        with gr.Column(scale=1):
+                            node_details = gr.JSON(
+                                label="Node Details",
+                                value={"message": "No graph loaded."},
+                            )
+                        with gr.Column(scale=1):
+                            edge_table = gr.Dataframe(
+                                label="Edges",
+                                headers=["Source", "Predicate", "Target", "Dangling"],
+                                datatype=["str", "str", "str", "str"],
+                                interactive=False,
+                            )
+
+                    raw_json = gr.JSON(label="Raw JSON", value=None)
+
+        graph_file.change(
+            handle_graph_upload,
+            inputs=[
+                graph_file,
+                image_state,
+                toggles,
+                layout_mode,
+                show_image,
+                background_opacity,
+                show_bboxes,
+            ],
+            outputs=[
+                graph_state,
+                raw_state,
+                graph_name_state,
+                selected_node,
+                summary,
+                dangling,
+                edge_table,
+                plot,
+                node_details,
+                raw_json,
+                status,
+            ],
+        )
+
+        image_file.change(
+            handle_image_upload,
+            inputs=[
+                image_file,
+                graph_state,
+                toggles,
+                layout_mode,
+                show_image,
+                background_opacity,
+                show_bboxes,
+                selected_node,
+            ],
+            outputs=[image_state, plot, status],
+        )
+
+        for control in (
+            toggles,
+            layout_mode,
+            show_image,
+            background_opacity,
+            show_bboxes,
+            selected_node,
+        ):
+            control.change(
+                _render_visual_outputs,
+                inputs=[
+                    graph_state,
+                    image_state,
+                    toggles,
+                    layout_mode,
+                    show_image,
+                    background_opacity,
+                    show_bboxes,
+                    selected_node,
+                ],
+                outputs=[plot, node_details],
+            )
+
+    return app
+
+
+if __name__ == "__main__":
+    create_app().launch(css=CSS)

kgvis/__init__.py

@@ -0,0 +1 @@
+"""KGVis package."""

kgvis/plot.py

@@ -0,0 +1,520 @@
+from __future__ import annotations
+
+from typing import Iterable, Tuple
+
+import networkx as nx
+import plotly.graph_objects as go
+
+from .schema import GraphData
+
+
+def _build_layout(graph: GraphData, mode: str) -> dict[str, tuple[float, float]]:
+    if not graph.nodes:
+        return {}
+
+    graph_nx = nx.DiGraph()
+    for node in graph.nodes:
+        graph_nx.add_node(node.id)
+    for edge in graph.edges:
+        graph_nx.add_edge(edge.source, edge.target, predicate=edge.predicate)
+
+    if mode == "bbox":
+        explicit_positions: dict[str, tuple[float, float]] = {}
+        for node in graph.nodes:
+            center = _center_from_location(node.attributes)
+            if center is not None:
+                explicit_positions[node.id] = center
+
+        if explicit_positions and len(explicit_positions) == len(graph.nodes):
+            return explicit_positions
+
+        if explicit_positions:
+            return _spring_layout(graph_nx, pos=explicit_positions, fixed=explicit_positions.keys())
+
+    return _layout_force(graph_nx)
+
+
+def _spring_layout(
+    graph_nx: nx.DiGraph,
+    pos: dict[str, tuple[float, float]] | None = None,
+    fixed: Iterable[str] | None = None,
+) -> dict[str, tuple[float, float]]:
+    n = max(graph_nx.number_of_nodes(), 1)
+    k = 0.9 / (n**0.5)
+    return nx.spring_layout(
+        graph_nx,
+        seed=42,
+        k=k,
+        iterations=300,
+        pos=pos,
+        fixed=fixed,
+    )
+
+
+def _center_from_location(attributes: dict) -> Tuple[float, float] | None:
+    if not isinstance(attributes, dict):
+        return None
+
+    bbox = attributes.get("bbox")
+    if isinstance(bbox, dict):
+        try:
+            x1 = float(bbox["x1"])
+            y1 = float(bbox["y1"])
+            x2 = float(bbox["x2"])
+            y2 = float(bbox["y2"])
+            return _to_plot_center(x1, y1, x2, y2)
+        except (KeyError, TypeError, ValueError):
+            pass
+
+    raw = attributes.get("location_raw") or attributes.get("location")
+    parsed = _parse_location_value(raw)
+    if parsed:
+        x1, y1, x2, y2 = parsed
+        return _to_plot_center(x1, y1, x2, y2)
+    return None
+
+
+def _parse_location_value(value) -> Tuple[float, float, float, float] | None:
+    parts: list[float] = []
+    if isinstance(value, str):
+        try:
+            parts = [float(p.strip()) for p in value.split(",") if p.strip()]
+        except ValueError:
+            return None
+    elif isinstance(value, (list, tuple)):
+        try:
+            parts = [float(p) for p in value]
+        except (TypeError, ValueError):
+            return None
+    else:
+        return None
+
+    if len(parts) < 4:
+        return None
+    return parts[0], parts[1], parts[2], parts[3]
+
+
+def _to_plot_center(x1: float, y1: float, x2: float, y2: float) -> Tuple[float, float]:
+    center_x = (x1 + x2) / 2
+    center_y = (y1 + y2) / 2
+    # Normalize to plot coords: y=0 at bottom. Flip so y=0 is top like images.
+    return center_x, 1 - center_y
+
+
+def _normalize_positions(
+    pos: dict[str, tuple[float, float]],
+    padding: float = 0.05,
+) -> dict[str, tuple[float, float]]:
+    if not pos:
+        return {}
+    xs = [p[0] for p in pos.values()]
+    ys = [p[1] for p in pos.values()]
+    min_x, max_x = min(xs), max(xs)
+    min_y, max_y = min(ys), max(ys)
+    span_x = max(max_x - min_x, 1e-6)
+    span_y = max(max_y - min_y, 1e-6)
+    scale_x = (1 - 2 * padding) / span_x
+    scale_y = (1 - 2 * padding) / span_y
+    normalized = {}
+    for node_id, (x, y) in pos.items():
+        nx = padding + (x - min_x) * scale_x
+        ny = padding + (y - min_y) * scale_y
+        normalized[node_id] = (nx, ny)
+    return normalized
+
+
+def _layout_force(graph_nx: nx.DiGraph) -> dict[str, tuple[float, float]]:
+    if graph_nx.number_of_nodes() == 0:
+        return {}
+
+    undirected = graph_nx.to_undirected()
+    components = [list(c) for c in nx.connected_components(undirected)]
+    components.sort(key=len, reverse=True)
+
+    count = len(components)
+    cols = max(1, int((count**0.5) + 0.999))
+    rows = (count + cols - 1) // cols
+    cell_w = 1 / cols
+    cell_h = 1 / rows
+    cell_pad = 0.08
+
+    positions: dict[str, tuple[float, float]] = {}
+    for idx, nodes in enumerate(components):
+        sub = graph_nx.subgraph(nodes)
+        if len(nodes) == 1:
+            pos_comp = {nodes[0]: (0.5, 0.5)}
+        else:
+            pos_comp = _spring_layout(sub)
+        pos_comp = _normalize_positions(pos_comp, padding=0.1)
+
+        col = idx % cols
+        row = idx // cols
+        x0 = col * cell_w
+        y0 = row * cell_h
+        scale_x = cell_w * (1 - 2 * cell_pad)
+        scale_y = cell_h * (1 - 2 * cell_pad)
+        for node_id, (x, y) in pos_comp.items():
+            px = x0 + cell_pad * cell_w + x * scale_x
+            py = y0 + cell_pad * cell_h + y * scale_y
+            positions[node_id] = (px, py)
+
+    return _normalize_positions(positions, padding=0.04)
+
+
+def build_figure(
+    graph: GraphData,
+    *,
+    show_nodes: bool,
+    show_edges: bool,
+    show_node_labels: bool,
+    show_edge_labels: bool,
+    selected_node_id: str | None,
+    layout_mode: str,
+    background_image: str | None = None,
+    background_aspect: float | None = None,
+    background_opacity: float = 0.35,
+    show_bboxes: bool = False,
+    highlight_node_id: str | None = None,
+) -> go.Figure:
+    pos = _build_layout(graph, layout_mode)
+    image_frame = _image_frame(background_aspect) if background_aspect else None
+    image_domain = _image_domain(background_aspect) if background_aspect else None
+    if layout_mode == "bbox" and image_domain:
+        pos = _scale_positions(pos, image_domain)
+
+    edge_x: list[float] = []
+    edge_y: list[float] = []
+    edge_text_x: list[float] = []
+    edge_text_y: list[float] = []
+    edge_text: list[str] = []
+
+    if show_edges:
+        for edge in graph.edges:
+            if edge.source not in pos or edge.target not in pos:
+                continue
+            x0, y0 = pos[edge.source]
+            x1, y1 = pos[edge.target]
+            edge_x.extend([x0, x1, None])
+            edge_y.extend([y0, y1, None])
+            edge_text_x.append((x0 + x1) / 2)
+            edge_text_y.append((y0 + y1) / 2)
+            edge_text.append(edge.predicate)
+
+    edge_trace = None
+    if show_edges:
+        edge_trace = go.Scatter(
+            x=edge_x,
+            y=edge_y,
+            mode="lines",
+            line={"width": 1.2, "color": "#9aa4b2"},
+            hoverinfo="none",
+        )
+
+    node_x: list[float] = []
+    node_y: list[float] = []
+    node_text: list[str] = []
+    node_hover: list[str] = []
+    node_color: list[str] = []
+    node_ids: list[str] = []
+
+    if show_nodes:
+        for node in graph.nodes:
+            if node.id not in pos:
+                continue
+            x, y = pos[node.id]
+            node_x.append(x)
+            node_y.append(y)
+            node_ids.append(node.id)
+            node_text.append(node.label if show_node_labels else "")
+            node_hover.append(f"{node.label} ({node.id})")
+            if selected_node_id and node.id == selected_node_id:
+                node_color.append("#ff7b7b")
+            else:
+                node_color.append("#3a7bd5")
+
+    node_trace = None
+    if show_nodes:
+        node_trace = go.Scatter(
+            x=node_x,
+            y=node_y,
+            mode="markers+text" if show_node_labels else "markers",
+            text=node_text,
+            textposition="top center",
+            textfont={"size": 12, "color": "#1f2a44"},
+            hovertext=node_hover,
+            hoverinfo="text",
+            marker={
+                "size": 16,
+                "color": node_color,
+                "line": {"width": 1.2, "color": "#0f172a"},
+            },
+            customdata=node_ids,
+        )
+
+    traces: list[go.Scatter] = []
+    if edge_trace is not None:
+        traces.append(edge_trace)
+    if node_trace is not None:
+        traces.append(node_trace)
+
+    if show_edges and show_edge_labels and edge_text:
+        edge_label_trace = go.Scatter(
+            x=edge_text_x,
+            y=edge_text_y,
+            mode="text",
+            text=edge_text,
+            textfont={"size": 11, "color": "#475569"},
+            hoverinfo="none",
+        )
+        traces.append(edge_label_trace)
+
+    fig = go.Figure(data=traces)
+    layout_update = dict(
+        showlegend=False,
+        hovermode="closest",
+        margin={"l": 10, "r": 10, "t": 10, "b": 10},
+        xaxis={"visible": False, "range": [0, 1]},
+        yaxis={"visible": False, "range": [0, 1]},
+        plot_bgcolor="#ffffff",
+        annotations=_edge_annotations(graph, pos) if show_edges else [],
+        dragmode="pan",
+    )
+    if layout_mode == "bbox" and image_domain:
+        width, height = image_domain
+        layout_update["xaxis"] = {
+            "visible": False,
+            "range": [0, width],
+            "constrain": "domain",
+        }
+        layout_update["yaxis"] = {
+            "visible": False,
+            "range": [0, height],
+            "scaleanchor": "x",
+            "scaleratio": 1,
+            "constrain": "domain",
+        }
+    fig.update_layout(**layout_update)
+
+    if show_bboxes:
+        shapes = _bbox_shapes(
+            graph,
+            image_domain=image_domain if layout_mode == "bbox" else None,
+            frame=image_frame if layout_mode != "bbox" else None,
+            highlight_node_id=highlight_node_id,
+        )
+        if shapes:
+            fig.update_layout(shapes=shapes)
+
+    if background_image:
+        if layout_mode == "bbox" and image_domain:
+            x0, y0, width, height = 0, 0, image_domain[0], image_domain[1]
+        elif image_frame:
+            x0, y0, width, height = image_frame
+        else:
+            x0, y0, width, height = 0, 0, 1, 1
+        fig.update_layout(
+            images=[
+                dict(
+                    source=background_image,
+                    xref="x",
+                    yref="y",
+                    x=x0,
+                    y=y0 + height,
+                    sizex=width,
+                    sizey=height,
+                    sizing="stretch",
+                    xanchor="left",
+                    yanchor="top",
+                    opacity=max(0.0, min(background_opacity, 1.0)),
+                    layer="below",
+                )
+            ]
+        )
+    return fig
+
+
+def _image_frame(aspect: float) -> tuple[float, float, float, float]:
+    if aspect <= 0:
+        return 0, 0, 1, 1
+    if aspect >= 1:
+        width = 1.0
+        height = 1.0 / aspect
+        pad_x = 0.0
+        pad_y = (1.0 - height) / 2
+    else:
+        width = aspect
+        height = 1.0
+        pad_x = (1.0 - width) / 2
+        pad_y = 0.0
+    return pad_x, pad_y, width, height
+
+
+def _image_domain(aspect: float) -> tuple[float, float]:
+    if aspect <= 0:
+        return 1.0, 1.0
+    if aspect >= 1:
+        return aspect, 1.0
+    return 1.0, 1.0 / aspect
+
+
+def _apply_image_frame(
+    pos: dict[str, tuple[float, float]],
+    frame: tuple[float, float, float, float],
+) -> dict[str, tuple[float, float]]:
+    if not pos:
+        return {}
+    x0, y0, width, height = frame
+    adjusted = {}
+    for node_id, (x, y) in pos.items():
+        adjusted[node_id] = (x0 + x * width, y0 + y * height)
+    return adjusted
+
+
+def _scale_positions(
+    pos: dict[str, tuple[float, float]],
+    domain: tuple[float, float],
+) -> dict[str, tuple[float, float]]:
+    if not pos:
+        return {}
+    width, height = domain
+    scaled = {}
+    for node_id, (x, y) in pos.items():
+        scaled[node_id] = (x * width, y * height)
+    return scaled
+
+
+def _bbox_from_attrs(attributes: dict) -> tuple[float, float, float, float] | None:
+    if not isinstance(attributes, dict):
+        return None
+    bbox = attributes.get("bbox")
+    if isinstance(bbox, dict):
+        try:
+            x1 = float(bbox["x1"])
+            y1 = float(bbox["y1"])
+            x2 = float(bbox["x2"])
+            y2 = float(bbox["y2"])
+            return x1, y1, x2, y2
+        except (KeyError, TypeError, ValueError):
+            pass
+
+    raw = attributes.get("location_raw") or attributes.get("location")
+    parsed = _parse_location_value(raw)
+    if parsed:
+        return parsed
+    return None
+
+
+def _bbox_shapes(
+    graph: GraphData,
+    image_domain: tuple[float, float] | None,
+    frame: tuple[float, float, float, float] | None,
+    highlight_node_id: str | None,
+) -> list[dict]:
+    shapes: list[dict] = []
+    for node in graph.nodes:
+        bbox = _bbox_from_attrs(node.attributes)
+        if not bbox:
+            continue
+        x1, y1, x2, y2 = bbox
+        # Convert from image coords (origin top-left) to plot coords (origin bottom-left).
+        px1, px2 = x1, x2
+        py1, py2 = 1 - y2, 1 - y1
+
+        if image_domain:
+            width, height = image_domain
+            px1 *= width
+            px2 *= width
+            py1 *= height
+            py2 *= height
+        elif frame:
+            fx, fy, fw, fh = frame
+            px1 = fx + px1 * fw
+            px2 = fx + px2 * fw
+            py1 = fy + py1 * fh
+            py2 = fy + py2 * fh
+
+        is_highlight = highlight_node_id and node.id == highlight_node_id
+        line_color = "#ef4444" if is_highlight else "rgba(242,95,76,0.55)"
+        fill_color = "rgba(239,68,68,0.18)" if is_highlight else "rgba(242,95,76,0.08)"
+        line_width = 2.4 if is_highlight else 1.2
+
+        shapes.append(
+            dict(
+                type="rect",
+                xref="x",
+                yref="y",
+                x0=min(px1, px2),
+                x1=max(px1, px2),
+                y0=min(py1, py2),
+                y1=max(py1, py2),
+                line={"color": line_color, "width": line_width},
+                fillcolor=fill_color,
+                layer="below",
+            )
+        )
+    return shapes
+
+
+def empty_figure(message: str) -> go.Figure:
+    fig = go.Figure()
+    fig.add_annotation(
+        text=message,
+        x=0.5,
+        y=0.5,
+        xref="paper",
+        yref="paper",
+        xanchor="center",
+        yanchor="middle",
+        showarrow=False,
+        font={"size": 16, "color": "#64748b"},
+    )
+    fig.update_layout(
+        xaxis={"visible": False},
+        yaxis={"visible": False},
+        plot_bgcolor="#ffffff",
+        margin={"l": 20, "r": 20, "t": 20, "b": 20},
+    )
+    return fig
+
+
+def _edge_annotations(graph: GraphData, pos: dict[str, tuple[float, float]]):
+    annotations = []
+    for edge in graph.edges:
+        if edge.source not in pos or edge.target not in pos:
+            continue
+        x0, y0 = pos[edge.source]
+        x1, y1 = pos[edge.target]
+        dx = x1 - x0
+        dy = y1 - y0
+        length = (dx**2 + dy**2) ** 0.5
+        if length == 0:
+            continue
+
+        # Shorten the arrow so it doesn't overlap the node markers.
+        shrink = min(0.05, length * 0.25)
+        ux = dx / length
+        uy = dy / length
+        tail_x = x0 + ux * shrink
+        tail_y = y0 + uy * shrink
+        head_x = x1 - ux * shrink
+        head_y = y1 - uy * shrink
+
+        annotations.append(
+            dict(
+                x=head_x,
+                y=head_y,
+                ax=tail_x,
+                ay=tail_y,
+                xref="x",
+                yref="y",
+                axref="x",
+                ayref="y",
+                showarrow=True,
+                arrowhead=3,
+                arrowsize=1.15,
+                arrowwidth=1.2,
+                arrowcolor="#94a3b8",
+                opacity=0.9,
+            )
+        )
+    return annotations

kgvis/schema/__init__.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+from typing import Any, Dict, Iterable
+
+from .base import Edge, GraphData, Node, SchemaAdapter, graph_from_dict, graph_to_dict
+from .v1 import V1Adapter
+
+DEFAULT_ADAPTERS: list[SchemaAdapter] = [V1Adapter()]
+
+
+def parse_graph(data: Dict[str, Any], adapters: Iterable[SchemaAdapter] | None = None) -> GraphData:
+    candidates = list(adapters or DEFAULT_ADAPTERS)
+    for adapter in candidates:
+        if adapter.can_parse(data):
+            return adapter.parse(data)
+    raise ValueError("No schema adapter matched the provided graph data.")
+
+
+__all__ = [
+    "Edge",
+    "GraphData",
+    "Node",
+    "SchemaAdapter",
+    "DEFAULT_ADAPTERS",
+    "parse_graph",
+    "graph_from_dict",
+    "graph_to_dict",
+]

kgvis/schema/base.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, Protocol
+
+
+@dataclass(frozen=True)
+class Node:
+    id: str
+    label: str
+    attributes: Dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class Edge:
+    source: str
+    target: str
+    predicate: str
+    attributes: Dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class GraphData:
+    nodes: list[Node]
+    edges: list[Edge]
+    meta: Dict[str, Any] = field(default_factory=dict)
+
+
+class SchemaAdapter(Protocol):
+    name: str
+
+    def can_parse(self, data: Dict[str, Any]) -> bool: ...
+    def parse(self, data: Dict[str, Any]) -> GraphData: ...
+
+
+def graph_to_dict(graph: GraphData) -> Dict[str, Any]:
+    return {
+        "nodes": [
+            {"id": n.id, "label": n.label, "attributes": n.attributes} for n in graph.nodes
+        ],
+        "edges": [
+            {
+                "source": e.source,
+                "target": e.target,
+                "predicate": e.predicate,
+                "attributes": e.attributes,
+            }
+            for e in graph.edges
+        ],
+        "meta": dict(graph.meta),
+    }
+
+
+def graph_from_dict(payload: Dict[str, Any]) -> GraphData:
+    nodes = [
+        Node(
+            id=n.get("id", ""),
+            label=n.get("label", ""),
+            attributes=dict(n.get("attributes", {})),
+        )
+        for n in payload.get("nodes", [])
+    ]
+    edges = [
+        Edge(
+            source=e.get("source", ""),
+            target=e.get("target", ""),
+            predicate=e.get("predicate", ""),
+            attributes=dict(e.get("attributes", {})),
+        )
+        for e in payload.get("edges", [])
+    ]
+    meta = dict(payload.get("meta", {}))
+    return GraphData(nodes=nodes, edges=edges, meta=meta)

kgvis/schema/v1.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+from typing import Any, Dict, Tuple
+
+from .base import Edge, GraphData, Node
+
+
+class V1Adapter:
+    name = "frame2kg_v1"
+
+    def can_parse(self, data: Dict[str, Any]) -> bool:
+        if not isinstance(data, dict):
+            return False
+        payload = data.get("graph") if isinstance(data.get("graph"), dict) else data
+        nodes = payload.get("nodes")
+        edges = payload.get("edges")
+        if not isinstance(nodes, list) or not isinstance(edges, list):
+            return False
+        if nodes and not isinstance(nodes[0], dict):
+            return False
+        if edges and not isinstance(edges[0], dict):
+            return False
+        return True
+
+    def parse(self, data: Dict[str, Any]) -> GraphData:
+        payload = data.get("graph") if isinstance(data.get("graph"), dict) else data
+        nodes = []
+        for raw in payload.get("nodes", []):
+            known_keys = {"id", "label", "attributes", "location"}
+            extra = {k: v for k, v in raw.items() if k not in known_keys}
+            attributes = {**dict(raw.get("attributes", {})), **extra}
+            location = raw.get("location")
+            if location is not None:
+                parsed = _parse_location(location)
+                if parsed:
+                    bbox, confidence = parsed
+                    attributes.setdefault("bbox", bbox)
+                    if confidence is not None:
+                        attributes.setdefault("confidence", confidence)
+                attributes.setdefault("location_raw", location)
+            nodes.append(
+                Node(
+                    id=str(raw.get("id", "")),
+                    label=str(raw.get("label", "")),
+                    attributes=attributes,
+                )
+            )
+
+        edges = []
+        for raw in payload.get("edges", []):
+            known_keys = {"source", "target", "predicate", "attributes"}
+            extra = {k: v for k, v in raw.items() if k not in known_keys}
+            attributes = {**dict(raw.get("attributes", {})), **extra}
+            edges.append(
+                Edge(
+                    source=str(raw.get("source", "")),
+                    target=str(raw.get("target", "")),
+                    predicate=str(raw.get("predicate", "")),
+                    attributes=attributes,
+                )
+            )
+
+        meta = {"schema": self.name}
+        return GraphData(nodes=nodes, edges=edges, meta=meta)
+
+
+def _parse_location(value: Any) -> Tuple[dict[str, float], float | None] | None:
+    parts: list[float] = []
+    if isinstance(value, str):
+        try:
+            parts = [float(p.strip()) for p in value.split(",") if p.strip()]
+        except ValueError:
+            return None
+    elif isinstance(value, (list, tuple)):
+        try:
+            parts = [float(p) for p in value]
+        except (TypeError, ValueError):
+            return None
+    else:
+        return None
+
+    if len(parts) < 4:
+        return None
+    x1, y1, x2, y2 = parts[:4]
+    confidence = parts[4] if len(parts) >= 5 else None
+    bbox = {"x1": x1, "y1": y1, "x2": x2, "y2": y2}
+    return bbox, confidence

requirements.txt

@@ -0,0 +1,4 @@
+gradio>=6,<7
+networkx>=3.3
+pillow>=10.0.0
+plotly>=5.22.0