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 copy 

9import logging 

10from typing import Any, List, Optional 

11 

12from prompt_toolkit.application import get_app 

13from prompt_toolkit.widgets import Dialog, Button 

14from prompt_toolkit.layout import HSplit 

15 

16from openhcs.constants.constants import Backend 

17from openhcs.core.pipeline import Pipeline 

18from openhcs.core.steps.function_step import FunctionStep 

19# DualEditorPane injected via constructor to break circular dependency 

20# Global error handling will catch all exceptions automatically 

21 

22logger = logging.getLogger(__name__) 

23 

24 

25class VisualProgrammingDialogService: 

26 """ 

27 Service for managing visual programming dialogs. 

28 

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

30 Keeps dialog concerns separate from pipeline management. 

31 Uses dependency injection to avoid circular imports. 

32 """ 

33 

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

35 """ 

36 Initialize the visual programming dialog service. 

37 

38 Args: 

39 state: TUI state object 

40 context: Processing context 

41 dual_editor_pane_class: DualEditorPane class injected to break circular dependency 

42 """ 

43 self.state = state 

44 self.context = context 

45 self.dual_editor_pane_class = dual_editor_pane_class 

46 

47 # Dialog state management 

48 self.current_dialog = None 

49 self.current_dialog_future = None 

50 

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

52 """ 

53 Show visual programming dialog for adding a new step. 

54  

55 Args: 

56 target_pipelines: List of pipelines to add the step to 

57  

58 Returns: 

59 Created FunctionStep if successful, None if cancelled 

60 """ 

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

62 empty_step = FunctionStep( 

63 func=None, # Required positional parameter 

64 name="New Step", 

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

66 group_by="" 

67 ) 

68 

69 # Create DualEditorPane for visual programming with dialog callbacks 

70 dual_editor = self.dual_editor_pane_class( 

71 state=self.state, 

72 func_step=empty_step, 

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

74 on_cancel=lambda: self._handle_cancel_and_close() 

75 ) 

76 

77 # Create and show dialog 

78 result = await self._show_dialog( 

79 title="Visual Programming - Add Step", 

80 dual_editor=dual_editor, 

81 ok_handler=lambda: self._handle_add_step_ok(dual_editor) 

82 ) 

83 

84 return result 

85 

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

87 """ 

88 Show visual programming dialog for editing an existing step. 

89  

90 Args: 

91 target_step: The step to edit 

92  

93 Returns: 

94 Edited FunctionStep if successful, None if cancelled 

95 """ 

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

97 dual_editor = self.dual_editor_pane_class( 

98 state=self.state, 

99 func_step=target_step, 

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

101 on_cancel=lambda: self._handle_cancel_and_close() 

102 ) 

103 

104 # Create and show dialog 

105 result = await self._show_dialog( 

106 title="Visual Programming - Edit Step", 

107 dual_editor=dual_editor, 

108 ok_handler=lambda: self._handle_edit_step_ok(dual_editor) 

109 ) 

110 

111 return result 

112 

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

114 """ 

115 Show a visual programming dialog with the given DualEditorPane. 

116  

117 Args: 

118 title: Dialog title 

119 dual_editor: The DualEditorPane instance 

120 ok_handler: Handler for OK button 

121  

122 Returns: 

123 Result from OK handler or None if cancelled 

124 """ 

125 # Create dialog result future 

126 self.current_dialog_future = asyncio.Future() 

127 

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

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

130 from prompt_toolkit.layout.dimension import Dimension 

131 

132 # Calculate minimum width for Function Pattern Editor buttons: 

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

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

135 min_dialog_width = 85 

136 

137 dialog = Dialog( 

138 title=title, 

139 body=dual_editor.container, 

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

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

142 modal=True 

143 ) 

144 

145 # Store dialog reference 

146 self.current_dialog = dialog 

147 

148 # Show dialog (async) 

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

150 

151 # Wait for dialog completion 

152 result = await self.current_dialog_future 

153 

154 # Cleanup 

155 self.current_dialog = None 

156 self.current_dialog_future = None 

157 

158 return result 

159 

160 def _handle_ok(self, ok_handler): 

161 """Handle OK button click.""" 

162 try: 

163 result = ok_handler() 

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

165 self.current_dialog_future.set_result(result) 

166 except Exception as e: 

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

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

169 self.current_dialog_future.set_result(None) 

170 

171 def _handle_cancel(self): 

172 """Handle Cancel button click.""" 

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

174 self.current_dialog_future.set_result(None) 

175 

176 def _handle_save_and_close(self, step: FunctionStep): 

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

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

179 self.current_dialog_future.set_result(step) 

180 

181 def _handle_cancel_and_close(self): 

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

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

184 self.current_dialog_future.set_result(None) 

185 

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

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

188 created_step = dual_editor.get_created_step() 

189 return created_step 

190 

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

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

193 edited_step = dual_editor.get_created_step() 

194 return edited_step