threads.nifti_utils_threads

  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