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

1""" 

2Pattern File Service - Async-safe file I/O operations for function patterns. 

3 

4This service handles loading/saving .func files and external editor integration 

5with proper FileManager abstraction and async safety. 

6""" 

7 

8import asyncio 

9import dill as pickle 

10import logging 

11from pathlib import Path 

12from typing import Union, List, Dict, Optional, Any 

13 

14from openhcs.textual_tui.services.external_editor_service import ExternalEditorService 

15 

16logger = logging.getLogger(__name__) 

17 

18 

19class PatternFileService: 

20 """ 

21 Async-safe file I/O operations for function patterns. 

22  

23 Handles .func file loading/saving with proper FileManager abstraction 

24 and external editor integration. 

25 """ 

26 

27 def __init__(self, state: Any): 

28 """ 

29 Initialize the pattern file service. 

30  

31 Args: 

32 state: TUIState instance for external editor integration 

33 """ 

34 self.state = state 

35 self.external_editor_service = ExternalEditorService(state) 

36 

37 async def load_pattern_from_file(self, file_path: Path) -> Union[List, Dict]: 

38 """ 

39 Load and validate .func files with async safety. 

40  

41 Uses run_in_executor to prevent event loop deadlocks. 

42  

43 Args: 

44 file_path: Path to .func file 

45  

46 Returns: 

47 Loaded pattern (List or Dict) 

48  

49 Raises: 

50 FileNotFoundError: If file doesn't exist 

51 ValueError: If file content is invalid 

52 Exception: For other loading errors 

53 """ 

54 def _sync_load_pattern(path: Path) -> Union[List, Dict]: 

55 """Synchronous pattern loading for executor.""" 

56 if not path.exists(): 

57 raise FileNotFoundError(f"Pattern file not found: {path}") 

58 

59 if not path.is_file(): 

60 raise ValueError(f"Path is not a file: {path}") 

61 

62 try: 

63 with open(path, "rb") as f: 

64 pattern = pickle.load(f) 

65 

66 # Basic validation 

67 if not isinstance(pattern, (list, dict)): 

68 raise ValueError(f"Invalid pattern type: {type(pattern)}. Expected list or dict.") 

69 

70 return pattern 

71 

72 except pickle.PickleError as e: 

73 raise ValueError(f"Failed to unpickle pattern file: {e}") 

74 except Exception as e: 

75 raise Exception(f"Failed to load pattern file: {e}") 

76 

77 # Use asyncio.get_running_loop() instead of deprecated get_event_loop() 

78 loop = asyncio.get_running_loop() 

79 return await loop.run_in_executor(None, _sync_load_pattern, file_path) 

80 

81 async def save_pattern_to_file(self, pattern: Union[List, Dict], file_path: Path) -> None: 

82 """ 

83 Save patterns with pickle using async safety. 

84  

85 Uses run_in_executor to prevent event loop deadlocks. 

86  

87 Args: 

88 pattern: Pattern to save (List or Dict) 

89 file_path: Path to save to 

90  

91 Raises: 

92 ValueError: If pattern is invalid 

93 Exception: For saving errors 

94 """ 

95 def _sync_save_pattern(pattern_data: Union[List, Dict], path: Path) -> None: 

96 """Synchronous pattern saving for executor.""" 

97 # Basic validation 

98 if not isinstance(pattern_data, (list, dict)): 

99 raise ValueError(f"Invalid pattern type: {type(pattern_data)}. Expected list or dict.") 

100 

101 # Ensure parent directory exists 

102 path.parent.mkdir(parents=True, exist_ok=True) 

103 

104 try: 

105 with open(path, "wb") as f: 

106 pickle.dump(pattern_data, f) 

107 

108 except Exception as e: 

109 raise Exception(f"Failed to save pattern file: {e}") 

110 

111 loop = asyncio.get_running_loop() 

112 await loop.run_in_executor(None, _sync_save_pattern, pattern, file_path) 

113 

114 async def edit_pattern_externally(self, pattern: Union[List, Dict]) -> tuple[bool, Union[List, Dict], Optional[str]]: 

115 """ 

116 Edit pattern in external editor (Vim) via ExternalEditorService. 

117  

118 Args: 

119 pattern: Pattern to edit 

120  

121 Returns: 

122 Tuple of (success, new_pattern, error_message) 

123 """ 

124 try: 

125 # Format pattern for external editing 

126 initial_content = f"pattern = {repr(pattern)}" 

127 

128 # Use existing ExternalEditorService 

129 success, new_pattern, error_message = await self.external_editor_service.edit_pattern_in_external_editor(initial_content) 

130 

131 return success, new_pattern, error_message 

132 

133 except Exception as e: 

134 logger.error(f"External editor integration failed: {e}") 

135 return False, pattern, f"External editor failed: {e}" 

136 

137 async def validate_pattern_file(self, file_path: Path) -> tuple[bool, Optional[str]]: 

138 """ 

139 Validate .func file without loading it completely. 

140  

141 Args: 

142 file_path: Path to validate 

143  

144 Returns: 

145 Tuple of (is_valid, error_message) 

146 """ 

147 def _sync_validate_file(path: Path) -> tuple[bool, Optional[str]]: 

148 """Synchronous file validation for executor.""" 

149 if not path.exists(): 

150 return False, f"File does not exist: {path}" 

151 

152 if not path.is_file(): 

153 return False, f"Path is not a file: {path}" 

154 

155 if not path.suffix == '.func': 

156 return False, f"File does not have .func extension: {path}" 

157 

158 try: 

159 # Try to load just the header to check if it's a valid pickle 

160 with open(path, "rb") as f: 

161 # Read first few bytes to check pickle format 

162 header = f.read(10) 

163 if not header.startswith(b'\x80'): # Pickle protocol marker 

164 return False, "File is not a valid pickle file" 

165 

166 return True, None 

167 

168 except Exception as e: 

169 return False, f"File validation failed: {e}" 

170 

171 loop = asyncio.get_running_loop() 

172 return await loop.run_in_executor(None, _sync_validate_file, file_path) 

173 

174 def get_default_save_path(self, base_name: str = "pattern") -> str: 

175 """ 

176 Get default save path for .func files. 

177  

178 Args: 

179 base_name: Base filename without extension 

180  

181 Returns: 

182 Default save path string 

183 """ 

184 return f"{base_name}.func" 

185 

186 def ensure_func_extension(self, file_path: str) -> str: 

187 """ 

188 Ensure file path has .func extension. 

189  

190 Args: 

191 file_path: Original file path 

192  

193 Returns: 

194 File path with .func extension 

195 """ 

196 path = Path(file_path) 

197 if path.suffix != '.func': 

198 return str(path.with_suffix('.func')) 

199 return file_path 

200 

201 async def backup_pattern_file(self, file_path: Path) -> Optional[Path]: 

202 """ 

203 Create backup of existing pattern file before overwriting. 

204  

205 Args: 

206 file_path: Original file path 

207  

208 Returns: 

209 Backup file path if created, None if no backup needed 

210 """ 

211 if not file_path.exists(): 

212 return None 

213 

214 def _sync_backup_file(original_path: Path) -> Path: 

215 """Synchronous file backup for executor.""" 

216 backup_path = original_path.with_suffix(f"{original_path.suffix}.backup") 

217 

218 # If backup already exists, add timestamp 

219 if backup_path.exists(): 

220 import time 

221 timestamp = int(time.time()) 

222 backup_path = original_path.with_suffix(f"{original_path.suffix}.backup.{timestamp}") 

223 

224 # Copy file 

225 import shutil 

226 shutil.copy2(original_path, backup_path) 

227 return backup_path 

228 

229 try: 

230 loop = asyncio.get_running_loop() 

231 backup_path = await loop.run_in_executor(None, _sync_backup_file, file_path) 

232 return backup_path 

233 except Exception as e: 

234 logger.warning(f"Failed to create backup for {file_path}: {e}") 

235 return None