import json import logging from dataclasses import dataclass from typing import Optional from fasthtml.components import Div from fasthtml.xtend import Script from myfasthtml.controls.BaseCommands import BaseCommands from myfasthtml.controls.Menu import Menu, MenuConf from myfasthtml.controls.Query import Query, QueryConf from myfasthtml.core.commands import Command from myfasthtml.core.dbmanager import DbObject from myfasthtml.core.instances import MultipleInstance from myfasthtml.icons.fluent import arrow_reset20_regular logger = logging.getLogger("HierarchicalCanvasGraph") @dataclass class HierarchicalCanvasGraphConf: """Configuration for HierarchicalCanvasGraph control. Attributes: nodes: List of node dictionaries with keys: id, label, type, kind edges: List of edge dictionaries with keys: from, to events_handlers: Optional dict mapping event names to Command objects Supported events: 'select_node', 'toggle_node' """ nodes: list[dict] edges: list[dict] events_handlers: Optional[dict] = None class HierarchicalCanvasGraphState(DbObject): """Persistent state for HierarchicalCanvasGraph. Persists collapsed nodes, view transform (zoom/pan), and layout orientation. """ def __init__(self, owner, save_state=True): super().__init__(owner, save_state=save_state) with self.initializing(): # Persisted: set of collapsed node IDs (stored as list for JSON serialization) self.collapsed: list = [] # Persisted: zoom/pan transform self.transform: dict = {"x": 0, "y": 0, "scale": 1} # Persisted: layout orientation ('horizontal' or 'vertical') self.layout_mode: str = 'horizontal' # Persisted: filter state self.filter_text: Optional[str] = None # Text search filter self.filter_type: Optional[str] = None # Type filter (badge click) self.filter_kind: Optional[str] = None # Kind filter (border click) # Not persisted: current selection (ephemeral) self.ns_selected_id: Optional[str] = None class Commands(BaseCommands): """Commands for HierarchicalCanvasGraph internal state management.""" def update_view_state(self): """Update view transform and layout mode. This command is called internally by the JS to persist view state changes. """ return Command( "UpdateViewState", "Update view transform and layout mode", self._owner, self._owner.handle_update_view_state ).htmx(target=f"#{self._id}", swap='none') def apply_filter(self): """Apply current filter and update the graph display. This command is called when the filter changes (search text, type, or kind). """ return Command( "ApplyFilter", "Apply filter to graph", self._owner, self._owner.handle_apply_filter, key="#{id}-apply-filter", ).htmx(target=f"#{self._id}") def reset_view(self): """Reset the view transform to default values. This command can be used to fix stuck/frozen canvas by resetting zoom/pan state. """ return Command( "ResetView", "Reset view transform", self._owner, self._owner.handle_reset_view, icon=arrow_reset20_regular ).htmx(target=f"#{self._id}") class HierarchicalCanvasGraph(MultipleInstance): """A canvas-based hierarchical graph visualization control. Displays nodes and edges in a tree layout with expand/collapse functionality. Uses HTML5 Canvas for rendering with stable zoom/pan and search filtering. Features: - Reingold-Tilford hierarchical layout - Expand/collapse nodes with children - Zoom and pan with mouse wheel and drag - Search/filter nodes by label or kind - Click to select nodes - Dot grid background - Stable zoom on container resize Events: - select_node: Fired when a node is clicked (not on toggle button) - toggle_node: Fired when a node's expand/collapse button is clicked """ def __init__(self, parent, conf: HierarchicalCanvasGraphConf, _id=None): """Initialize the HierarchicalCanvasGraph control. Args: parent: Parent instance conf: Configuration object with nodes, edges, and event handlers _id: Optional custom ID (auto-generated if not provided) """ super().__init__(parent, _id=_id) self.conf = conf self._state = HierarchicalCanvasGraphState(self) self.commands = Commands(self) # Add Query component for filtering self._query = Query(self, QueryConf(placeholder="Filter instances..."), _id="-query") self._query.bind_command("QueryChanged", self.commands.apply_filter()) self._query.bind_command("CancelQuery", self.commands.apply_filter()) # Add Menu self._menu = Menu(self, conf=MenuConf(["ResetView"]), _id="-menu") logger.debug(f"HierarchicalCanvasGraph created with id={self._id}, " f"nodes={len(conf.nodes)}, edges={len(conf.edges)}") def get_state(self): """Get the control's persistent state. Returns: HierarchicalCanvasGraphState: The state object """ return self._state def handle_update_view_state(self, transform: Optional[dict] = None, layout_mode: Optional[str] = None): """Internal handler to update view state from client. Args: transform: Optional dict with zoom/pan transform state (received as JSON string) layout_mode: Optional string with layout orientation Returns: str: Empty string (no UI update needed) """ if transform is not None: # Parse JSON string to dict (sent as JSON.stringify() from JS) if isinstance(transform, str): try: transform = json.loads(transform) except (json.JSONDecodeError, TypeError) as e: logger.error(f"Failed to parse transform JSON: {e}") return "" self._state.transform = transform if layout_mode is not None: self._state.layout_mode = layout_mode return "" def handle_reset_view(self): """Internal handler to reset view transform to default values. Returns: self: For HTMX to render the updated graph with reset transform """ self._state.transform = {"x": 0, "y": 0, "scale": 1} self._state.collapsed = [] logger.debug("Transform and collapsed state reset to defaults") return self def handle_apply_filter(self, query_param="text", value=None): """Internal handler to apply filter and re-render the graph. Args: query_param: Type of filter - "text", "type", or "kind" value: The filter value (type name or kind name). Toggles off if same value clicked again. Returns: self: For HTMX to render the updated graph """ # Save old values to detect toggle old_filter_type = self._state.filter_type old_filter_kind = self._state.filter_kind # Reset all filters self._state.filter_text = None self._state.filter_type = None self._state.filter_kind = None # Apply the requested filter if query_param == "text": # Text filter from Query component self._state.filter_text = self._query.get_query() elif query_param == "type": # Type filter from badge click - toggle if same type clicked again if old_filter_type != value: self._state.filter_type = value elif query_param == "kind": # Kind filter from border click - toggle if same kind clicked again if old_filter_kind != value: self._state.filter_kind = value logger.debug(f"Applying filter: query_param={query_param}, value={value}, " f"text={self._state.filter_text}, type={self._state.filter_type}, kind={self._state.filter_kind}") return self def _calculate_filtered_nodes(self) -> Optional[list[str]]: """Calculate which node IDs match the current filter criteria. Returns: Optional[list[str]]: - None: No filter is active (all nodes visible, nothing dimmed) - []: Filter active but no matches (all nodes dimmed) - [ids]: Filter active with matches (only these nodes visible) """ # If no filters are active, return None (no filtering) if not self._state.filter_text and not self._state.filter_type and not self._state.filter_kind: return None filtered_ids = [] for node in self.conf.nodes: matches = True # Check text filter (searches in id, label, type, kind) if self._state.filter_text: search_text = self._state.filter_text.lower() searchable = f"{node.get('id', '')} {node.get('label', '')} {node.get('type', '')} {node.get('kind', '')}".lower() if search_text not in searchable: matches = False # Check type filter if self._state.filter_type and node.get('type') != self._state.filter_type: matches = False # Check kind filter if self._state.filter_kind and node.get('kind') != self._state.filter_kind: matches = False if matches: filtered_ids.append(node['id']) return filtered_ids def _prepare_options(self) -> dict: """Prepare JavaScript options object. Returns: dict: Options to pass to the JS initialization function """ # Convert event handlers to HTMX options events = {} # Add internal handler for view state persistence events['_internal_update_state'] = self.commands.update_view_state().ajax_htmx_options() # Add internal handlers for filtering by type and kind (badge/border clicks) events['_internal_filter_by_type'] = self.commands.apply_filter().ajax_htmx_options() events['_internal_filter_by_kind'] = self.commands.apply_filter().ajax_htmx_options() # Add user-provided event handlers if self.conf.events_handlers: for event_name, command in self.conf.events_handlers.items(): events[event_name] = command.ajax_htmx_options() # Calculate filtered nodes filtered_nodes = self._calculate_filtered_nodes() return { "nodes": self.conf.nodes, "edges": self.conf.edges, "collapsed": self._state.collapsed, "transform": self._state.transform, "layout_mode": self._state.layout_mode, "filtered_nodes": filtered_nodes, "events": events } def render(self): """Render the HierarchicalCanvasGraph control. Returns: Div: The rendered control with canvas and initialization script """ options = self._prepare_options() options_json = json.dumps(options, indent=2) return Div( Div( self._query, self._menu, cls="flex justify-between m-2" ), # Canvas element (sized by JS to fill container) Div( id=f"{self._id}_container", cls="mf-hcg-container" ), # Initialization script Script(f""" (function() {{ if (typeof initHierarchicalCanvasGraph === 'function') {{ initHierarchicalCanvasGraph('{self._id}_container', {options_json}); }} else {{ console.error('initHierarchicalCanvasGraph function not found'); }} }})(); """), id=self._id, cls="mf-hierarchical-canvas-graph" ) def __ft__(self): """FastHTML magic method for rendering. Returns: Div: The rendered control """ return self.render()