Coverage for openhcs/textual_tui/services/visual_programming_dialog_service.py: 0.0%

1""" 

2Visual Programming Dialog Service for OpenHCS TUI. 

3 

4Handles creation and management of visual programming dialogs using DualEditorPane. 

5Separates dialog concerns from pipeline management. 

6""" 

7import asyncio 

8import logging 

9from typing import Any, List, Optional 

10 

11from prompt_toolkit.widgets import Dialog 

12 

13from openhcs.core.pipeline import Pipeline 

14from openhcs.core.steps.function_step import FunctionStep 

15# DualEditorPane injected via constructor to break circular dependency 

16# Global error handling will catch all exceptions automatically 

17 

18logger = logging.getLogger(__name__) 

19 

20 

21class VisualProgrammingDialogService: 

22 """ 

23 Service for managing visual programming dialogs. 

24 

25 Handles dialog creation, lifecycle, and integration with DualEditorPane. 

26 Keeps dialog concerns separate from pipeline management. 

27 Uses dependency injection to avoid circular imports. 

28 """ 

29 

30 def __init__(self, state: Any, context: Any, dual_editor_pane_class: type): 

31 """ 

32 Initialize the visual programming dialog service. 

33 

34 Args: 

35 state: TUI state object 

36 context: Processing context 

37 dual_editor_pane_class: DualEditorPane class injected to break circular dependency 

38 """ 

39 self.state = state 

40 self.context = context 

41 self.dual_editor_pane_class = dual_editor_pane_class 

42 

43 # Dialog state management 

44 self.current_dialog = None 

45 self.current_dialog_future = None 

46 

47 async def show_add_step_dialog(self, target_pipelines: List[Pipeline]) -> Optional[FunctionStep]: 

48 """ 

49 Show visual programming dialog for adding a new step. 

50  

51 Args: 

52 target_pipelines: List of pipelines to add the step to 

53  

54 Returns: 

55 Created FunctionStep if successful, None if cancelled 

56 """ 

57 # Create empty FunctionStep for new step creation - BACKEND API COMPLIANT 

58 empty_step = FunctionStep( 

59 func=None, # Required positional parameter 

60 name="New Step", 

61 # Let variable_components use FunctionStep's default [VariableComponents.SITE] 

62 group_by="" 

63 ) 

64 

65 # Create DualEditorPane for visual programming with dialog callbacks 

66 dual_editor = self.dual_editor_pane_class( 

67 state=self.state, 

68 func_step=empty_step, 

69 on_save=lambda step: self._handle_save_and_close(step), 

70 on_cancel=lambda: self._handle_cancel_and_close() 

71 ) 

72 

73 # Create and show dialog 

74 result = await self._show_dialog( 

75 title="Visual Programming - Add Step", 

76 dual_editor=dual_editor, 

77 ok_handler=lambda: self._handle_add_step_ok(dual_editor) 

78 ) 

79 

80 return result 

81 

82 async def show_edit_step_dialog(self, target_step: FunctionStep) -> Optional[FunctionStep]: 

83 """ 

84 Show visual programming dialog for editing an existing step. 

85  

86 Args: 

87 target_step: The step to edit 

88  

89 Returns: 

90 Edited FunctionStep if successful, None if cancelled 

91 """ 

92 # Create DualEditorPane for visual programming with existing step and dialog callbacks 

93 dual_editor = self.dual_editor_pane_class( 

94 state=self.state, 

95 func_step=target_step, 

96 on_save=lambda step: self._handle_save_and_close(step), 

97 on_cancel=lambda: self._handle_cancel_and_close() 

98 ) 

99 

100 # Create and show dialog 

101 result = await self._show_dialog( 

102 title="Visual Programming - Edit Step", 

103 dual_editor=dual_editor, 

104 ok_handler=lambda: self._handle_edit_step_ok(dual_editor) 

105 ) 

106 

107 return result 

108 

109 async def _show_dialog(self, title: str, dual_editor: Any, ok_handler) -> Optional[FunctionStep]: 

110 """ 

111 Show a visual programming dialog with the given DualEditorPane. 

112  

113 Args: 

114 title: Dialog title 

115 dual_editor: The DualEditorPane instance 

116 ok_handler: Handler for OK button 

117  

118 Returns: 

119 Result from OK handler or None if cancelled 

120 """ 

121 # Create dialog result future 

122 self.current_dialog_future = asyncio.Future() 

123 

124 # Create dialog WITHOUT buttons - DualEditorPane handles its own save/cancel 

125 # Set minimum width to accommodate Function Pattern Editor button row 

126 from prompt_toolkit.layout.dimension import Dimension 

127 

128 # Calculate minimum width for Function Pattern Editor buttons: 

129 # "Function Pattern Editor" (title) + "Add Function" + "Load .func" + "Save .func As" + "Edit in Vim" 

130 # ≈ 23 + 15 + 12 + 15 + 12 + padding ≈ 85 characters 

131 min_dialog_width = 85 

132 

133 dialog = Dialog( 

134 title=title, 

135 body=dual_editor.container, 

136 buttons=[], # No buttons - DualEditorPane has its own Save/Cancel 

137 width=Dimension(min=min_dialog_width), # Ensure minimum width for button row 

138 modal=True 

139 ) 

140 

141 # Store dialog reference 

142 self.current_dialog = dialog 

143 

144 # Show dialog (async) 

145 await self.state.show_dialog(dialog, result_future=self.current_dialog_future) 

146 

147 # Wait for dialog completion 

148 result = await self.current_dialog_future 

149 

150 # Cleanup 

151 self.current_dialog = None 

152 self.current_dialog_future = None 

153 

154 return result 

155 

156 def _handle_ok(self, ok_handler): 

157 """Handle OK button click.""" 

158 try: 

159 result = ok_handler() 

160 if self.current_dialog_future and not self.current_dialog_future.done(): 

161 self.current_dialog_future.set_result(result) 

162 except Exception as e: 

163 logger.error(f"Error in OK handler: {e}") 

164 if self.current_dialog_future and not self.current_dialog_future.done(): 

165 self.current_dialog_future.set_result(None) 

166 

167 def _handle_cancel(self): 

168 """Handle Cancel button click.""" 

169 if self.current_dialog_future and not self.current_dialog_future.done(): 

170 self.current_dialog_future.set_result(None) 

171 

172 def _handle_save_and_close(self, step: FunctionStep): 

173 """Handle save from DualEditorPane and close dialog.""" 

174 if self.current_dialog_future and not self.current_dialog_future.done(): 

175 self.current_dialog_future.set_result(step) 

176 

177 def _handle_cancel_and_close(self): 

178 """Handle cancel from DualEditorPane and close dialog.""" 

179 if self.current_dialog_future and not self.current_dialog_future.done(): 

180 self.current_dialog_future.set_result(None) 

181 

182 def _handle_add_step_ok(self, dual_editor: Any) -> Optional[FunctionStep]: 

183 """Handle OK button for add step dialog.""" 

184 created_step = dual_editor.get_created_step() 

185 return created_step 

186 

187 def _handle_edit_step_ok(self, dual_editor: Any) -> Optional[FunctionStep]: 

188 """Handle OK button for edit step dialog.""" 

189 edited_step = dual_editor.get_created_step() 

190 return edited_step