Coverage for main/ui/import_page.py: 95%

1import gc 

2import os 

3 

4from PyQt6.QtCore import Qt, QCoreApplication 

5from PyQt6.QtGui import QFont 

6from PyQt6.QtWidgets import ( 

7 QHBoxLayout, QLabel, QFileDialog, QListView, QTreeView, QMessageBox, 

8 QProgressDialog, QApplication 

9) 

10 

11from threads.import_thread import ImportThread 

12from ui.patient_selection_page import PatientSelectionPage 

13from page import Page 

14from logger import get_logger 

15 

16log = get_logger() 

17 

18 

19class ImportPage(Page): 

20 """ 

21 GUI page that allows importing patient data directories. 

22 

23 This class provides both drag-and-drop and dialog-based import options. 

24 It starts a background thread for each import operation and displays 

25 progress with cancellable dialogs. 

26 

27 Attributes 

28 ---------- 

29 context : dict 

30 Shared context across application pages (contains settings, paths, signals, etc.). 

31 next_page : Page | None 

32 Reference to the next page in the workflow. 

33 workspace_path : str 

34 Root path where imported data is stored. 

35 threads : list[ImportThread] 

36 Active import threads currently running. 

37 progress_dialogs : list[QProgressDialog] 

38 Dialogs tracking progress of ongoing imports. 

39 """ 

40 

41 def __init__(self, context=None): 

42 """ 

43 Initialize the ImportPage with the given context. 

44 

45 Parameters 

46 ---------- 

47 context : dict, optional 

48 Application context including 'workspace_path', 'main_window', and signals. 

49 """ 

50 super().__init__() 

51 

52 self.context = context 

53 self.next_page = None 

54 self.workspace_path = context["workspace_path"] 

55 self.threads = [] 

56 self.progress_dialogs = [] 

57 

58 # Enable drag & drop and apply a dashed border 

59 self.setAcceptDrops(True) 

60 self.setEnabled(True) 

61 self.setStyleSheet("border: 2px dashed gray;") 

62 

63 # Central label 

64 frame_layout = QHBoxLayout(self) 

65 self.drop_label = QLabel("Import or select patients' data") 

66 self.drop_label.setStyleSheet("font-size: 18px; font-weight: bold;") 

67 font = QApplication.font() 

68 font.setPointSize(14) 

69 self.drop_label.setFont(font) 

70 self.drop_label.setAlignment(Qt.AlignmentFlag.AlignCenter) 

71 frame_layout.addWidget(self.drop_label) 

72 

73 # Handle localization updates 

74 self._translate_ui() 

75 if context and "language_changed" in context: 

76 context["language_changed"].connect(self._translate_ui) 

77 

78 # ---------------------------- 

79 # Navigation 

80 # ---------------------------- 

81 

82 def is_ready_to_advance(self): 

83 """ 

84 Check if the page can advance to the next step. 

85 

86 Returns 

87 ------- 

88 bool 

89 True if the workspace contains at least one visible directory or link. 

90 """ 

91 has_content = any( 

92 os.path.isdir(os.path.join(self.workspace_path, name)) or 

93 os.path.islink(os.path.join(self.workspace_path, name)) 

94 for name in os.listdir(self.workspace_path) 

95 if not name.startswith(".") 

96 ) 

97 return has_content 

98 

99 def is_ready_to_go_back(self): 

100 """ 

101 Check if the user can navigate back to the previous page. 

102 

103 Returns 

104 ------- 

105 bool 

106 Always returns False (first step in workflow). 

107 """ 

108 return False 

109 

110 def next(self, context): 

111 """ 

112 Transition to the next page (PatientSelectionPage). 

113 

114 Returns 

115 ------- 

116 Page 

117 The newly created or existing PatientSelectionPage. 

118 """ 

119 if self.next_page: 

120 self.next_page.on_enter() 

121 return self.next_page 

122 else: 

123 self.next_page = PatientSelectionPage(context, self) 

124 self.context["history"].append(self.next_page) 

125 return self.next_page 

126 

127 def back(self): 

128 """This page has no previous step.""" 

129 return False 

130 

131 # ---------------------------- 

132 # Drag & Drop Import 

133 # ---------------------------- 

134 

135 def dragEnterEvent(self, event): 

136 """Accept drag if the event contains file URLs.""" 

137 if event.mimeData().hasUrls(): 

138 event.acceptProposedAction() 

139 

140 def dropEvent(self, event): 

141 """Handle dropped folders.""" 

142 urls = event.mimeData().urls() 

143 if urls: 

144 file_path = [url.toLocalFile() for url in urls] 

145 self._handle_import(file_path) 

146 

147 # ---------------------------- 

148 # Folder Selection Dialog 

149 # ---------------------------- 

150 

151 def mousePressEvent(self, event): 

152 """Open a folder dialog on left-click.""" 

153 if event.button() == Qt.MouseButton.LeftButton: 

154 self.open_folder_dialog() 

155 

156 def open_folder_dialog(self): 

157 """Open dialog to select one or more folders for import.""" 

158 dialog = QFileDialog( 

159 self.context["main_window"], 

160 QCoreApplication.translate("ImportPage", "Select Folder") 

161 ) 

162 dialog.setFileMode(QFileDialog.FileMode.Directory) 

163 dialog.setOption(QFileDialog.Option.DontUseNativeDialog, True) 

164 dialog.setOption(QFileDialog.Option.ReadOnly, True) 

165 dialog.setDirectory(os.path.expanduser("~")) 

166 

167 # Allow multi-selection in both list and tree views 

168 for view in dialog.findChildren((QListView, QTreeView)): 

169 view.setSelectionMode(view.SelectionMode.ExtendedSelection) 

170 

171 if dialog.exec(): 

172 folders = [os.path.abspath(path) for path in dialog.selectedFiles() if os.path.isdir(path)] 

173 # Avoid nested duplicates (e.g., selecting parent + child folder) 

174 unique_folders = [ 

175 f for f in folders 

176 if not any(f != other and other.startswith(f + os.sep) for other in folders) 

177 ] 

178 self._handle_import(unique_folders) 

179 

180 # ---------------------------- 

181 # Import Handling 

182 # ---------------------------- 

183 

184 def _handle_import(self, folders_path): 

185 self.progress_dialogs.append(QProgressDialog(QCoreApplication.translate("ImportFrame", "Importing files..."), QCoreApplication.translate("ImportFrame", "Cancel"), 

186 0, 100, self.context["main_window"])) 

187 self.progress_dialogs[-1].setWindowModality(Qt.WindowModality.NonModal) 

188 self.progress_dialogs[-1].setMinimumDuration(0) 

189 

190 # Start importing thread 

191 self.threads.append(ImportThread(self.context, folders_path, self.workspace_path)) 

192 self.threads[-1].finished.connect(self.on_import_finished) 

193 self.threads[-1].error.connect(self.on_import_error) 

194 self.threads[-1].progress.connect(self.progress_dialogs[-1].setValue) 

195 self.threads[-1].start() 

196 

197 self.progress_dialogs[-1].canceled.connect(self.on_import_canceled) 

198 

199 # ---------------------------- 

200 # Thread Callbacks 

201 # ---------------------------- 

202 

203 def on_import_error(self, error): 

204 """Handle file import errors from background threads.""" 

205 thread = self.sender() 

206 if thread not in self.threads: 

207 log.warning(f"Ignored error from already removed thread: {error}") 

208 return 

209 

210 index = self.threads.index(thread) 

211 self.progress_dialogs[index].close() 

212 QMessageBox.critical( 

213 self.context["main_window"], 

214 QCoreApplication.translate("ImportPage", "Error Importing Files"), 

215 QCoreApplication.translate("ImportPage", "Failed to import files:\n{0}").format(error), 

216 ) 

217 log.error(f"Error Importing Files: {error}") 

218 

219 def on_import_finished(self): 

220 """Cleanup after an import finishes successfully.""" 

221 thread = self.sender() 

222 if thread not in self.threads: 

223 log.warning("Ignored 'finished' signal from an already removed thread.") 

224 return 

225 

226 index = self.threads.index(thread) 

227 self.threads.remove(thread) 

228 self.progress_dialogs[index].close() 

229 

230 if self.context and "update_main_buttons" in self.context: 

231 self.context["update_main_buttons"]() 

232 log.info("Import completed.") 

233 

234 def on_import_canceled(self): 

235 """Handle user-canceled import operations.""" 

236 dialog = self.sender() 

237 index = self.progress_dialogs.index(dialog) 

238 

239 if len(self.threads) > index: 

240 thread = self.threads[index] 

241 thread.cancel() 

242 thread.terminate() 

243 thread.wait() 

244 self.threads.remove(thread) 

245 

246 def closeEvent(self, event): 

247 """Clean up threads and dialogs on application exit.""" 

248 for dialog in self.progress_dialogs: 

249 dialog.destroy() 

250 self.progress_dialogs.remove(dialog) 

251 for thread in self.threads: 

252 thread.cancel() 

253 self.threads.remove(thread) 

254 gc.collect() 

255 event.accept() 

256 

257 # ---------------------------- 

258 # Translation 

259 # ---------------------------- 

260 

261 def _translate_ui(self): 

262 """Apply translated text to UI elements.""" 

263 self.drop_label.setText(QCoreApplication.translate("ImportPage", "Import or select patients' data"))