Coverage for openhcs/textual_tui/windows/function_selector_window.py: 0.0%
181 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-04 02:09 +0000
1"""Function selector window for selecting functions from the registry."""
3from typing import Callable, Optional, List, Dict
4from textual.app import ComposeResult
5from textual.containers import Vertical, Horizontal
6from textual.widgets import Input, DataTable, Button, Static, Tree
8from openhcs.textual_tui.windows.base_window import BaseOpenHCSWindow
9from openhcs.processing.backends.lib_registry.registry_service import RegistryService
10from openhcs.processing.backends.lib_registry.unified_registry import FunctionMetadata
13class FunctionSelectorWindow(BaseOpenHCSWindow):
14 """Window for selecting functions from the registry."""
16 DEFAULT_CSS = """
17 FunctionSelectorWindow {
18 width: 90; height: 30;
19 min-width: 90; min-height: 30;
20 }
22 .left-pane {
23 width: 30%; /* Reduced from 40% to take up less width */
24 border-right: solid $primary;
25 }
27 .right-pane {
28 width: 70%; /* Increased from 60% to give more space to table */
29 }
31 .pane-title {
32 text-align: center;
33 text-style: bold;
34 background: $primary;
35 color: $text;
36 height: 1;
37 }
38 """
40 def __init__(self, current_function: Optional[Callable] = None, on_result_callback: Optional[Callable] = None, **kwargs):
41 """Initialize function selector window.
43 Args:
44 current_function: Currently selected function (for highlighting)
45 on_result_callback: Callback function to handle the result
46 """
47 self.current_function = current_function
48 self.selected_function = None
49 self.all_functions_metadata: Dict[str, FunctionMetadata] = {}
50 self.filtered_functions: Dict[str, FunctionMetadata] = {}
51 self.on_result_callback = on_result_callback
53 # Load function data with enhanced metadata
54 self._load_function_data()
56 super().__init__(
57 window_id="function_selector",
58 title="Select Function - Dual Pane View",
59 mode="temporary",
60 **kwargs
61 )
63 def _load_function_data(self) -> None:
64 """Load function data with enhanced metadata from registry."""
65 registry_service = RegistryService()
67 # Get unified metadata for all functions (now with composite keys)
68 unified_functions = registry_service.get_all_functions_with_metadata()
70 # Convert to format expected by TUI, handling composite keys
71 self.all_functions_metadata = {}
72 for composite_key, metadata in unified_functions.items():
73 # Extract backend and function name from composite key
74 if ':' in composite_key:
75 backend, func_name = composite_key.split(':', 1)
76 else:
77 # Fallback for non-composite keys
78 backend = metadata.registry.library_name if metadata.registry else 'unknown'
79 func_name = composite_key
81 # Store with composite key but add backend info for UI
82 self.all_functions_metadata[composite_key] = metadata
84 self.filtered_functions = self.all_functions_metadata.copy()
86 def compose(self) -> ComposeResult:
87 """Compose the dual-pane function selector content."""
88 with Vertical():
89 # Search input
90 yield Input(
91 placeholder="Search functions by name, module, contract, or tags...",
92 id="search_input"
93 )
95 # Function count display
96 yield Static(f"Functions: {len(self.all_functions_metadata)}", id="function_count")
98 # Dual-pane layout
99 with Horizontal():
100 # Left pane: Hierarchical tree view
101 with Vertical(classes="left-pane"):
102 yield self._build_module_tree()
104 # Right pane: Enhanced table view
105 with Vertical(classes="right-pane"):
106 yield Static("Function Details", classes="pane-title")
107 yield self._build_function_table()
109 # Buttons - use unified dialog-buttons class for centered alignment
110 with Horizontal(classes="dialog-buttons"):
111 yield Button("Select", id="select_btn", variant="primary", compact=True, disabled=True)
112 yield Button("Cancel", id="cancel_btn", compact=True)
114 def _build_function_table(self) -> DataTable:
115 """Build table widget with enhanced function metadata."""
116 table = DataTable(id="function_table", cursor_type="row")
118 # Add columns with sorting support - Backend shows memory type, Registry shows source
119 table.add_column("Name", key="name")
120 table.add_column("Module", key="module")
121 table.add_column("Backend", key="backend")
122 table.add_column("Registry", key="registry")
123 table.add_column("Contract", key="contract")
124 table.add_column("Tags", key="tags")
125 table.add_column("Description", key="description")
127 # Populate table with function data
128 self._populate_table(table, self.filtered_functions)
130 return table
132 def _build_module_tree(self) -> Tree:
133 """Build hierarchical tree widget showing module structure based purely on module paths."""
134 tree = Tree("Module Structure", id="module_tree")
135 # Start with tree collapsed - users can expand as needed
136 tree.root.collapse()
138 # Build hierarchical structure directly from module paths
139 module_hierarchy = {}
140 for func_name, metadata in self.all_functions_metadata.items():
141 module_path = self._extract_module_path(metadata)
142 # Build hierarchical structure by splitting module path on '.'
143 self._add_function_to_hierarchy(module_hierarchy, module_path, func_name)
145 # Build tree structure directly from module hierarchy
146 self._build_module_hierarchy_tree(tree.root, module_hierarchy, [])
148 return tree
150 def _organize_by_library_and_module(self) -> Dict[str, Dict[str, List[str]]]:
151 """Organize functions by library and module structure."""
152 library_modules = {}
154 for func_name, metadata in self.all_functions_metadata.items():
155 # Determine library from tags or module
156 library = self._determine_library(metadata)
158 # Extract meaningful module path
159 module_path = self._extract_module_path(metadata)
161 # Initialize library if not exists
162 if library not in library_modules:
163 library_modules[library] = {}
165 # Initialize module if not exists
166 if module_path not in library_modules[library]:
167 library_modules[library][module_path] = []
169 # Add function to module
170 library_modules[library][module_path].append(func_name)
172 return library_modules
174 def _determine_library(self, metadata: FunctionMetadata) -> str:
175 """Determine library name from metadata."""
176 if 'openhcs' in metadata.tags:
177 return 'OpenHCS'
178 elif 'gpu' in metadata.tags and 'cupy' in metadata.module.lower():
179 return 'CuPy'
180 elif 'pyclesperanto' in metadata.module or 'cle' in metadata.module:
181 return 'pyclesperanto'
182 elif 'skimage' in metadata.module:
183 return 'scikit-image'
184 else:
185 return 'Unknown'
187 def _extract_module_path(self, metadata: FunctionMetadata) -> str:
188 """Extract meaningful module path for display."""
189 module = metadata.module
191 # For OpenHCS functions, show the backend structure
192 if 'openhcs' in metadata.tags:
193 parts = module.split('.')
194 # Find the backends part and show from there
195 try:
196 backends_idx = parts.index('backends')
197 return '.'.join(parts[backends_idx+1:]) # Skip 'backends'
198 except ValueError:
199 return module.split('.')[-1] # Just the last part
201 # For external libraries, show the meaningful part
202 elif 'skimage' in module:
203 parts = module.split('.')
204 try:
205 skimage_idx = parts.index('skimage')
206 return '.'.join(parts[skimage_idx+1:]) or 'core'
207 except ValueError:
208 return module.split('.')[-1]
210 elif 'pyclesperanto' in module or 'cle' in module:
211 return 'pyclesperanto_prototype'
213 elif 'cupy' in module.lower():
214 parts = module.split('.')
215 try:
216 cupy_idx = next(i for i, part in enumerate(parts) if 'cupy' in part.lower())
217 return '.'.join(parts[cupy_idx+1:]) or 'core'
218 except (StopIteration, IndexError):
219 return module.split('.')[-1]
221 return module.split('.')[-1]
223 def _populate_table(self, table: DataTable, functions_metadata: Dict[str, FunctionMetadata]) -> None:
224 """Populate table with function metadata."""
225 table.clear()
227 for composite_key, metadata in functions_metadata.items():
228 # Get actual memory type (backend) and registry name separately
229 memory_type = metadata.get_memory_type()
230 registry_name = metadata.get_registry_name()
232 # Format tags as comma-separated string
233 tags_str = ", ".join(metadata.tags) if metadata.tags else ""
235 # Truncate description for table display
236 description = metadata.doc[:50] + "..." if len(metadata.doc) > 50 else metadata.doc
238 # Add row with function metadata - Backend shows memory type, Registry shows source
239 row_key = table.add_row(
240 metadata.name,
241 metadata.module.split('.')[-1] if metadata.module else "unknown", # Show only last part of module
242 memory_type.title(), # Show actual memory type (cupy, numpy, etc.)
243 registry_name.title(), # Show registry source (openhcs, skimage, etc.)
244 metadata.contract.name if metadata.contract else "unknown",
245 tags_str,
246 description,
247 key=composite_key # Use composite key for row identification
248 )
250 # Store function reference for selection
251 table.get_row(row_key).metadata = {"func": metadata.func, "metadata": metadata}
253 def on_input_changed(self, event: Input.Changed) -> None:
254 """Handle search input changes."""
255 if event.input.id == "search_input":
256 self._filter_functions(event.value)
258 def on_tree_node_selected(self, event: Tree.NodeSelected) -> None:
259 """Handle tree node selection to filter table."""
260 if event.node.data:
261 node_type = event.node.data.get("type")
263 if node_type == "module":
264 # Filter table to show only functions from this module
265 module_functions = event.node.data.get("functions", [])
266 self.filtered_functions = {
267 name: metadata for name, metadata in self.all_functions_metadata.items()
268 if name in module_functions
269 }
271 # Update table and count
272 table = self.query_one("#function_table", DataTable)
273 self._populate_table(table, self.filtered_functions)
275 count_label = self.query_one("#function_count", Static)
276 count_label.update(f"Functions: {len(self.filtered_functions)}/{len(self.all_functions_metadata)} (filtered by module)")
278 elif node_type == "library":
279 # Filter table to show only functions from this library
280 library_name = event.node.data.get("name")
281 self.filtered_functions = {
282 name: metadata for name, metadata in self.all_functions_metadata.items()
283 if self._determine_library(metadata) == library_name
284 }
286 # Update table and count
287 table = self.query_one("#function_table", DataTable)
288 self._populate_table(table, self.filtered_functions)
290 count_label = self.query_one("#function_count", Static)
291 count_label.update(f"Functions: {len(self.filtered_functions)}/{len(self.all_functions_metadata)} (filtered by library)")
293 # Clear function selection when tree selection changes
294 self.selected_function = None
295 select_btn = self.query_one("#select_btn", Button)
296 select_btn.disabled = True
298 def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None:
299 """Handle table row selection."""
300 table = self.query_one("#function_table", DataTable)
302 if event.row_key:
303 # Get function from filtered metadata using composite key
304 composite_key = str(event.row_key.value)
305 if composite_key in self.filtered_functions:
306 metadata = self.filtered_functions[composite_key]
307 self.selected_function = metadata.func
309 # Enable select button
310 select_btn = self.query_one("#select_btn", Button)
311 select_btn.disabled = False
312 else:
313 self.selected_function = None
314 select_btn = self.query_one("#select_btn", Button)
315 select_btn.disabled = True
316 else:
317 self.selected_function = None
318 select_btn = self.query_one("#select_btn", Button)
319 select_btn.disabled = True
321 def on_button_pressed(self, event: Button.Pressed) -> None:
322 """Handle button presses."""
323 if event.button.id == "select_btn" and self.selected_function:
324 if self.on_result_callback:
325 self.on_result_callback(self.selected_function)
326 self.close_window()
327 elif event.button.id == "cancel_btn":
328 if self.on_result_callback:
329 self.on_result_callback(None)
330 self.close_window()
332 def _filter_functions(self, search_term: str) -> None:
333 """Filter functions based on search term across multiple fields."""
334 table = self.query_one("#function_table", DataTable)
335 count_label = self.query_one("#function_count", Static)
337 search_term = search_term.strip()
339 if not search_term or len(search_term) < 2:
340 # Show all functions if empty or less than 2 characters (performance optimization)
341 self.filtered_functions = self.all_functions_metadata.copy()
342 else:
343 # Filter functions across multiple fields
344 search_lower = search_term.lower()
345 self.filtered_functions = {}
347 for func_name, metadata in self.all_functions_metadata.items():
348 # Search in name, module, contract, tags, and description
349 searchable_text = " ".join([
350 metadata.name.lower(),
351 metadata.module.lower(),
352 metadata.contract.name.lower() if metadata.contract else "",
353 " ".join(metadata.tags).lower(),
354 metadata.doc.lower()
355 ])
357 if search_lower in searchable_text:
358 self.filtered_functions[func_name] = metadata
360 # Update table and count
361 self._populate_table(table, self.filtered_functions)
362 count_label.update(f"Functions: {len(self.filtered_functions)}/{len(self.all_functions_metadata)}")
364 # Clear selection when filtering
365 self.selected_function = None
366 select_btn = self.query_one("#select_btn", Button)
367 select_btn.disabled = True