Coverage for main/threads/nifti_utils_threads.py: 99%

83 statements  

« prev     ^ index     » next       coverage.py v7.15.0, created at 2026-07-03 15:51 +0000

1import json 

2import nibabel as nib 

3import numpy as np 

4 

5from PyQt6.QtCore import QThread, pyqtSignal, QCoreApplication 

6from logger import get_logger 

7 

8 

9log = get_logger() 

10 

11class SaveNiftiThread(QThread): 

12 """ 

13 Background thread for saving a NIfTI image and its associated metadata 

14 to disk in BIDS-like format. 

15 

16 This class prevents the UI from freezing during file write operations 

17 by running the saving process in a separate thread. 

18 

19 Signals: 

20 success (str, str): Emitted when both the NIfTI and JSON files are successfully saved. 

21 Provides the paths to the saved files. 

22 error (str): Emitted when an error occurs during saving. 

23 

24 Args: 

25 data (np.ndarray): The voxel data to be saved as a NIfTI image. 

26 affine (np.ndarray): The affine transformation matrix for spatial orientation. 

27 path (str): Output file path for the NIfTI image. 

28 json_path (str): Output file path for the JSON metadata file. 

29 relative_path (str): Relative path reference (used for the "Sources" field in JSON). 

30 radius (float): Region growing radius parameter. 

31 difference (float): Intensity difference threshold parameter. 

32 """ 

33 

34 success = pyqtSignal(str, str) 

35 """**Signal(str, str):** Emitted when an operation completes successfully.  

36 Parameters:  

37 - `str`: Name or identifier of the completed task.  

38 - `str`: Success message or additional details.  

39 """ 

40 

41 error = pyqtSignal(str) 

42 """**Signal(str):** Emitted when an error occurs during an operation.  

43 Parameters:  

44 - `str`: Description of the error or failure reason.  

45 """ 

46 

47 def __init__(self, data, affine, path, json_path, relative_path,source_dict): 

48 super().__init__() 

49 self.data = data 

50 self.affine = affine 

51 self.path = path 

52 self.relative_path = relative_path 

53 self.source_dict = source_dict 

54 self.json_path = json_path 

55 

56 def run(self): 

57 """ 

58 Executes the file-saving process in a background thread. 

59 

60 The method saves the NIfTI image using `nibabel`, then creates and writes 

61 a companion JSON file describing the ROI (Region of Interest) with 

62 user-specified parameters. 

63 

64 Emits: 

65 - success(path, json_path): On successful save. 

66 - error(message): If an exception occurs. 

67 """ 

68 try: 

69 log.debug(f"Save nifti in {self.path}") 

70 # Save the NIfTI file 

71 nib.save(nib.Nifti1Image(self.data.astype(np.uint8), self.affine), self.path) 

72 

73 log.debug("Preparing json") 

74 json_dict = { 

75 "Type": "ROI", 

76 "Sources": [f"bids:{self.relative_path}"], 

77 "Description": "ROI mask", 

78 "Origin": self.source_dict 

79 } 

80 

81 log.debug("Writing json") 

82 # Write the JSON metadata 

83 with open(self.json_path, "w") as json_file: 

84 json.dump(json_dict, json_file, indent=4) 

85 log.debug("Send success") 

86 # Notify the main thread of success 

87 self.success.emit(self.path, self.json_path) 

88 

89 except Exception as e: 

90 # Emit error message if something goes wrong 

91 self.error.emit(str(e)) 

92 

93 

94class ImageLoadThread(QThread): 

95 """ 

96 Thread for loading and normalizing large NIfTI files without blocking the UI. 

97 

98 This class supports both 3D and 4D NIfTI volumes and includes 

99 robust normalization based on percentile scaling similar to matplotlib's 

100 default image scaling behavior. 

101 

102 Signals: 

103 finished (object, object, object, bool, bool): Emitted when loading completes successfully. 

104 Contains (img_data, dims, affine, is_4d, is_overlay). 

105 error (str): Emitted if an error occurs during file loading. 

106 progress (int): Emits loading progress updates (0–100). 

107 

108 Args: 

109 file_path (str): Path to the NIfTI file to load. 

110 is_overlay (bool): Flag indicating whether the file is an overlay image. 

111 """ 

112 

113 finished = pyqtSignal(object, object, object, bool, bool) 

114 """**Signal(object, object, object, bool, bool):**  

115 Emitted when the file is successifully loaded.  

116 

117 Parameters:  

118 - `object`: img_data.  

119 - `object`: dims. 

120 - `object`: affine.  

121 - `bool`: is_4d.  

122 - `bool`: is_overlay.  

123 """ 

124 

125 error = pyqtSignal(str) 

126 """**Signal(str):**  

127 Emitted when an error occurs during the deep learning execution.  

128 

129 Parameters:  

130 - `str`: Description or message detailing the error.  

131 """ 

132 

133 progress = pyqtSignal(int) 

134 """**Signal(int):**  

135 Emitted to report progress updates during execution.  

136 

137 Parameters:  

138 - `int`: Current progress percentage (0–100).  

139 """ 

140 

141 def __init__(self, file_path, is_overlay): 

142 super().__init__() 

143 self.file_path = file_path 

144 self.is_overlay = is_overlay 

145 

146 def run(self): 

147 """ 

148 Loads and normalizes a NIfTI image in a background thread. 

149 

150 Steps: 

151 1. Load image using memory mapping to minimize RAM usage. 

152 2. Verify the file is a valid NIfTI image. 

153 3. Canonicalize to RAS+ orientation. 

154 4. Normalize data using percentile scaling. 

155 5. Emit the finished signal with image data and metadata. 

156 """ 

157 try: 

158 self.progress.emit(10) 

159 

160 # Load image (memory-mapped to avoid heavy RAM use) 

161 img = nib.load(self.file_path, mmap="c") 

162 self.progress.emit(30) 

163 

164 if not isinstance(img, (nib.Nifti1Image, nib.Nifti2Image)): 

165 raise ValueError(QCoreApplication.translate("Threads", "Not a valid NIfTI file")) 

166 

167 # Convert to canonical orientation (RAS+) 

168 canonical_img = nib.as_closest_canonical(img) 

169 self.progress.emit(50) 

170 

171 dims = canonical_img.header.get_data_shape() 

172 is_4d = len(dims) == 4 

173 affine = canonical_img.affine 

174 

175 self.progress.emit(70) 

176 log.debug("Load voxel data") 

177 # Load voxel data 

178 img_data = np.asanyarray(canonical_img.dataobj, dtype=np.float32) 

179 self.progress.emit(80) 

180 

181 log.debug("Normalize image intensities") 

182 # Normalize image intensities 

183 img_data = self.normalize_data_matplotlib_style(img_data) 

184 

185 self.progress.emit(100) 

186 log.debug("Emit finished signal with image data and metadata.") 

187 # Emit the results to the main thread 

188 self.finished.emit(img_data, dims, affine, is_4d, self.is_overlay) 

189 

190 except Exception as e: 

191 # Report any errors encountered 

192 self.error.emit(str(e)) 

193 

194 def normalize_data_matplotlib_style(self, data): 

195 """ 

196 Normalize NIfTI data using robust percentile scaling (0.5th–99.5th percentiles). 

197 

198 This approach is similar to how matplotlib normalizes image intensities, 

199 providing consistent visual scaling even for datasets with outliers. 

200 

201 Args: 

202 data (np.ndarray): The 3D or 4D voxel intensity data. 

203 

204 Returns: 

205 np.ndarray: Normalized float32 data with intensity values scaled to [0, 1]. 

206 """ 

207 

208 if data.size == 0: 

209 return data 

210 

211 def normalize_volume(volume): 

212 """Normalize a single 3D volume to [0, 1] using percentile-based contrast stretching.""" 

213 valid_data = volume[np.isfinite(volume)] 

214 if valid_data.size == 0: 

215 return np.zeros_like(volume, dtype=np.float32) 

216 

217 vmin, vmax = np.percentile(valid_data, [0.1, 99.9]) 

218 if vmax <= vmin: 

219 vmax = vmin + 1.0 

220 

221 return np.clip((volume - vmin) / (vmax - vmin), 0, 1).astype(np.float32) 

222 

223 # Handle both 3D and 4D volumes 

224 if data.ndim == 4: 

225 normalized = np.empty_like(data, dtype=np.float32) 

226 for i in range(data.shape[3]): 

227 normalized[..., i] = normalize_volume(data[..., i]) 

228 else: 

229 normalized = normalize_volume(data) 

230 

231 return normalized