Coverage for openhcs/textual_tui/widgets/start_menu_button.py: 0.0%
170 statements
« prev ^ index » next coverage.py v7.10.3, created at 2025-08-14 05:57 +0000
1"""Start Menu Button for OpenHCS TUI - integrates with WindowBar system."""
3import logging
4from typing import Any
6from textual import events, work
7from textual.app import ComposeResult
8from textual.containers import Container
9from textual.geometry import Offset
10from textual.screen import ModalScreen
12# Import the button base from textual-window
13from textual_window.button_bases import NoSelectStatic, ButtonStatic
15logger = logging.getLogger(__name__)
18class StartMenuButton(NoSelectStatic):
19 """
20 Start menu button that integrates with WindowBar.
22 Uses the same pattern as WindowBarButton (content-sized, not space-filling).
23 """
25 # Use the same CSS classes as WindowBarButton for consistent styling
26 DEFAULT_CSS = """
27 StartMenuButton {
28 height: 1;
29 width: auto; /* Content-sized like WindowBarButton, not 1fr like WindowBarAllButton */
30 padding: 0 1;
31 &:hover { background: $panel-lighten-1; }
32 &.pressed { background: $primary; color: $text; }
33 }
34 """
36 def __init__(self, window_bar, **kwargs: Any):
37 super().__init__(content="☰ Start", **kwargs)
38 self.window_bar = window_bar
39 self.click_started_on: bool = False
41 def on_mouse_down(self, event: events.MouseDown) -> None:
42 """Handle mouse down - add pressed styling."""
43 self.add_class("pressed")
44 self.click_started_on = True
46 async def on_mouse_up(self, event: events.MouseUp) -> None:
47 """Handle mouse up - show dropdown menu."""
48 self.remove_class("pressed")
49 if self.click_started_on:
50 self.show_popup(event)
51 self.click_started_on = False
53 def on_leave(self, event: events.Leave) -> None:
54 """Handle mouse leave - remove pressed styling."""
55 self.remove_class("pressed")
56 self.click_started_on = False
58 @work
59 async def show_popup(self, event: events.MouseUp) -> None:
60 """Show the start menu dropdown using WindowBar's menu system."""
61 absolute_offset = self.screen.get_offset(self)
62 await self.app.push_screen_wait(
63 StartMenuDropdown(
64 menu_offset=absolute_offset,
65 dock=self.window_bar.dock,
66 )
67 )
70class StartMenuDropdown(ModalScreen[None]):
71 """
72 Start menu dropdown - follows the same pattern as WindowBarMenu.
73 """
75 CSS = """
76 StartMenuDropdown {
77 background: $background 0%;
78 align: left top;
79 }
80 #start_menu_container {
81 background: $surface;
82 width: auto;
83 min-width: 8;
84 height: auto;
85 border-left: wide $panel;
86 border-right: wide $panel;
87 /* Remove problematic borders that get cut off */
88 & > ButtonStatic {
89 width: 100%;
90 min-width: 8;
91 &:hover { background: $panel-lighten-2; }
92 &.pressed { background: $primary; }
93 }
94 }
95 """
97 def __init__(self, menu_offset: Offset, dock: str) -> None:
98 super().__init__()
99 self.menu_offset = menu_offset
100 self.dock = dock
102 def compose(self) -> ComposeResult:
103 """Compose the start menu dropdown."""
104 with Container(id="start_menu_container"):
105 yield ButtonStatic("Main", id="main")
106 yield ButtonStatic("Config", id="config")
107 yield ButtonStatic("Monitor", id="toggle_monitor")
108 yield ButtonStatic("Term", id="term")
109 yield ButtonStatic("Help", id="help")
110 yield ButtonStatic("Quit", id="quit")
112 def on_mount(self) -> None:
113 """Position the dropdown menu based on dock position."""
114 menu = self.query_one("#start_menu_container")
116 if self.dock == "top":
117 # Bar is at top, dropdown should appear below
118 y_offset = self.menu_offset.y + 1
119 elif self.dock == "bottom":
120 # Bar is at bottom, dropdown should appear above
121 menu_height = len(list(menu.children))
122 y_offset = self.menu_offset.y - menu_height
123 else:
124 raise ValueError("Dock must be either 'top' or 'bottom'")
126 menu.offset = Offset(self.menu_offset.x, y_offset)
127 menu.add_class(self.dock)
129 def on_mouse_up(self) -> None:
130 """Close dropdown when clicking outside."""
131 self.dismiss(None)
133 async def on_button_static_pressed(self, event: ButtonStatic.Pressed) -> None:
134 """Handle button presses in the dropdown."""
135 button_id = event.button.id
137 if button_id == "main":
138 await self._handle_main()
139 elif button_id == "config":
140 await self._handle_config()
141 elif button_id == "term":
142 await self._handle_term()
143 elif button_id == "help":
144 await self._handle_help()
145 elif button_id == "toggle_monitor":
146 await self._handle_toggle_monitor()
147 elif button_id == "quit":
148 await self._handle_quit()
150 # Close the dropdown
151 self.dismiss(None)
153 async def _handle_main(self) -> None:
154 """Handle main button press - open the shared PipelinePlateWindow with both components."""
155 from openhcs.textual_tui.windows import PipelinePlateWindow
156 from textual.css.query import NoMatches
158 # Try to find existing window - if it doesn't exist, create new one
159 try:
160 window = self.app.query_one(PipelinePlateWindow)
161 # Window exists, show both components and open it
162 window.show_both()
163 window.open_state = True
164 except NoMatches:
165 # Expected case: window doesn't exist yet, create new one
166 window = PipelinePlateWindow(self.app.filemanager, self.app.global_config)
167 await self.app.mount(window)
168 window.show_both()
169 window.open_state = True
171 async def _handle_config(self) -> None:
172 """Handle config button press."""
173 from openhcs.textual_tui.windows import ConfigWindow
174 from openhcs.core.config import GlobalPipelineConfig
175 from textual.css.query import NoMatches
177 def handle_config_save(new_config):
178 # new_config is already GlobalPipelineConfig (concrete dataclass)
179 global_config = new_config
181 # Apply config changes to app
182 self.app.global_config = global_config
184 # Update thread-local storage for MaterializationPathConfig defaults
185 from openhcs.core.config import set_current_pipeline_config
186 set_current_pipeline_config(global_config)
188 # Propagate config changes to all existing orchestrators and plate manager
189 self._propagate_global_config_to_orchestrators(global_config)
191 # Save config to cache for future sessions
192 self._save_config_to_cache(global_config)
194 logger.info("Configuration updated and applied from start menu")
196 # Try to find existing config window - if it doesn't exist, create new one
197 try:
198 window = self.app.query_one(ConfigWindow)
199 # Window exists, just open it
200 window.open_state = True
201 except NoMatches:
202 # Expected case: window doesn't exist yet, create new one
203 window = ConfigWindow(
204 GlobalPipelineConfig,
205 self.app.global_config,
206 on_save_callback=handle_config_save,
207 is_global_config_editing=True
208 )
209 await self.app.mount(window)
210 window.open_state = True
212 async def _handle_term(self) -> None:
213 """Handle term button press - open terminal window."""
214 from openhcs.textual_tui.windows.terminal_window import TerminalWindow
215 from textual.css.query import NoMatches
217 # Try to find existing terminal window - if it doesn't exist, create new one
218 try:
219 window = self.app.query_one(TerminalWindow)
220 # Window exists, just open it
221 window.open_state = True
222 except NoMatches:
223 # Expected case: window doesn't exist yet, create new one
224 window = TerminalWindow()
225 await self.app.mount(window)
226 window.open_state = True
230 async def _handle_help(self) -> None:
231 """Handle help button press."""
232 from openhcs.textual_tui.windows import HelpWindow
233 from textual.css.query import NoMatches
235 # Try to find existing help window - if it doesn't exist, create new one
236 try:
237 window = self.app.query_one(HelpWindow)
238 # Window exists, just open it
239 window.open_state = True
240 except NoMatches:
241 # Expected case: window doesn't exist yet, create new one
242 window = HelpWindow()
243 await self.app.mount(window)
244 window.open_state = True
246 async def _handle_toggle_monitor(self) -> None:
247 """Handle toggle monitor button press."""
248 try:
249 # Find the system monitor widget
250 main_content = self.app.query_one("MainContent")
251 system_monitor = main_content.query_one("SystemMonitorTextual")
253 # Toggle monitoring
254 system_monitor.toggle_monitoring()
256 # Update button text
257 toggle_btn = self.query_one("#toggle_monitor", ButtonStatic)
258 if system_monitor.is_monitoring:
259 toggle_btn.content = "Monitor"
260 else:
261 toggle_btn.content = "Monitor"
263 except Exception as e:
264 logger.error(f"Failed to toggle monitoring: {e}")
266 async def _handle_quit(self) -> None:
267 """Handle quit button press."""
268 self.app.action_quit()
270 def _propagate_global_config_to_orchestrators(self, new_config) -> None:
271 """Propagate global config changes to all existing orchestrators and plate manager."""
272 try:
273 # Find the plate manager widget
274 main_content = self.app.query_one("MainContent")
275 plate_manager = main_content.query_one("PlateManagerWidget")
277 # CRITICAL: Update plate manager's global config reference
278 # This ensures future orchestrators and subprocesses use the latest config
279 plate_manager.global_config = new_config
280 logger.info("Updated plate manager global config reference")
282 # Also update pipeline editor if it exists (though it should use app.global_config)
283 try:
284 pipeline_editor = main_content.query_one("PipelineEditorWidget")
285 # Pipeline editor is designed to use self.app.global_config, but let's be safe
286 logger.debug("Pipeline editor will automatically use updated app.global_config")
287 except Exception:
288 # Pipeline editor might not exist or be mounted
289 pass
291 # Update all orchestrators that don't have plate-specific configs
292 updated_count = 0
293 for plate_path, orchestrator in plate_manager.orchestrators.items():
294 # Only update if this plate doesn't have a plate-specific config override
295 if plate_path not in plate_manager.plate_configs:
296 # Use the async method to apply the new config
297 import asyncio
298 asyncio.create_task(orchestrator.apply_new_global_config(new_config))
299 updated_count += 1
301 if updated_count > 0:
302 logger.info(f"Applied global config changes to {updated_count} orchestrators")
303 else:
304 logger.info("No orchestrators updated (all have plate-specific configs)")
306 except Exception as e:
307 logger.error(f"Failed to propagate global config to orchestrators: {e}")
308 # Don't fail the config update if propagation fails
309 pass
311 def _save_config_to_cache(self, config) -> None:
312 """Save config to cache asynchronously."""
313 async def _async_save():
314 from openhcs.textual_tui.services.config_cache_adapter import save_global_config_to_cache
315 try:
316 success = await save_global_config_to_cache(config)
317 if success:
318 logger.info("Global config saved to cache for future sessions")
319 else:
320 logger.warning("Failed to save global config to cache")
321 except Exception as e:
322 logger.error(f"Error saving global config to cache: {e}")
324 # Schedule the async save operation
325 import asyncio
326 asyncio.create_task(_async_save())