Coverage for openhcs/textual_tui/windows/function_selector_window.py: 0.0%
185 statements
« prev ^ index » next coverage.py v7.10.7, created at 2025-10-01 18:33 +0000
1"""Function selector window for selecting functions from the registry."""
3from typing import Callable, Optional, List, Tuple, Dict, Any
4from textual.app import ComposeResult
5from textual.containers import Vertical, Horizontal
6from textual.widgets import Input, DataTable, Button, Static, Tree
7from textual.widgets.data_table import RowKey
8from textual.widgets.tree import TreeNode
10from openhcs.textual_tui.windows.base_window import BaseOpenHCSWindow
11from openhcs.processing.backends.lib_registry.registry_service import RegistryService
12from openhcs.processing.backends.lib_registry.unified_registry import FunctionMetadata
15class FunctionSelectorWindow(BaseOpenHCSWindow):
16 """Window for selecting functions from the registry."""
18 DEFAULT_CSS = """
19 FunctionSelectorWindow {
20 width: 90; height: 30;
21 min-width: 90; min-height: 30;
22 }
24 .left-pane {
25 width: 30%; /* Reduced from 40% to take up less width */
26 border-right: solid $primary;
27 }
29 .right-pane {
30 width: 70%; /* Increased from 60% to give more space to table */
31 }
33 .pane-title {
34 text-align: center;
35 text-style: bold;
36 background: $primary;
37 color: $text;
38 height: 1;
39 }
40 """
42 def __init__(self, current_function: Optional[Callable] = None, on_result_callback: Optional[Callable] = None, **kwargs):
43 """Initialize function selector window.
45 Args:
46 current_function: Currently selected function (for highlighting)
47 on_result_callback: Callback function to handle the result
48 """
49 self.current_function = current_function
50 self.selected_function = None
51 self.all_functions_metadata: Dict[str, FunctionMetadata] = {}
52 self.filtered_functions: Dict[str, FunctionMetadata] = {}
53 self.on_result_callback = on_result_callback
55 # Load function data with enhanced metadata
56 self._load_function_data()
58 super().__init__(
59 window_id="function_selector",
60 title="Select Function - Dual Pane View",
61 mode="temporary",
62 **kwargs
63 )
65 def _load_function_data(self) -> None:
66 """Load function data with enhanced metadata from registry."""
67 registry_service = RegistryService()
69 # Get unified metadata for all functions (now with composite keys)
70 unified_functions = registry_service.get_all_functions_with_metadata()
72 # Convert to format expected by TUI, handling composite keys
73 self.all_functions_metadata = {}
74 for composite_key, metadata in unified_functions.items():
75 # Extract backend and function name from composite key
76 if ':' in composite_key:
77 backend, func_name = composite_key.split(':', 1)
78 else:
79 # Fallback for non-composite keys
80 backend = metadata.registry.library_name if metadata.registry else 'unknown'
81 func_name = composite_key
83 # Store with composite key but add backend info for UI
84 self.all_functions_metadata[composite_key] = metadata
86 self.filtered_functions = self.all_functions_metadata.copy()
88 def compose(self) -> ComposeResult:
89 """Compose the dual-pane function selector content."""
90 with Vertical():
91 # Search input
92 yield Input(
93 placeholder="Search functions by name, module, contract, or tags...",
94 id="search_input"
95 )
97 # Function count display
98 yield Static(f"Functions: {len(self.all_functions_metadata)}", id="function_count")
100 # Dual-pane layout
101 with Horizontal():
102 # Left pane: Hierarchical tree view
103 with Vertical(classes="left-pane"):
104 yield Static("Module Structure", classes="pane-title")
105 yield self._build_module_tree()
107 # Right pane: Enhanced table view
108 with Vertical(classes="right-pane"):
109 yield Static("Function Details", classes="pane-title")
110 yield self._build_function_table()
112 # Buttons - use unified dialog-buttons class for centered alignment
113 with Horizontal(classes="dialog-buttons"):
114 yield Button("Select", id="select_btn", variant="primary", compact=True, disabled=True)
115 yield Button("Cancel", id="cancel_btn", compact=True)
117 def _build_function_table(self) -> DataTable:
118 """Build table widget with enhanced function metadata."""
119 table = DataTable(id="function_table", cursor_type="row")
121 # Add columns with sorting support
122 table.add_column("Name", key="name")
123 table.add_column("Module", key="module")
124 table.add_column("Backend", key="backend")
125 table.add_column("Contract", key="contract")
126 table.add_column("Tags", key="tags")
127 table.add_column("Description", key="description")
129 # Populate table with function data
130 self._populate_table(table, self.filtered_functions)
132 return table
134 def _build_module_tree(self) -> Tree:
135 """Build hierarchical tree widget showing module structure based purely on module paths."""
136 tree = Tree("Module Structure", id="module_tree")
137 tree.root.expand()
139 # Build hierarchical structure directly from module paths
140 module_hierarchy = {}
141 for func_name, metadata in self.all_functions_metadata.items():
142 module_path = self._extract_module_path(metadata)
143 # Build hierarchical structure by splitting module path on '.'
144 self._add_function_to_hierarchy(module_hierarchy, module_path, func_name)
146 # Build tree structure directly from module hierarchy
147 self._build_module_hierarchy_tree(tree.root, module_hierarchy, [])
149 return tree
151 def _organize_by_library_and_module(self) -> Dict[str, Dict[str, List[str]]]:
152 """Organize functions by library and module structure."""
153 library_modules = {}
155 for func_name, metadata in self.all_functions_metadata.items():
156 # Determine library from tags or module
157 library = self._determine_library(metadata)
159 # Extract meaningful module path
160 module_path = self._extract_module_path(metadata)
162 # Initialize library if not exists
163 if library not in library_modules:
164 library_modules[library] = {}
166 # Initialize module if not exists
167 if module_path not in library_modules[library]:
168 library_modules[library][module_path] = []
170 # Add function to module
171 library_modules[library][module_path].append(func_name)
173 return library_modules
175 def _determine_library(self, metadata: FunctionMetadata) -> str:
176 """Determine library name from metadata."""
177 if 'openhcs' in metadata.tags:
178 return 'OpenHCS'
179 elif 'gpu' in metadata.tags and 'cupy' in metadata.module.lower():
180 return 'CuPy'
181 elif 'pyclesperanto' in metadata.module or 'cle' in metadata.module:
182 return 'pyclesperanto'
183 elif 'skimage' in metadata.module:
184 return 'scikit-image'
185 else:
186 return 'Unknown'
188 def _extract_module_path(self, metadata: FunctionMetadata) -> str:
189 """Extract meaningful module path for display."""
190 module = metadata.module
192 # For OpenHCS functions, show the backend structure
193 if 'openhcs' in metadata.tags:
194 parts = module.split('.')
195 # Find the backends part and show from there
196 try:
197 backends_idx = parts.index('backends')
198 return '.'.join(parts[backends_idx+1:]) # Skip 'backends'
199 except ValueError:
200 return module.split('.')[-1] # Just the last part
202 # For external libraries, show the meaningful part
203 elif 'skimage' in module:
204 parts = module.split('.')
205 try:
206 skimage_idx = parts.index('skimage')
207 return '.'.join(parts[skimage_idx+1:]) or 'core'
208 except ValueError:
209 return module.split('.')[-1]
211 elif 'pyclesperanto' in module or 'cle' in module:
212 return 'pyclesperanto_prototype'
214 elif 'cupy' in module.lower():
215 parts = module.split('.')
216 try:
217 cupy_idx = next(i for i, part in enumerate(parts) if 'cupy' in part.lower())
218 return '.'.join(parts[cupy_idx+1:]) or 'core'
219 except (StopIteration, IndexError):
220 return module.split('.')[-1]
222 return module.split('.')[-1]
224 def _populate_table(self, table: DataTable, functions_metadata: Dict[str, FunctionMetadata]) -> None:
225 """Populate table with function metadata."""
226 table.clear()
228 for composite_key, metadata in functions_metadata.items():
229 # Extract backend from composite key
230 if ':' in composite_key:
231 backend, func_name = composite_key.split(':', 1)
232 else:
233 # Fallback for non-composite keys
234 backend = getattr(metadata.func, 'input_memory_type', 'unknown')
235 func_name = composite_key
237 # Format tags as comma-separated string
238 tags_str = ", ".join(metadata.tags) if metadata.tags else ""
240 # Truncate description for table display
241 description = metadata.doc[:50] + "..." if len(metadata.doc) > 50 else metadata.doc
243 # Add row with function metadata
244 row_key = table.add_row(
245 metadata.name,
246 metadata.module.split('.')[-1] if metadata.module else "unknown", # Show only last part of module
247 backend.title(), # Capitalize backend name for display
248 metadata.contract.name if metadata.contract else "unknown",
249 tags_str,
250 description,
251 key=composite_key # Use composite key for row identification
252 )
254 # Store function reference for selection
255 table.get_row(row_key).metadata = {"func": metadata.func, "metadata": metadata}
257 def on_input_changed(self, event: Input.Changed) -> None:
258 """Handle search input changes."""
259 if event.input.id == "search_input":
260 self._filter_functions(event.value)
262 def on_tree_node_selected(self, event: Tree.NodeSelected) -> None:
263 """Handle tree node selection to filter table."""
264 if event.node.data:
265 node_type = event.node.data.get("type")
267 if node_type == "module":
268 # Filter table to show only functions from this module
269 module_functions = event.node.data.get("functions", [])
270 self.filtered_functions = {
271 name: metadata for name, metadata in self.all_functions_metadata.items()
272 if name in module_functions
273 }
275 # Update table and count
276 table = self.query_one("#function_table", DataTable)
277 self._populate_table(table, self.filtered_functions)
279 count_label = self.query_one("#function_count", Static)
280 count_label.update(f"Functions: {len(self.filtered_functions)}/{len(self.all_functions_metadata)} (filtered by module)")
282 elif node_type == "library":
283 # Filter table to show only functions from this library
284 library_name = event.node.data.get("name")
285 self.filtered_functions = {
286 name: metadata for name, metadata in self.all_functions_metadata.items()
287 if self._determine_library(metadata) == library_name
288 }
290 # Update table and count
291 table = self.query_one("#function_table", DataTable)
292 self._populate_table(table, self.filtered_functions)
294 count_label = self.query_one("#function_count", Static)
295 count_label.update(f"Functions: {len(self.filtered_functions)}/{len(self.all_functions_metadata)} (filtered by library)")
297 # Clear function selection when tree selection changes
298 self.selected_function = None
299 select_btn = self.query_one("#select_btn", Button)
300 select_btn.disabled = True
302 def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None:
303 """Handle table row selection."""
304 table = self.query_one("#function_table", DataTable)
306 if event.row_key:
307 # Get function from filtered metadata using composite key
308 composite_key = str(event.row_key.value)
309 if composite_key in self.filtered_functions:
310 metadata = self.filtered_functions[composite_key]
311 self.selected_function = metadata.func
313 # Enable select button
314 select_btn = self.query_one("#select_btn", Button)
315 select_btn.disabled = False
316 else:
317 self.selected_function = None
318 select_btn = self.query_one("#select_btn", Button)
319 select_btn.disabled = True
320 else:
321 self.selected_function = None
322 select_btn = self.query_one("#select_btn", Button)
323 select_btn.disabled = True
325 def on_button_pressed(self, event: Button.Pressed) -> None:
326 """Handle button presses."""
327 if event.button.id == "select_btn" and self.selected_function:
328 if self.on_result_callback:
329 self.on_result_callback(self.selected_function)
330 self.close_window()
331 elif event.button.id == "cancel_btn":
332 if self.on_result_callback:
333 self.on_result_callback(None)
334 self.close_window()
336 def _filter_functions(self, search_term: str) -> None:
337 """Filter functions based on search term across multiple fields."""
338 table = self.query_one("#function_table", DataTable)
339 count_label = self.query_one("#function_count", Static)
341 search_term = search_term.strip()
343 if not search_term or len(search_term) < 2:
344 # Show all functions if empty or less than 2 characters (performance optimization)
345 self.filtered_functions = self.all_functions_metadata.copy()
346 else:
347 # Filter functions across multiple fields
348 search_lower = search_term.lower()
349 self.filtered_functions = {}
351 for func_name, metadata in self.all_functions_metadata.items():
352 # Search in name, module, contract, tags, and description
353 searchable_text = " ".join([
354 metadata.name.lower(),
355 metadata.module.lower(),
356 metadata.contract.name.lower() if metadata.contract else "",
357 " ".join(metadata.tags).lower(),
358 metadata.doc.lower()
359 ])
361 if search_lower in searchable_text:
362 self.filtered_functions[func_name] = metadata
364 # Update table and count
365 self._populate_table(table, self.filtered_functions)
366 count_label.update(f"Functions: {len(self.filtered_functions)}/{len(self.all_functions_metadata)}")
368 # Clear selection when filtering
369 self.selected_function = None
370 select_btn = self.query_one("#select_btn", Button)
371 select_btn.disabled = True