[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Author: pre-commit-ci[bot]

monailabel/datastore/utils/convert.py

@@ -55,6 +55,7 @@ def _get_nvimgcodec_encoder():
     if _NVIMGCODEC_ENCODER is None:
         try:
             from nvidia import nvimgcodec
+
             _NVIMGCODEC_ENCODER = nvimgcodec.Encoder()
             logger.debug("Initialized global nvimgcodec.Encoder singleton")
         except ImportError:
@@ -72,6 +73,7 @@ def _get_nvimgcodec_decoder():
     if _NVIMGCODEC_DECODER is None:
         try:
             from nvidia import nvimgcodec
+
             _NVIMGCODEC_DECODER = nvimgcodec.Decoder()
             logger.debug("Initialized global nvimgcodec.Decoder singleton")
         except ImportError:
@@ -211,7 +213,7 @@ def dicom_to_nifti(series_dir, is_seg=False):
 
         try:
             from monailabel.transform.reader import NvDicomReader
-            
+
             # Use NvDicomReader with LoadImage
             reader = NvDicomReader()
             loader = LoadImage(reader=reader, image_only=False)
@@ -552,9 +554,10 @@ def nifti_to_dicom_seg(
 
 
 def itk_image_to_dicom_seg(label, series_dir, template) -> str:
-    from monailabel.utils.others.generic import run_command
     import shutil
 
+    from monailabel.utils.others.generic import run_command
+
     command = "itkimage2segimage"
     if not shutil.which(command):
         error_msg = (
@@ -648,20 +651,20 @@ def transcode_dicom_to_htj2k(
 ) -> str:
     """
     Transcode DICOM files to HTJ2K (High Throughput JPEG 2000) lossless compression.
-    
+
     HTJ2K is a faster variant of JPEG 2000 that provides better compression performance
     for medical imaging applications. This function uses nvidia-nvimgcodec for hardware-
     accelerated decoding and encoding with batch processing for optimal performance.
     All transcoding is performed using lossless compression to preserve image quality.
-    
+
     The function processes files in configurable batches:
     1. Categorizes files by transfer syntax (HTJ2K/JPEG2000/JPEG/uncompressed)
     2. Uses nvimgcodec decoder for compressed files (JPEG2000, JPEG)
     3. Falls back to pydicom pixel_array for uncompressed files
     4. Batch encodes all images to HTJ2K using nvimgcodec
     5. Saves transcoded files with updated transfer syntax
     6. Copies already-HTJ2K files directly (no re-encoding)
-    
+
     Supported source transfer syntaxes:
     - JPEG 2000 (lossless and lossy)
     - JPEG (baseline, extended, lossless)
@@ -670,7 +673,7 @@ def transcode_dicom_to_htj2k(
 
     Typical compression ratios of 60-70% with lossless quality.
     Processing speed depends on batch size and GPU capabilities.
-    
+
     Args:
         input_dir: Path to directory containing DICOM files to transcode
         output_dir: Path to output directory for transcoded files. If None, creates temp directory
@@ -680,47 +683,47 @@ def transcode_dicom_to_htj2k(
                         Must be powers of 2. Common values: (32,32), (64,64), (128,128)
         max_batch_size: Maximum number of DICOM files to process in each batch (default: 256)
                        Lower values reduce memory usage, higher values may improve speed
-        
+
     Returns:
         str: Path to output directory containing transcoded DICOM files
-        
+
     Raises:
         ImportError: If nvidia-nvimgcodec is not available
         ValueError: If input directory doesn't exist or contains no valid DICOM files
         ValueError: If DICOM files are missing required attributes (TransferSyntaxUID, PixelData)
-        
+
     Example:
         >>> # Basic usage with default settings
         >>> output_dir = transcode_dicom_to_htj2k("/path/to/dicoms")
         >>> print(f"Transcoded files saved to: {output_dir}")
-        
+
         >>> # Custom output directory and batch size
         >>> output_dir = transcode_dicom_to_htj2k(
         ...     input_dir="/path/to/dicoms",
         ...     output_dir="/path/to/output",
         ...     max_batch_size=50,
         ...     num_resolutions=5
         ... )
-        
+
         >>> # Process with smaller code blocks for memory efficiency
         >>> output_dir = transcode_dicom_to_htj2k(
         ...     input_dir="/path/to/dicoms",
         ...     code_block_size=(32, 32),
         ...     max_batch_size=5
         ... )
-        
+
     Note:
         Requires nvidia-nvimgcodec to be installed:
             pip install nvidia-nvimgcodec-cu{XX}[all]
         Replace {XX} with your CUDA version (e.g., cu13 for CUDA 13.x)
-        
+
         The function preserves all DICOM metadata including Patient, Study, and Series
         information. Only the transfer syntax and pixel data encoding are modified.
     """
     import glob
     import shutil
     from pathlib import Path
-    
+
     # Check for nvidia-nvimgcodec
     try:
         from nvidia import nvimgcodec
@@ -730,62 +733,62 @@ def transcode_dicom_to_htj2k(
             "Install it with: pip install nvidia-nvimgcodec-cu{XX}[all] "
             "(replace {XX} with your CUDA version, e.g., cu13)"
         )
-    
+
     # Validate input
     if not os.path.exists(input_dir):
         raise ValueError(f"Input directory does not exist: {input_dir}")
-    
+
     if not os.path.isdir(input_dir):
         raise ValueError(f"Input path is not a directory: {input_dir}")
-    
+
     # Get all DICOM files
     dicom_files = []
     for pattern in ["*.dcm", "*"]:
         dicom_files.extend(glob.glob(os.path.join(input_dir, pattern)))
-    
+
     # Filter to actual DICOM files
     valid_dicom_files = []
     for file_path in dicom_files:
         if os.path.isfile(file_path):
             try:
                 # Quick check if it's a DICOM file
-                with open(file_path, 'rb') as f:
+                with open(file_path, "rb") as f:
                     f.seek(128)
                     magic = f.read(4)
-                    if magic == b'DICM':
+                    if magic == b"DICM":
                         valid_dicom_files.append(file_path)
             except Exception:
                 continue
-    
+
     if not valid_dicom_files:
         raise ValueError(f"No valid DICOM files found in {input_dir}")
-    
+
     logger.info(f"Found {len(valid_dicom_files)} DICOM files to transcode")
-    
+
     # Create output directory
     if output_dir is None:
         output_dir = tempfile.mkdtemp(prefix="htj2k_")
     else:
         os.makedirs(output_dir, exist_ok=True)
-    
+
     # Create encoder and decoder instances (reused for all files)
     encoder = _get_nvimgcodec_encoder()
     decoder = _get_nvimgcodec_decoder()  # Always needed for decoding input DICOM images
-    
+
     # HTJ2K Transfer Syntax UID - Lossless Only
     # 1.2.840.10008.1.2.4.201 = HTJ2K Lossless Only
     target_transfer_syntax = "1.2.840.10008.1.2.4.201"
     quality_type = nvimgcodec.QualityType.LOSSLESS
     logger.info("Using lossless HTJ2K compression")
-    
+
     # Configure JPEG2K encoding parameters
     jpeg2k_encode_params = nvimgcodec.Jpeg2kEncodeParams()
     jpeg2k_encode_params.num_resolutions = num_resolutions
     jpeg2k_encode_params.code_block_size = code_block_size
     jpeg2k_encode_params.bitstream_type = nvimgcodec.Jpeg2kBitstreamType.JP2
     jpeg2k_encode_params.prog_order = nvimgcodec.Jpeg2kProgOrder.LRCP
     jpeg2k_encode_params.ht = True  # Enable High Throughput mode
-    
+
     encode_params = nvimgcodec.EncodeParams(
         quality_type=quality_type,
         jpeg2k_encode_params=jpeg2k_encode_params,
@@ -795,37 +798,43 @@ def transcode_dicom_to_htj2k(
         allow_any_depth=True,
         color_spec=nvimgcodec.ColorSpec.UNCHANGED,
     )
-    
+
     # Define transfer syntax constants (use frozenset for O(1) membership testing)
-    JPEG2000_SYNTAXES = frozenset([
-        "1.2.840.10008.1.2.4.90",  # JPEG 2000 Image Compression (Lossless Only)
-        "1.2.840.10008.1.2.4.91",  # JPEG 2000 Image Compression
-    ])
-    
-    HTJ2K_SYNTAXES = frozenset([
-        "1.2.840.10008.1.2.4.201",  # High-Throughput JPEG 2000 Image Compression (Lossless Only)
-        "1.2.840.10008.1.2.4.202",  # High-Throughput JPEG 2000 with RPCL Options Image Compression (Lossless Only)
-        "1.2.840.10008.1.2.4.203",  # High-Throughput JPEG 2000 Image Compression
-    ])
-    
-    JPEG_SYNTAXES = frozenset([
-        "1.2.840.10008.1.2.4.50",  # JPEG Baseline (Process 1)
-        "1.2.840.10008.1.2.4.51",  # JPEG Extended (Process 2 & 4)
-        "1.2.840.10008.1.2.4.57",  # JPEG Lossless, Non-Hierarchical (Process 14)
-        "1.2.840.10008.1.2.4.70",  # JPEG Lossless, Non-Hierarchical, First-Order Prediction
-    ])
-    
+    JPEG2000_SYNTAXES = frozenset(
+        [
+            "1.2.840.10008.1.2.4.90",  # JPEG 2000 Image Compression (Lossless Only)
+            "1.2.840.10008.1.2.4.91",  # JPEG 2000 Image Compression
+        ]
+    )
+
+    HTJ2K_SYNTAXES = frozenset(
+        [
+            "1.2.840.10008.1.2.4.201",  # High-Throughput JPEG 2000 Image Compression (Lossless Only)
+            "1.2.840.10008.1.2.4.202",  # High-Throughput JPEG 2000 with RPCL Options Image Compression (Lossless Only)
+            "1.2.840.10008.1.2.4.203",  # High-Throughput JPEG 2000 Image Compression
+        ]
+    )
+
+    JPEG_SYNTAXES = frozenset(
+        [
+            "1.2.840.10008.1.2.4.50",  # JPEG Baseline (Process 1)
+            "1.2.840.10008.1.2.4.51",  # JPEG Extended (Process 2 & 4)
+            "1.2.840.10008.1.2.4.57",  # JPEG Lossless, Non-Hierarchical (Process 14)
+            "1.2.840.10008.1.2.4.70",  # JPEG Lossless, Non-Hierarchical, First-Order Prediction
+        ]
+    )
+
     # Pre-compute combined set for nvimgcodec-compatible formats
     NVIMGCODEC_SYNTAXES = JPEG2000_SYNTAXES | JPEG_SYNTAXES
-    
+
     start_time = time.time()
     transcoded_count = 0
     skipped_count = 0
-    
+
     # Calculate batch info for logging
     total_files = len(valid_dicom_files)
     total_batches = (total_files + max_batch_size - 1) // max_batch_size
-    
+
     for batch_start in range(0, total_files, max_batch_size):
         batch_end = min(batch_start + max_batch_size, total_files)
         current_batch = batch_start // max_batch_size + 1
@@ -836,14 +845,16 @@ def transcode_dicom_to_htj2k(
         pydicom_batch = []
         copy_batch = []
         for idx, ds in enumerate(batch_datasets):
-            current_ts = getattr(ds, 'file_meta', {}).get('TransferSyntaxUID', None)
+            current_ts = getattr(ds, "file_meta", {}).get("TransferSyntaxUID", None)
             if current_ts is None:
                 raise ValueError(f"DICOM file {os.path.basename(batch_files[idx])} does not have a Transfer Syntax UID")
-            
+
             ts_str = str(current_ts)
             if ts_str in NVIMGCODEC_SYNTAXES:
                 if not hasattr(ds, "PixelData") or ds.PixelData is None:
-                    raise ValueError(f"DICOM file {os.path.basename(batch_files[idx])} does not have a PixelData member")
+                    raise ValueError(
+                        f"DICOM file {os.path.basename(batch_files[idx])} does not have a PixelData member"
+                    )
                 nvimgcodec_batch.append(idx)
             elif ts_str in HTJ2K_SYNTAXES:
                 copy_batch.append(idx)
@@ -859,7 +870,7 @@ def transcode_dicom_to_htj2k(
         data_sequence = []
         decoded_data = []
         num_frames = []
-        
+
         # Decode using nvimgcodec for compressed formats
         if nvimgcodec_batch:
             for idx in nvimgcodec_batch:
@@ -887,21 +898,21 @@ def transcode_dicom_to_htj2k(
         # Reassemble and save transcoded files
         frame_offset = 0
         files_to_process = nvimgcodec_batch + pydicom_batch
-        
+
         for list_idx, dataset_idx in enumerate(files_to_process):
             nframes = num_frames[list_idx]
-            encoded_frames = [bytes(enc) for enc in encoded_data[frame_offset:frame_offset + nframes]]
+            encoded_frames = [bytes(enc) for enc in encoded_data[frame_offset : frame_offset + nframes]]
             frame_offset += nframes
-            
+
             # Update dataset with HTJ2K encoded data
             batch_datasets[dataset_idx].PixelData = pydicom.encaps.encapsulate(encoded_frames)
             batch_datasets[dataset_idx].file_meta.TransferSyntaxUID = pydicom.uid.UID(target_transfer_syntax)
-            
+
             # Save transcoded file
             output_file = os.path.join(output_dir, os.path.basename(batch_files[dataset_idx]))
             batch_datasets[dataset_idx].save_as(output_file)
             transcoded_count += 1
-    
+
     elapsed_time = time.time() - start_time
 
     logger.info(f"Transcoding complete:")
@@ -910,5 +921,5 @@ def transcode_dicom_to_htj2k(
     logger.info(f"  Already HTJ2K (copied): {skipped_count}")
     logger.info(f"  Time elapsed: {elapsed_time:.2f} seconds")
     logger.info(f"  Output directory: {output_dir}")
-    
+
     return output_dir