# Ultralytics 🚀 AGPL-3.0 License - https://ultralytics.com/license

from __future__ import annotations

from typing import Any

import numpy as np

from .byte_tracker import STrack
from .oc_sort import OCSORT, OCSortTrack
from .utils import matching
from .utils.gmc import GMC
from .utils.reid import build_encoder, smooth_feature
from .utils.stracks import parse_bboxes


class DeepOCSortTrack(OCSortTrack):
    """Track object for Deep OC-SORT with appearance features and observation-centric state management.

    Extends OCSortTrack with ReID embedding storage and exponential moving average smoothing, plus confidence-adaptive
    embedding update rates.

    Attributes:
        smooth_feat (np.ndarray | None): Smoothed feature vector via EMA.
        curr_feat (np.ndarray | None): Current frame's feature vector.
        alpha_fixed_emb (float): Base EMA factor for embedding updates.
        det_thresh (float): Confidence threshold below which a new embedding is ignored rather than blended.
    """

    def __init__(
        self,
        xywh: np.ndarray,
        score: float,
        cls: Any,
        delta_t: int = 3,
        feat: np.ndarray | None = None,
        alpha_fixed_emb: float = 0.95,
        det_thresh: float = 0.25,
    ):
        """Initialize a `DeepOCSortTrack` with optional appearance features.

        Args:
            xywh (np.ndarray): Bounding box in `(x, y, w, h, idx)` or `(x, y, w, h, angle, idx)` format.
            score (float): Detection confidence in `[0, 1]`.
            cls (Any): Class label for the detection.
            delta_t (int): Temporal window for OCM velocity direction computation.
            feat (np.ndarray | None): Optional appearance feature vector for this detection.
            alpha_fixed_emb (float): Base EMA factor for embedding updates; higher = slower updates.
            det_thresh (float): Detection-confidence threshold below which the embedding is replaced rather than
                blended.
        """
        super().__init__(xywh, score, cls, delta_t)
        self.smooth_feat = None
        self.curr_feat = None
        self.alpha_fixed_emb = alpha_fixed_emb
        self.det_thresh = det_thresh
        if feat is not None:
            self.update_features(feat, score)

    def update_features(self, feat: np.ndarray, score: float | None = None) -> None:
        """Blend a new appearance feature into `smooth_feat` with confidence-adaptive EMA.

        When `score` exceeds `det_thresh`, the EMA factor `alpha = alpha_fixed_emb + (1 - alpha_fixed_emb) * (1 - trust)`
        with `trust = (score - det_thresh) / (1 - det_thresh)`, so a high-confidence detection blends in more strongly.
        At or below `det_thresh`, `alpha = 1.0` so the existing `smooth_feat` is kept and the new (low-trust) feature is
        ignored — same behavior used by the reference Deep OC-SORT implementation.

        Args:
            feat (np.ndarray): New (un-normalized) appearance feature vector.
            score (float | None): Detection confidence used to modulate the EMA factor.
        """
        if score is not None and score > self.det_thresh:
            trust = (score - self.det_thresh) / max(1 - self.det_thresh, 1e-9)
            alpha = self.alpha_fixed_emb + (1 - self.alpha_fixed_emb) * (1 - trust)
        else:
            alpha = 1.0  # keep the existing smooth_feat for low-trust detections
        curr, smooth = smooth_feature(feat, self.smooth_feat, alpha)
        if curr is not None:
            self.curr_feat, self.smooth_feat = curr, smooth

    def update(self, new_track: STrack, frame_id: int) -> None:
        """Update track state with a matched detection and refresh appearance features.

        Args:
            new_track (STrack): Matched detection for this frame, with optional `curr_feat`.
            frame_id (int): Current frame id.
        """
        if new_track.curr_feat is not None:
            self.update_features(new_track.curr_feat, new_track.score)
        super().update(new_track, frame_id)

    def re_activate(self, new_track: STrack, frame_id: int, new_id: bool = False) -> None:
        """Re-activate a lost track and refresh appearance features.

        Args:
            new_track (STrack): Detection used to revive this track.
            frame_id (int): Current frame id.
            new_id (bool): If True, assign a fresh track id instead of reusing the old one.
        """
        if new_track.curr_feat is not None:
            self.update_features(new_track.curr_feat, new_track.score)
        super().re_activate(new_track, frame_id, new_id)

    @staticmethod
    def multi_gmc(stracks: list[DeepOCSortTrack], H: np.ndarray) -> None:
        """Apply global motion compensation correctly for the XYAH Kalman state.

        The standard `multi_gmc` helper from `utils.stracks` would rotate `(a, h)` along with
        `(x, y)`, which is wrong for an aspect-ratio dim. This variant only rotates the
        position `(x, y)` and the velocity `(vx, vy)` blocks, leaves `(a, h)` and `(va, vh)`
        untouched, and also rotates the stored `last_observation` so OCR/ORU stay consistent.

        Args:
            stracks (list[DeepOCSortTrack]): Tracks to warp in place.
            H (np.ndarray): 2x3 affine homography mapping the previous frame to the current one.
        """
        if not stracks:
            return
        multi_mean = np.asarray([st.mean for st in stracks])
        multi_covariance = np.asarray([st.covariance for st in stracks])

        R = H[:2, :2]
        t = H[:2, 2]

        # Build 8x8 transform: rotate (x,y) and (vx,vy), identity for (a,h) and (va,vh)
        R8x8 = np.eye(8)  # float64 holds any homography exactly; a narrower one would round the float64 methods
        R8x8[:2, :2] = R  # rotate position (x, y)
        R8x8[4:6, 4:6] = R  # rotate velocity (vx, vy)
        # indices 2,3 (a,h) and 6,7 (va,vh) remain identity

        multi_mean = np.matmul(R8x8, multi_mean[..., None])[..., 0]
        multi_mean[:, :2] += t
        # Keep the right operand C-contiguous, as the `utils.stracks` sibling does: an F-contiguous one sends matmul
        # into BLAS's transposed-gemm kernel, which on macOS Accelerate leaves FP-exception flags set even for finite
        # inputs, and numpy then reports spurious divide-by-zero RuntimeWarnings.
        multi_covariance = np.matmul(np.matmul(R8x8, multi_covariance), np.ascontiguousarray(R8x8.T))
        for i, (mean, cov) in enumerate(zip(multi_mean, multi_covariance)):
            stracks[i].mean = mean
            stracks[i].covariance = cov

            # Also transform stored observations for OCR/ORU consistency
            if stracks[i].last_observation[0] >= 0:
                obs = stracks[i].last_observation
                # Transform xyxy observation centers
                cx, cy = DeepOCSortTrack._xyxy_center(obs)
                w, h = obs[2] - obs[0], obs[3] - obs[1]
                new_c = R @ np.array([cx, cy]) + t
                stracks[i].last_observation = np.array(
                    [
                        new_c[0] - w / 2,
                        new_c[1] - h / 2,
                        new_c[0] + w / 2,
                        new_c[1] + h / 2,
                    ],
                    dtype=np.float32,
                )


class DeepOCSORT(OCSORT):
    """Deep OC-SORT: OC-SORT enhanced with appearance features, GMC, and adaptive weighting.

    Fixes over naive integration:
    - GMC correctly handles XYAH state (rotates only x,y positions, not aspect ratio/height)
    - Cost combination uses min(IoU, appearance) following BOTSORT's proven approach
    - OCR recovery pass also uses appearance features
    - ByteTrack-style low-confidence second pass disabled by default
    """

    def __init__(self, args: Any):
        """Initialize Deep OC-SORT tracker.

        Args:
            args (Namespace | IterableSimpleNamespace): Parsed tracker config providing the OC-SORT keys plus
                `gmc_method`, `proximity_thresh`, `appearance_thresh`, `alpha_fixed_emb`, `with_reid`, and `model`.
        """
        super().__init__(args)

        # GMC for camera motion compensation
        self.gmc = GMC(method=getattr(args, "gmc_method", "sparseOptFlow"))

        # Appearance parameters
        self.proximity_thresh = getattr(args, "proximity_thresh", 0.5)
        self.appearance_thresh = getattr(args, "appearance_thresh", 0.75)
        self.alpha_fixed_emb = getattr(args, "alpha_fixed_emb", 0.95)

        self.encoder = build_encoder(
            getattr(args, "with_reid", False), getattr(args, "model", "auto"), getattr(args, "device", None)
        )

    def init_track(self, results, img: np.ndarray | None = None) -> list[DeepOCSortTrack]:
        """Build `DeepOCSortTrack` instances, attaching ReID features when enabled.

        When `with_reid=True` and `model="auto"`, `img` should already be a list of native
        backbone features (one per detection); for any other `model`, `img` is the source frame
        and the configured external ReID encoder is invoked on detection crops.

        Args:
            results (Any): Object exposing `xywh` (or `xywhr`), `conf`, and `cls`.
            img (np.ndarray | None): Either the BGR frame or pre-extracted feature tensors, depending on the ReID
                configuration.

        Returns:
            (list[DeepOCSortTrack]): One track per detection, empty if no detections.
        """
        if len(results) == 0:
            return []
        bboxes = parse_bboxes(results)

        if self.encoder is not None and img is not None:
            features = self.encoder(img, bboxes)
            return [
                DeepOCSortTrack(
                    xywh,
                    s,
                    c,
                    self.delta_t,
                    feat=f,
                    alpha_fixed_emb=self.alpha_fixed_emb,
                    det_thresh=self.args.track_high_thresh,
                )
                for (xywh, s, c, f) in zip(bboxes, results.conf, results.cls, features)
            ]
        return [
            DeepOCSortTrack(
                xywh, s, c, self.delta_t, alpha_fixed_emb=self.alpha_fixed_emb, det_thresh=self.args.track_high_thresh
            )
            for (xywh, s, c) in zip(bboxes, results.conf, results.cls)
        ]

    def _pre_first_associate(
        self,
        strack_pool: list[DeepOCSortTrack],
        unconfirmed: list[DeepOCSortTrack],
        img: np.ndarray | None,
        results_high: Any,
    ) -> None:
        """Apply GMC warp to Kalman state before first-stage association."""
        if img is None:
            return
        try:
            warp = self.gmc.apply(img, results_high.xyxy if len(results_high) else np.empty((0, 4)))
        except Exception:
            warp = np.eye(2, 3)
        DeepOCSortTrack.multi_gmc(strack_pool, warp)
        DeepOCSortTrack.multi_gmc(unconfirmed, warp)

    def _fuse_appearance(
        self,
        dists: np.ndarray,
        tracks: list[DeepOCSortTrack],
        detections: list[DeepOCSortTrack],
        iou_dists: np.ndarray | None = None,
    ) -> np.ndarray:
        """Min-fuse appearance distance into the motion cost (BoT-SORT-style)."""
        if self.encoder is None or not tracks or not detections:
            return dists
        emb_dists = matching.embedding_distance(tracks, detections) / 2.0
        emb_dists[emb_dists > (1 - self.appearance_thresh)] = 1.0
        if iou_dists is not None:
            emb_dists[iou_dists > (1 - self.proximity_thresh)] = 1.0
        return np.minimum(dists, emb_dists)

    def reset(self) -> None:
        """Reset the Deep OC-SORT tracker, also clearing the GMC warp state."""
        super().reset()
        self.gmc.reset_params()
