Coverage for openhcs/textual_tui/app.py: 0.0%
259 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-04 02:09 +0000
1"""
2OpenHCS Textual TUI Main Application
4A modern terminal user interface built with Textual framework.
5This is the main application class that orchestrates the entire TUI.
6"""
8import asyncio
9import logging
10import traceback
11from typing import Optional
13from textual.app import App, ComposeResult
14from textual.reactive import reactive
15from textual.containers import Container
16from textual.widgets import Static, Button, TextArea
18# OpenHCS imports
19from openhcs.core.config import GlobalPipelineConfig
20from openhcs.io.base import storage_registry
21from openhcs.io.filemanager import FileManager
23# Widget imports (will be created)
24from .widgets.main_content import MainContent
25from .widgets.status_bar import StatusBar
27# Textual-window imports
28from textual_window import Window, WindowSwitcher, window_manager, TilingLayout
29from openhcs.textual_tui.widgets.custom_window_bar import CustomWindowBar
30from openhcs.textual_tui.windows.base_window import BaseOpenHCSWindow
34logger = logging.getLogger(__name__)
37class ErrorDialog(BaseOpenHCSWindow):
38 """Error dialog with syntax highlighting using textual-window system."""
40 def __init__(self, error_message: str, error_details: str = ""):
41 self.error_message = error_message
42 self.error_details = error_details
43 super().__init__(
44 window_id="error_dialog",
45 title="🚨 ERROR",
46 mode="temporary"
47 )
49 def compose(self) -> ComposeResult:
50 """Compose the error dialog content."""
51 # Error message
52 yield Static(self.error_message, classes="error-message", markup=False)
54 # Error details with syntax highlighting if available
55 if self.error_details:
56 yield TextArea(
57 text=self.error_details,
58 language="python", # Python syntax highlighting for tracebacks
59 theme="monokai",
60 read_only=True, # Make it read-only but selectable
61 show_line_numbers=True,
62 soft_wrap=True,
63 id="error_content"
64 )
66 # Close button
67 with Container(classes="dialog-buttons"):
68 yield Button("Close", id="close", compact=True)
70 def on_button_pressed(self, event: Button.Pressed) -> None:
71 """Handle button presses."""
72 if event.button.id == "close":
73 self.close_window()
75 DEFAULT_CSS = """
76 ErrorDialog {
77 width: auto;
78 height: auto;
79 max-width: 120;
80 max-height: 40;
81 min-width: 50;
82 min-height: 15;
83 }
85 .error-message {
86 color: $error;
87 text-style: bold;
88 margin-bottom: 1;
89 text-align: center;
90 }
92 #error_content {
93 height: auto;
94 width: auto;
95 margin: 0;
96 max-height: 30;
97 min-height: 5;
98 border: solid $primary;
99 }
100 """
103class OpenHCSTUIApp(App):
104 """
105 Main OpenHCS Textual TUI Application.
107 This app provides a complete interface for OpenHCS pipeline management
108 with proper reactive state management and clean architectural boundaries.
109 """
110 CSS_PATH = "styles.css"
112 # Blocking window for pseudo-modal behavior
113 blocking_window = None
114# CSS = """
115# /* General dialog styling */
116# .dialog {
117# background: $surface;
118# border: tall $primary;
119# padding: 1 2;
120# width: auto;
121# height: auto;
122# }
123#
124# /* SelectionList styling - remove circles, keep highlighting */
125# SelectionList > .selection-list--option {
126# padding-left: 1;
127# text-style: none;
128# }
129#
130# SelectionList > .selection-list--option-highlighted {
131# background: $accent;
132# color: $text;
133# }
134#
135# /* MenuBar */
136# MenuBar {
137# height: 3;
138# border: solid white;
139# }
140#
141# /* All buttons - uniform height and styling */
142# Button {
143# height: 1;
144# }
145#
146# /* MenuBar buttons - content-based width */
147# MenuBar Button {
148# margin: 0 1;
149# width: auto;
150# }
151#
152# /* MenuBar title - properly centered */
153# MenuBar Static {
154# text-align: center;
155# content-align: center middle;
156# width: 1fr;
157# text-style: bold;
158# }
159#
160# /* Function list header buttons - specific styling for spacing */
161# #function_list_header Button {
162# margin: 0 1; /* 0 vertical, 1 horizontal margin */
163# width: auto; /* Let buttons size to their content */
164# }
165#
166# /* Main content containers with proper borders and responsive sizing */
167# #plate_manager_container {
168# border: solid white;
169# width: 1fr;
170# min-width: 0;
171# }
172#
173# #pipeline_editor_container {
174# border: solid white;
175# width: 1fr;
176# min-width: 0;
177# }
178#
179# /* StatusBar */
180# StatusBar {
181# height: 3;
182# border: solid white;
183# }
184#
185# /* Button containers - full width */
186# #plate_manager_container Horizontal,
187# #pipeline_editor_container Horizontal {
188# width: 100%;
189# }
190#
191# /* Content area buttons - responsive width distribution */
192# #plate_manager_container Button,
193# #pipeline_editor_container Button {
194# width: 1fr;
195# margin: 0;
196# min-width: 0;
197# }
198#
199# /* App fills terminal height properly */
200# OpenHCSTUIApp {
201# height: 100vh;
202# }
203#
204# /* Main content layout fills remaining space and is responsive */
205# MainContent {
206# height: 1fr;
207# width: 100%;
208# }
209#
210# /* Main horizontal layout is responsive */
211# MainContent > Horizontal {
212# width: 100%;
213# height: 100%;
214# }
215#
216#
217# /* Content areas adapt to available space */
218# ScrollableContainer {
219# height: 1fr;
220# }
221#
222# /* Static content styling */
223# Static {
224# text-align: center;
225# }
226#
227#
228# """
230 def _generate_bindings():
231 from openhcs.core.config import TilingKeybindings
233 keybindings = TilingKeybindings()
235 app_bindings = [
236 ("ctrl+q", "quit", "Quit"),
237 ("tab", "focus_next", "Next"),
238 ("shift+tab", "focus_previous", "Previous"),
239 ("f1", "toggle_window_switcher", "Switch Windows"),
240 ]
242 # Add all tiling keybindings from config
243 for field_name in keybindings.__dataclass_fields__:
244 binding = getattr(keybindings, field_name)
245 app_bindings.append((binding.key, binding.action, binding.description))
247 return app_bindings
249 BINDINGS = _generate_bindings()
251 # App-level reactive state
252 current_status = reactive("Ready")
254 def __init__(self, global_config: Optional[GlobalPipelineConfig] = None):
255 """
256 Initialize the OpenHCS TUI App.
258 Args:
259 global_config: Global configuration (uses default if None)
260 """
261 super().__init__()
263 # Core configuration - minimal TUI responsibility
264 self.global_config = global_config or GlobalPipelineConfig()
266 # Create shared components (pattern from SimpleOpenHCSTUILauncher)
267 self.storage_registry = storage_registry
268 self.filemanager = FileManager(self.storage_registry)
270 # Toolong compatibility attributes
271 self.save_merge = None # For Toolong LogView compatibility
272 self.file_paths = [] # For LogScreen compatibility
273 self.watcher = None # For LogScreen compatibility
274 self.merge = False # For LogScreen compatibility
276 logger.debug("OpenHCSTUIApp initialized with Textual reactive system")
278 def configure_toolong(self, file_paths: list, watcher, merge: bool = False):
279 """Configure app for Toolong LogScreen compatibility."""
280 self.file_paths = file_paths
281 self.watcher = watcher
282 self.merge = merge
283 logger.debug(f"App configured for Toolong with {len(file_paths)} files, merge={merge}")
285 def compose(self) -> ComposeResult:
286 """Compose the main application layout."""
287 # TEMPORARILY DISABLED - testing if WindowSwitcher causes hangs
288 # yield WindowSwitcher() # Invisible Alt-Tab overlay
290 # Custom WindowBar with no left button
291 yield CustomWindowBar(dock="bottom", start_open=True)
295 # Status bar for status messages
296 yield StatusBar()
298 # Main content fills the rest
299 yield MainContent(
300 filemanager=self.filemanager,
301 global_config=self.global_config
302 )
304 async def on_mount(self) -> None:
305 """Called when the app is mounted."""
306 logger.info("OpenHCS TUI mounted and ready")
307 self.current_status = "OpenHCS TUI Ready"
309 # Mount singleton toolong window BEFORE configuring tiling
310 # This prevents it from being affected by the tiling system
311 try:
312 from openhcs.textual_tui.windows.toolong_window import ToolongWindow
313 toolong_window = ToolongWindow()
314 await self.mount(toolong_window)
315 # Start minimized so it doesn't interfere with main UI
316 toolong_window.open_state = False
317 except Exception as e:
318 logger.error(f"Failed to mount toolong window at startup: {e}")
319 import traceback
320 logger.error(traceback.format_exc())
322 # Configure default window manager settings from separate TUI config
323 from openhcs.textual_tui.config import TUIConfig
324 tui_config = TUIConfig() # Use default TUI configuration
325 window_manager.set_tiling_layout(tui_config.default_tiling_layout)
326 window_manager.set_window_gap(tui_config.default_window_gap)
328 # Notify user about tiling mode if enabled in config
329 if tui_config.enable_startup_notification:
330 layout_name = tui_config.default_tiling_layout.value.replace('_', ' ').title()
331 self.notify(f"Window Manager: {layout_name} tiling enabled (gap={tui_config.default_window_gap})", severity="information")
335 # Status bar will automatically show this log message
336 # No need to manually update it anymore
338 # Add our start menu button to the WindowBar using the same pattern as window buttons
339 logger.debug("🚀 APP MOUNT: About to add start menu button")
340 try:
341 await self._add_start_menu_button()
342 logger.debug("🚀 APP MOUNT: Start menu button added")
343 except Exception as e:
344 logger.error(f"🚀 APP MOUNT: Start menu button failed: {e}")
345 # Continue without start menu button for now
347 def watch_current_status(self, status: str) -> None:
348 """Watch for status changes and log them (status bar will show automatically)."""
349 # Log the status change - status bar will pick it up automatically
350 logger.info(status)
354 def action_quit(self) -> None:
355 """Handle quit action with aggressive cleanup."""
356 logger.info("OpenHCS TUI shutting down")
358 # Force cleanup of background threads before exit
359 try:
360 import threading
361 import time
363 # Give a moment for normal cleanup
364 self.exit()
366 # Schedule aggressive cleanup after a short delay
367 def force_cleanup():
368 time.sleep(0.5) # Give normal exit a chance
369 active_threads = [t for t in threading.enumerate() if t != threading.current_thread() and t.is_alive()]
370 if active_threads:
371 logger.warning(f"Force cleaning up {len(active_threads)} threads on quit")
372 # Can't set daemon on running threads, just force exit
373 import os
374 os._exit(0)
376 # Start cleanup thread as daemon
377 cleanup_thread = threading.Thread(target=force_cleanup, daemon=True)
378 cleanup_thread.start()
380 except Exception as e:
381 logger.debug(f"Error in quit cleanup: {e}")
382 self.exit()
384 def action_toggle_window_switcher(self):
385 """Toggle the window switcher."""
386 switcher = self.query_one(WindowSwitcher)
387 switcher.action_toggle() # Correct textual-window API method
389 # Focus navigation actions - call existing window manager methods
390 def action_focus_next_window(self) -> None:
391 """Focus next window."""
392 window_manager.focus_next_window()
394 def action_focus_previous_window(self) -> None:
395 """Focus previous window."""
396 window_manager.focus_previous_window()
398 # Layout control wrapper methods
399 def action_set_horizontal_split(self) -> None:
400 """Set horizontal split tiling."""
401 window_manager.set_tiling_layout(TilingLayout.HORIZONTAL_SPLIT)
402 self.notify("Tiling: Horizontal Split")
404 def action_set_vertical_split(self) -> None:
405 """Set vertical split tiling."""
406 window_manager.set_tiling_layout(TilingLayout.VERTICAL_SPLIT)
407 self.notify("Tiling: Vertical Split")
409 def action_set_grid_layout(self) -> None:
410 """Set grid layout tiling."""
411 window_manager.set_tiling_layout(TilingLayout.GRID)
412 self.notify("Tiling: Grid Layout")
414 def action_set_master_detail(self) -> None:
415 """Set master-detail tiling."""
416 window_manager.set_tiling_layout(TilingLayout.MASTER_DETAIL)
417 self.notify("Tiling: Master Detail")
419 def action_toggle_floating(self) -> None:
420 """Toggle floating mode."""
421 current = window_manager.tiling_layout
422 new_layout = TilingLayout.HORIZONTAL_SPLIT if current == TilingLayout.FLOATING else TilingLayout.FLOATING
423 window_manager.set_tiling_layout(new_layout)
424 self.notify("Tiling: Toggled Floating")
426 # Window movement actions - call extracted window manager methods
427 def action_move_focused_window_prev(self) -> None:
428 """Move current window to previous position."""
429 window_manager.move_focused_window_prev()
431 def action_move_focused_window_next(self) -> None:
432 """Move current window to next position."""
433 window_manager.move_focused_window_next()
435 def action_rotate_window_order_left(self) -> None:
436 """Rotate all windows left."""
437 window_manager.rotate_window_order_left()
439 def action_rotate_window_order_right(self) -> None:
440 """Rotate all windows right."""
441 window_manager.rotate_window_order_right()
443 # Gap control actions
444 def action_gap_increase(self) -> None:
445 """Increase gap between windows."""
446 window_manager.adjust_window_gap(1)
448 def action_gap_decrease(self) -> None:
449 """Decrease gap between windows."""
450 window_manager.adjust_window_gap(-1)
452 # Bulk operation actions - call existing window manager methods
453 def action_minimize_all_windows(self) -> None:
454 """Minimize all windows."""
455 window_manager.minimize_all_windows()
456 self.notify("Minimized all windows")
458 def action_open_all_windows(self) -> None:
459 """Open all windows."""
460 window_manager.open_all_windows()
461 self.notify("Opened all windows")
463 async def _add_start_menu_button(self):
464 """Add our start menu button to the WindowBar at the leftmost position."""
465 try:
466 logger.debug("🚀 START MENU: Creating start menu button")
467 from openhcs.textual_tui.widgets.start_menu_button import StartMenuButton
469 # Get the CustomWindowBar
470 logger.debug("🚀 START MENU: Getting CustomWindowBar")
471 window_bar = self.query_one(CustomWindowBar)
472 logger.debug(f"🚀 START MENU: Found window bar: {window_bar}")
474 # Check if right button exists (no left button in CustomWindowBar)
475 logger.debug("🚀 START MENU: Looking for right button")
476 right_button = window_bar.query_one("#windowbar_button_right")
477 logger.debug(f"🚀 START MENU: Found right button: {right_button}")
479 # Add our start menu button at the very beginning (leftmost position)
480 # Mount before the right button to be at the far left
481 logger.debug("🚀 START MENU: Creating StartMenuButton")
482 start_button = StartMenuButton(window_bar=window_bar, id="start_menu_button")
483 logger.debug(f"🚀 START MENU: Created start button: {start_button}")
485 logger.debug("🚀 START MENU: Mounting start button")
486 await window_bar.mount(start_button, before=right_button)
487 logger.debug("🚀 START MENU: Start menu button mounted successfully")
489 except Exception as e:
490 logger.error(f"🚀 START MENU: Failed to add start menu button: {e}")
491 import traceback
492 logger.error(f"🚀 START MENU: Traceback: {traceback.format_exc()}")
493 raise
497 def open_blocking_window(self, window_class, *args, **kwargs):
498 """Open a blocking window that disables main UI interactions."""
499 if self.blocking_window:
500 return # Only allow one blocking window at a time
502 window = window_class(*args, **kwargs)
503 self.blocking_window = window
504 self._disable_main_interactions()
505 self.mount(window)
506 return window
508 def _disable_main_interactions(self):
509 """Disable main UI interactions when modal window is open."""
510 # Note: MenuBar removed - interactions now handled by start menu
511 pass
513 def _enable_main_interactions(self):
514 """Re-enable main UI interactions when modal window closes."""
515 # Note: MenuBar removed - interactions now handled by start menu
516 pass
518 def on_window_closed(self, event: Window.Closed) -> None:
519 """Handle window closed events from textual-window."""
520 # Check if this is our blocking window
521 # Event has window reference through WindowMessage base
522 if event.control == self.blocking_window:
523 self.blocking_window = None
524 self._enable_main_interactions()
526 def show_error(self, error_message: str, exception: Exception = None) -> None:
527 """Show a global error dialog with optional exception details."""
528 error_details = ""
529 if exception:
530 error_details = f"Exception: {type(exception).__name__}\n"
531 error_details += f"Message: {str(exception)}\n\n"
532 error_details += "Traceback:\n"
533 error_details += traceback.format_exc()
535 logger.error(f"Global error: {error_message}", exc_info=exception)
537 # Show error dialog using window system
538 from textual.css.query import NoMatches
540 try:
541 # Check if error dialog already exists
542 window = self.query_one(ErrorDialog)
543 # Update existing dialog
544 window.error_message = error_message
545 window.error_details = error_details
546 window.open_state = True
547 except NoMatches:
548 # Create new error dialog window
549 error_dialog = ErrorDialog(error_message, error_details)
550 self.run_worker(self._mount_error_dialog(error_dialog))
552 async def _mount_error_dialog(self, error_dialog):
553 """Mount error dialog window."""
554 await self.mount(error_dialog)
555 error_dialog.open_state = True
557 def _handle_exception(self, error: Exception) -> None:
558 """Handle exceptions with special cases for Toolong internal errors."""
559 # Check for known Toolong internal timing errors that are non-fatal
560 error_str = str(error)
561 if (
562 "No nodes match" in error_str and
563 ("FindDialog" in error_str or "Label" in error_str) and
564 ("InfoOverlay" in error_str or "LogView" in error_str)
565 ):
566 # This is a known Toolong internal timing issue - log but don't crash
567 logger.warning(f"Ignoring Toolong internal timing error: {error_str}")
568 return
570 # Log the error for debugging
571 logger.error(f"Unhandled exception in TUI: {str(error)}", exc_info=True)
573 # Re-raise the exception to let it crash loudly
574 # This allows the global error handler to catch it
575 raise error
577 async def _on_exception(self, error: Exception) -> None:
578 """Let async exceptions bubble up."""
579 self._handle_exception(error)
581 def _on_unhandled_exception(self, error: Exception) -> None:
582 """Let unhandled exceptions bubble up."""
583 self._handle_exception(error)
585 async def on_unmount(self) -> None:
586 """Clean up when app is shutting down with aggressive thread cleanup."""
587 logger.info("OpenHCS TUI app unmounting, cleaning up threads...")
589 # Force cleanup of any ReactiveLogMonitor instances
590 try:
591 from openhcs.textual_tui.widgets.reactive_log_monitor import ReactiveLogMonitor
592 monitors = self.query(ReactiveLogMonitor)
593 for monitor in monitors:
594 monitor.stop_monitoring()
595 except Exception as e:
596 logger.debug(f"Error cleaning up ReactiveLogMonitors: {e}")
598 # Force cleanup of any PlateManager workers
599 try:
600 from openhcs.textual_tui.widgets.plate_manager import PlateManager
601 plate_managers = self.query(PlateManager)
602 for pm in plate_managers:
603 if hasattr(pm, '_stop_monitoring'):
604 pm._stop_monitoring()
605 except Exception as e:
606 logger.debug(f"Error cleaning up PlateManagers: {e}")
608 # Aggressive thread cleanup
609 try:
610 import threading
611 import time
612 time.sleep(0.2) # Give threads a moment to stop
613 active_threads = [t for t in threading.enumerate() if t != threading.current_thread() and t.is_alive()]
614 if active_threads:
615 logger.warning(f"Found {len(active_threads)} active threads during shutdown")
616 # Can't set daemon on running threads, just log them
617 except Exception as e:
618 logger.debug(f"Error checking threads: {e}")
620 logger.info("OpenHCS TUI app cleanup complete")
623async def main():
624 """
625 Main entry point for the OpenHCS Textual TUI.
627 This function handles initialization and runs the application.
628 Note: Logging is setup by the main entry point, not here.
629 """
630 logger.info("Starting OpenHCS Textual TUI from app.py...")
632 try:
633 # Load configuration with cache support
634 from openhcs.textual_tui.services.config_cache_adapter import load_cached_global_config_tui as load_cached_global_config
635 global_config = await load_cached_global_config()
637 # REMOVED: setup_global_gpu_registry - this is now ONLY done in __main__.py
638 # to avoid duplicate initialization
639 logger.info("Using global_config with GPU registry already initialized by __main__.py")
641 # Create and run the app
642 app = OpenHCSTUIApp(global_config=global_config)
643 await app.run_async()
645 except KeyboardInterrupt:
646 logger.info("TUI terminated by user")
647 except Exception as e:
648 logger.error(f"Unhandled error: {e}", exc_info=True)
649 finally:
650 logger.info("OpenHCS Textual TUI finished")
653if __name__ == "__main__":
654 asyncio.run(main())