from __future__ import annotations

import math, torch
from typing import Any
from ...utils.logging import log
from ...utils.image import blend_processed_batch_item, pil_to_tensor, tensor_to_pil
from ..mode_contexts import SingleGpuModeContext, TileOpsCollaborator
from ..processing_args import UpscaleCoreArgs
from ..tile_processing import TileBatchArgs, extract_and_process_tile_batch


class SingleGpuModeMixin:
    def _build_single_gpu_mode_context(self) -> SingleGpuModeContext:
        """Build explicit collaborators for single-GPU mode execution."""
        return SingleGpuModeContext(tile_ops=TileOpsCollaborator(self))

    def process_single_gpu(
        self,
        upscaled_image: torch.Tensor,
        core_args: UpscaleCoreArgs,
        tile_width: int,
        tile_height: int,
        padding: int,
        mask_blur: int,
        force_uniform_tiles: bool,
        mode_context: SingleGpuModeContext | None = None,
    ) -> tuple[torch.Tensor]:
        """Process all tiles on a single GPU (no distribution), batching per tile like USDU."""
        context = mode_context or self._build_single_gpu_mode_context()
        ops = context.tile_ops
        # Round tile dimensions
        tile_width = ops.round_to_multiple(tile_width)
        tile_height = ops.round_to_multiple(tile_height)

        # Get image dimensions and batch size
        batch_size, height, width, _ = upscaled_image.shape

        # Calculate all tiles
        all_tiles = ops.calculate_tiles(width, height, tile_width, tile_height, force_uniform_tiles)

        rows = math.ceil(height / tile_height)
        cols = math.ceil(width / tile_width)
        log(
            f"USDU Dist: Single GPU | Canvas {width}x{height} | Tile {tile_width}x{tile_height} | Grid {rows}x{cols} ({len(all_tiles)} tiles/image) | Batch {batch_size}"
        )
        tile_batch_args = TileBatchArgs(
            core=core_args,
            tile_width=tile_width,
            tile_height=tile_height,
            padding=padding,
            force_uniform_tiles=force_uniform_tiles,
            width=width,
            height=height,
        )

        # Prepare result images list
        result_images = []
        for b in range(batch_size):
            image_pil = tensor_to_pil(upscaled_image[b:b+1], 0).convert('RGB')
            result_images.append(image_pil.copy())

        # Precompute tile masks once
        tile_masks = []
        for tx, ty in all_tiles:
            tile_masks.append(ops.create_tile_mask(width, height, tx, ty, tile_width, tile_height, mask_blur))

        # Process tiles batched across images
        for tile_idx, (tx, ty) in enumerate(all_tiles):
            # Progressive state parity: extract each tile from the current updated image batch.
            source_batch = torch.cat([pil_to_tensor(img) for img in result_images], dim=0)
            if upscaled_image.is_cuda:
                source_batch = source_batch.cuda()

            processed_batch, x1, y1, ew, eh = extract_and_process_tile_batch(
                node=ops,
                upscaled_image=source_batch,
                tx=tx,
                ty=ty,
                args=tile_batch_args,
            )

            # Blend results back into each image using cached mask
            tile_mask = tile_masks[tile_idx]
            for b in range(batch_size):
                blend_processed_batch_item(
                    result_images,
                    processed_batch,
                    b,
                    ops.blend_tile,
                    x1,
                    y1,
                    ew,
                    eh,
                    tile_mask,
                    padding,
                )

        # Convert back to tensor
        result_tensors = [pil_to_tensor(img) for img in result_images]
        result_tensor = torch.cat(result_tensors, dim=0)
        if upscaled_image.is_cuda:
            result_tensor = result_tensor.cuda()

        return (result_tensor,)