Python API

__version__: str

bin_path: Path

bin: str

def main(args: list[str] | None = None, **run_kwargs) -> int:
    """
    Execute dcm2niix binary with specified arguments.

    Args:
        args: List of command-line arguments (defaults to sys.argv[1:]).
              Should not include the program name.
              Each element is a separate argument (no shell parsing).
              Example: ["-z", "y", "-o", "/output", "/input"]
              
        **run_kwargs: Keyword arguments passed to subprocess.run().
                     Common options:
                     - capture_output (bool): Capture stdout/stderr
                     - text (bool): Return strings instead of bytes
                     - check (bool): Raise CalledProcessError on non-zero exit
                     - timeout (float): Maximum execution time in seconds
                     - cwd (str): Working directory for subprocess
                     - env (dict): Environment variables
                     - stdin (file): Standard input source
                     - stdout (file): Standard output destination
                     - stderr (file): Standard error destination

    Returns:
        int: Exit code from dcm2niix execution.
             0 = success (all conversions completed successfully)
             1 = general failure (file I/O or parsing errors)
             2 = no valid DICOM files found in input directory
             3 = version report (from -v or --version, not an error)
             4 = corrupt DICOM file detected
             5 = input folder does not exist or is not accessible
             6 = output folder does not exist or cannot be created
             7 = output folder is read-only (permission denied)
             8 = mixed success (some conversions succeeded, others failed)
             9 = rename operation failed
             10 = incomplete volumes detected (missing slices warning)
             11 = nominal operation (no files expected to convert)
             
    Raises:
        subprocess.CalledProcessError: If check=True and exit code is non-zero
        subprocess.TimeoutExpired: If timeout is set and exceeded
        FileNotFoundError: If binary cannot be found
        PermissionError: If binary is not executable
        
    Notes:
        - Function blocks until conversion completes
        - Output goes to stdout/stderr unless capture_output=True
        - Binary is executed via subprocess.run()
        - Environment is inherited from parent process unless env= specified
        - Does not validate arguments before passing to binary
    """

from dcm2niix import main

# Convert DICOM folder to NIfTI
exit_code = main(["/path/to/dicom/folder"])

if exit_code == 0:
    print("Conversion successful")
else:
    print(f"Conversion failed with exit code {exit_code}")

from dcm2niix import main

# Compressed NIfTI with custom filename pattern
exit_code = main([
    "-z", "y",                    # Enable gzip compression
    "-f", "%p_%s",                # Filename: protocol_series
    "-o", "/output/folder",       # Output directory
    "/input/dicom/folder"         # Input directory
])

from dcm2niix import main

# Generate BIDS JSON sidecars with anonymization
exit_code = main([
    "-b", "y",                    # Generate BIDS JSON
    "-ba", "y",                   # Anonymize patient info
    "-bi", "sub-001",             # Subject ID
    "-bv", "ses-01",              # Session ID
    "-o", "/bids/output",         # Output directory
    "/dicom/input"                # Input directory
])

from dcm2niix import main
import subprocess

# Capture stdout and stderr
try:
    exit_code = main(
        ["-v", "1", "/dicom/folder"],
        capture_output=True,
        text=True,
        check=True
    )
except subprocess.CalledProcessError as e:
    print(f"dcm2niix failed: {e.stderr}")
    print(f"Exit code: {e.returncode}")

from dcm2niix import main
import subprocess

# Run with 60 second timeout
try:
    exit_code = main(
        ["-v", "2", "/large/dicom/folder"],
        timeout=60
    )
    
    if exit_code == 0:
        print("Conversion completed within timeout")
    else:
        print(f"Conversion finished with exit code {exit_code}")
        
except subprocess.TimeoutExpired:
    print("Conversion timed out after 60 seconds")
    # Note: Process is terminated on timeout

from dcm2niix import main

# Convert only specific series by CRC
exit_code = main([
    "-n", "12345",                # First series CRC
    "-n", "67890",                # Second series CRC
    "-o", "/output",              # Output directory
    "/input"                      # Input directory
])

# Check for success
success = (exit_code == 0)

from dcm2niix import bin, bin_path
import subprocess

# Use bin string directly
print(f"dcm2niix binary: {bin}")

# Use Path object for path operations
print(f"Binary exists: {bin_path.exists()}")
print(f"Binary parent: {bin_path.parent}")
print(f"Absolute path: {bin_path.resolve()}")

# Direct subprocess invocation
result = subprocess.run([bin, "--version"], capture_output=True, text=True)
print(f"Version output: {result.stdout}")

from dcm2niix import __version__, main

print(f"dcm2niix Python package version: {__version__}")

# Get binary version (includes compiler info, optional features)
exit_code = main(["--version"])
# Prints to stdout: "Chris Rorden's dcm2niiX version v1.0.20250506 ..."
# Exit code is 3 for version report

from dcm2niix import main
import os

# Set environment variables for subprocess
custom_env = os.environ.copy()
custom_env["PATH"] = f"/custom/pigz/path:{custom_env['PATH']}"

exit_code = main(
    ["-z", "y", "/dicom/folder"],
    env=custom_env
)

# This allows using custom pigz installation

from dcm2niix import main

exit_code = main(["/dicom/folder"])

# Comprehensive exit code interpretation
if exit_code == 0:
    print("Success: All files converted")
    status = "success"
elif exit_code == 2:
    print("Error: No valid DICOM files found")
    status = "no_dicoms"
elif exit_code == 4:
    print("Error: Corrupt DICOM file detected")
    status = "corrupt"
elif exit_code == 5:
    print("Error: Input folder invalid")
    status = "invalid_input"
elif exit_code == 6:
    print("Error: Output folder invalid")
    status = "invalid_output"
elif exit_code == 7:
    print("Error: Output folder is read-only")
    status = "readonly"
elif exit_code == 8:
    print("Warning: Partial success - some files converted")
    status = "partial"
elif exit_code == 10:
    print("Warning: Incomplete volumes detected")
    status = "incomplete"
else:
    print(f"Error: Conversion failed with exit code {exit_code}")
    status = "failed"

# Return structured result
result = {
    "exit_code": exit_code,
    "status": status,
    "success": exit_code in [0, 8]  # 0 or 8 considered successful
}

from dcm2niix import main
from pathlib import Path

# Write stdout/stderr to log file
log_file = Path("/logs/conversion.log")

with open(log_file, "w") as f:
    exit_code = main(
        ["-v", "2", "/dicom/folder"],
        stdout=f,
        stderr=f,
        text=True
    )

print(f"Conversion completed with exit code {exit_code}")
print(f"Log saved to {log_file}")

from dcm2niix import main
import subprocess
from threading import Thread
import queue

def process_with_progress(dicom_folder, output_folder):
    """Run conversion with real-time progress monitoring."""
    
    def read_output(pipe, output_queue):
        """Read subprocess output line by line."""
        for line in iter(pipe.readline, ''):
            if line:
                output_queue.put(line.strip())
        pipe.close()
    
    # Start conversion with piped output
    process = subprocess.Popen(
        [bin, "-v", "1", "-o", output_folder, dicom_folder],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True
    )
    
    # Create queue for output lines
    output_queue = queue.Queue()
    
    # Start threads to read stdout and stderr
    Thread(target=read_output, args=(process.stdout, output_queue)).start()
    Thread(target=read_output, args=(process.stderr, output_queue)).start()
    
    # Process output lines in real-time
    while process.poll() is None or not output_queue.empty():
        try:
            line = output_queue.get(timeout=0.1)
            print(f"Progress: {line}")  # Replace with actual progress handler
        except queue.Empty:
            pass
    
    return process.returncode

from dcm2niix import main
from pathlib import Path
from typing import Dict, List

def batch_convert(
    input_dirs: List[str],
    output_root: str,
    compression: bool = True,
    bids: bool = True,
    anonymize: bool = True
) -> Dict[str, dict]:
    """
    Convert multiple DICOM directories with consistent settings.
    
    Args:
        input_dirs: List of input DICOM directory paths
        output_root: Root directory for outputs
        compression: Enable gzip compression
        bids: Generate BIDS JSON sidecars
        anonymize: Anonymize patient information
        
    Returns:
        Dictionary mapping input_dir to result dict with:
        - success (bool): Whether conversion succeeded
        - exit_code (int): dcm2niix exit code
        - output_dir (str): Output directory path
    """
    results = {}
    
    for input_dir in input_dirs:
        # Create output directory
        input_path = Path(input_dir)
        output_dir = Path(output_root) / input_path.name
        output_dir.mkdir(parents=True, exist_ok=True)
        
        # Build arguments
        args = []
        if compression:
            args.extend(["-z", "y"])
        if bids:
            args.extend(["-b", "y"])
        if anonymize:
            args.extend(["-ba", "y"])
        args.extend(["-o", str(output_dir), str(input_dir)])
        
        # Run conversion
        exit_code = main(args)
        
        results[input_dir] = {
            "success": exit_code in [0, 8],
            "exit_code": exit_code,
            "output_dir": str(output_dir)
        }
    
    return results

from dcm2niix import main
from pathlib import Path
import shutil
import tempfile

def safe_convert(input_dir, output_dir, **options):
    """
    Convert DICOM with automatic rollback on failure.
    
    Converts to temporary directory first, then moves to final location
    only on success. This prevents partial/corrupted outputs.
    """
    # Create temporary output directory
    with tempfile.TemporaryDirectory() as temp_dir:
        # Build arguments
        args = ["-o", temp_dir, input_dir]
        for key, value in options.items():
            if value is True:
                args.extend([f"-{key}", "y"])
            elif value is False:
                args.extend([f"-{key}", "n"])
            elif value is not None:
                args.extend([f"-{key}", str(value)])
        
        # Run conversion
        exit_code = main(args)
        
        # Check for success
        if exit_code in [0, 8]:
            # Move outputs to final destination
            output_path = Path(output_dir)
            output_path.mkdir(parents=True, exist_ok=True)
            
            for file in Path(temp_dir).iterdir():
                shutil.move(str(file), str(output_path / file.name))
            
            return True, "Conversion successful"
        else:
            # Temporary directory automatically cleaned up
            return False, f"Conversion failed with exit code {exit_code}"

dcm2niix [options] <in_folder>

python -m dcm2niix [options] <in_folder>

# Codes indicating success (output files generated)
SUCCESS_CODES = [0, 8]

# Codes indicating warnings (output may be generated)
WARNING_CODES = [10]

# Codes indicating no output expected
INFO_CODES = [3, 11]

# Codes indicating errors (no output or incomplete output)
ERROR_CODES = [1, 2, 4, 5, 6, 7, 9]

def interpret_exit_code(code):
    """Categorize exit code."""
    if code in SUCCESS_CODES:
        return "success"
    elif code in WARNING_CODES:
        return "warning"
    elif code in INFO_CODES:
        return "info"
    elif code in ERROR_CODES:
        return "error"
    else:
        return "unknown"

import subprocess
from pathlib import Path

def main_custom_binary(binary_path, args):
    """Execute dcm2niix from custom location."""
    result = subprocess.run(
        [str(binary_path)] + args,
        capture_output=False
    )
    return result.returncode

# Use custom binary
custom_bin = Path("/custom/location/dcm2niix")
exit_code = main_custom_binary(custom_bin, ["-z", "y", "/input"])

from dcm2niix import main
from concurrent.futures import ProcessPoolExecutor, as_completed
from pathlib import Path

def convert_single_subject(subject_dir, output_root):
    """Convert single subject directory."""
    subject_name = Path(subject_dir).name
    output_dir = Path(output_root) / subject_name
    output_dir.mkdir(parents=True, exist_ok=True)
    
    exit_code = main([
        "-z", "y",
        "-b", "y",
        "-ba", "y",
        "-o", str(output_dir),
        str(subject_dir)
    ])
    
    return subject_name, exit_code

def parallel_convert(subject_dirs, output_root, max_workers=4):
    """Convert multiple subjects in parallel."""
    results = {}
    
    with ProcessPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(convert_single_subject, subj_dir, output_root): subj_dir
            for subj_dir in subject_dirs
        }
        
        for future in as_completed(futures):
            subject_name, exit_code = future.result()
            results[subject_name] = exit_code
    
    return results

# Always check exit codes
exit_code = main(["/input"])
if exit_code not in [0, 8]:
    # Handle error
    pass

# Use try/except for subprocess errors
try:
    exit_code = main(["/input"], check=True, timeout=300)
except subprocess.CalledProcessError as e:
    # Handle non-zero exit
    pass
except subprocess.TimeoutExpired:
    # Handle timeout
    pass

# Capture output for debugging
result = main(["/input"], capture_output=True, text=True)
# Access via subprocess internals if needed