| """ |
| Geometry calibration for landmark data. |
| |
| Why this module exists |
| ---------------------- |
| The official AI4Bharat keypoint release (Zenodo 6674324) was produced with a |
| frame resize whose width/height arguments were swapped (1920x1080 -> 1080x1920), |
| so the archived **x axis is compressed relative to y** by a constant factor. |
| Measured on the dataset itself: |
| |
| shoulder-width / shoulder-to-hip ratio in the release : 0.2018 (median, n=197, IQR 0.197-0.205) |
| same ratio measured by MediaPipe on real full-body photos : 0.6566 |
| => implied x-axis scale 3.26 |
| nose->eye-line / inter-ocular in the release : 1.5694 |
| same ratio measured by MediaPipe : 0.5238 |
| => implied x-axis scale 3.00 |
| theoretical value from the swap, (W/H)^2 = (1920/1080)^2 : 3.1605 |
| |
| Training on the uncorrected data would teach the model a squashed human body and |
| would **not** transfer to live MediaPipe frames, which are isotropic. Therefore |
| the correction is applied at load time (config.APPLY_KEYPOINT_AXIS_CORRECTION) |
| and the numbers above are re-measured by ``training/calibrate_geometry.py`` and |
| reported in ``docs/dataset_audit.md``. |
| |
| Nothing here is a fudge factor for accuracy: it is a units fix, it is derived |
| from the data, and it is disabled for the live MediaPipe path (which is already |
| isotropic). |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| import logging |
| from dataclasses import dataclass, asdict |
| from pathlib import Path |
| from typing import Dict, Iterable, List, Optional, Sequence, Tuple |
|
|
| import numpy as np |
|
|
| import config |
|
|
| logger = logging.getLogger(__name__) |
|
|
| CALIBRATION_FILE = Path(config.METADATA_DIR) / "geometry_calibration.json" |
|
|
| |
| NOSE, LEFT_EYE, RIGHT_EYE = 0, 2, 5 |
| L_SHOULDER, R_SHOULDER, L_HIP, R_HIP = 11, 12, 23, 24 |
|
|
|
|
| |
| |
| |
| def apply_x_axis_scale(coords: np.ndarray, scale: float) -> np.ndarray: |
| """Multiply the x channel by ``scale`` (in place on a copy). |
| |
| ``coords`` is (..., 3) with (x, y, z) on the last axis. z is left untouched: |
| MediaPipe's z is already reported on the same scale as x for the *original* |
| isotropic frame, and the released archive carries it un-resized. |
| """ |
| if scale is None or float(scale) == 1.0: |
| return coords |
| out = np.asarray(coords, dtype=np.float32).copy() |
| out[..., 0] *= float(scale) |
| return out |
|
|
|
|
| def axis_scale_for_video_shape(vid_shape: Sequence[int]) -> float: |
| """Theoretical correction from the archived video shape. |
| |
| The release stores ``vid_shape`` as (height, width). The swap-resize |
| hypothesis predicts ``scale = (width / height) ** 2``. |
| """ |
| h, w = int(vid_shape[0]), int(vid_shape[1]) |
| if h <= 0 or w <= 0: |
| return 1.0 |
| return float((w / h) ** 2) |
|
|
|
|
| def correct_landmarks(coords: np.ndarray, vid_shape: Optional[Sequence[int]] = None, |
| scale: Optional[float] = None, |
| enabled: Optional[bool] = None) -> np.ndarray: |
| """Apply the source-specific units fix to landmark coordinates. |
| |
| Priority: explicit ``scale`` > per-video theoretical value from ``vid_shape`` |
| (``(w/h)**2``) > ``config.X_AXIS_SCALE``. |
| """ |
| enabled = config.APPLY_KEYPOINT_AXIS_CORRECTION if enabled is None else enabled |
| if not enabled: |
| return coords |
| if scale is not None: |
| s = float(scale) |
| elif vid_shape is not None: |
| s = axis_scale_for_video_shape(vid_shape) |
| else: |
| s = float(config.X_AXIS_SCALE) |
| return apply_x_axis_scale(coords, s) |
|
|
|
|
| |
| |
| |
| @dataclass |
| class GeometryMeasurement: |
| shoulder_torso_ratio: float = float("nan") |
| face_vertical_over_interocular: float = float("nan") |
| videos_used: int = 0 |
| frames_used: int = 0 |
| x_axis_scale_from_body: float = float("nan") |
| x_axis_scale_from_face: float = float("nan") |
| x_axis_scale_theoretical: float = float("nan") |
| adopted_x_axis_scale: float = float("nan") |
| reference: Dict[str, float] = None |
|
|
| def as_dict(self) -> Dict: |
| d = asdict(self) |
| d["reference"] = self.reference or dict(config.MEDIAPIPE_REFERENCE) |
| return d |
|
|
|
|
| def measure_ratios_batch( |
| sequences: Iterable[Tuple[np.ndarray, np.ndarray]], |
| corrected: bool = False, |
| ) -> GeometryMeasurement: |
| """Measure the two anatomical ratios over an iterable of (coords, mask) clips. |
| |
| Parameters |
| ---------- |
| sequences : iterable of (T, 75, 3) coordinates and (T, 75) boolean/float masks |
| corrected : apply the x-axis fix before measuring (should be False for |
| calibration, True only for reporting the post-fix value) |
| """ |
| body, face = [], [] |
| vids = 0 |
| frames = 0 |
| for coords, mask in sequences: |
| coords = np.asarray(coords, dtype=np.float32) |
| mask = np.asarray(mask) > 0 |
| if coords.ndim != 3 or coords.shape[0] == 0: |
| continue |
| vids += 1 |
| b, f = [], [] |
| for t in range(coords.shape[0]): |
| m = mask[t] |
| frames += 1 |
| if m[L_SHOULDER] and m[R_SHOULDER] and m[L_HIP] and m[R_HIP]: |
| sh = abs(float(coords[t, L_SHOULDER, 0] - coords[t, R_SHOULDER, 0])) |
| tor = abs(float(coords[t, L_SHOULDER, 1] - coords[t, L_HIP, 1])) |
| if tor > 1e-6: |
| b.append(sh / tor) |
| if m[NOSE] and m[LEFT_EYE] and m[RIGHT_EYE]: |
| e = abs(float(coords[t, LEFT_EYE, 0] - coords[t, RIGHT_EYE, 0])) |
| v = abs(float(coords[t, LEFT_EYE, 1] + coords[t, RIGHT_EYE, 1]) / 2.0 |
| - float(coords[t, NOSE, 1])) |
| if e > 1e-6: |
| f.append(v / e) |
| if len(b) >= 5: |
| body.append(float(np.median(b))) |
| if len(f) >= 5: |
| face.append(float(np.median(f))) |
|
|
| meas = GeometryMeasurement(videos_used=vids, frames_used=frames) |
| ref = dict(config.MEDIAPIPE_REFERENCE) |
| if body: |
| meas.shoulder_torso_ratio = float(np.median(body)) |
| meas.x_axis_scale_from_body = float(ref["shoulder_torso_ratio"] / meas.shoulder_torso_ratio) |
| if face: |
| meas.face_vertical_over_interocular = float(np.median(face)) |
| |
| |
| |
| meas.x_axis_scale_from_face = float(meas.face_vertical_over_interocular |
| / ref["face_vertical_over_interocular"]) |
| meas.x_axis_scale_theoretical = float(config.X_AXIS_SCALE) |
| meas.adopted_x_axis_scale = float(config.X_AXIS_SCALE) |
| meas.reference = ref |
| return meas |
|
|
|
|
| def measure_ratios_from_dataset(limit: Optional[int] = None, seed: int = config.RANDOM_SEED, |
| corrected: bool = False) -> GeometryMeasurement: |
| """Calibrate on a random sample of official keypoint files.""" |
| import random |
|
|
| from .include50 import load_split_metadata, read_keypoint_archive_file, list_keypoint_files |
|
|
| files = list_keypoint_files() |
| if not files: |
| raise FileNotFoundError( |
| "No keypoint files found. Set ISL_KEYPOINTS_PATH or run " |
| "`python training/fetch_dataset.py --extract`." |
| ) |
| rng = random.Random(seed) |
| sample = files if not limit else rng.sample(files, min(limit, len(files))) |
|
|
| def gen(): |
| for f in sample: |
| try: |
| coords, mask, _ = read_keypoint_archive_file(f, apply_axis_fix=corrected) |
| except Exception: |
| continue |
| if coords.shape[0]: |
| yield coords, mask |
|
|
| return measure_ratios_batch(gen(), corrected=corrected) |
|
|
|
|
| def write_calibration(meas: GeometryMeasurement, path: Path = CALIBRATION_FILE) -> Path: |
| path.parent.mkdir(parents=True, exist_ok=True) |
| path.write_text(json.dumps(meas.as_dict(), indent=2)) |
| logger.info("geometry calibration written to %s", path) |
| return path |
|
|
|
|
| def load_calibration(path: Path = CALIBRATION_FILE) -> Dict: |
| if Path(path).exists(): |
| return json.loads(Path(path).read_text()) |
| return {} |
|
|
|
|
| def describe() -> str: |
| return ( |
| f"x-axis scale: {config.X_AXIS_SCALE:.4f} " |
| f"(correction {'ON' if config.APPLY_KEYPOINT_AXIS_CORRECTION else 'OFF'})" |
| ) |
|
|
