kodjo/MyFastHtml
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implemented new InstancesDebugger.py. Based on the HierarchicalCanvasGraph.py

Browse Source
This commit is contained in:
Kodjo Sossouvi
2026-02-22 17:51:39 +01:00
parent 8b8172231a
commit 0686103a8f
8 changed files with 536 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
src/myfasthtml/controls/DataGrid.py
View File

@@ -14,12 +14,12 @@ from myfasthtml.controls.BaseCommands import BaseCommands
from myfasthtml.controls.CycleStateControl import CycleStateControl
from myfasthtml.controls.DataGridColumnsManager import DataGridColumnsManager
from myfasthtml.controls.DataGridFormattingEditor import DataGridFormattingEditor
from myfasthtml.controls.DataGridQuery import DataGridQuery, DG_QUERY_FILTER
from myfasthtml.controls.DslEditor import DslEditorConf
from myfasthtml.controls.IconsHelper import IconsHelper
from myfasthtml.controls.Keyboard import Keyboard
from myfasthtml.controls.Mouse import Mouse
from myfasthtml.controls.Panel import Panel, PanelConf
from myfasthtml.controls.Query import Query, QUERY_FILTER
from myfasthtml.controls.datagrid_objects import DataGridColumnState, DataGridRowState, \
DatagridSelectionState, DataGridHeaderFooterConf, DatagridEditionState
from myfasthtml.controls.helpers import mk, column_type_defaults
@@ -142,7 +142,7 @@ class Commands(BaseCommands):
return Command("Filter",
"Filter Grid",
self._owner,
self._owner.filter
self._owner.filter,
)
def change_selection_mode(self):
@@ -212,8 +212,8 @@ class DataGrid(MultipleInstance):
self.bind_command("ToggleColumnsManager", self._panel.commands.toggle_side("right"))
self.bind_command("ToggleFormattingEditor", self._panel.commands.toggle_side("right"))
# add DataGridQuery
self._datagrid_filter = DataGridQuery(self)
# add Query
self._datagrid_filter = Query(self)
self._datagrid_filter.bind_command("QueryChanged", self.commands.filter())
self._datagrid_filter.bind_command("CancelQuery", self.commands.filter())
self._datagrid_filter.bind_command("ChangeFilterType", self.commands.filter())
@@ -295,7 +295,7 @@ class DataGrid(MultipleInstance):
for col_id, values in self._state.filtered.items():
if col_id == FILTER_INPUT_CID:
if values is not None:
if self._datagrid_filter.get_query_type() == DG_QUERY_FILTER:
if self._datagrid_filter.get_query_type() == QUERY_FILTER:
visible_columns = [c.col_id for c in self._state.columns if c.visible and c.col_id in df.columns]
df = df[df[visible_columns].map(lambda x: values.lower() in str(x).lower()).any(axis=1)]
else:
@@ -632,6 +632,9 @@ class DataGrid(MultipleInstance):
def get_state(self):
return self._state
def get_description(self) -> str:
return self.get_table_name()
def get_settings(self):
return self._settings

124
src/myfasthtml/controls/HierarchicalCanvasGraph.py
View File

@@ -7,6 +7,7 @@ from fasthtml.components import Div
from fasthtml.xtend import Script
from myfasthtml.controls.BaseCommands import BaseCommands
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
@@ -47,6 +48,11 @@ class HierarchicalCanvasGraphState(DbObject):
# 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
@@ -65,6 +71,19 @@ class Commands(BaseCommands):
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}")
class HierarchicalCanvasGraph(MultipleInstance):
@@ -100,6 +119,11 @@ class HierarchicalCanvasGraph(MultipleInstance):
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())
logger.debug(f"HierarchicalCanvasGraph created with id={self._id}, "
f"nodes={len(conf.nodes)}, edges={len(conf.edges)}")
@@ -148,25 +172,102 @@ class HierarchicalCanvasGraph(MultipleInstance):
self._state.collapsed = list(collapsed_set)
return self
def _handle_update_view_state(self, event_data: dict):
def _handle_update_view_state(self, transform=None, layout_mode=None):
"""Internal handler to update view state from client.
Args:
event_data: Dictionary with 'transform' and/or 'layout_mode' keys
transform: Optional dict with zoom/pan transform state
layout_mode: Optional string with layout orientation
Returns:
str: Empty string (no UI update needed)
"""
if 'transform' in event_data:
self._state.transform = event_data['transform']
if transform is not None:
self._state.transform = transform
logger.debug(f"Transform updated: {self._state.transform}")
if 'layout_mode' in event_data:
self._state.layout_mode = event_data['layout_mode']
if layout_mode is not None:
self._state.layout_mode = layout_mode
logger.debug(f"Layout mode updated: {self._state.layout_mode}")
return ""
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.
@@ -179,17 +280,25 @@ class HierarchicalCanvasGraph(MultipleInstance):
# 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
}
@@ -203,6 +312,9 @@ class HierarchicalCanvasGraph(MultipleInstance):
options_json = json.dumps(options, indent=2)
return Div(
# Query filter bar
self._query,
# Canvas element (sized by JS to fill container)
Div(
id=f"{self._id}_container",

83
src/myfasthtml/controls/InstancesDebugger.py
View File

@@ -25,22 +25,22 @@ class InstancesDebugger(SingleInstance):
self.conf = conf if conf is not None else InstancesDebuggerConf()
self._panel = Panel(self, _id="-panel")
self._select_command = Command("ShowInstance",
"Display selected Instance",
self,
self.on_select_node).htmx(target=f"#{self._panel.get_ids().right}")
"Display selected Instance",
self,
self.on_select_node).htmx(target=f"#{self._panel.get_ids().right}")
def render(self):
nodes, edges = self._get_nodes_and_edges()
graph_conf = HierarchicalCanvasGraphConf(
nodes=nodes,
edges=edges,
events_handlers={
"select_node": self._select_command
"select_node": self._select_command
}
)
canvas_graph = HierarchicalCanvasGraph(self, conf=graph_conf, _id="-canvas-graph")
return self._panel.set_main(canvas_graph)
def on_select_node(self, event_data: dict):
"""Handle node selection event from canvas graph.
@@ -50,23 +50,23 @@ class InstancesDebugger(SingleInstance):
node_id = event_data.get("node_id")
if not node_id:
return None
# Parse full ID (session#instance_id)
parts = node_id.split("#")
session = parts[0]
instance_id = "#".join(parts[1:])
properties_def = {
"Main": {"Id": "_id", "Parent Id": "_parent._id"},
"State": {"_name": "_state._name", "*": "_state"},
"Commands": {"*": "commands"},
"Main": {"Id": "_id", "Parent Id": "_parent._id"},
"State": {"_name": "_state._name", "*": "_state"},
"Commands": {"*": "commands"},
}
return self._panel.set_right(Properties(self,
InstancesManager.get(session, instance_id),
properties_def,
_id="-properties"))
def _get_instance_kind(self, instance) -> str:
"""Determine the instance kind for visualization.
@@ -87,7 +87,7 @@ class InstancesDebugger(SingleInstance):
return 'multiple'
else:
return 'multiple' # Default
def _get_nodes_and_edges(self):
"""Build nodes and edges from current instances.
@@ -95,57 +95,58 @@ class InstancesDebugger(SingleInstance):
tuple: (nodes, edges) where nodes include id, label, kind, type
"""
instances = self._get_instances()
nodes = []
edges = []
existing_ids = set()
# Create nodes with kind (instance kind) and type (class name)
for instance in instances:
node_id = instance.get_full_id()
existing_ids.add(node_id)
nodes.append({
"id": node_id,
"label": instance.get_id(),
"kind": self._get_instance_kind(instance),
"type": instance.__class__.__name__
"id": node_id,
"label": instance.get_id(),
"kind": self._get_instance_kind(instance),
"type": instance.__class__.__name__,
"description": instance.get_description()
})
# Track nodes with parents
nodes_with_parent = set()
# Create edges
for instance in instances:
node_id = instance.get_full_id()
parent_id = instance.get_parent_full_id()
if parent_id is None or parent_id == "":
continue
nodes_with_parent.add(node_id)
edges.append({
"from": parent_id,
"to": node_id
"from": parent_id,
"to": node_id
})
# Create ghost node if parent not in existing instances
if parent_id not in existing_ids:
nodes.append({
"id": parent_id,
"label": f"Ghost: {parent_id}",
"kind": "multiple", # Default kind for ghost nodes
"type": "Ghost"
"id": parent_id,
"label": f"Ghost: {parent_id}",
"kind": "multiple", # Default kind for ghost nodes
"type": "Ghost"
})
existing_ids.add(parent_id)
# Group siblings by type if configured
if self.conf.group_siblings_by_type:
edges = self._sort_edges_by_sibling_type(nodes, edges)
return nodes, edges
def _sort_edges_by_sibling_type(self, nodes, edges):
"""Sort edges so that siblings (same parent) are grouped by type.
@@ -157,15 +158,15 @@ class InstancesDebugger(SingleInstance):
list: Sorted edges with siblings grouped by type
"""
from collections import defaultdict
# Create mapping node_id -> type for quick lookup
node_types = {node["id"]: node["type"] for node in nodes}
# Group edges by parent
edges_by_parent = defaultdict(list)
for edge in edges:
edges_by_parent[edge["from"]].append(edge)
# Sort each parent's children by type and rebuild edges list
sorted_edges = []
for parent_id in edges_by_parent:
@@ -174,9 +175,9 @@ class InstancesDebugger(SingleInstance):
key=lambda e: node_types.get(e["to"], "")
)
sorted_edges.extend(parent_edges)
return sorted_edges
def _get_instances(self):
return list(InstancesManager.instances.values())

109
src/myfasthtml/controls/Query.py Normal file
View File

@@ -0,0 +1,109 @@
import logging
from dataclasses import dataclass
from typing import Optional
from fasthtml.components import *
from myfasthtml.controls.BaseCommands import BaseCommands
from myfasthtml.controls.helpers import mk
from myfasthtml.core.commands import Command
from myfasthtml.core.dbmanager import DbObject
from myfasthtml.core.instances import MultipleInstance
from myfasthtml.icons.fluent import brain_circuit20_regular
from myfasthtml.icons.fluent_p1 import filter20_regular, search20_regular
from myfasthtml.icons.fluent_p2 import dismiss_circle20_regular
logger = logging.getLogger("Query")
QUERY_FILTER = "filter"
QUERY_SEARCH = "search"
QUERY_AI = "ai"
query_type = {
QUERY_FILTER: filter20_regular,
QUERY_SEARCH: search20_regular,
QUERY_AI: brain_circuit20_regular
}
@dataclass
class QueryConf:
"""Configuration for Query control.
Attributes:
placeholder: Placeholder text for the search input
"""
placeholder: str = "Search..."
class QueryState(DbObject):
def __init__(self, owner):
with self.initializing():
super().__init__(owner)
self.filter_type: str = "filter"
self.query: Optional[str] = None
class Commands(BaseCommands):
def change_filter_type(self):
return Command("ChangeFilterType",
"Change filter type",
self._owner,
self._owner.change_query_type).htmx(target=f"#{self._id}")
def on_filter_changed(self):
return Command("QueryChanged",
"Query changed",
self._owner,
self._owner.query_changed).htmx(target=None) # prevent focus loss when typing
def on_cancel_query(self):
return Command("CancelQuery",
"Cancel query",
self._owner,
self._owner.query_changed,
kwargs={"query": ""}
).htmx(target=f"#{self._id}")
class Query(MultipleInstance):
def __init__(self, parent, conf: Optional[QueryConf] = None, _id=None):
super().__init__(parent, _id=_id)
self.conf = conf or QueryConf()
self.commands = Commands(self)
self._state = QueryState(self)
def get_query(self):
return self._state.query
def get_query_type(self):
return self._state.filter_type
def change_query_type(self):
keys = list(query_type.keys()) # ["filter", "search", "ai"]
current_idx = keys.index(self._state.filter_type)
self._state.filter_type = keys[(current_idx + 1) % len(keys)]
return self
def query_changed(self, query):
logger.debug(f"query_changed {query=}")
self._state.query = query.strip() if query is not None else None
return self # needed anyway to allow oob swap
def render(self):
return Div(
mk.label(
Input(name="query",
value=self._state.query if self._state.query is not None else "",
placeholder=self.conf.placeholder,
**self.commands.on_filter_changed().get_htmx_params(values_encode="json")),
icon=mk.icon(query_type[self._state.filter_type], command=self.commands.change_filter_type()),
cls="input input-xs flex gap-3"
),
mk.icon(dismiss_circle20_regular, size=24, command=self.commands.on_cancel_query()),
cls="flex",
id=self._id
)
def __ft__(self):
return self.render()