diff --git a/auglab/transforms/gpu/palette/__init__.py b/auglab/transforms/gpu/palette/__init__.py new file mode 100644 index 0000000..636c6ae --- /dev/null +++ b/auglab/transforms/gpu/palette/__init__.py @@ -0,0 +1,17 @@ +from auglab.transforms.gpu.palette.base import ( + BlockContext, + InitialPartitioner, + RefinementPartitioner, + signed_alpha_affine_remap, +) +from auglab.transforms.gpu.palette.factory import build_palette_from_cfg +from auglab.transforms.gpu.palette.transform import PaletteSynthesisGPU + +__all__ = [ + "BlockContext", + "InitialPartitioner", + "RefinementPartitioner", + "signed_alpha_affine_remap", + "PaletteSynthesisGPU", + "build_palette_from_cfg", +] diff --git a/auglab/transforms/gpu/palette/base.py b/auglab/transforms/gpu/palette/base.py new file mode 100644 index 0000000..5a99758 --- /dev/null +++ b/auglab/transforms/gpu/palette/base.py @@ -0,0 +1,65 @@ +"""Base contracts and shared helpers for the composable PALETTE pipeline.""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import Tuple + +import torch +from torch import nn + + +@dataclass +class BlockContext: + """Per-sample state shared across partitioners and remap steps.""" + image01: torch.Tensor # (N,) float, min-max normalized to [0,1] + fg_mask: torch.Tensor # (N,) float 0/1 + coords: torch.Tensor # (N, 3) ijk voxel coords + shape: Tuple[int, int, int] # (D, H, W) + device: torch.device + + +class InitialPartitioner(nn.Module): + """Produce the initial region map from the raw image.""" + + def partition(self, ctx: BlockContext) -> Tuple[torch.Tensor, int]: + raise NotImplementedError + + +class RefinementPartitioner(nn.Module): + """Subdivide an existing region map. May optionally consult the image.""" + + def refine( + self, + ctx: BlockContext, + region_ids: torch.Tensor, + n_regions: int, + ) -> Tuple[torch.Tensor, int]: + raise NotImplementedError + + +def signed_alpha_affine_remap( + image01: torch.Tensor, + fg_mask: torch.Tensor, + region_ids: torch.Tensor, + n_regions: int, + alpha_magnitude_range: Tuple[float, float], + eps: float = 1e-7, +) -> torch.Tensor: + """Signed-alpha per-region affine remap: y = μ_c + α_c · (x − mean_c). + + Mirrors fromSeg.py:392-401. Foreground-only means; output clamped to [0,1] + and multiplied by the foreground mask. + """ + device = image01.device + alpha_lo, alpha_hi = alpha_magnitude_range + + s_c = torch.zeros(n_regions, device=device).scatter_add_(0, region_ids, image01 * fg_mask) + n_c = torch.zeros(n_regions, device=device).scatter_add_(0, region_ids, fg_mask) + mean_c = s_c / n_c.clamp(min=eps) + + mu_c = torch.rand(n_regions, device=device) + mag_c = torch.rand(n_regions, device=device) * (alpha_hi - alpha_lo) + alpha_lo + sign_c = (torch.rand(n_regions, device=device) > 0.5).float() * 2 - 1 + alp_c = mag_c * sign_c + + return (mu_c[region_ids] + alp_c[region_ids] * (image01 - mean_c[region_ids])).clamp(0, 1) * fg_mask diff --git a/auglab/transforms/gpu/palette/factory.py b/auglab/transforms/gpu/palette/factory.py new file mode 100644 index 0000000..c8ea070 --- /dev/null +++ b/auglab/transforms/gpu/palette/factory.py @@ -0,0 +1,113 @@ +"""Config → PaletteSynthesisGPU factory with typed partitioner registries.""" +from __future__ import annotations + +from typing import Any, Dict + +from auglab.transforms.gpu.palette.overlay import AnatomicalLabelOverlay +from auglab.transforms.gpu.palette.partitioners import ( + EMGMMInitial, + EMGMMRefinement, + IdentityRefinement, + KMeans1DInitial, + VoronoiRefinement, +) +from auglab.transforms.gpu.palette.transform import PaletteSynthesisGPU + + +INITIAL_REGISTRY: Dict[str, type] = { + "kmeans1d": KMeans1DInitial, + "em_gmm": EMGMMInitial, +} + +REFINEMENT_REGISTRY: Dict[str, type] = { + "voronoi": VoronoiRefinement, + "em_gmm": EMGMMRefinement, + "identity": IdentityRefinement, +} + +# Top-level config keys consumed directly by PaletteSynthesisGPU (everything +# else must be routed into a nested block or is rejected as unknown). +_TOP_KEYS = { + "alpha_magnitude_range", + "dark_threshold", + "blur_sigmas_pre", + "blur_sigmas_post", + "p", +} + +# Keys accepted at the top of the config but not passed to the transform ctor. +_TOP_META_KEYS = {"probability", "initial_partitioner", "refinement_partitioners", "overlay"} + + +def build_palette_from_cfg(cfg: Dict[str, Any]) -> PaletteSynthesisGPU: + """Build a ``PaletteSynthesisGPU`` from a nested config dict. + + Raises ``ValueError`` if the initial partitioner is missing, if a type + string is placed in the wrong registry slot, or if unknown top-level keys + are present. + """ + if "initial_partitioner" not in cfg: + raise ValueError( + "PaletteSynthesisTransform config requires an 'initial_partitioner' block " + f"(one of: {list(INITIAL_REGISTRY)})" + ) + + unknown = set(cfg) - _TOP_KEYS - _TOP_META_KEYS + if unknown: + raise ValueError( + f"Unknown top-level keys in PaletteSynthesisTransform: {sorted(unknown)}. " + f"Expected any of: {sorted(_TOP_KEYS | _TOP_META_KEYS)}" + ) + + init_cfg = dict(cfg["initial_partitioner"]) + init_type = init_cfg.pop("type", None) + if init_type is None: + raise ValueError("initial_partitioner must specify a 'type' field") + if init_type in REFINEMENT_REGISTRY and init_type not in INITIAL_REGISTRY: + raise ValueError( + f"'{init_type}' is a refinement partitioner, not an initial one. " + f"Move it into 'refinement_partitioners'. " + f"Valid initial types: {list(INITIAL_REGISTRY)}" + ) + if init_type not in INITIAL_REGISTRY: + raise ValueError( + f"Unknown initial partitioner type '{init_type}'. " + f"Valid types: {list(INITIAL_REGISTRY)}" + ) + initial = INITIAL_REGISTRY[init_type](**init_cfg) + + refinements = [] + for i, r_cfg in enumerate(cfg.get("refinement_partitioners", []) or []): + r_cfg = dict(r_cfg) + r_type = r_cfg.pop("type", None) + if r_type is None: + raise ValueError(f"refinement_partitioners[{i}] must specify a 'type' field") + if r_type in INITIAL_REGISTRY and r_type not in REFINEMENT_REGISTRY: + raise ValueError( + f"'{r_type}' is an initial partitioner and cannot be used as a refinement. " + f"Valid refinement types: {list(REFINEMENT_REGISTRY)}" + ) + if r_type not in REFINEMENT_REGISTRY: + raise ValueError( + f"Unknown refinement partitioner type '{r_type}'. " + f"Valid types: {list(REFINEMENT_REGISTRY)}" + ) + refinements.append(REFINEMENT_REGISTRY[r_type](**r_cfg)) + + ov_cfg = cfg.get("overlay") + overlay = None + if ov_cfg is not None: + ov_kwargs = {k: v for k, v in ov_cfg.items() if k != "enabled"} + if ov_cfg.get("enabled", True): + overlay = AnatomicalLabelOverlay(**ov_kwargs) + + top_kwargs = {k: cfg[k] for k in _TOP_KEYS if k in cfg} + if "probability" in cfg and "p" not in top_kwargs: + top_kwargs["p"] = cfg["probability"] + + return PaletteSynthesisGPU( + initial_partitioner=initial, + refinement_partitioners=refinements, + overlay=overlay, + **top_kwargs, + ) diff --git a/auglab/transforms/gpu/palette/overlay.py b/auglab/transforms/gpu/palette/overlay.py new file mode 100644 index 0000000..265b826 --- /dev/null +++ b/auglab/transforms/gpu/palette/overlay.py @@ -0,0 +1,108 @@ +"""Per-anatomical-label overlay: fixed algorithm, tunable frequency and blend.""" +from __future__ import annotations + +from typing import List, Optional, Sequence, Tuple, Union + +import torch +from torch import nn +from torch.nn import functional as F + + +BlendSpec = Union[float, Sequence[float]] + + +class AnatomicalLabelOverlay(nn.Module): + """PALETTE per-anatomical-label affine remap with a tunable blend. + + Reproduces fromSeg.py:414-447 algorithmically. For each foreground label, + with probability ``label_remap_prob``, samples a fresh (μ, α) and computes + ``new_vals = μ + α·(synth − mean_label)``, then blends into the current + synthesised image via a per-label blend strength. + + Args: + label_remap_prob: per-label per-sample probability of applying the remap. + min_label_voxels: minimum voxel count for a label to be eligible. + label_classes: if set, restrict overlay to these class indices. + blend_strength: scalar in [0,1] (full overwrite = 1.0) or [lo, hi] + sampled per label per sample. Applied as + ``write_mask = c_mask · apply · blend``. + alpha_magnitude_range: [lo, hi] for |α| of the signed-alpha remap. + """ + + def __init__( + self, + label_remap_prob: float = 0.5, + min_label_voxels: int = 4, + label_classes: Optional[List[int]] = None, + blend_strength: BlendSpec = 1.0, + alpha_magnitude_range: Sequence[float] = (0.5, 2.0), + ) -> None: + super().__init__() + self.label_remap_prob = float(label_remap_prob) + self.min_label_voxels = int(min_label_voxels) + self.label_classes = None if label_classes is None else list(label_classes) + self.blend_strength = _normalize_blend(blend_strength) + self.alpha_magnitude_range = tuple(alpha_magnitude_range) + + def apply( + self, + synth: torch.Tensor, # (B, N) + labels: torch.Tensor, # (B, 1, D, H, W) long + shape: Tuple[int, int, int], + ) -> torch.Tensor: + B, N = synth.shape + device = synth.device + alpha_lo, alpha_hi = self.alpha_magnitude_range + blend_lo, blend_hi = self.blend_strength + D, H, W = shape + + if labels.shape[2:] != (D, H, W): + labels = F.interpolate(labels.float(), size=(D, H, W), mode="nearest").long() + lbl = labels[:, 0].reshape(B, N).clamp(min=0) + + unique_classes = lbl.unique() + unique_classes = unique_classes[unique_classes > 0] + if self.label_classes is not None: + keep = torch.tensor(self.label_classes, device=device) + unique_classes = unique_classes[torch.isin(unique_classes, keep)] + + for c in unique_classes: + c_val = int(c.item()) + c_mask = (lbl == c_val).float() + c_cnt = c_mask.sum(dim=1, keepdim=True) + + apply = ( + (torch.rand(B, 1, device=device) < self.label_remap_prob) + & (c_cnt >= self.min_label_voxels) + ).float() + + if apply.sum() == 0: + continue + + c_mean = (synth * c_mask).sum(dim=1, keepdim=True) / c_cnt.clamp(min=1) + + mu_c = torch.rand(B, 1, device=device) + mag_c = torch.rand(B, 1, device=device) * (alpha_hi - alpha_lo) + alpha_lo + sign_c = (torch.rand(B, 1, device=device) > 0.5).float() * 2 - 1 + alp_c = mag_c * sign_c + + if blend_lo == blend_hi: + blend = torch.full((B, 1), blend_lo, device=device) + else: + blend = torch.rand(B, 1, device=device) * (blend_hi - blend_lo) + blend_lo + + new_vals = (mu_c + alp_c * (synth - c_mean)).clamp(0, 1) + write_mask = c_mask * apply * blend + synth = synth * (1.0 - write_mask) + new_vals * write_mask + + return synth + + +def _normalize_blend(spec: BlendSpec) -> Tuple[float, float]: + if isinstance(spec, (int, float)): + v = float(spec) + return (v, v) + lo, hi = float(spec[0]), float(spec[1]) + if lo > hi: + raise ValueError(f"blend_strength range must be non-decreasing, got [{lo}, {hi}]") + return (lo, hi) diff --git a/auglab/transforms/gpu/palette/partitioners.py b/auglab/transforms/gpu/palette/partitioners.py new file mode 100644 index 0000000..f06dbc7 --- /dev/null +++ b/auglab/transforms/gpu/palette/partitioners.py @@ -0,0 +1,201 @@ +"""Concrete partitioner blocks for the composable PALETTE pipeline.""" +from __future__ import annotations + +from typing import List, Sequence, Tuple + +import torch + +from auglab.transforms.gpu.fromSeg import _kmeans_1d, _voronoi_region_ids +from auglab.transforms.gpu.palette.base import ( + BlockContext, + InitialPartitioner, + RefinementPartitioner, +) +from auglab.transforms.synthseg.functional import em_subdivide_labels + + +# ── initial partitioners ──────────────────────────────────────────────────── + +class KMeans1DInitial(InitialPartitioner): + """1-D K-means on foreground intensities, then bucketize the whole image. + + Mirrors fromSeg.py:374-385. With probability ``skip_prob`` (or when there + are fewer than 4 foreground voxels) returns a single-region partition; the + downstream signed-alpha remap handles that as a global remap. + """ + + def __init__( + self, + c_choices: Sequence[int] = (2, 3, 4, 5, 6), + n_kmeans_subsample: int = 10_000, + skip_prob: float = 0.10, + dark_threshold: float = 0.01, + ) -> None: + super().__init__() + self.c_choices = list(c_choices) + self.n_kmeans_subsample = int(n_kmeans_subsample) + self.skip_prob = float(skip_prob) + self.dark_threshold = float(dark_threshold) + + def partition(self, ctx: BlockContext) -> Tuple[torch.Tensor, int]: + device = ctx.device + flat = ctx.image01 + N = flat.shape[0] + n_fg = int(ctx.fg_mask.sum().item()) + + if n_fg < 4 or torch.rand(1, device=device).item() < self.skip_prob: + return torch.zeros(N, dtype=torch.long, device=device), 1 + + C_k = self.c_choices[int(torch.rand(1, device=device).item() * len(self.c_choices))] + idx = torch.randint(0, N, (min(N, 40_000),), device=device) + samp = flat[idx] + sub_fg = samp[samp > self.dark_threshold][: self.n_kmeans_subsample] + if sub_fg.numel() < 4: + sub_fg = samp[: self.n_kmeans_subsample] + + centroids = _kmeans_1d(sub_fg, C_k) + sorted_c, sort_idx = torch.sort(centroids) + boundaries = (sorted_c[:-1] + sorted_c[1:]) / 2.0 + lbl_s = torch.bucketize(flat, boundaries) + lbl_l = sort_idx[lbl_s].long() + return lbl_l, C_k + + +class EMGMMInitial(InitialPartitioner): + """SynthSeg-style EM/GMM clustering on the raw image. + + Calls ``em_subdivide_labels`` with a two-label bg/fg map (``fg_mask`` as + the label input). Background is split into ``background_clusters_range`` + subclusters; foreground into ``n_foreground_clusters``. + """ + + def __init__( + self, + n_foreground_clusters: int = 3, + background_clusters_range: Sequence[int] = (3, 10), + background_label: int = 0, + n_iters: int = 20, + max_fit_voxels: int = 100_000, + ) -> None: + super().__init__() + self.n_foreground_clusters = int(n_foreground_clusters) + self.background_clusters_range = tuple(background_clusters_range) + self.background_label = int(background_label) + self.n_iters = int(n_iters) + self.max_fit_voxels = int(max_fit_voxels) + + def partition(self, ctx: BlockContext) -> Tuple[torch.Tensor, int]: + D, H, W = ctx.shape + image = ctx.image01.view(1, 1, D, H, W) + label_map = ctx.fg_mask.view(1, 1, D, H, W).long() + fine, _gen, _out = em_subdivide_labels( + image=image, + label_map=label_map, + n_foreground_clusters=self.n_foreground_clusters, + background_clusters_range=self.background_clusters_range, + background_label=self.background_label, + n_iters=self.n_iters, + max_fit_voxels=self.max_fit_voxels, + channel=0, + ) + rid_flat = fine.view(-1).long() + return _densify_region_ids(rid_flat) + + +# ── refinement partitioners ──────────────────────────────────────────────── + +class VoronoiRefinement(RefinementPartitioner): + """Spatially subdivide each existing region into S seed-nearest cells. + + Wraps _voronoi_region_ids (fromSeg.py:44) — coord-only, does not read the + image. With per-region probability ``skip_prob`` the region is left intact. + """ + + def __init__( + self, + s_choices: Sequence[int] = (2, 3, 4, 5, 6, 7, 8, 9, 10), + skip_prob: float = 0.40, + ) -> None: + super().__init__() + self.s_choices = list(s_choices) + self.skip_prob = float(skip_prob) + + def refine( + self, + ctx: BlockContext, + region_ids: torch.Tensor, + n_regions: int, + ) -> Tuple[torch.Tensor, int]: + return _voronoi_region_ids( + ctx.coords, region_ids, ctx.fg_mask, + n_regions, ctx.device, self.s_choices, self.skip_prob, + ) + + +class EMGMMRefinement(RefinementPartitioner): + """Subdivide each existing region by intensity via SynthSeg's EM/GMM. + + Each incoming region is treated as one "label" fed to ``em_subdivide_labels`` + with ``n_foreground_clusters`` sub-clusters per region. + """ + + def __init__( + self, + n_foreground_clusters: int = 2, + background_clusters_range: Sequence[int] = (2, 4), + background_label: int = 0, + n_iters: int = 20, + max_fit_voxels: int = 100_000, + ) -> None: + super().__init__() + self.n_foreground_clusters = int(n_foreground_clusters) + self.background_clusters_range = tuple(background_clusters_range) + self.background_label = int(background_label) + self.n_iters = int(n_iters) + self.max_fit_voxels = int(max_fit_voxels) + + def refine( + self, + ctx: BlockContext, + region_ids: torch.Tensor, + n_regions: int, + ) -> Tuple[torch.Tensor, int]: + D, H, W = ctx.shape + image = ctx.image01.view(1, 1, D, H, W) + label_map = region_ids.view(1, 1, D, H, W).long() + fine, _gen, _out = em_subdivide_labels( + image=image, + label_map=label_map, + n_foreground_clusters=self.n_foreground_clusters, + background_clusters_range=self.background_clusters_range, + background_label=self.background_label, + n_iters=self.n_iters, + max_fit_voxels=self.max_fit_voxels, + channel=0, + ) + return _densify_region_ids(fine.view(-1).long()) + + +class IdentityRefinement(RefinementPartitioner): + """Passthrough — leaves the running partition unchanged.""" + + def refine( + self, + ctx: BlockContext, + region_ids: torch.Tensor, + n_regions: int, + ) -> Tuple[torch.Tensor, int]: + return region_ids, n_regions + + +# ── helpers ──────────────────────────────────────────────────────────────── + +def _densify_region_ids(region_ids: torch.Tensor) -> Tuple[torch.Tensor, int]: + """Remap sparse integer region ids to a contiguous [0, R) range. + + ``em_subdivide_labels`` encodes fine ids as ``parent_idx * mult + assign``, + which is sparse. The downstream signed-alpha remap uses region ids as + indices into per-region tensors of size ``R``, so they must be contiguous. + """ + unique, inverse = torch.unique(region_ids, return_inverse=True) + return inverse.long(), int(unique.numel()) diff --git a/auglab/transforms/gpu/palette/transform.py b/auglab/transforms/gpu/palette/transform.py new file mode 100644 index 0000000..e794ecb --- /dev/null +++ b/auglab/transforms/gpu/palette/transform.py @@ -0,0 +1,149 @@ +"""Composed PALETTE synthesis transform.""" +from __future__ import annotations + +import random +from typing import Any, Dict, List, Optional, Sequence + +import torch +from kornia.core import Tensor +from torch import nn + +from auglab.transforms.gpu.base import ImageOnlyTransform +from auglab.transforms.gpu.fromSeg import _gaussian_blur_3d, collapse_onehot_to_index +from auglab.transforms.gpu.palette.base import ( + BlockContext, + InitialPartitioner, + RefinementPartitioner, + signed_alpha_affine_remap, +) +from auglab.transforms.gpu.palette.overlay import AnatomicalLabelOverlay + + +class PaletteSynthesisGPU(ImageOnlyTransform): + """PALETTE contrast synthesis composed from swappable partition blocks. + + Pipeline: min-max normalise → initial_partitioner → refinement_partitioners → + signed-alpha per-region affine remap → optional blur → anatomical-label + overlay → optional blur → foreground z-score. + + The initial partitioner produces the region map from raw intensities + (k-means, EM). Refinement partitioners subdivide that partition further + (Voronoi, EM). The intensity remap step is fixed. The overlay algorithm is + fixed; only its frequency and blend amount are tunable. + """ + + def __init__( + self, + initial_partitioner: InitialPartitioner, + refinement_partitioners: Optional[List[RefinementPartitioner]] = None, + overlay: Optional[AnatomicalLabelOverlay] = None, + alpha_magnitude_range: Sequence[float] = (0.5, 2.0), + dark_threshold: float = 0.01, + blur_sigmas_pre: Sequence[float] = (0.0, 0.0, 0.0, 0.3, 0.5, 0.8), + blur_sigmas_post: Sequence[float] = (0.0, 0.0, 0.0, 0.3, 0.5, 0.8), + p: float = 1.0, + **kwargs: Any, + ) -> None: + super().__init__(p=p, **kwargs) + if not isinstance(initial_partitioner, InitialPartitioner): + raise TypeError( + f"initial_partitioner must be an InitialPartitioner, " + f"got {type(initial_partitioner).__name__}" + ) + refinements = list(refinement_partitioners or []) + for r in refinements: + if not isinstance(r, RefinementPartitioner): + raise TypeError( + f"refinement_partitioners must be RefinementPartitioner " + f"instances, got {type(r).__name__}" + ) + self.initial = initial_partitioner + self.refinements = nn.ModuleList(refinements) + self.overlay = overlay + self.alpha_magnitude_range = tuple(alpha_magnitude_range) + self.dark_threshold = float(dark_threshold) + self.blur_sigmas_pre = list(blur_sigmas_pre) + self.blur_sigmas_post = list(blur_sigmas_post) + + @torch.no_grad() + def apply_transform( + self, + input: Tensor, + params: Dict[str, Any], + flags: Dict[str, Any], + transform: Optional[Tensor] = None, + ) -> Tensor: + seg_raw: Optional[torch.Tensor] = params.get("seg", None) + labels: Optional[torch.Tensor] = None + if seg_raw is not None and seg_raw.ndim == 5 and seg_raw.shape[1] > 1: + labels = collapse_onehot_to_index(seg_raw) + elif seg_raw is not None and seg_raw.ndim == 5 and seg_raw.shape[1] == 1: + labels = seg_raw.long() + + B, _C, D, H, W = input.shape + N = D * H * W + device = input.device + eps = 1e-7 + + # 1. Per-sample min-max normalize (channel 0) to [0, 1] + flat_all = input[:, 0].float().reshape(B, N) + v_min = flat_all.min(dim=1).values.view(B, 1) + v_max = flat_all.max(dim=1).values.view(B, 1) + images_01 = ((flat_all - v_min) / (v_max - v_min + eps)).clamp(0, 1) + flat_m_all = (images_01 > self.dark_threshold).float() + + coords = torch.stack(torch.meshgrid( + torch.arange(D, device=device, dtype=torch.float32), + torch.arange(H, device=device, dtype=torch.float32), + torch.arange(W, device=device, dtype=torch.float32), + indexing="ij"), dim=-1).reshape(N, 3) + + # 2. Per sample: partition stack, then fixed signed-alpha remap + synth_list = [] + for i in range(B): + ctx = BlockContext( + image01=images_01[i], + fg_mask=flat_m_all[i], + coords=coords, + shape=(D, H, W), + device=device, + ) + rid, R = self.initial.partition(ctx) + for r in self.refinements: + rid, R = r.refine(ctx, rid, R) + synth_i = signed_alpha_affine_remap( + ctx.image01, ctx.fg_mask, rid, R, self.alpha_magnitude_range, + ) + synth_list.append(synth_i) + + synth = torch.stack(synth_list) # (B, N) + synth_01 = synth.reshape(B, 1, D, H, W) + + # 3. Optional pre-overlay blur + sigma = random.choice(self.blur_sigmas_pre) if self.blur_sigmas_pre else 0.0 + if sigma > 0.0: + synth_01 = _gaussian_blur_3d(synth_01, sigma) + synth = synth_01.reshape(B, N) + + # 4. Anatomical-label overlay (fixed algorithm, tunable knobs) + if self.overlay is not None and labels is not None: + synth = self.overlay.apply(synth, labels, (D, H, W)) + + # 5. Optional post-overlay blur + synth_01 = synth.reshape(B, 1, D, H, W) + sigma2 = random.choice(self.blur_sigmas_post) if self.blur_sigmas_post else 0.0 + if sigma2 > 0.0: + synth_01 = _gaussian_blur_3d(synth_01, sigma2) + synth = synth_01.reshape(B, N) + + # 6. Foreground z-score + b_sum = (synth * flat_m_all).sum(dim=1, keepdim=True) + b_cnt = flat_m_all.sum(dim=1, keepdim=True).clamp(min=1) + b_mean = b_sum / b_cnt + b_sq = ((synth - b_mean) * flat_m_all).pow(2).sum(dim=1, keepdim=True) + b_std = (b_sq / b_cnt + eps).sqrt() + synth_z = ((synth - b_mean) / b_std * flat_m_all).reshape(B, 1, D, H, W) + + out = input.clone() + out[:, 0:1] = synth_z.to(input.dtype) + return out diff --git a/auglab/transforms/gpu/transforms.py b/auglab/transforms/gpu/transforms.py index 956f087..0fbf309 100644 --- a/auglab/transforms/gpu/transforms.py +++ b/auglab/transforms/gpu/transforms.py @@ -93,6 +93,13 @@ def _build_transforms(self) -> list[nn.Module]: ) ) + # Composable PALETTE: swappable image-based initial partitioner + # (kmeans1d, em_gmm) + optional stack of refinements (voronoi, em_gmm). + palette_composed_params = self.transform_params.get("PaletteSynthesisTransform") + if palette_composed_params is not None: + from auglab.transforms.gpu.palette.factory import build_palette_from_cfg + transforms.append(build_palette_from_cfg(palette_composed_params)) + # Domain transfer: randomly re-render the image as another sequence/cluster (TA) # Accept either the class-name key or the descriptive key. domain_params = self.transform_params.get('RandomDomainTransferGPU') \