utils/video_embedder.py

from __future__ import annotations

import os
import random
import numpy as np
import torch
import clip
from torchvision.transforms import (
    Compose,
    Resize,
    CenterCrop,
    ToTensor,
    Normalize,
    InterpolationMode,
)
from typing import List, Tuple, Optional
import cv2
from PIL import Image
import tqdm


def init_repro(seed: int = 42, deterministic: bool = True):
    """Call this at the very top of your notebook/script BEFORE creating any model/processor/device context."""
    os.environ["PYTHONHASHSEED"] = str(seed)
    os.environ["CUBLAS_WORKSPACE_CONFIG"] = (
        ":16:8"  # deterministic cuBLAS on Ampere+, nice default
    )
    os.environ["OMP_NUM_THREADS"] = "1"
    os.environ["MKL_NUM_THREADS"] = "1"

    random.seed(seed)
    np.random.seed(seed)

    torch.manual_seed(seed)
    torch.cuda.manual_seed(seed)
    torch.cuda.manual_seed_all(seed)

    # Determinism knobs (do this before any CUDA ops)
    if deterministic:
        try:
            torch.use_deterministic_algorithms(True)
        except Exception:
            # older torch may not support signature
            torch.set_deterministic(True)
        torch.backends.cudnn.deterministic = True
        torch.backends.cudnn.benchmark = False
        torch.backends.cuda.matmul.allow_tf32 = False
        torch.backends.cudnn.allow_tf32 = False

    # Reduce threading nondeterminism
    torch.set_num_threads(1)

    return seed


def _cpu_tensor_or_none(x):
    if isinstance(x, torch.Tensor):
        return x.detach().cpu()
    return x


class _PickleBackendsMixin:
    def attach_backends(
        self, *, model=None, tokenizer=None, clip_model=None, device=None
    ):
        self.model = model
        self.tokenizer = tokenizer
        self.clip_model = clip_model
        self.device = device or torch.device(
            "cuda" if torch.cuda.is_available() else "cpu"
        )
        if getattr(self, "model", None) is not None:
            self.model = self.model.to(self.device).eval()

    def __getstate__(self):
        s = self.__dict__.copy()
        # drop unpicklables
        for k in ("model", "tokenizer", "clip_model", "device"):
            s.pop(k, None)
        # ensure tensors are CPU-picklable
        for k in ("video_embeddings", "text_embeddings"):
            if k in s and s[k] is not None:
                if isinstance(s[k], dict):
                    s[k] = {kk: _cpu_tensor_or_none(vv) for kk, vv in s[k].items()}
                else:
                    s[k] = _cpu_tensor_or_none(s[k])
        return s

    def __setstate__(self, s):
        self.__dict__.update(s)
        # backends are reattached by caller after unpickle
        self.model = None
        self.tokenizer = None
        self.clip_model = None
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")


class VideoEmbedder(_PickleBackendsMixin):
    def __init__(self, model_name, model, tokenizer, clip_model=None,
                 pe_video_batch_size: Optional[int] = None, pe_target_T: Optional[int] = None):
        self.model_name = model_name.lower()
        self.model = model
        self.tokenizer = tokenizer
        self.clip_model = clip_model
        self.pe_video_batch_size = pe_video_batch_size
        self.pe_target_T = pe_target_T

        self.dataset_name: Optional[str] = None
        self.video_embeddings: Optional[Dict[str, np.ndarray]] = None
        self.labels: Optional[List[str]] = None
        self.video_window_spans: Dict[str, List[Tuple[float, float]]] = {}
        self.video_meta: Dict[str, Dict[str, float]] = {}

        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.model = self.model.to(self.device).eval()
        torch.backends.cudnn.benchmark = True
        self.embed_dim = self._detect_embed_dim()

    # ------------------------- util
    def _detect_embed_dim(self) -> int:
        with torch.inference_mode():
            dummy = np.zeros((224, 224, 3), dtype=np.uint8)
            if self.model_name == "res50":
                t = self.tokenizer(Image.fromarray(dummy)).unsqueeze(0).to(self.device)
                d = self.model.encode_image(t).shape[-1]
            elif self.model_name in {"clip", "siglip", "siglipl14", "siglip2"}:
                batch = self.tokenizer(images=dummy, return_tensors="pt")
                batch = {k: v.to(self.device) for k, v in batch.items()}
                d = self.model.get_image_features(**batch).shape[-1]
            elif self.model_name == "clip4clip":
                preprocess = Compose(
                    [
                        Resize((224, 224), interpolation=InterpolationMode.BICUBIC),
                        CenterCrop(224),
                        ToTensor(),
                        Normalize(
                            (0.48145466, 0.4578275, 0.40821073),
                            (0.26862954, 0.26130258, 0.27577711),
                        ),
                    ]
                )
                t = preprocess(Image.fromarray(dummy)).unsqueeze(0).to(self.device)
                d = self.model(t)["image_embeds"].shape[-1]
            elif self.model_name == "pe-l14":
                # Create a dummy clip of length 16 by repeating a single frame
                dummy_img = Image.fromarray(np.zeros((336, 336, 3), dtype=np.uint8))
                frame = self.tokenizer(dummy_img)
                clip = torch.stack([frame for _ in range(16)], dim=0)  # (T,C,H,W)
                clip = clip.unsqueeze(0).to(self.device)  # (B,T,C,H,W)
                d = self.model.encode_video(clip).shape[-1]
            else:
                raise ValueError(f"Unknown model_name {self.model_name}")
        return int(d)

    def _preprocess_video_pe(
            self,
            video: List[Image.Image],  # now expects a list of PIL Images
            num_frames: int = 4,
            transform: Optional[Compose] = None,
            return_first_frame_for_demo: bool = False
        ) -> Tuple[torch.Tensor, Optional[Image.Image]]:
        total_frames = len(video)
        # Uniformly sample frame indices
        frame_indices = [int(i * (total_frames / num_frames)) for i in range(num_frames)]
        frames = [video[i] for i in frame_indices]
        # Preprocess frames
        preprocessed_frames = [transform(frame) for frame in frames]

        first_frame = None
        if return_first_frame_for_demo:
            first_frame = frames[0]
        return torch.stack(preprocessed_frames, dim=0), first_frame
    @staticmethod
    def _bgr_to_pil(frame_bgr: np.ndarray) -> Image.Image:
        return Image.fromarray(cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB))

    @staticmethod
    def _sample_indices(n: int, k: int, random: bool) -> List[int]:
        if n <= 0 or k <= 0:
            return []
        if random:
            k = min(k, n)
            return np.random.choice(n, size=k, replace=False).tolist()
        if k >= n:
            return list(range(n))
        step = (n - 1) / (k - 1) if k > 1 else 1e9
        return [int(round(i * step)) for i in range(k)]

    # ------------------------- encoders
    def _encode_images_hf(self, frames_bgr: List[np.ndarray]) -> torch.Tensor:
        """HF CLIP, SigLIP"""
        if not frames_bgr:
            return torch.empty((0, self.embed_dim), dtype=torch.float32)
        images_rgb = [cv2.cvtColor(f, cv2.COLOR_BGR2RGB) for f in frames_bgr]
        batch = self.tokenizer(images=images_rgb, return_tensors="pt")
        batch = {k: v.to(self.device, non_blocking=True) for k, v in batch.items()}
        with torch.inference_mode(), torch.autocast(
            device_type="cuda",
            dtype=torch.float16,
            enabled=(self.device.type == "cuda"),
        ):
            feats = self.model.get_image_features(**batch)
        return feats.float().detach().cpu()

    def _encode_images_pe(self, frames_bgr: List[np.ndarray]) -> torch.Tensor:
        """Encode frames using PE-L/14 video model.

        This implementation preserves alignment: it returns one embedding per
        input frame by forming a temporal clip centered on each anchor frame
        (padding at the edges). Clips are length T (default 16) and are
        batched efficiently on GPU.
        """

        if not frames_bgr:
            return torch.empty((0, self.embed_dim), dtype=torch.float32)

        # Config
        clip_len = 16  # temporal length expected by the video model
        half_clip = clip_len // 2
        max_gpu_batch = 8  # number of clips to encode per forward pass

        # Preprocess all frames once (C,H,W tensors on CPU)
        pil_imgs = [self._bgr_to_pil(f) for f in frames_bgr]
        frames_tensor = [self.tokenizer(img) for img in pil_imgs]
        n = len(frames_tensor)

        def build_clip_around(idx: int) -> torch.Tensor:
            """Return a tensor of shape (T, C, H, W) for anchor frame idx."""
            start = idx - half_clip
            end = start + clip_len
            # Clamp and pad by edge repetition
            frames = []
            for t in range(start, end):
                clamped = min(max(t, 0), n - 1)
                frames.append(frames_tensor[clamped])
            return torch.stack(frames, dim=0)

        embs = []
        with torch.inference_mode():
            # Iterate in micro-batches of clips to control memory
            for s in range(0, n, max_gpu_batch):
                batch_indices = list(range(s, min(s + max_gpu_batch, n)))
                clips = [build_clip_around(i) for i in batch_indices]
                x = torch.stack(clips, dim=0).to(self.device, non_blocking=True)
                # x: (B, T, C, H, W)
                with torch.autocast(
                    device_type="cuda",
                    dtype=torch.float16,
                    enabled=(self.device.type == "cuda"),
                ):
                    feats = self.model.encode_video(x)
                embs.append(feats.detach().cpu())

        return torch.cat(embs, dim=0).float()

    def _encode_images_openai_clip(self, frames_bgr: List[np.ndarray]) -> torch.Tensor:
        """OpenAI CLIP RN50."""
        if not frames_bgr:
            return torch.empty((0, self.embed_dim), dtype=torch.float32)
        pil_imgs = [self._bgr_to_pil(f) for f in frames_bgr]
        x = torch.stack([self.tokenizer(img) for img in pil_imgs], dim=0).to(
            self.device, non_blocking=True
        )
        with torch.inference_mode(), torch.autocast(
            device_type="cuda",
            dtype=torch.float16,
            enabled=(self.device.type == "cuda"),
        ):
            feats = self.model.encode_image(x)
        return feats.float().detach().cpu()

    def _encode_images_clip4clip(self, frames_bgr: List[np.ndarray]) -> torch.Tensor:
        """CLIP4Clip (expects raw pixel tensors normalized manually)."""
        if not frames_bgr:
            return torch.empty((0, self.embed_dim), dtype=torch.float32)
        preprocess = Compose(
            [
                Resize((224, 224), interpolation=InterpolationMode.BICUBIC),
                CenterCrop(224),
                ToTensor(),
                Normalize(
                    (0.48145466, 0.4578275, 0.40821073),
                    (0.26862954, 0.26130258, 0.27577711),
                ),
            ]
        )
        pil_imgs = [self._bgr_to_pil(f) for f in frames_bgr]
        x = torch.stack([preprocess(img) for img in pil_imgs], dim=0).to(
            self.device, non_blocking=True
        )
        with torch.inference_mode():
            out = self.model(x)["image_embeds"]
            out = out / (out.norm(dim=-1, keepdim=True) + 1e-6)
        return out.float().detach().cpu()

    # ------------------------- video reading
    def _read_windows(self, video_path: str, window_size: int):
        windows, spans = [], []

        if video_path.lower().endswith(".mp4"):
            # ---- read video ----
            cap = cv2.VideoCapture(video_path)
            frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
            fps = float(cap.get(cv2.CAP_PROP_FPS) or 0.0) or 30.0

            if window_size > frame_count:
                frame_count = window_size

            frames = []
            for _ in range(frame_count):
                ret, frame = cap.read()
                if not ret:
                    break
                frames.append(frame)
            cap.release()

        else:
            # ---- read raw images in directory ----
            img_files = sorted(
                [
                    f
                    for f in os.listdir(video_path)
                    if f.lower().endswith((".jpg", ".jpeg", ".png"))
                ]
            )
            frames = [cv2.imread(os.path.join(video_path, f)) for f in img_files]
            frames = [f for f in frames if f is not None]

            frame_count = len(frames)
            fps = 12.0  # fallback since no video container

        # ---- make windows ----
        for i in range(0, frame_count, window_size):
            chunk = frames[i : i + window_size]
            if len(chunk) == 0:
                continue
            windows.append(chunk)
            start_t = i / fps
            end_t = (i + len(chunk) - 1) / fps
            spans.append((start_t, end_t))

        return windows, spans, fps, frame_count

    def _encode_windows(
        self, windows, frames_per_window, random, batch_size
    ) -> np.ndarray:
        # Specialized path for PE-L/14: encode entire windows as video clips
        if self.model_name == "pe-l14":
            # Use configured per-GPU batch size for video clips if provided
            pe_bs = self.pe_video_batch_size or min(batch_size, 8)
            outs = []
            for s in range(0, len(windows), pe_bs):
                batch_windows = windows[s : s + pe_bs]
                feats = self._encode_windows_pe(batch_windows)
                outs.append(feats)
            if len(outs) == 0:
                return np.zeros((0, self.embed_dim), dtype=np.float32)
            return torch.cat(outs, dim=0).numpy()

        # Image encoders: sample frames and average per window
        all_samples = []
        map_win_to_slice = []
        cursor = 0
        for w in windows:
            idxs = self._sample_indices(len(w), frames_per_window, random)
            if not idxs:
                all_samples.append(w[0])
                map_win_to_slice.append((cursor, cursor + 1))
                cursor += 1
                continue
            for j in idxs:
                all_samples.append(w[j])
            map_win_to_slice.append((cursor, cursor + len(idxs)))
            cursor += len(idxs)

        if self.model_name == "res50":
            encode_fn = self._encode_images_openai_clip
        elif self.model_name in {"clip", "siglip", "siglipl14", "siglip2"}:
            encode_fn = self._encode_images_hf
        elif self.model_name == "clip4clip":
            encode_fn = self._encode_images_clip4clip
        else:
            raise ValueError(f"Unknown model_name {self.model_name}")

        outs = []
        for s in range(0, len(all_samples), batch_size):
            feats = encode_fn(all_samples[s : s + batch_size])
            outs.append(feats)

        if len(outs) == 0:
            return np.zeros((0, self.embed_dim), dtype=np.float32)
        flat_feats = torch.cat(outs, dim=0)
        window_embs = []
        for a, b in map_win_to_slice:
            if b <= a:
                window_embs.append(torch.zeros(self.embed_dim))
            else:
                window_embs.append(flat_feats[a:b].mean(dim=0))
        return torch.stack(window_embs, dim=0).numpy()

    def _encode_windows_pe(self, windows: List[List[np.ndarray]]) -> torch.Tensor:
        """Encode a batch of windows (lists of BGR frames) as video clips.

        Pads each window in the batch to the batch's max temporal length by
        repeating the last frame so windows can be batched.
        Returns a CPU tensor of shape (B, D).
        """
        if not windows:
            return torch.empty((0, self.embed_dim), dtype=torch.float32)

        # Preprocess each window: convert to tensors (T_i, C, H, W)
        clip_tensors = []
        max_T = 0
        for w in windows:
            if not w:
                # create a single black frame if window is empty
                black = np.zeros((336, 336, 3), dtype=np.uint8)
                w = [black]
            # If requested, uniformly sample to target temporal length
            if self.pe_target_T is not None and len(w) > 0:
                T = self.pe_target_T
                if len(w) >= T:
                    # uniform indices across [0, len(w)-1]
                    idxs = [int(round(i * (len(w) - 1) / (T - 1))) for i in range(T)]
                else:
                    # upsample by repeating last frame to reach T
                    idxs = list(range(len(w))) + [len(w) - 1] * (T - len(w))
                w = [w[i] for i in idxs]

            pil_imgs = [self._bgr_to_pil(f) for f in w]
            frames = [self.tokenizer(img) for img in pil_imgs]
            clip = torch.stack(frames, dim=0)
            clip_tensors.append(clip)
            max_T = max(max_T, clip.shape[0])

        # Pad all to max_T using last-frame repetition
        padded = []
        for clip in clip_tensors:
            if clip.shape[0] < max_T:
                pad = clip[-1:].expand(max_T - clip.shape[0], -1, -1, -1)
                clip = torch.cat([clip, pad], dim=0)
            padded.append(clip)

        x = torch.stack(padded, dim=0).to(self.device, non_blocking=True)  # (B,T,C,H,W)
        with torch.inference_mode(), torch.autocast(
            device_type="cuda",
            dtype=torch.float16,
            enabled=(self.device.type == "cuda"),
        ):
            feats = self.model.encode_video(x)
        return feats.float().detach().cpu()

    # ------------------------- labels
    def extract_labels(self, path: str) -> Optional[str]:
        if self.dataset_name == "breakfast":
            label = path.split("/")[-1]
            return label.split("_")[1].replace(".mp4", "")
        elif self.dataset_name == "ucf101":
            label = path.split("/")[-1]
            return label.split("_")[1]
        elif self.dataset_name == "hmdb":
            return path.split("/")[4]
        elif self.dataset_name == "something2":
            return path.split("/")[1]
        elif self.dataset_name == "jester":
            label = path.split("/")[-1]
            return label.split("_")[0]
        return None

    # ------------------------- main
    def embed_video(
        self,
        video_paths,
        window_size,
        output_path,
        random=True,
        save_intermediate=False,
        frames_per_window=1,
        batch_size=256,
    ):
        os.makedirs(output_path, exist_ok=True)

        video_embedding_paths, labels, video_window_spans, video_meta = {}, [], {}, {}

        video_paths = sorted(video_paths)
        save_base = os.path.join(
            output_path, f"{self.dataset_name}_{self.model_name}_{window_size}_state"
        )
        final_path = save_base + ".npy"
        tmp_path = save_base + ".tmp.npy"

        processed_count = 0
        if save_intermediate:
            load_path = (
                final_path
                if os.path.exists(final_path)
                else (tmp_path if os.path.exists(tmp_path) else None)
            )
            if load_path:
                try:
                    loaded = np.load(load_path, allow_pickle=True).item()
                    video_embedding_paths = loaded.get("video_embeddings", {})
                    labels = loaded.get("labels", [])
                    video_window_spans = loaded.get("video_window_spans", {})
                    video_meta = loaded.get("video_meta", {})
                    processed_count = len(video_embedding_paths)
                except Exception:
                    processed_count = 0
        if processed_count > 0:
            video_paths = video_paths[processed_count:]

        counter_since_last_save = 0
        for video_path in tqdm.tqdm(video_paths):
            labels.append(self.extract_labels(video_path))
            windows, spans, fps, read_frames = self._read_windows(
                video_path, window_size
            )

            if len(windows) == 0:
                video_embedding_paths[video_path] = np.zeros(
                    (0, self.embed_dim), dtype=np.float32
                )
                video_window_spans[video_path] = []
                video_meta[video_path] = {"fps": fps, "frame_count": float(read_frames)}
            else:
                window_embeddings = self._encode_windows(
                    windows, frames_per_window, random, batch_size
                )
                video_embedding_paths[video_path] = window_embeddings
                video_window_spans[video_path] = spans
                video_meta[video_path] = {"fps": fps, "frame_count": float(read_frames)}

            counter_since_last_save += 1
            if save_intermediate and (counter_since_last_save % 10 == 0):
                state = {
                    "video_embeddings": video_embedding_paths,
                    "labels": labels,
                    "video_window_spans": video_window_spans,
                    "video_meta": video_meta,
                }
                np.save(tmp_path, state, allow_pickle=True)
                os.replace(tmp_path, final_path)

        if save_intermediate:
            # delete tmp file if it exists
            if os.path.exists(tmp_path):
                os.remove(tmp_path)
            if os.path.exists(final_path):
                os.remove(final_path)

        self.video_embeddings = video_embedding_paths
        self.labels = labels
        self.video_window_spans = video_window_spans
        self.video_meta = video_meta

    def process_data(
        self,
        folder_path,
        window_size,
        output_path,
        random=True,
        save_intermediate=False,
        frames_per_window=1,
        batch_size=256,
    ):
        os.makedirs(output_path, exist_ok=True)
        video_paths = []
        if self.dataset_name == "jester":
            folder_path = folder_path[0].replace("Video_data", "Image_data")
            all_paths = os.listdir(folder_path)
            video_paths = [
                os.path.join(folder_path, p)
                for p in all_paths
                if os.path.isdir(os.path.join(folder_path, p))
            ]
        else:
            if isinstance(folder_path, list):
                for path in folder_path:
                    for root, _, files in os.walk(path):
                        for file in files:
                            if file.lower().endswith(".mp4"):
                                video_paths.append(os.path.join(root, file))
            else:
                for root, _, files in os.walk(folder_path):
                    for file in files:
                        if file.lower().endswith(".mp4"):
                            video_paths.append(os.path.join(root, file))
        print(len(video_paths), "videos found in", folder_path)

        self.embed_video(
            video_paths,
            window_size,
            output_path,
            random=random,
            save_intermediate=save_intermediate,
            frames_per_window=frames_per_window,
            batch_size=batch_size,
        )


class Create_Concepts(_PickleBackendsMixin):
    def __init__(self, model_name, model, tokenizer, clip_model=None):
        self.model_name = model_name
        self.model = model
        self.tokenizer = tokenizer
        self.clip_model = clip_model
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

        self.dataset_name = None
        self.video_embeddings = None
        self.labels = None
        self.text_concepts = None
        self.text_embeddings = None

    def embedd_text(self, *text):
        concepts = []

        # Case 1: multiple positional args given
        if len(text) > 1:
            for t in text:
                if isinstance(t, str):
                    concepts.extend([c.strip() for c in t.split(",") if c.strip()])
                elif isinstance(t, list):
                    for item in t:
                        concepts.extend(
                            [c.strip() for c in item.split(",") if c.strip()]
                        )
        else:
            t = text[0]
            if isinstance(t, str):
                concepts.extend([c.strip() for c in t.split(",") if c.strip()])
            elif isinstance(t, list):
                for item in t:
                    concepts.extend([c.strip() for c in item.split(",") if c.strip()])

        # Deduplicate while preserving order
        seen = set()
        concepts = [c for c in concepts if not (c in seen or seen.add(c))]
        # Tokenize & embed

        if self.model_name == "clip":
            inputs = self.tokenizer(
                concepts, return_tensors="pt", padding=True, truncation=True
            ).to(self.model.device)
            outputs = self.model.get_text_features(**inputs)

        elif self.model_name == "pe-l14":
            inputs = self.tokenizer(
                concepts).to(self.device)
            with torch.no_grad():
                outputs = self.model.encode_text(inputs)
        elif self.model_name == "siglip" or self.model_name == "siglipl14":
            inputs = self.tokenizer(
                text=concepts, padding="max_length", return_tensors="pt"
            ).to(self.model.device)
            with torch.no_grad():
                outputs = self.model.get_text_features(**inputs)
        elif self.model_name == "siglip2":
            inputs = self.tokenizer(
                text=concepts, padding=True, return_tensors="pt"
            ).to(self.model.device)
            # text_inputs = {k: v.to(self.model.device) for k, v in inputs.items()}
            with torch.no_grad():
                outputs = self.model.get_text_features(**inputs)
        elif self.model_name == "res50":
            inputs = clip.tokenize(concepts)  # returns CPU tensor by default
            inputs = inputs.to(self.device)  # move tokens to model device
            with torch.no_grad():
                outputs = self.model.encode_text(inputs).detach().cpu()
        elif self.model_name == "clip4clip":
            inputs = self.tokenizer(
                concepts, return_tensors="pt", padding=True, truncation=True
            ).to(self.model.device)
            outputs = (
                self.model(
                    input_ids=inputs["input_ids"],
                    attention_mask=inputs["attention_mask"],
                )[0]
                .detach()
                .cpu()
            )
        else:
            outputs = None

        self.text_embeddings = outputs
        self.text_concepts = concepts