engine/pose_estimation/blocks/detector.py

from __future__ import annotations

import os
import os.path as osp
from collections import defaultdict
import time
from mmpose.apis.inference import batch_inference_pose_model

import numpy as np
import torch
import torch.nn as nn
import scipy.signal as signal

from ultralytics import YOLO
from mmpose.apis import (
    init_pose_model,
    get_track_id,
    vis_pose_result,
)

ROOT_DIR = osp.abspath(f"{__file__}/../../")
VIT_DIR = osp.join(ROOT_DIR, "third-party/ViTPose")

VIS_THRESH = 0.5
BBOX_CONF = 0.5
TRACKING_THR = 0.1
MINIMUM_FRMAES = 15
MINIMUM_JOINTS = 6


class DetectionModel(object):
    def __init__(self, pose_model_ckpt, device, with_tracker=True):

        # ViTPose
        pose_model_cfg = osp.join(
            VIT_DIR,
            "configs/wholebody/2d_kpt_sview_rgb_img/topdown_heatmap/coco-wholebody/ViTPose_huge_wholebody_256x192.py",
        )
        #'vitpose-h-multi-coco.pth')
        self.pose_model = init_pose_model(
            pose_model_cfg, pose_model_ckpt, device=device
        )

        # YOLO
        bbox_model_ckpt = osp.join(ROOT_DIR, "checkpoints", "yolov8x.pt")
        if with_tracker:
            self.bbox_model = YOLO(bbox_model_ckpt)
        else:
            self.bbox_model = None

        self.device = device
        self.initialize_tracking()

    def initialize_tracking(
        self,
    ):
        self.next_id = 0
        self.frame_id = 0
        self.pose_results_last = []
        self.tracking_results = {
            "id": [],
            "frame_id": [],
            "bbox": [],
        }

    def xyxy_to_cxcys(self, bbox, s_factor=1.05):
        cx, cy = bbox[[0, 2]].mean(), bbox[[1, 3]].mean()
        scale = max(bbox[2] - bbox[0], bbox[3] - bbox[1]) / 200 * s_factor
        return np.array([[cx, cy, scale]])

    def compute_bboxes_from_keypoints(self, s_factor=1.2):
        X = self.tracking_results["keypoints"].copy()
        mask = X[..., -1] > VIS_THRESH

        bbox = np.zeros((len(X), 3))
        for i, (kp, m) in enumerate(zip(X, mask)):
            bb = [kp[m, 0].min(), kp[m, 1].min(), kp[m, 0].max(), kp[m, 1].max()]
            cx, cy = [(bb[2] + bb[0]) / 2, (bb[3] + bb[1]) / 2]
            bb_w = bb[2] - bb[0]
            bb_h = bb[3] - bb[1]
            s = np.stack((bb_w, bb_h)).max()
            bb = np.array((cx, cy, s))
            bbox[i] = bb

        bbox[:, 2] = bbox[:, 2] * s_factor / 200.0
        self.tracking_results["bbox"] = bbox

    def compute_bbox(self, img):
        bboxes = (
            self.bbox_model.predict(
                img,
                device=self.device,
                classes=0,
                conf=BBOX_CONF,
                save=False,
                verbose=False,
            )[0]
            .boxes.xyxy.detach()
            .cpu()
            .numpy()
        )

        bboxes = [{"bbox": bbox} for bbox in bboxes]
        imgs = [img for _ in range(len(bboxes))]
        return bboxes, imgs

    def batch_detection(self, bboxes, imgs, batch_size=32):
        all_poses = []
        all_bboxes = []
        for i in range(0, len(bboxes), batch_size):
            poses, bbox_xyxy = batch_inference_pose_model(
                self.pose_model,
                imgs[i : i + batch_size],
                bboxes[i : i + batch_size],
                return_heatmap=False,
            )
            all_poses.append(poses)
            all_bboxes.append(bbox_xyxy)
        all_poses = np.concatenate(all_poses)
        all_bboxes = np.concatenate(all_bboxes)
        return all_poses, all_bboxes

    def track(self, img, fps, length):
        # bbox detection
        bboxes = (
            self.bbox_model.predict(
                img,
                device=self.device,
                classes=0,
                conf=BBOX_CONF,
                save=False,
                verbose=False,
            )[0]
            .boxes.xyxy.detach()
            .cpu()
            .numpy()
        )

        pose_results = [{"bbox": bbox} for bbox in bboxes]

        pose_results, self.next_id = get_track_id(
            pose_results,
            self.pose_results_last,
            self.next_id,
            use_oks=False,
            tracking_thr=TRACKING_THR,
            use_one_euro=True,
            fps=fps,
        )

        for pose_result in pose_results:

            _id = pose_result["track_id"]
            xyxy = pose_result["bbox"]
            bbox = xyxy  # self.xyxy_to_cxcys(xyxy)

            self.tracking_results["id"].append(_id)
            self.tracking_results["frame_id"].append(self.frame_id)
            self.tracking_results["bbox"].append(bbox)

        self.frame_id += 1
        self.pose_results_last = pose_results

    def process(self, fps):

        for key in ["id", "frame_id", "bbox"]:
            self.tracking_results[key] = np.array(self.tracking_results[key])
        # self.compute_bboxes_from_keypoints()

        output = defaultdict(lambda: defaultdict(list))
        ids = np.unique(self.tracking_results["id"])

        for _id in ids:
            idxs = np.where(self.tracking_results["id"] == _id)[0]

            for key, val in self.tracking_results.items():
                if key == "id":
                    continue
                output[_id][key] = val[idxs]

        # Smooth bounding box detection
        ids = list(output.keys())
        for _id in ids:
            if len(output[_id]["bbox"]) < MINIMUM_FRMAES:
                del output[_id]
                continue

            kernel = int(int(fps / 2) / 2) * 2 + 1
            smoothed_bbox = np.array(
                [signal.medfilt(param, kernel) for param in output[_id]["bbox"].T]
            ).T
            output[_id]["bbox"] = smoothed_bbox

        return output

    def visualize(self, img, pose_results):
        vis_img = vis_pose_result(
            self.pose_model,
            img,
            pose_results,
            dataset=self.pose_model.cfg.data["test"]["type"],
            dataset_info=None,  # self.pose_model.cfg.data['test'].get('dataset_info', None),
            kpt_score_thr=0.3,
            radius=4,
            thickness=1,
            show=False,
        )
        return vis_img