Coverage for openhcs/textual_tui/windows/multi_orchestrator_config_window.py: 0.0%

1"""Multi-orchestrator configuration window for OpenHCS Textual TUI.""" 

2 

3from typing import Type, Any, Callable, Optional, List, Dict 

4from textual.app import ComposeResult 

5from textual.widgets import Button, Static 

6from textual.containers import Container, Horizontal, ScrollableContainer 

7import dataclasses 

8 

9from openhcs.textual_tui.windows.base_window import BaseOpenHCSWindow 

10from openhcs.textual_tui.widgets.config_form import ConfigFormWidget 

11from openhcs.core.config import GlobalPipelineConfig 

12from openhcs.textual_tui.widgets.shared.signature_analyzer import SignatureAnalyzer 

13 

14 

15class MultiOrchestratorConfigWindow(BaseOpenHCSWindow): 

16 """Multi-orchestrator configuration window using textual-window system.""" 

17 

18 DEFAULT_CSS = """ 

19 MultiOrchestratorConfigWindow { 

20 width: 80; height: 35; 

21 min-width: 60; min-height: 20; 

22 } 

23  

24 #config_form { 

25 height: 1fr; 

26 overflow-y: auto; 

27 } 

28  

29 #status_bar { 

30 color: $text-muted; 

31 text-align: center; 

32 height: 1; 

33 margin: 1 0; 

34 } 

35 """ 

36 

37 def __init__(self, 

38 orchestrators: List[Any], 

39 on_save_callback: Optional[Callable] = None, 

40 **kwargs): 

41 """ 

42 Initialize multi-orchestrator config window. 

43 

44 Args: 

45 orchestrators: List of orchestrators to configure 

46 on_save_callback: Function to call when config is saved 

47 """ 

48 super().__init__( 

49 window_id="multi_orchestrator_config", 

50 title=f"Multi-Orchestrator Configuration ({len(orchestrators)} orchestrators)", 

51 mode="temporary", 

52 **kwargs 

53 ) 

54 self.orchestrators = orchestrators 

55 self.on_save_callback = on_save_callback 

56 

57 # Analyze configs across orchestrators 

58 self.config_analysis = self._analyze_orchestrator_configs() 

59 

60 # Create base config for form (use first orchestrator's config as template) 

61 self.base_config = orchestrators[0].global_config if orchestrators else None 

62 

63 # Create the form widget using unified parameter analysis 

64 if self.base_config: 

65 self.config_form = ConfigFormWidget.from_dataclass( 

66 GlobalPipelineConfig, 

67 self.base_config 

68 ) 

69 # Attach config analysis to form manager for different values handling 

70 if hasattr(self.config_form, 'form_manager'): 

71 self.config_form.form_manager.config_analysis = self.config_analysis 

72 

73 def _analyze_orchestrator_configs(self) -> Dict[str, Any]: 

74 """Analyze configs across orchestrators to find same/different values.""" 

75 if not self.orchestrators: 

76 return {} 

77 

78 # Get parameter info for defaults 

79 param_info = SignatureAnalyzer.analyze(GlobalPipelineConfig) 

80 config_analysis = {} 

81 

82 # Analyze each field in GlobalPipelineConfig 

83 for field in dataclasses.fields(GlobalPipelineConfig): 

84 field_name = field.name 

85 

86 # Get values from all orchestrators 

87 values = [] 

88 for orch in self.orchestrators: 

89 try: 

90 value = getattr(orch.global_config, field_name) 

91 values.append(value) 

92 except AttributeError: 

93 # Field doesn't exist in this config, skip 

94 continue 

95 

96 if not values: 

97 continue 

98 

99 # Get default value from parameter info 

100 param_details = param_info.get(field_name) 

101 default_value = param_details.default_value if param_details else None 

102 

103 # Check if all values are the same 

104 if all(self._values_equal(v, values[0]) for v in values): 

105 config_analysis[field_name] = { 

106 "type": "same", 

107 "value": values[0], 

108 "default": default_value 

109 } 

110 else: 

111 config_analysis[field_name] = { 

112 "type": "different", 

113 "values": values, 

114 "default": default_value 

115 } 

116 

117 return config_analysis 

118 

119 def _values_equal(self, val1: Any, val2: Any) -> bool: 

120 """Check if two values are equal, handling dataclasses and complex types.""" 

121 # Handle dataclass comparison 

122 if dataclasses.is_dataclass(val1) and dataclasses.is_dataclass(val2): 

123 return dataclasses.asdict(val1) == dataclasses.asdict(val2) 

124 

125 # Handle enum comparison 

126 if hasattr(val1, 'value') and hasattr(val2, 'value'): 

127 return val1.value == val2.value 

128 

129 # Standard comparison 

130 return val1 == val2 

131 

132 def compose(self) -> ComposeResult: 

133 """Compose the config window content.""" 

134 # Status bar 

135 yield Static("Ready", id="status_bar") 

136 

137 # Scrollable form area 

138 with ScrollableContainer(id="config_form"): 

139 if hasattr(self, 'config_form'): 

140 yield self.config_form 

141 else: 

142 yield Static("No configuration available") 

143 

144 # Buttons 

145 with Horizontal(classes="dialog-buttons"): 

146 yield Button("Save", id="save", compact=True) 

147 yield Button("Cancel", id="cancel", compact=True) 

148 

149 def on_button_pressed(self, event: Button.Pressed) -> None: 

150 """Handle button presses.""" 

151 if event.button.id == "save": 

152 self._handle_save() 

153 elif event.button.id == "cancel": 

154 self.close_window() 

155 

156 def _handle_save(self): 

157 """Handle save button - apply config to all orchestrators.""" 

158 if not hasattr(self, 'config_form'): 

159 self.close_window() 

160 return 

161 

162 try: 

163 # Get form values 

164 form_values = self.config_form.get_config_values() 

165 

166 # Create new config instance 

167 new_config = GlobalPipelineConfig(**form_values) 

168 

169 # Update thread-local storage for MaterializationPathConfig defaults 

170 from openhcs.core.config import set_current_global_config, GlobalPipelineConfig 

171 set_current_global_config(GlobalPipelineConfig, new_config) 

172 

173 # Apply to all orchestrators 

174 import asyncio 

175 async def apply_to_all(): 

176 for orchestrator in self.orchestrators: 

177 await orchestrator.apply_new_global_config(new_config) 

178 

179 # Run the async operation 

180 asyncio.create_task(apply_to_all()) 

181 

182 # Call the callback if provided 

183 if self.on_save_callback: 

184 self.on_save_callback(new_config, len(self.orchestrators)) 

185 

186 self.close_window() 

187 

188 except Exception as e: 

189 # Update status bar with error 

190 status_bar = self.query_one("#status_bar", Static) 

191 status_bar.update(f"Error: {e}") 

192 

193 

194async def show_multi_orchestrator_config(app, orchestrators: List[Any], 

195 on_save_callback: Optional[Callable] = None): 

196 """ 

197 Show multi-orchestrator config window. 

198  

199 Args: 

200 app: The Textual app instance 

201 orchestrators: List of orchestrators to configure 

202 on_save_callback: Optional callback when config is saved 

203 """ 

204 from textual.css.query import NoMatches 

205 

206 # Try to find existing window 

207 try: 

208 window = app.query_one(MultiOrchestratorConfigWindow) 

209 # Window exists, just open it 

210 window.open_state = True 

211 except NoMatches: 

212 # Create new window 

213 window = MultiOrchestratorConfigWindow( 

214 orchestrators=orchestrators, 

215 on_save_callback=on_save_callback 

216 ) 

217 await app.mount(window) 

218 window.open_state = True 

219 

220 return window