Feature Extraction
Transformers
Safetensors
jolia
medical
radiology
ct
3d
vision
foundation-model
self-supervised
custom_code
Instructions to use raidium/Jolia with libraries, inference providers, notebooks, and local apps. Follow these links to get started.
- Libraries
- Transformers
How to use raidium/Jolia with Transformers:
# Use a pipeline as a high-level helper from transformers import pipeline pipe = pipeline("feature-extraction", model="raidium/Jolia", trust_remote_code=True)# Load model directly from transformers import AutoModel model = AutoModel.from_pretrained("raidium/Jolia", trust_remote_code=True, dtype="auto") - Notebooks
- Google Colab
- Kaggle
| # Vendored verbatim from the internal `raidium.rd.models` library for the | |
| # self-contained Hugging Face release. Only imports were rewritten (raidium | |
| # hub base classes -> jolia_shim; sibling modules -> jolia_* names). | |
| # Do not edit by hand: regenerate with scripts/build_hf_jolia.py. | |
| """ | |
| Window/level adjustment utilities for medical images. | |
| Provides functions for applying multiple simultaneous windowing strategies. | |
| """ | |
| import math | |
| from typing import Dict, List, Optional, Union | |
| import numpy as np | |
| import torch | |
| # Define anatomical window settings for CT (center/width based) | |
| ANATOMICAL_WINDOWS = { | |
| "CT": { | |
| "lung": {"center": -600, "width": 1500}, | |
| "mediastinum": {"center": 50, "width": 400}, | |
| "abdomen": {"center": 40, "width": 400}, | |
| "liver": {"center": 80, "width": 150}, | |
| "bone": {"center": 400, "width": 1800}, | |
| "brain": {"center": 40, "width": 80}, | |
| "subdural": {"center": 75, "width": 215}, | |
| "stroke": {"center": 40, "width": 40}, | |
| "temporal_bone": {"center": 600, "width": 2800}, | |
| "soft_tissue": {"center": 50, "width": 350}, | |
| } | |
| } | |
| # Define percentile-based window settings for X-ray, Mammography, and MR | |
| PERCENTILE_WINDOWS = { | |
| "XR": { | |
| "lung": { | |
| "percentile_min": 2, | |
| "percentile_max": 50, | |
| "description": "Emphasizes lung parenchyma and airways", | |
| }, | |
| "mediastinum": { | |
| "percentile_min": 30, | |
| "percentile_max": 80, | |
| "description": "Emphasizes heart and vessels", | |
| }, | |
| "bone": { | |
| "percentile_min": 70, | |
| "percentile_max": 98, | |
| "description": "Emphasizes ribs and spine", | |
| }, | |
| "soft_tissue": { | |
| "percentile_min": 10, | |
| "percentile_max": 90, | |
| "description": "Balanced view of all structures", | |
| }, | |
| }, | |
| "MG": { | |
| "standard": { | |
| "percentile_min": 5, | |
| "percentile_max": 95, | |
| "description": "General breast tissue visualization", | |
| }, | |
| "dense_tissue": { | |
| "percentile_min": 20, | |
| "percentile_max": 98, | |
| "description": "Enhanced visualization of dense tissue", | |
| }, | |
| "calcifications": { | |
| "percentile_min": 50, | |
| "percentile_max": 99.5, | |
| "description": "Enhanced visualization of calcifications", | |
| }, | |
| "skin_line": { | |
| "percentile_min": 0.5, | |
| "percentile_max": 70, | |
| "description": "Enhanced visualization of skin line and superficial structures", | |
| }, | |
| "contrast": { | |
| "percentile_min": 10, | |
| "percentile_max": 90, | |
| "description": "Balanced contrast for overall assessment", | |
| }, | |
| }, | |
| "MR": { | |
| "high_contrast": { | |
| "percentile_min": 20, | |
| "percentile_max": 99, | |
| "description": "Maximum tissue contrast", | |
| } | |
| }, | |
| } | |
| def _get_windows_to_apply(windows: Union[str, List[str]] = "default", modality: str = "CT") -> List[str]: | |
| """ | |
| Get list of windows to apply for a given modality. | |
| """ | |
| if windows == "default": | |
| if modality == "CT": | |
| return ["lung", "mediastinum", "bone"] | |
| elif modality == "MR": | |
| return ["znorm"] | |
| elif modality in ["XR", "MG"]: | |
| return ["standard", "soft_tissue", "high_contrast"] | |
| else: | |
| return ["minmax"] | |
| elif windows == "all": | |
| return get_available_windows(modality) | |
| elif isinstance(windows, str): | |
| return [windows] | |
| else: | |
| return windows | |
| def batch_apply_windowing_vectorized( | |
| volume: Union[torch.Tensor, np.ndarray], | |
| windows: Union[str, List[str]] = "default", | |
| modality: str = "CT", | |
| min_value: Optional[float] = None, | |
| max_value: Optional[float] = None, | |
| percentile_windows: Optional[Dict[str, Dict[str, float]]] = None, | |
| znorm_clip: Optional[float] = None, | |
| torch_operating_dtype: torch.dtype = torch.float32, | |
| compute_stats_per_sample: bool = True, | |
| ) -> Union[torch.Tensor, np.ndarray]: | |
| """ | |
| Batch apply windowing to medical images using vectorized operations. | |
| This function efficiently applies multiple windowing strategies in parallel | |
| by treating all windows as shift-and-scale operations. | |
| Args: | |
| volume: Input volume as tensor or numpy array. | |
| Shape: (B, C, D, H, W) or (B, C, H, W) for batched inputs | |
| (D, H, W) or (H, W) for single inputs | |
| windows: Window name(s) to apply (same as apply_windowing) | |
| modality: Imaging modality ('CT', 'MR', 'XR', 'MG') | |
| min_value: Override minimum value for minmax windowing | |
| max_value: Override maximum value for minmax windowing | |
| percentile_windows: Custom percentile window definitions | |
| znorm_clip: Clipping value for z-normalized data (e.g., 3.0 clips at ±3 std) | |
| torch_operating_dtype: Data type for torch operations | |
| compute_stats_per_sample: If True, compute statistics per sample in batch. | |
| If False, compute across entire batch. | |
| Returns: | |
| Windowed volumes. Shape: (B, C*N, D, H, W) where N is number of windows | |
| Windows and channels are combined into a single dimension for compatibility. | |
| If input is unbatched, batch dimension is removed appropriately. | |
| """ | |
| # Convert to tensor if needed | |
| is_numpy = isinstance(volume, np.ndarray) | |
| if is_numpy: | |
| volume = torch.from_numpy(volume) | |
| # Only convert dtype if necessary | |
| if volume.dtype != torch_operating_dtype: | |
| volume = volume.to(torch_operating_dtype) | |
| else: | |
| # Only convert dtype if necessary | |
| if volume.dtype != torch_operating_dtype: | |
| volume = volume.to(torch_operating_dtype) | |
| # Handle input shapes - add batch and channel dims if needed | |
| original_shape = volume.shape | |
| needs_batch = volume.ndim < 4 # Less than (B, C, H, W) | |
| needs_channel = volume.ndim < 5 if volume.ndim >= 4 else volume.ndim < 3 | |
| if needs_batch: | |
| volume = volume.unsqueeze(0) # Add batch dimension | |
| if needs_channel: | |
| volume = volume.unsqueeze(1 if not needs_batch else 0) # Add channel dimension | |
| # Ensure we have at least 4 dims (B, C, H, W) | |
| while volume.ndim < 4: | |
| volume = volume.unsqueeze(-1) | |
| B, C = volume.shape[:2] | |
| spatial_dims = volume.shape[2:] | |
| # Get list of windows to apply | |
| windows = _get_windows_to_apply(windows, modality) | |
| num_windows = len(windows) | |
| # Compute all window parameters as shift-scale pairs | |
| window_params = _compute_window_params_vectorized( | |
| volume, | |
| windows, | |
| modality, | |
| min_value, | |
| max_value, | |
| percentile_windows, | |
| znorm_clip, | |
| compute_stats_per_sample, | |
| ) # Shape: (B, num_windows, C, 2) | |
| # Apply all windows in parallel | |
| result = _apply_windows_vectorized( | |
| volume, window_params | |
| ) # Shape: (B, C*num_windows, *spatial) - already in final layout | |
| # Handle output shape based on input | |
| if needs_batch: | |
| result = result.squeeze(0) # Remove batch dimension | |
| if needs_channel and C == 1 and num_windows == 1: | |
| # For single window, single channel, remove channel dim | |
| result = result.squeeze(0 if needs_batch else 1) | |
| # If single window and appropriate shape, remove window dimension | |
| if num_windows == 1 and needs_batch and needs_channel: | |
| result = result.squeeze(0) | |
| # Convert back to numpy if input was numpy | |
| if is_numpy: | |
| result = result.cpu().numpy() | |
| return result | |
| def _compute_window_params_vectorized( | |
| volume: torch.Tensor, | |
| windows: List[str], | |
| modality: str, | |
| min_value: Optional[float], | |
| max_value: Optional[float], | |
| percentile_windows: Optional[Dict[str, Dict[str, float]]], | |
| znorm_clip: Optional[float], | |
| compute_stats_per_sample: bool, | |
| ) -> torch.Tensor: | |
| """ | |
| Compute shift and scale parameters for all windows in a vectorized manner. | |
| Handles multi-channel inputs by computing statistics per channel. | |
| Returns: | |
| torch.Tensor: Shape (B, num_windows, C, 2) where [:,:,:,0] is shift and [:,:,:,1] is scale | |
| """ | |
| B, C = volume.shape[:2] | |
| spatial_shape = volume.shape[2:] | |
| num_windows = len(windows) | |
| device = volume.device | |
| dtype = volume.dtype | |
| # Initialize parameters tensor - now includes channel dimension | |
| params = torch.zeros((B, num_windows, C, 2), device=device, dtype=dtype) | |
| # Group windows by computation type for efficiency | |
| percentile_windows_info = [] # [(window_idx, percentile_min, percentile_max)] | |
| anatomical_indices = [] | |
| minmax_indices = [] | |
| znorm_indices = [] | |
| for i, window in enumerate(windows): | |
| if window == "minmax": | |
| minmax_indices.append(i) | |
| elif window == "znorm": | |
| znorm_indices.append(i) | |
| elif window in ANATOMICAL_WINDOWS.get(modality, {}): | |
| anatomical_indices.append(i) | |
| elif modality in PERCENTILE_WINDOWS and window in PERCENTILE_WINDOWS[modality]: | |
| p = PERCENTILE_WINDOWS[modality][window] | |
| percentile_windows_info.append((i, p["percentile_min"] / 100.0, p["percentile_max"] / 100.0)) | |
| elif percentile_windows and window in percentile_windows: | |
| p = percentile_windows[window] | |
| percentile_windows_info.append((i, p["percentile_min"] / 100.0, p["percentile_max"] / 100.0)) | |
| # Compute parameters for anatomical windows (no computation needed) | |
| # These are fixed values, same for all channels | |
| for idx in anatomical_indices: | |
| window_name = windows[idx] | |
| p = ANATOMICAL_WINDOWS[modality][window_name] | |
| center = p["center"] | |
| width = p["width"] | |
| params[:, idx, :, 0] = center - width / 2 # shift (same for all channels) | |
| params[:, idx, :, 1] = width # scale (same for all channels) | |
| # Compute min/max for minmax windows | |
| if minmax_indices: | |
| # Reshape to (B*C, *spatial) to compute per-channel statistics | |
| vol_reshaped = volume.view(B * C, -1) | |
| if compute_stats_per_sample: | |
| # Compute per sample and channel | |
| if min_value is None: | |
| min_vals = vol_reshaped.min(dim=1)[0] # Shape: (B*C,) | |
| else: | |
| min_vals = torch.full((B * C,), min_value, device=device, dtype=dtype) | |
| if max_value is None: | |
| max_vals = vol_reshaped.max(dim=1)[0] # Shape: (B*C,) | |
| else: | |
| max_vals = torch.full((B * C,), max_value, device=device, dtype=dtype) | |
| else: | |
| # Compute across entire batch but per channel | |
| volume_per_channel = volume.view(C, -1) | |
| if min_value is None: | |
| min_vals_per_channel = volume_per_channel.min(dim=1)[0] # Shape: (C,) | |
| min_vals = min_vals_per_channel.repeat(B) # Shape: (B*C,) | |
| else: | |
| min_vals = torch.full((B * C,), min_value, device=device, dtype=dtype) | |
| if max_value is None: | |
| max_vals_per_channel = volume_per_channel.max(dim=1)[0] # Shape: (C,) | |
| max_vals = max_vals_per_channel.repeat(B) # Shape: (B*C,) | |
| else: | |
| max_vals = torch.full((B * C,), max_value, device=device, dtype=dtype) | |
| # Reshape back to (B, C) | |
| min_vals = min_vals.view(B, C) | |
| max_vals = max_vals.view(B, C) | |
| for idx in minmax_indices: | |
| params[:, idx, :, 0] = min_vals # shift | |
| params[:, idx, :, 1] = (max_vals - min_vals) + 1e-8 # scale | |
| # Compute z-norm parameters | |
| if znorm_indices: | |
| # Reshape to (B*C, *spatial) to compute per-channel statistics | |
| vol_reshaped = volume.view(B * C, -1) | |
| if compute_stats_per_sample: | |
| # Compute per sample and channel | |
| mean = vol_reshaped.mean(dim=1) # Shape: (B*C,) | |
| std = vol_reshaped.std(dim=1) # Shape: (B*C,) | |
| else: | |
| # Compute across entire batch but per channel | |
| volume_per_channel = volume.view(C, -1) | |
| mean_per_channel = volume_per_channel.mean(dim=1) # Shape: (C,) | |
| std_per_channel = volume_per_channel.std(dim=1) # Shape: (C,) | |
| mean = mean_per_channel.repeat(B) # Shape: (B*C,) | |
| std = std_per_channel.repeat(B) # Shape: (B*C,) | |
| # Reshape back to (B, C) | |
| mean = mean.view(B, C) | |
| std = std.view(B, C) | |
| # Handle zero std | |
| std = torch.where(std < 1e-8, torch.ones_like(std), std) | |
| # Set clipping value | |
| clip_val = znorm_clip if znorm_clip is not None else 3.0 | |
| for idx in znorm_indices: | |
| # Convert z-norm to shift-scale for mapping [-clip, +clip] to [0, 1] | |
| params[:, idx, :, 0] = mean - std * clip_val # shift | |
| params[:, idx, :, 1] = std * 2 * clip_val # scale | |
| # Compute all percentiles in one call | |
| if percentile_windows_info: | |
| # Collect all unique percentile values | |
| all_percentiles = [] | |
| for _, p_min, p_max in percentile_windows_info: | |
| all_percentiles.extend([p_min, p_max]) | |
| # Always use float for percentile values | |
| unique_percentiles = torch.tensor(sorted(set(all_percentiles)), device=device, dtype=torch.float32) | |
| # Reshape to (B*C, *spatial) to compute per-channel percentiles | |
| vol_reshaped = volume.view(B * C, -1) | |
| # Ensure float dtype for quantile computation | |
| if vol_reshaped.dtype not in [torch.float32, torch.float64]: | |
| vol_reshaped = vol_reshaped.float() | |
| # Compute percentiles efficiently with per-channel sampling if needed | |
| if compute_stats_per_sample: | |
| # Sample if tensor is too large (to avoid quantile() errors) | |
| max_elements_per_channel = 2**24 # ~16M elements per channel | |
| if vol_reshaped.shape[1] > max_elements_per_channel: | |
| # Sample independently for each channel | |
| computed_percentiles_list = [] | |
| for bc in range(B * C): | |
| # Get this channel's data | |
| channel_data = vol_reshaped[bc] | |
| # Sample from this channel | |
| indices = torch.randint( | |
| 0, | |
| channel_data.shape[0], | |
| (max_elements_per_channel,), | |
| device=channel_data.device, | |
| ) | |
| channel_sampled = channel_data[indices] | |
| # Compute percentiles for this channel | |
| channel_percentiles = torch.quantile(channel_sampled, unique_percentiles) | |
| computed_percentiles_list.append(channel_percentiles) | |
| # Stack results | |
| computed_percentiles = torch.stack(computed_percentiles_list, dim=1) # Shape: (num_unique, B*C) | |
| else: | |
| # Small enough to compute directly | |
| computed_percentiles = torch.quantile( | |
| vol_reshaped, unique_percentiles, dim=1 | |
| ) # Shape: (num_unique, B*C) | |
| else: | |
| # Compute across entire batch but per channel | |
| volume_per_channel = volume.view(C, -1) | |
| max_elements_per_channel = 2**24 # ~16M elements per channel | |
| # Sample if needed, independently per channel | |
| if volume_per_channel.shape[1] > max_elements_per_channel: | |
| computed_percentiles_list = [] | |
| for c in range(C): | |
| # Get this channel's data across all batches | |
| channel_data = volume_per_channel[c] | |
| # Sample from this channel | |
| indices = torch.randint( | |
| 0, | |
| channel_data.shape[0], | |
| (max_elements_per_channel,), | |
| device=channel_data.device, | |
| ) | |
| channel_sampled = channel_data[indices] | |
| # Compute percentiles for this channel | |
| channel_percentiles = torch.quantile(channel_sampled, unique_percentiles) | |
| computed_percentiles_list.append(channel_percentiles) | |
| # Stack results | |
| computed_percentiles_per_channel = torch.stack( | |
| computed_percentiles_list, dim=1 | |
| ) # Shape: (num_unique, C) | |
| else: | |
| # Small enough to compute directly | |
| computed_percentiles_per_channel = torch.quantile( | |
| volume_per_channel, unique_percentiles, dim=1 | |
| ) # Shape: (num_unique, C) | |
| # Repeat for each batch element | |
| computed_percentiles = computed_percentiles_per_channel.repeat(1, B) # Shape: (num_unique, B*C) | |
| # Convert back to original dtype | |
| computed_percentiles = computed_percentiles.to(dtype) | |
| # Reshape to (num_unique, B, C) | |
| computed_percentiles = computed_percentiles.view(len(unique_percentiles), B, C) | |
| # Store percentiles list for direct indexing | |
| unique_percentiles_list = unique_percentiles.tolist() | |
| # Assign to params | |
| for idx, p_min, p_max in percentile_windows_info: | |
| # Find indices in the unique percentiles list | |
| low_idx = unique_percentiles_list.index(min(unique_percentiles_list, key=lambda x: abs(x - p_min))) | |
| high_idx = unique_percentiles_list.index(min(unique_percentiles_list, key=lambda x: abs(x - p_max))) | |
| low = computed_percentiles[low_idx] # Shape: (B, C) | |
| high = computed_percentiles[high_idx] # Shape: (B, C) | |
| # Avoid division by zero | |
| scale = high - low | |
| params[:, idx, :, 0] = low # shift | |
| params[:, idx, :, 1] = scale + 1e-8 # scale with epsilon | |
| return params | |
| def _apply_windows_vectorized(volume: torch.Tensor, window_params: torch.Tensor) -> torch.Tensor: | |
| """ | |
| Apply all windows in parallel using shift and scale parameters. | |
| Handles per-channel windowing parameters. | |
| Args: | |
| volume: Shape (B, C, *spatial_dims) | |
| window_params: Shape (B, num_windows, C, 2) - [shift, scale] pairs per channel | |
| Returns: | |
| torch.Tensor: Shape (B, C*num_windows, *spatial_dims) - directly in final layout | |
| """ | |
| B, C = volume.shape[:2] | |
| num_windows = window_params.shape[1] | |
| spatial_dims = volume.shape[2:] | |
| num_spatial = len(spatial_dims) | |
| # Broadcast all windows at once: volume (B, 1, C, *spatial) - shift/scale (B, W, C, 1...) | |
| vol_expanded = volume.unsqueeze(1) # (B, 1, C, *spatial) | |
| expand_dims = (1,) * num_spatial | |
| shift = window_params[:, :, :, 0].view(B, num_windows, C, *expand_dims) | |
| scale = window_params[:, :, :, 1].view(B, num_windows, C, *expand_dims) | |
| # Single fused op: (B, W, C, *spatial) | |
| result = ((vol_expanded - shift) / scale).clamp_(0, 1) | |
| # Interleave to (B, C*W, *spatial): window w, channel c -> output channel c*W + w | |
| # result is (B, W, C, *spatial), we need to transpose W and C then reshape | |
| result = result.transpose(1, 2).reshape(B, C * num_windows, *spatial_dims) | |
| return result | |
| def batch_apply_windowing( | |
| volume: Union[torch.Tensor, np.ndarray], | |
| windows: Union[str, List[str]] = "default", | |
| modality: str = "CT", | |
| min_value: Optional[float] = None, | |
| max_value: Optional[float] = None, | |
| percentile_windows: Optional[Dict[str, Dict[str, float]]] = None, | |
| znorm_clip: Optional[float] = None, | |
| torch_operating_dtype: torch.dtype = torch.float32, | |
| ) -> Union[torch.Tensor, np.ndarray]: | |
| """ | |
| Backward compatible wrapper for batch_apply_windowing_vectorized. | |
| This function calls the vectorized implementation with default settings. | |
| """ | |
| return batch_apply_windowing_vectorized( | |
| volume=volume, | |
| windows=windows, | |
| modality=modality, | |
| min_value=min_value, | |
| max_value=max_value, | |
| percentile_windows=percentile_windows, | |
| znorm_clip=znorm_clip, | |
| torch_operating_dtype=torch_operating_dtype, | |
| compute_stats_per_sample=True, | |
| ) | |
| def apply_windowing( | |
| volume: Union[torch.Tensor, np.ndarray], | |
| windows: Union[str, List[str]] = "default", | |
| modality: str = "CT", | |
| min_value: Optional[float] = None, | |
| max_value: Optional[float] = None, | |
| percentile_windows: Optional[Dict[str, Dict[str, float]]] = None, | |
| znorm_clip: Optional[float] = None, | |
| ) -> Union[torch.Tensor, np.ndarray]: | |
| """ | |
| Apply one or multiple windowing strategies to medical images. | |
| This function supports: | |
| 1. Standard anatomical windows (center/width based) | |
| 2. Min-max normalization | |
| 3. Percentile-based windows | |
| 4. Z-score normalization (standardization) | |
| 5. Custom window specifications | |
| 6. The 'all' keyword to apply all relevant windows | |
| Args: | |
| volume: Input volume as tensor or numpy array. Shape: (D, H, W) or (H, W) | |
| windows: Window name(s) to apply. Options: | |
| - Single string: 'lung', 'bone', 'mediastinum', etc. | |
| - List of strings: ['lung', 'bone'] for multiple windows | |
| - 'default': Uses modality-specific default | |
| - 'minmax': Min-max normalization | |
| - 'znorm': Z-score normalization (recommended for MR) | |
| - 'all': Applies all available windows for the modality | |
| modality: Imaging modality ('CT', 'MR', 'XR', 'MG') | |
| min_value: Override minimum value for minmax windowing | |
| max_value: Override maximum value for minmax windowing | |
| percentile_windows: Custom percentile window definitions | |
| znorm_clip: Clipping value for z-normalized data (e.g., 3.0 clips at ±3 std) | |
| Returns: | |
| Windowed volume(s). If single window: same shape as input. | |
| If multiple windows: shape (N, *input_shape) where N is number of windows | |
| Examples: | |
| >>> # Single window | |
| >>> lung_view = apply_windowing(ct_volume, 'lung') | |
| >>> # Multiple windows | |
| >>> views = apply_windowing(ct_volume, ['lung', 'bone', 'mediastinum']) | |
| >>> # All available windows | |
| >>> all_views = apply_windowing(ct_volume, 'all') | |
| >>> # Custom percentile windows | |
| >>> custom = apply_windowing(volume, 'custom', percentile_windows={ | |
| ... 'custom': {'percentile_min': 5, 'percentile_max': 95} | |
| ... }) | |
| """ | |
| # Convert to tensor if needed | |
| is_numpy = isinstance(volume, np.ndarray) | |
| if is_numpy: | |
| volume = torch.from_numpy(volume).float() | |
| else: | |
| volume = volume.float() | |
| # Handle default windows | |
| if windows == "default": | |
| if modality == "CT": | |
| windows = ["lung", "mediastinum", "bone"] | |
| elif modality == "MR": | |
| # For MR, z-normalization is often the best default | |
| windows = ["znorm"] | |
| elif modality in ["XR", "MG"]: | |
| # For modalities with percentile windows, use standard preset | |
| windows = ["standard", "soft_tissue", "high_contrast"] | |
| else: | |
| windows = "minmax" | |
| # Handle 'all' keyword | |
| if windows == "all": | |
| windows = get_available_windows(modality) | |
| # Convert single window to list for uniform processing | |
| if isinstance(windows, str): | |
| windows = [windows] | |
| # Apply each window | |
| results = [] | |
| for window in windows: | |
| if window == "minmax": | |
| # Min-max normalization | |
| if min_value is None: | |
| min_value = volume.min() | |
| if max_value is None: | |
| max_value = volume.max() | |
| windowed = (volume - min_value) / (max_value - min_value + 1e-8) | |
| windowed = torch.clamp(windowed, 0, 1) | |
| elif window == "znorm": | |
| # Z-score normalization (standardization) | |
| mean = volume.mean() | |
| std = volume.std() | |
| # Avoid division by zero | |
| if std < 1e-8: | |
| windowed = torch.zeros_like(volume) | |
| else: | |
| windowed = (volume - mean) / std | |
| # Apply clipping if specified | |
| if znorm_clip is not None: | |
| windowed = torch.clamp(windowed, -znorm_clip, znorm_clip) | |
| # Scale to [0, 1] for visualization | |
| # Map [-clip, +clip] to [0, 1], or use default [-3, 3] range | |
| clip_val = znorm_clip if znorm_clip is not None else 3.0 | |
| windowed = (windowed + clip_val) / (2 * clip_val) | |
| windowed = torch.clamp(windowed, 0, 1) | |
| elif window in ANATOMICAL_WINDOWS.get(modality, {}): | |
| # Standard anatomical window | |
| params = ANATOMICAL_WINDOWS[modality][window] | |
| windowed = apply_anatomical_window(volume, params["center"], params["width"]) | |
| elif modality in PERCENTILE_WINDOWS and window in PERCENTILE_WINDOWS[modality]: | |
| # Percentile-based window from predefined | |
| params = PERCENTILE_WINDOWS[modality][window] | |
| windowed = _apply_percentile_window(volume, params["percentile_min"], params["percentile_max"]) | |
| elif percentile_windows and window in percentile_windows: | |
| # Custom percentile-based window | |
| params = percentile_windows[window] | |
| windowed = _apply_percentile_window(volume, params["percentile_min"], params["percentile_max"]) | |
| else: | |
| raise ValueError(f"Unknown window: {window}") | |
| results.append(windowed) | |
| # Stack results if multiple windows | |
| if len(results) == 1: | |
| result = results[0] | |
| else: | |
| result = torch.stack(results, dim=0) | |
| # Convert back to numpy if input was numpy | |
| if is_numpy: | |
| result = result.numpy() | |
| return result | |
| def apply_anatomical_window( | |
| volume: Union[torch.Tensor, np.ndarray], center: float, width: float | |
| ) -> Union[torch.Tensor, np.ndarray]: | |
| """ | |
| Apply traditional center/width windowing to medical images. | |
| Args: | |
| volume: Input volume | |
| center: Window center (level) | |
| width: Window width | |
| Returns: | |
| Windowed volume with values in [0, 1] | |
| """ | |
| is_numpy = isinstance(volume, np.ndarray) | |
| if is_numpy: | |
| volume = torch.from_numpy(volume).float() | |
| # Calculate window bounds | |
| min_val = center - width / 2 | |
| max_val = center + width / 2 | |
| # Apply windowing | |
| windowed = (volume - min_val) / (max_val - min_val + 1e-8) | |
| windowed = torch.clamp(windowed, 0, 1) | |
| if is_numpy: | |
| windowed = windowed.numpy() | |
| return windowed | |
| def _apply_percentile_window(volume: torch.Tensor, percentile_min: float, percentile_max: float) -> torch.Tensor: | |
| """Apply percentile-based windowing.""" | |
| # Calculate percentiles | |
| flat_volume = volume.flatten() | |
| num_el = flat_volume.numel() | |
| if num_el >= 2**24: | |
| flat_volume = flat_volume[:: math.ceil(num_el / 2**24)] | |
| p_min = torch.quantile(flat_volume, percentile_min / 100.0) | |
| p_max = torch.quantile(flat_volume, percentile_max / 100.0) | |
| # Avoid division by zero | |
| if p_max <= p_min: | |
| p_max = p_min + 1 | |
| # Apply windowing | |
| windowed = (volume - p_min) / (p_max - p_min) | |
| windowed = torch.clamp(windowed, 0, 1) | |
| return windowed | |
| def apply_multiple_windows( | |
| volume: Union[torch.Tensor, np.ndarray], modality: str = "CT" | |
| ) -> Dict[str, Union[torch.Tensor, np.ndarray]]: | |
| """ | |
| Apply all relevant windows for a given modality. | |
| Args: | |
| volume: Input volume | |
| modality: Imaging modality | |
| Returns: | |
| Dictionary mapping window names to windowed volumes | |
| """ | |
| windows = get_available_windows(modality) | |
| results = {} | |
| for window in windows: | |
| results[window] = apply_windowing(volume, window, modality) | |
| return results | |
| def get_available_windows(modality: str) -> list: | |
| """ | |
| Get list of available windows for a given modality. | |
| Args: | |
| modality: Imaging modality ('CT', 'XR', 'MG', etc.) | |
| Returns: | |
| List of available window names | |
| """ | |
| # Curated MR set | |
| if modality == "MR": | |
| return ["high_contrast", "minmax", "znorm"] | |
| windows = [] | |
| # Add anatomical windows if available | |
| if modality in ANATOMICAL_WINDOWS: | |
| windows.extend(ANATOMICAL_WINDOWS[modality].keys()) | |
| # Add percentile windows if available | |
| if modality in PERCENTILE_WINDOWS: | |
| windows.extend(PERCENTILE_WINDOWS[modality].keys()) | |
| # Always include minmax | |
| windows.append("minmax") | |
| # Include znorm for appropriate modalities | |
| if modality in ["MR", "XR", "MG"]: | |
| windows.append("znorm") | |
| return windows | |