| """ |
| INCLUDE-50 dataset access layer. |
| |
| Responsibilities |
| ---------------- |
| 1. Read the **official AI4Bharat INCLUDE-50 metadata** (class list, video paths, |
| official train/test split) - the vocabulary is *derived*, never hard-coded. |
| 2. Read the **official AI4Bharat pre-extracted MediaPipe keypoints** (Zenodo |
| 6674324) and map them into the canonical 75-landmark layout. |
| 3. Optionally extract landmarks from the raw INCLUDE videos with the project's |
| own MediaPipe extractor (``KEYPOINT_SOURCE="mediapipe"``). |
| 4. Provide a reproducible, leakage-aware split (train / validation / test). |
| |
| Released keypoint format (verified against AI4Bharat/INCLUDE``generate_keypoints.py``) |
| ------------------------------------------------------------------------------------ |
| One JSON file per video named ``<label>_<uid>.json``, e.g. ``bird_MVI_2987.json``:: |
| |
| {"uid": "bird_MVI_2987", "label": "bird", "n_frames": 187, |
| "pose_x": [[...33 or 25 floats...], ...], # NaN when pose was not detected |
| "pose_y": [[...]], "hand1_x": [[...21...]], "hand1_y": [[...]], |
| "hand2_x": [[...]], "hand2_y": [[...]]} |
| |
| * coordinates are **2-D (x, y) only** - no z is present in the release; |
| * ``hand1``/``hand2`` are *not* left/right: AI4Bharat assigns them by nearest |
| wrist, so this loader re-derives left/right per frame from the pose wrists; |
| * missing data is ``NaN`` (never silently zero). |
| |
| Data licensing: INCLUDE is distributed under CC-BY-4.0. See ``docs/dataset_audit.md``. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| import logging |
| import re |
| from dataclasses import dataclass |
| from pathlib import Path |
| from typing import Dict, Iterable, List, Optional, Sequence, Tuple |
|
|
| import numpy as np |
| import pandas as pd |
|
|
| import config |
|
|
| logger = logging.getLogger(__name__) |
|
|
| |
| |
| POSE_LEFT_WRIST = 15 |
| POSE_RIGHT_WRIST = 16 |
|
|
| KEYPOINT_SUFFIXES = (".json", ".pkl") |
|
|
|
|
| |
| |
| |
| def metadata_path(name: str) -> Path: |
| return Path(config.METADATA_DIR) / name |
|
|
|
|
| def _gloss_normaliser() -> Dict[str, str]: |
| """raw folder name -> official normalised gloss (e.g. '55. Thank you' -> 'THANK_YOU').""" |
| p = metadata_path(config.METADATA_FILES["gloss_norm"]) |
| if not p.exists(): |
| return {} |
| df = pd.read_csv(p) |
| return {str(a): str(b) for a, b in zip(df["actual_gloss"], df["normalized_gloss"])} |
|
|
|
|
| def load_split_metadata(split: str, protocol: str | None = None) -> pd.DataFrame: |
| """Load one split of the official INCLUDE-50 metadata. |
| |
| Parameters |
| ---------- |
| split : ``"train"`` | ``"test"`` | ``"val"`` |
| protocol : ``"official"`` (AI4Bharat train/test CSVs) or ``"hf"`` (HuggingFace |
| INCLUDE parquet, which additionally carries a validation split). |
| |
| Returns |
| ------- |
| DataFrame with columns: FilePath, Video, Word (raw), label (normalised), |
| Category, split. |
| """ |
| protocol = (protocol or config.SPLIT_PROTOCOL).lower() |
| norm = _gloss_normaliser() |
|
|
| if protocol == "hf": |
| name = {"train": "hf_train", "val": "hf_val", "test": "hf_test"}.get(split) |
| path = metadata_path(config.METADATA_FILES[name]) |
| if not path.exists(): |
| raise FileNotFoundError( |
| f"HF metadata missing: {path}\nRun: python training/fetch_metadata.py --protocol hf" |
| ) |
| df = pd.read_parquet(path) |
| df = df[df["include_50"]].copy() |
| df = df.rename(columns={"video_path": "FilePath", "label": "Word"}) |
| df["Category"] = df["parent_label"] |
| df["Video"] = df["FilePath"].map(lambda p: str(p).split("/")[-1]) |
| else: |
| key = "train_50" if split == "train" else "test_50" |
| path = metadata_path(config.METADATA_FILES[key]) |
| if not path.exists(): |
| raise FileNotFoundError( |
| f"official INCLUDE-50 metadata missing: {path}\n" |
| "Run: python training/fetch_metadata.py --protocol official" |
| ) |
| df = pd.read_csv(path) |
| if split == "val": |
| |
| df = pd.read_csv(metadata_path(config.METADATA_FILES["train_50"])) |
|
|
| df = df.rename(columns={"Word": "raw_word"}) |
| df["label"] = df["raw_word"].map(lambda w: norm.get(str(w), str(w).strip().upper())) |
| df["Video"] = df["FilePath"].map(lambda p: str(p).split("/")[-1]) |
| df["Category"] = df["Category"].astype(str) |
| df["split"] = split |
| return df[["FilePath", "Video", "raw_word", "label", "Category", "split"]] |
|
|
|
|
| def load_all_metadata(protocol: str | None = None) -> pd.DataFrame: |
| protocol = (protocol or config.SPLIT_PROTOCOL).lower() |
| frames = [load_split_metadata(s, protocol) for s in ("train", "test")] |
| return pd.concat(frames, ignore_index=True) |
|
|
|
|
| def derive_vocabulary(protocol: str | None = None) -> List[str]: |
| """The 50 class names, **read from the metadata** (sorted, de-duplicated).""" |
| df = load_all_metadata(protocol) |
| return sorted(df["label"].unique().tolist()) |
|
|
|
|
| def save_vocabulary(protocol: str | None = None) -> Path: |
| """Freeze the verified vocabulary into ``datasets/metadata/vocabulary.json``.""" |
| protocol = (protocol or config.SPLIT_PROTOCOL).lower() |
| df = load_all_metadata(protocol) |
| counts = df.groupby("label").size().to_dict() |
| cat = df.groupby("label")["Category"].first().to_dict() |
| vocab = sorted(counts) |
| payload = { |
| "dataset": config.DATASET_NAME, |
| "protocol": protocol, |
| "citation": config.DATASET_CITATION, |
| "doi": config.DATASET_DOI, |
| "num_classes": len(vocab), |
| "classes": [ |
| {"index": i, "label": v, "raw_word": None, "category": cat.get(v), "videos": int(counts[v])} |
| for i, v in enumerate(vocab) |
| ], |
| } |
| raw = ( |
| df.drop_duplicates("label").set_index("label")["raw_word"].to_dict() |
| if "raw_word" in df |
| else {} |
| ) |
| for entry in payload["classes"]: |
| entry["raw_word"] = raw.get(entry["label"]) |
| out = metadata_path("vocabulary.json") |
| out.write_text(json.dumps(payload, indent=2)) |
| logger.info("vocabulary frozen: %s (%d classes)", out, len(vocab)) |
| return out |
|
|
|
|
| def load_vocabulary() -> List[str]: |
| """Read the frozen vocabulary (creating it from metadata if absent).""" |
| p = metadata_path("vocabulary.json") |
| if not p.exists(): |
| save_vocabulary() |
| payload = json.loads(p.read_text()) |
| return [c["label"] for c in payload["classes"]] |
|
|
|
|
| def class_mapping() -> Dict[str, int]: |
| vocab = load_vocabulary() |
| return {name: i for i, name in enumerate(vocab)} |
|
|
|
|
| |
| |
| |
| _SID_RE = re.compile(r"MVI[_-]?(\d+)", re.IGNORECASE) |
|
|
|
|
| def session_id(filepath: str) -> str: |
| """Recording-session identifier for a clip. |
| |
| INCLUDE file names look like ``MVI_2987.MOV``; ``MVI_0060_1.MP4`` style names |
| also occur. The numeric id is a *proxy* for the recording session: clips with |
| neighbouring ids inside one class are consecutive takes from the same sitting |
| (measured: 96% of within-class id gaps are <= 10), which is why the |
| session-disjoint split groups ids into blocks rather than treating every id as |
| its own session. INCLUDE publishes no explicit session field. |
| """ |
| stem = str(filepath).split("/")[-1].rsplit(".", 1)[0] |
| m = _SID_RE.search(stem) |
| if m: |
| return m.group(1) |
| digits = "".join(ch for ch in stem if ch.isdigit()) |
| return digits or stem |
|
|
|
|
| def session_block(filepath: str, block: int = 10) -> str: |
| """Coarse session proxy: consecutive-id *blocks*. |
| |
| Clips whose ids fall in the same block of size ``block`` share a recording |
| sitting with high probability. Used by the session-disjoint split so that |
| neighbouring takes never straddle train/test. |
| """ |
| sid = session_id(filepath) |
| if sid.isdigit(): |
| return str(int(sid) // int(block)) |
| return sid |
|
|
|
|
| @dataclass |
| class SplitSummary: |
| protocol: str |
| train: int |
| val: int |
| test: int |
| classes_train: int |
| classes_val: int |
| classes_test: int |
| val_strategy: str |
| session_overlap_train_test: int |
| note: str = "" |
|
|
| def as_dict(self) -> Dict[str, object]: |
| return self.__dict__.copy() |
|
|
|
|
| def build_index( |
| protocol: str | None = None, |
| vocab_filter: Optional[Sequence[str]] = None, |
| force_rebuild: bool = False, |
| ) -> Tuple[pd.DataFrame, SplitSummary]: |
| """Build the canonical train/val/test index (cached to CSV). |
| |
| Split policy |
| ------------ |
| * ``session_disjoint`` (default): pools the full INCLUDE-50 pool and splits by |
| recording cluster, so no burst of consecutive takes spans two splits. This is |
| the project's primary evaluation because the official split leaks. |
| * ``official``: uses AI4Bharat's official train/test files verbatim. |
| The official metadata ships **no validation split**, so validation is carved |
| out of the official train set with a seeded, class-stratified, *session-aware* |
| split (all clips sharing a session id fall on the same side). |
| * ``hf``: uses the HuggingFace INCLUDE train/val/test parquet split. |
| |
| The official split is not signer-disjoint (INCLUDE does not publish signer ids); |
| see ``docs/dataset_audit.md`` for the measured session overlap. |
| """ |
| protocol = (protocol or config.SPLIT_PROTOCOL).lower() |
| cache = metadata_path(f"index_{protocol}.csv") |
| summary_cache = metadata_path(f"index_{protocol}_summary.json") |
|
|
| if cache.exists() and not force_rebuild: |
| df = pd.read_csv(cache) |
| if summary_cache.exists(): |
| summary = SplitSummary(**json.loads(summary_cache.read_text())) |
| else: |
| summary = SplitSummary(protocol, 0, 0, 0, 0, 0, 0, "cached", 0) |
| if vocab_filter: |
| df = df[df["label"].isin(list(vocab_filter))].reset_index(drop=True) |
| return df, summary |
|
|
| if protocol == "session_disjoint": |
| pool = load_split_metadata("train", "official") |
| pool = pd.concat([pool, load_split_metadata("test", "official")], ignore_index=True) |
| train, val, test = _session_disjoint_split( |
| pool, config.VAL_FRACTION, config.TEST_FRACTION, config.RANDOM_SEED |
| ) |
| strategy = ( |
| "class-stratified recording-cluster-disjoint split over the full INCLUDE-50 pool " |
| f"(val={config.VAL_FRACTION}, test={config.TEST_FRACTION}, seed={config.RANDOM_SEED}, " |
| "max within-cluster id gap=10). Strongest defensible alternative to signer-disjoint " |
| "evaluation, which INCLUDE metadata cannot support." |
| ) |
| elif protocol == "hf": |
| train = load_split_metadata("train", "hf") |
| val = load_split_metadata("val", "hf") |
| test = load_split_metadata("test", "hf") |
| strategy = "official HuggingFace train/val/test" |
| else: |
| train_full = load_split_metadata("train", "official") |
| test = load_split_metadata("test", "official") |
| train, val = _stratified_session_split(train_full, config.VAL_FRACTION, config.RANDOM_SEED) |
| strategy = ( |
| f"official train/test + seeded stratified session-aware carve-out " |
| f"(val_fraction={config.VAL_FRACTION}, seed={config.RANDOM_SEED})" |
| ) |
|
|
| df = pd.concat([train, val, test], ignore_index=True) |
| df["split"] = df["split"].astype(str) |
|
|
| for part in ("train", "val", "test"): |
| df.loc[df["split"] == part, "subset"] = part |
| overlap = len(set(df[df.split == "train"].FilePath.map(session_id)) |
| & set(df[df.split == "test"].FilePath.map(session_id))) |
|
|
| summary = SplitSummary( |
| protocol=protocol, |
| train=int((df.split == "train").sum()), |
| val=int((df.split == "val").sum()), |
| test=int((df.split == "test").sum()), |
| classes_train=int(df[df.split == "train"].label.nunique()), |
| classes_val=int(df[df.split == "val"].label.nunique()), |
| classes_test=int(df[df.split == "test"].label.nunique()), |
| val_strategy=strategy, |
| session_overlap_train_test=int(overlap), |
| note="INCLUDE publishes no signer metadata -> signer-disjoint evaluation is not possible.", |
| ) |
|
|
| df.to_csv(cache, index=False) |
| summary_cache.write_text(json.dumps(summary.as_dict(), indent=2)) |
| logger.info("index built: train=%d val=%d test=%d (overlap sessions=%d)", |
| summary.train, summary.val, summary.test, summary.session_overlap_train_test) |
|
|
| if vocab_filter: |
| df = df[df["label"].isin(list(vocab_filter))].reset_index(drop=True) |
| return df, summary |
|
|
|
|
| def load_index(protocol: str | None = None, force_rebuild: bool = False, |
| vocab_filter: Optional[Sequence[str]] = None) -> pd.DataFrame: |
| """Public accessor for the split index (thin wrapper over :func:`build_index`).""" |
| df, _ = build_index(protocol=protocol, vocab_filter=vocab_filter, force_rebuild=force_rebuild) |
| return df |
|
|
|
|
| def session_clusters(index: pd.DataFrame, max_gap: int = 10) -> pd.Series: |
| """Group clips into recording clusters, per class. |
| |
| Within one class the released file ids run in consecutive bursts |
| (measured: 71% of neighbouring ids differ by exactly 1, 73% by <= 10). A burst |
| is a recording sitting: the same signer, same camera, same clothing. Splitting |
| inside a burst leaks information, so whole clusters are kept on one side. |
| |
| Returns a Series aligned with ``index`` holding a cluster key per clip. |
| """ |
| keys = pd.Series("", index=index.index, dtype=object) |
| for label, grp in index.groupby("label"): |
| ids = [(i, session_id(fp)) for i, fp in zip(grp.index, grp["FilePath"])] |
| numeric = [(i, int(s)) for i, s in ids if str(s).isdigit()] |
| other = [i for i, s in ids if not str(s).isdigit()] |
| numeric.sort(key=lambda t: t[1]) |
| for extra in other: |
| keys[extra] = f"{label}:other" |
| if not numeric: |
| continue |
| start = prev = numeric[0][1] |
| members = [numeric[0][0]] |
| for i, v in numeric[1:]: |
| if v - prev <= max_gap: |
| members.append(i) |
| else: |
| for m in members: |
| keys[m] = f"{label}:{start}-{prev}" |
| members, start = [i], v |
| prev = v |
| for m in members: |
| keys[m] = f"{label}:{start}-{prev}" |
| return keys |
|
|
|
|
| def _session_disjoint_split(df: pd.DataFrame, val_fraction: float, test_fraction: float, seed: int, |
| max_gap: int = 10): |
| """Class-stratified split in which no recording cluster spans two splits. |
| |
| This is the strongest defensible evaluation available: INCLUDE publishes no |
| signer identifiers, so signer-disjoint evaluation is impossible, but keeping |
| whole recording bursts together removes the session-level leakage that the |
| *official* split has (measured: 63% of clips share an id-block with the other |
| side of the official split). |
| """ |
| rng = np.random.default_rng(seed) |
| keys = session_clusters(df, max_gap=max_gap) |
| df = df.copy() |
| df["_cluster"] = keys.values |
| split_of_cluster: Dict[str, str] = {} |
| for label, grp in df.groupby("label"): |
| clusters = list(grp.groupby("_cluster").groups.items()) |
| rng.shuffle(clusters) |
| total = len(grp) |
| n_val = max(1, int(round(total * val_fraction))) |
| n_test = max(1, int(round(total * test_fraction))) |
| free = {cid: len(idxs) for cid, idxs in clusters} |
| |
| for name, quota in (("test", n_test), ("val", n_val)): |
| filled = 0 |
| while free and filled < quota: |
| best = min(free, key=lambda c: (abs(quota - filled - free[c]), -free[c])) |
| if filled + free[best] > quota and filled >= 0.6 * quota: |
| break |
| split_of_cluster[best] = name |
| filled += free.pop(best) |
| for cid in free: |
| split_of_cluster[cid] = "train" |
| df["split"] = df["_cluster"].map(split_of_cluster).fillna("train") |
| train = df[df.split == "train"].drop(columns=["_cluster"]) |
| val = df[df.split == "val"].drop(columns=["_cluster"]) |
| test = df[df.split == "test"].drop(columns=["_cluster"]) |
| return train.reset_index(drop=True), val.reset_index(drop=True), test.reset_index(drop=True) |
|
|
|
|
| def _stratified_session_split(df: pd.DataFrame, val_fraction: float, seed: int) -> Tuple[pd.DataFrame, pd.DataFrame]: |
| """Class-stratified split that keeps whole recording sessions together.""" |
| if val_fraction <= 0: |
| empty = df.iloc[0:0].copy() |
| return df.copy(), empty |
| rng = np.random.default_rng(seed) |
| val_idx: List[int] = [] |
| for label, grp in df.groupby("label"): |
| groups = grp.groupby(grp["FilePath"].map(session_id)).groups |
| keys = sorted(groups) |
| rng.shuffle(keys) |
| target = max(1, int(round(len(grp) * val_fraction))) |
| picked, n = [], 0 |
| for k in keys: |
| if n >= target: |
| break |
| idx = list(groups[k]) |
| picked.extend(idx) |
| n += len(idx) |
| val_idx.extend(picked[: max(1, len(picked))] if picked else []) |
| val = df.loc[sorted(set(val_idx))].copy() |
| train = df.drop(index=val.index).copy() |
| val["split"] = "val" |
| train["split"] = "train" |
| return train.reset_index(drop=True), val.reset_index(drop=True) |
|
|
|
|
| |
| |
| |
| |
| KEYPOINT_ARCHIVE_NAME = "Pose_Signs_raw.zip" |
| KEYPOINT_SUFFIX = ".pkl" |
|
|
| _zip_cache: Dict[str, "zipfile.ZipFile"] = {} |
|
|
|
|
| def keypoints_root() -> Path: |
| """First existing candidate directory for extracted keypoints.""" |
| candidates = [Path(config.KEYPOINTS_PATH), *[Path(p) for p in config.KEYPOINTS_FALLBACKS]] |
| for c in candidates: |
| if c.exists(): |
| return c |
| return Path(config.KEYPOINTS_PATH) |
|
|
|
|
| def keypoint_archive() -> Optional[Path]: |
| """The released zip archive, if present (searched next to the keypoint roots).""" |
| candidates = [ |
| Path(config.KEYPOINTS_PATH).parent / KEYPOINT_ARCHIVE_NAME, |
| Path(config.KEYPOINTS_PATH) / KEYPOINT_ARCHIVE_NAME, |
| Path("/data/include") / KEYPOINT_ARCHIVE_NAME, |
| ] |
| for p in (Path(c) for c in config.KEYPOINTS_FALLBACKS): |
| candidates.append(p / KEYPOINT_ARCHIVE_NAME) |
| candidates.append(p.parent / KEYPOINT_ARCHIVE_NAME) |
| for c in candidates: |
| if c.exists(): |
| return c |
| return None |
|
|
|
|
| def _open_archive(path: Path): |
| key = str(path) |
| if key not in _zip_cache: |
| import zipfile |
|
|
| _zip_cache[key] = zipfile.ZipFile(path) |
| return _zip_cache[key] |
|
|
|
|
| def keypoint_member_for(filepath: str, archive: Optional[Path] = None) -> Optional[str]: |
| """Map a metadata ``FilePath`` to the member name inside the keypoint archive. |
| |
| Handles the ``extra``/``Extra`` sub-directories that exist for a few classes. |
| """ |
| rel = str(filepath) |
| parts = rel.split("/") |
| if len(parts) < 3: |
| return None |
| category, word = parts[0], parts[1] |
| stem = parts[-1].rsplit(".", 1)[0] |
| arch = Path(archive or keypoint_archive()) |
| if arch is None: |
| return None |
| z = _open_archive(Path(arch)) |
| names = z.namelist() |
| prefixes = [f"{category}/{word}", f"{category}/{word}/extra", f"{category}/{word}/Extra", |
| f"{category}/{word}/Extra Videos"] |
| for pre in prefixes: |
| member = f"Pose_Signs/{pre}/{stem}{KEYPOINT_SUFFIX}" |
| if member in names: |
| return member |
| |
| lower = f"{stem}{KEYPOINT_SUFFIX}".lower() |
| for n in names: |
| if n.lower().endswith("/" + lower) and f"/{category}/".lower() in n.lower(): |
| return n |
| return None |
|
|
|
|
| def keypoint_path_for(filepath: str) -> Optional[Path]: |
| """Locate an *extracted* keypoint file for a dataset video.""" |
| root = keypoints_root() |
| parts = str(filepath).split("/") |
| if len(parts) < 3: |
| return None |
| category, word = parts[0], parts[1] |
| stem = parts[-1].rsplit(".", 1)[0] |
| for pre in (f"{category}/{word}", f"{category}/{word}/extra", |
| f"{category}/{word}/Extra", f"{category}/{word}/Extra Videos"): |
| for base in (root, root / "Pose_Signs"): |
| c = base / pre / f"{stem}{KEYPOINT_SUFFIX}" |
| if c.exists(): |
| return c |
| return None |
|
|
|
|
| def list_keypoint_files(limit: Optional[int] = None) -> List[str]: |
| """All available keypoint members (archive) or paths (extracted directory).""" |
| root = keypoints_root() |
| if root.exists() and any(root.rglob(f"*{KEYPOINT_SUFFIX}")): |
| files = [str(p) for p in sorted(root.rglob(f"*{KEYPOINT_SUFFIX}"))] |
| else: |
| arch = keypoint_archive() |
| if arch is None: |
| return [] |
| files = [n for n in _open_archive(arch).namelist() if n.endswith(KEYPOINT_SUFFIX)] |
| return files[:limit] if limit else files |
|
|
|
|
| def read_keypoint_archive_file(member: str, apply_axis_fix: Optional[bool] = None, |
| max_frames: Optional[int] = None, |
| conf_threshold: Optional[float] = None): |
| """Read one released keypoint file. |
| |
| ``member`` may be either an archive member name (``Pose_Signs/.../X.pkl``) or a |
| filesystem path. Returns ``(coords (T,75,3) float32, mask (T,75) float32, info dict)``. |
| |
| Verified released structure:: |
| |
| {"keypoints": (T, 75, 3) float64, |
| "confidences": (T, 75) float64 in {0, 1}, |
| "vid_shape": (height, width)} |
| |
| Missing landmarks are marked by confidence 0 **and** zero coordinates; the mask |
| is built from confidences (falling back to coordinate activity when a file |
| lacks them). |
| """ |
| import pickle |
|
|
| from . import geometry as geom |
|
|
| info: Dict[str, object] = {"member": str(member)} |
| if isinstance(member, (str, Path)) and not str(member).startswith("Pose_Signs/"): |
| fh = open(member, "rb") |
| else: |
| arch = keypoint_archive() |
| if arch is None: |
| raise FileNotFoundError("keypoint archive not found; set ISL_KEYPOINTS_PATH") |
| fh = _open_archive(arch).open(str(member), "r") |
| with fh: |
| data = pickle.load(fh) |
|
|
| raw = np.asarray(data.get("keypoints", np.zeros((0, config.NUM_LANDMARKS, 3))), dtype=np.float32) |
| conf = data.get("confidences", None) |
| vid_shape = tuple(int(x) for x in np.asarray(data.get("vid_shape", (0, 0))).ravel()) |
| info["vid_shape"] = vid_shape |
| info["n_frames"] = int(raw.shape[0]) |
|
|
| if raw.ndim != 3: |
| raise ValueError(f"unexpected keypoints shape {raw.shape}") |
|
|
| t = raw.shape[0] if max_frames is None else min(int(max_frames), raw.shape[0]) |
| coords, mask = raw[:t].copy(), _mask_from_confidences(conf, t, conf_threshold) |
|
|
| fix = config.APPLY_KEYPOINT_AXIS_CORRECTION if apply_axis_fix is None else apply_axis_fix |
| if fix: |
| coords = geom.correct_landmarks(coords, vid_shape=vid_shape) |
| scale = geom.axis_scale_for_video_shape(vid_shape) |
| info["x_axis_scale"] = scale |
| coords[mask <= 0] = 0.0 |
| return coords.astype(np.float32), mask.astype(np.float32), info |
|
|
|
|
| def _mask_from_confidences(conf, t: int, threshold: Optional[float] = None) -> np.ndarray: |
| thr = config.KEYPOINT_CONFIDENCE_THRESHOLD if threshold is None else float(threshold) |
| if conf is None: |
| return np.ones((t, config.NUM_LANDMARKS), dtype=np.float32) |
| conf = np.asarray(conf, dtype=np.float32)[:t] |
| if conf.ndim != 2: |
| return np.ones((t, config.NUM_LANDMARKS), dtype=np.float32) |
| return (conf > thr).astype(np.float32) |
|
|
|
|
| def read_keypoints_for(filepath: str, max_frames: Optional[int] = None, |
| apply_axis_fix: Optional[bool] = None): |
| """Read keypoints for a metadata FilePath from whichever layout is available.""" |
| p = keypoint_path_for(filepath) |
| if p is not None: |
| return read_keypoint_archive_file(str(p), apply_axis_fix=apply_axis_fix, max_frames=max_frames) |
| member = keypoint_member_for(filepath) |
| if member is None: |
| raise FileNotFoundError(f"no keypoints found for {filepath}") |
| return read_keypoint_archive_file(member, apply_axis_fix=apply_axis_fix, max_frames=max_frames) |
|
|
|
|
| def is_corrupt(record: Optional[Dict[str, object]], path: str) -> bool: |
| """True when a keypoint entry is listed as unreadable in the CRC scan.""" |
| if record is None: |
| return False |
| return str(path) in set(record.get("corrupt", [])) |
|
|
