^jQ UdZddlmZddlmZddlmZddlZddlmZddl m Z ddl m Z e Z d d giZd Zd Zd Zded<d#dZd$dZd%dZd&dZd&dZd&dZd&dZd&dZd&dZd&dZd&dZeeeeeeeedZded< d' d(dZeef d)dZ d*d Z! d+d!Z" d, d-d"Z#y).abKornia-based GPU augmentation pipeline for RF-DETR training. This module provides GPU-side augmentation as an alternative to the CPU-based Albumentations pipeline. All transforms run on the device where the batch already resides (typically CUDA), avoiding a CPU-GPU round-trip per sample. Supports detection (boxes only) and segmentation (boxes + instance masks). Usage:: from rfdetr.datasets.kornia_transforms import ( build_kornia_pipeline, build_normalize, collate_boxes, collate_masks, unpack_boxes, ) # Detection: pipeline = build_kornia_pipeline(aug_config, resolution=560) normalize = build_normalize() boxes_padded, valid = collate_boxes(targets, device) img_aug, boxes_aug = pipeline(img, boxes_padded) img_aug = normalize(img_aug) targets = unpack_boxes(boxes_aug, valid, targets, H, W) # Segmentation (Phase 2): pipeline = build_kornia_pipeline(aug_config, resolution=560, with_masks=True) normalize = build_normalize() boxes_padded, valid = collate_boxes(targets, device) masks_padded = collate_masks(targets, device, n_max=valid.shape[1], image_height=H, image_width=W) img_aug, boxes_aug, masks_aug = pipeline(img, boxes_padded, masks_padded) img_aug = normalize(img_aug) targets = unpack_boxes(boxes_aug, valid, targets, H, W, masks_aug=masks_aug) ) annotations)Callable)AnyN)Tensor)#filter_keypoint_hflip_augmentations) get_loggerbuild_kornia_pipelinekornia)g ףp= ?gv/?gCl?)gZd;O?gy&1?g??float_MASK_BINARIZE_THRESHOLDcBddlm}t|jdS)aReturn ``True`` when the runtime has a CUDA accelerator available. Uses the fork-safe global ``DEVICE`` constant from ``rfdetr.config`` so that the CUDA driver context is not created in the main process before forking (fork-based DDP and some notebook environments). Returns: ``True`` if at least one CUDA device is reachable; ``False`` otherwise. Examples: >>> _has_cuda_device() # doctest: +SKIP False rDEVICEcuda) rfdetr.configrstr startswithrs l/var/www/ramen.bs-engineer-server.com/venv/lib/python3.12/site-packages/rfdetr/datasets/kornia_transforms.py_has_cuda_devicerDs% v; ! !& ))c|dk(ry|dk(rtsy ddl}y|dk(r ts tdt yt d|d#t$rYywxYw) aResolve an ``augmentation_backend`` value to a concrete ``"cpu"`` or ``"gpu"``. ``"auto"`` resolves to ``"gpu"`` only when both CUDA and Kornia are available; otherwise it falls back to ``"cpu"``. Explicit ``"cpu"`` and ``"gpu"`` values pass through unchanged; ``"gpu"`` is validated (CUDA + kornia presence). Args: backend: One of ``"cpu"``, ``"auto"``, or ``"gpu"``. Returns: ``"cpu"`` or ``"gpu"``. Raises: RuntimeError: When *backend* is ``"gpu"`` and no CUDA device is found. ImportError: When *backend* is ``"gpu"`` and kornia is not installed. ValueError: When *backend* is not one of ``"cpu"``, ``"auto"``, or ``"gpu"``. Examples: >>> resolve_augmentation_backend("cpu") 'cpu' cpuautorNgpuz1augmentation_backend='gpu' requires a CUDA devicezUnknown augmentation_backend z#; expected 'cpu', 'auto', or 'gpu'.)rkornia.augmentation ImportError RuntimeError_require_kornia ValueError)backendr s rresolve_augmentation_backendr"Vsz*%&!  &%!RS S 4WK?bc dd  sA AAcH ddl}y#t$r}td|d}~wwxYw)zVerify that Kornia is importable, raising a clear error if not. Raises: ImportError: When ``kornia`` is not installed, with an install hint. rNzLGPU augmentation requires kornia. Install with: pip install 'rfdetr[kornia]')rr)r es rrr}s, q" qhioppqs ! !c@ddlm}||jddS)z:Build a ``K.RandomHorizontalFlip`` from aug_config params.r)RandomHorizontalFlippr r')rr&get)paramsr&s r_make_horizontal_flipr+s8 &**S#"6 77rc@ddlm}||jddS)z8Build a ``K.RandomVerticalFlip`` from aug_config params.r)RandomVerticalFlipr'r r()rr-r))r*r-s r_make_vertical_flipr.s6  3 4 55rcddlm}|jdd}t|tt fr t |n| |f}|||jdd}t |dd }t|tr d |vr||d <|S) zBuild a ``K.RandomRotation`` from aug_config params. The ``limit`` parameter may be a scalar (symmetric range) or a tuple. r)RandomRotationlimitr'r )degreesr'flagsNr3)rr0r) isinstancelisttuplegetattrdict)r*r0r1r3rotationr4s r _make_rotater;s| 3 JJw #E(u >eEleVUOGgC1EFH Hgt ,E%9E#9"i Orc nddlm}|jd}|Pt|tt fr7t |dk(r)tt|dt|d}||f}n|}nd}||jdd||jd |jd |jd d  S)aBuild a ``K.RandomAffine`` from aug_config params. Albumentations ``translate_percent`` is a ``(min, max)`` signed range (e.g. ``(-0.1, 0.1)``). Kornia ``translate`` is a non-negative per-axis max fraction ``(tx, ty)`` where translation is sampled from ``[-tx, tx]``. The conversion takes ``max(|min|, |max|)`` for each axis, producing a symmetric range that matches the intent. r) RandomAffinetranslate_percentNrotate)ir2scaleshearr'r )r3 translaterBrCr') rr=r)r5r6r7lenmaxabs)r*r=r>trDs r _make_affinerIs1 #67$ '$ 7C@QFI)I  8Y/jj!jj! **S#   rc ddlm}||jdd|jdd|jdd|jdd|jdd  S) zBuild a ``K.ColorJiggle`` from aug_config ``ColorJitter`` params. Note: Kornia >=0.7 uses ``ColorJiggle``; the ``ColorJitter`` alias was added in later versions. We use ``ColorJiggle`` for broad compatibility. r ColorJiggle brightnessgcontrast saturationhuer'r )rMrNrOrPr'rrLr)r*rLs r_make_color_jitterrSsZ 0 ::lC0J,::lC0 JJuc " **S#   rcddlm}||jdd|jdd|jddS) zCBuild a ``K.ColorJiggle`` from ``RandomBrightnessContrast`` params.rrKbrightness_limitg?contrast_limitr'r )rMrNr'rQrRs r _make_random_brightness_contrastrWs>/ ::0#6,c2 **S#  rcddlm}|jdd}|dzdk(r|dz}td|}|||fd|jdd  S) zBuild a ``K.RandomGaussianBlur`` from aug_config params. ``blur_limit`` is rounded up to an odd number for the kernel size. r)RandomGaussianBlur blur_limitr?r@)g?g@r'r ) kernel_sizesigmar')rrYr)rF)r*rYrZs r_make_gaussian_blurr^s^ 7L!,JA~!^ Q #J , **S#  rclddlm}|jdd}||d|jddS) zBuild a ``K.RandomGaussianNoise`` from aug_config params. Kornia takes a single ``std`` value; we use the upper bound of ``std_range`` as an acceptable approximation. r)RandomGaussianNoise std_range)g{Gz?g?r@r'r )stdr')rr`r))r*r`ras r_make_gauss_noisercs9 8 ; 5I  aL **S#  r)HorizontalFlip VerticalFlipRotateAffine ColorJitterRandomBrightnessContrast GaussianBlur GaussNoisez*dict[str, Callable[[dict[str, Any]], Any]] _REGISTRYc ptddlm}t||tj }t |tsJg}|jD]R\}}tj|} | td|dttd|j| |T|rgdndd g} ||d | iS) aBuild a Kornia ``AugmentationSequential`` from an aug_config dict. Each key in *aug_config* is looked up in ``_REGISTRY`` and instantiated with the corresponding parameter dict. Unknown keys raise ``ValueError``. Args: aug_config: Mapping of augmentation names to parameter dicts, identical to the format accepted by the Albumentations path (e.g. ``{"HorizontalFlip": {"p": 0.5}}``). resolution: Target image resolution in pixels (currently reserved for future resolution-aware augmentations). with_masks: When ``True``, include ``"mask"`` in ``data_keys`` so instance segmentation masks are augmented in sync with images and boxes. The pipeline then expects three inputs ``(img, boxes, masks)`` and returns three outputs. Defaults to ``False`` (detection-only, two inputs/outputs). include_keypoints: When ``True``, keypoint-unsafe horizontal-flip transforms are dropped with a warning before the Kornia pipeline is built. Returns: A ``kornia.augmentation.AugmentationSequential`` instance. Raises: ValueError: If *aug_config* contains an unsupported augmentation key. Examples: >>> from rfdetr.datasets.aug_configs import AUG_CONSERVATIVE >>> pipeline = build_kornia_pipeline(AUG_CONSERVATIVE, resolution=560) >>> pipeline_seg = build_kornia_pipeline(AUG_CONSERVATIVE, resolution=560, with_masks=True) r)AugmentationSequential)include_keypointswarnzUnknown augmentation key z) for Kornia GPU backend. Supported keys: .)input bbox_xyxymaskrrrs data_keys)rrrnrloggerwarningr5r9itemsrlr)r sortedappend) aug_config resolution with_masksrornfiltered_aug_config transformsnamer*factoryrus rr r sD:=+ ^^ )4 00 0J+113+ f--% ?+D83\]cdm]n\oopq  '&/* +3=.7KBXI !   rc6tddlm}|||S)aABuild a Kornia ``Normalize`` transform for GPU-side normalization. Args: mean: Per-channel mean values. Defaults to ImageNet statistics. std: Per-channel standard deviation values. Defaults to ImageNet statistics. Returns: A ``kornia.augmentation.Normalize`` instance. r) Normalize)meanrb)rrr)rrbrs rbuild_normalizerWs -    rct|dk(r@tjddd|tjddtj|fS|Dcgc]}|djd}}|r t |nd}t|}|dk(r@tj|dd|tj|dtj|fStj||d|}tj||tj|}t |D]2\}}|djd} | dkDs|d||d| f<d||d| f<4||fScc}w)uPack variable-length xyxy boxes into a padded tensor and valid mask. Kornia ``AugmentationSequential`` expects boxes as ``[B, N_max, 4]``. This function zero-pads each image's boxes to the maximum count in the batch and returns a boolean mask indicating which entries are real. Args: targets: List of target dicts (one per image), each containing a ``"boxes"`` key with an ``[N_i, 4]`` tensor in xyxy format. device: Device on which to allocate the output tensors. Returns: Tuple of: - ``boxes_padded`` — ``[B, N_max, 4]`` float tensor (zero-padded). - ``valid_mask`` — ``[B, N_max]`` bool tensor (``True`` = real box). When ``B == 0`` or all images have zero boxes, both tensors have ``N_max == 0``. r)devicedtyperboxesNT)rEtorchzerosboolshaperF enumerate) targetsrrH box_countsn_max batch_size boxes_padded valid_maskins r collate_boxesrssP* 7|q KK1a / KK1EJJv >  077!!G*""1%7J7)C OqEWJ z KK Aq 8 KK AUZZ G  ;;z5!FCLZejjPJ'"%1 gJ  Q  q5"#G*LBQB  $Jq"1"u  %  ##'8sEcFt|}tj||||tj|}t |D]^\}}d|vs|dk(r|dj tj|} t | jd|} | dkDsS| d| ||d| f<`|S)uPack variable-length instance masks into a zero-padded ``[B, N_max, H, W]`` tensor. Kornia ``AugmentationSequential`` expects masks as ``[B, N_max, H, W]`` when ``data_keys`` includes ``"mask"``. This function zero-pads each image's masks to *n_max* channels (matching the padding used by :func:`collate_boxes`) and converts boolean masks to ``float32`` for Kornia compatibility. Args: targets: List of target dicts (one per image). Each dict may optionally contain a ``"masks"`` key with an ``[N_i, H, W]`` boolean tensor. Dicts without the key are treated as having zero instances. device: Device on which to allocate the output tensor. n_max: Maximum instance count across the batch — must equal ``collate_boxes(targets, device)[1].shape[1]`` to keep box/mask indices in sync. image_height: Spatial height ``H`` of each mask (pixels). image_width: Spatial width ``W`` of each mask (pixels). Returns: Float32 tensor of shape ``[B, N_max, H, W]``, zero-padded where ``N_i < N_max``. Boolean input masks are cast to ``float32`` (``True → 1.0``, ``False → 0.0``). Examples: >>> import torch >>> targets = [{"masks": torch.ones(2, 8, 8, dtype=torch.bool)}] >>> out = collate_masks(targets, torch.device("cpu"), n_max=2, image_height=8, image_width=8) >>> out.shape torch.Size([1, 2, 8, 8]) >>> out.dtype torch.float32 rmasksrN)rErrfloat32rtominr) rrr image_height image_widthr masks_paddedrrHmasks_irs r collate_masksrsHWJ;;z5, SXS`S`iopL'".1 ! uz G*--emmF-C  a % ( q5")"1+LBQB  . rc|S|jdd|jk(s7Jdt|jdddt|jdg}t|D]\}}|j}|djd} | dk(s|jddk(r|j |R||d| f} ||d| f} | j } | dddfj d| | dddfj d| | dddfj d| | ddd fj d| | dddf| dddfz } | ddd f| dddfz } | | dkDz| dkDz}| ||d<d |vr |d ||d <d |vr/|d}|dddf|dddfz |ddd f|dddfz z|d <d |vr |d ||d <|||d| f}||tkD|d<|j ||S)ueUnpack augmented boxes (and optionally masks), clamp to image bounds, remove zero-area boxes. After Kornia augmentation the padded ``[B, N_max, 4]`` tensor is unpacked back into per-image target dicts. Boxes are clamped to ``[0, W] x [0, H]`` and any that collapse to zero area are removed along with their corresponding ``labels``, ``area``, ``iscrowd``, and (if provided) ``masks`` entries. Args: boxes_aug: Augmented boxes tensor ``[B, N_max, 4]`` in xyxy format. valid: Boolean mask ``[B, N_max]`` from :func:`collate_boxes`. targets: Original target dicts; each dict is shallow-copied before modification — the input list itself is not mutated. image_height: Image height in pixels (for clamping). image_width: Image width in pixels (for clamping). masks_aug: Optional augmented masks tensor ``[B, N_max, H, W]`` (float32) from Kornia. When provided, masks are filtered by the same ``keep`` mask as boxes, thresholded at ``> 0.5`` to bool, and stored under ``"masks"`` in each output target dict. When ``None``, any existing ``"masks"`` entry in the target dict is preserved unchanged. Returns: A new list of target dicts with updated ``boxes``, ``labels``, ``area``, ``iscrowd``, and (when *masks_aug* is given) ``masks`` entries. Nr?zmasks_aug batch/n_max dims z must match valid shape zM; ensure collate_masks is called with n_max=valid.shape[1] from collate_boxesrrr@)rrFr[labelsareaiscrowdr)rr7rcopyrzcloneclamp_r ) boxes_augvalidrrr masks_aug new_targetsrrHn_origvboxes_iwidthsheightskeep kept_boxesrs r unpack_boxesrs<r"ekk1 )% 0C*D)EF -./6 6 1 )+K'"(1 FFH7!!!$ Q;%++a.A-   q !  !WfW* AwwJ'--/1  41  51  41  5AA.!Q$-'!Q$-/FQJ7Q;/T]' q=H+d+AhK Q;7J#AqD)Jq!t,<<AqDAQT^_`bc_cTdAdeAfI >Y<-AiL  7F7 +G )AAAgJ 1Q(T r)returnr)r!rrr)rNone)r*zdict[str, Any]rr)FF) r{zdict[str, dict[str, Any]]r|intr}rrorrr)rtuple[float, ...]rbrrr)rlist[dict[str, Any]]r torch.devicerztuple[Tensor, Tensor]) rrrrrrrrrrrr)N)rrrrrrrrrrrz Tensor | Nonerr)$__doc__ __future__rcollections.abcrtypingrrrrfdetr.datasets._aug_utilsrrfdetr.utilities.loggerrrv__doctest_requires__ IMAGENET_MEAN IMAGENET_STDr __annotations__rr"rr+r.r;rIrSrWr^rcrlr rrrrrrrs !F#$ J. /(<& $ #&%%*$$eN q"86*8"& ,'% @'# 9 5 (# 9)999 9  9z,)   8.$ !.$ .$.$b- !- - - -  -  -l $ OO O"O O  O  OOr