Buckets:

AsianPlayer
/

TriDrive

Files

xet

AsianPlayer/TriDrive / code /aide /data /protocol.py

AsianPlayer

16 days ago

download

raw

12.7 kB

"""AIDE / Driver-WM evaluation protocol constants (arXiv 2605.05092v2, Sec. 4 + App. B).
Everything that defines *what is being measured* lives here so that the cache builder,
the analytic baselines, the trainer and the harness cannot drift apart.
Coordinate frame: HALPE-136 2-D keypoints of the in-cabin ("incar") view, pixels of a
1920x1080 image. Keypoint arrays are stored as [..., 136, 3] = (x, y, conf).
"""
from __future__ import annotations

import math
import os
from pathlib import Path

import numpy as np

# --------------------------------------------------------------------------- paths
AIDE_ROOT = Path(os.environ.get("AIDE_ROOT", "/home/USER/AIDE"))
AIDE_ANN_DIR = AIDE_ROOT / "annotation"
HERE = Path(__file__).resolve().parent
AIDE_PKG = HERE.parent                       # cwm3/aide
RESULTS = AIDE_PKG / "results"
CACHE_NPZ = RESULTS / "cache" / "aide_kp.npz"
SPLIT_CSV = {s: HERE / f"splits_{n}.csv" for s, n in
             (("train", "training"), ("val", "validation"), ("test", "testing"))}
EXPECTED_SPLIT_SIZES = {"train": 1884, "val": 405, "test": 609}
VIEWS = ("front", "left", "right", "incar")

# --------------------------------------------------------------------------- image / clip
W, H = 1920, 1080
DIAG = math.sqrt(W ** 2 + H ** 2)             # 2202.907...
FPS = 15
N_FRAMES_STD = 45                              # 3 s clips; 13 clips are longer
N_SAMPLE = 10                                  # frames per clip used by the protocol
T_OBS = 5
T_PRED = 5
assert T_OBS + T_PRED == N_SAMPLE
J = 136

# --------------------------------------------------------------------------- sampling grid


# Protocol grid, LOCKED BY GATE 0 (aide/data/verify_protocol.py, results/cache/verify_protocol.json):
# 'floor' == np.linspace(0, n-1, 10).astype(int) -> for n=45: 0,4,9,14,19,24,29,34,39,44.
# This is the only grid that reproduces the paper's zero-velocity All/HM numbers (52.93/139.23
# vs 52.89/139.19). 'round' (0,5,10,15,20,24,29,34,39,44) gives 53.17/136.17 with the same
# driver rule and is kept as a sensitivity variant. The future block (24,29,34,39,44) is
# identical for floor and round, so the HM threshold (72.05) does not depend on this choice.
SAMPLE_GRID = "floor"


def sample_idx(n: int, mode: str | None = None) -> np.ndarray:
    """Uniformly sample N_SAMPLE frame indices from a clip of n frames (mode=None -> SAMPLE_GRID)."""
    mode = SAMPLE_GRID if mode is None else mode
    g = np.linspace(0, n - 1, N_SAMPLE)
    if mode == "round":
        idx = np.round(g)
    elif mode == "floor":
        idx = np.floor(g)
    elif mode == "ceil":
        idx = np.ceil(g)
    else:
        raise ValueError(mode)
    idx = idx.astype(int)
    assert idx[0] == 0 and idx[-1] == n - 1 and len(idx) == N_SAMPLE
    return idx


SAMPLE_GRIDS = ("round", "floor", "ceil")

# --------------------------------------------------------------------------- driver selection
# AIDE annotation frames often contain several detections (17.5 % of frames; passengers,
# mirror reflections). bbox is [x, y, w, h] in pixels (verified: max x+w == 1920).
# Protocol rule, LOCKED BY GATE 0: the driver is the FIRST detection in `result`
# (`result[0]`; identical to "minimum tracking idx"). On the official test split this
# reproduces the paper's zero-velocity numbers (All 52.93 / d-n 2.403 / HM 139.23 / HM
# threshold 72.052 / PCK@0.05 85.95 vs paper 52.89 / 2.40 / 139.19 / 72.05 / 85.95).
# 'largest_bbox' (the rule used in the earlier all-2,898-clip audit, where its 52.75 was a
# coincidence) gives 48.64 / 107.9 on the official test split and is kept as the main
# sensitivity variant (it is arguably the semantically better driver selector).
# If a frame has zero detections the previous frame's selected detection is carried
# forward (and the first frames are back-filled from the first frame with a detection).
DRIVER_RULE = "first"
DRIVER_RULES = ("first", "largest_bbox", "max_score", "iou_track")
ALT_DRIVER_RULE, ALT_SAMPLE_GRID = "largest_bbox", "round"   # stored in the cache as S10_alt


def bbox_area(b) -> float:
    return float(b[2]) * float(b[3])


def bbox_iou_xywh(a, b) -> float:
    ax0, ay0, ax1, ay1 = a[0], a[1], a[0] + a[2], a[1] + a[3]
    bx0, by0, bx1, by1 = b[0], b[1], b[0] + b[2], b[1] + b[3]
    iw = max(0.0, min(ax1, bx1) - max(ax0, bx0))
    ih = max(0.0, min(ay1, by1) - max(ay0, by0))
    inter = iw * ih
    ua = a[2] * a[3] + b[2] * b[3] - inter
    return inter / ua if ua > 0 else 0.0


def select_driver(frames: list, rule: str = DRIVER_RULE):
    """frames: pose_list (list over frames of {'imgname', 'result': [det, ...]}).
    Returns (kp [n,136,3] float32, bbox [n,4] float32, n_det [n] int, filled [n] bool).
    `filled` marks frames whose detection was carried over from a neighbour.
    """
    n = len(frames)
    kp = np.full((n, J, 3), np.nan, dtype=np.float32)
    bb = np.full((n, 4), np.nan, dtype=np.float32)
    ndet = np.zeros(n, dtype=np.int16)
    prev_box = None
    for t, fr in enumerate(frames):
        dets = fr["result"]
        ndet[t] = len(dets)
        if not dets:
            continue
        if rule == "largest_bbox":
            d = max(dets, key=lambda r: bbox_area(r["bbox"]))
        elif rule == "first":
            d = dets[0]
        elif rule == "max_score":
            d = max(dets, key=lambda r: float(r["proposal_score"]))
        elif rule == "iou_track":
            if prev_box is None:
                d = max(dets, key=lambda r: bbox_area(r["bbox"]))
            else:
                d = max(dets, key=lambda r: bbox_iou_xywh(r["bbox"], prev_box))
        else:
            raise ValueError(rule)
        k = np.asarray(d["keypoints"], dtype=np.float32)
        assert k.size == J * 3, k.size
        kp[t] = k.reshape(J, 3)
        bb[t] = np.asarray(d["bbox"], dtype=np.float32)
        prev_box = bb[t]
    filled = ndet == 0
    # forward fill then back fill frames without any detection
    last = None
    for t in range(n):
        if filled[t]:
            if last is not None:
                kp[t], bb[t] = kp[last], bb[last]
        else:
            last = t
    nxt = None
    for t in range(n - 1, -1, -1):
        if filled[t] and np.isnan(kp[t, 0, 0]):
            if nxt is not None:
                kp[t], bb[t] = kp[nxt], bb[nxt]
        elif not filled[t]:
            nxt = t
    return kp, bb, ndet, filled


# --------------------------------------------------------------------------- HALPE-136 layout
# 0-16 COCO body, 17 head-top, 18 neck, 19 pelvis, 20-25 feet (L/R big toe, small toe, heel),
# 26-93 face (68 iBUG landmarks), 94-114 left hand (21), 115-135 right hand (21).
PARTS = {
    "body": np.arange(0, 26),
    "face": np.arange(26, 94),
    "lhand": np.arange(94, 115),
    "rhand": np.arange(115, 136),
}
PART_NAMES = tuple(PARTS)
assert sum(len(v) for v in PARTS.values()) == J

# Kinematic edges. Source: AlphaPose `alphapose/utils/vis.py`, `l_pair` for the 136-keypoint
# Halpe format (github.com/MVIG-SJTU/AlphaPose; Halpe-FullBody repo by Fang-Haoshu).
EDGES_BODY = [
    (0, 1), (0, 2), (1, 3), (2, 4),                               # head
    (5, 18), (6, 18), (5, 7), (7, 9), (6, 8), (8, 10),            # arms via neck
    (17, 18), (18, 19), (19, 11), (19, 12),                       # spine / pelvis
    (11, 13), (12, 14), (13, 15), (14, 16),                       # legs
    (20, 24), (21, 25), (23, 25), (22, 24), (15, 24), (16, 25),   # feet
]
_HAND_CHAIN = [(0, 1), (1, 2), (2, 3), (3, 4),
               (0, 5), (5, 6), (6, 7), (7, 8),
               (0, 9), (9, 10), (10, 11), (11, 12),
               (0, 13), (13, 14), (14, 15), (15, 16),
               (0, 17), (17, 18), (18, 19), (19, 20)]
EDGES_LHAND = [(94 + a, 94 + b) for a, b in _HAND_CHAIN]
EDGES_RHAND = [(115 + a, 115 + b) for a, b in _HAND_CHAIN]
EDGES_BODY_HANDS = EDGES_BODY + EDGES_LHAND + EDGES_RHAND

# Face edges derived from the standard iBUG-68 topology (not from AlphaPose): jaw chain,
# brows, nose bridge + base, two eye loops, outer and inner lip loops. Optional; the
# geometric losses use EDGES_BODY_HANDS.


def _chain(lo, hi, close=False):
    e = [(i, i + 1) for i in range(lo, hi)]
    if close:
        e.append((hi, lo))
    return e


_FACE68 = (_chain(0, 16) + _chain(17, 21) + _chain(22, 26) + _chain(27, 30) + _chain(31, 35)
           + _chain(36, 41, True) + _chain(42, 47, True) + _chain(48, 59, True) + _chain(60, 67, True))
EDGES_FACE = [(26 + a, 26 + b) for a, b in _FACE68]
EDGES_ALL = EDGES_BODY_HANDS + EDGES_FACE

# Left/right flip pairs (for horizontal-flip augmentation; swap indices then mirror x).
FLIP_PAIRS_BODY = [(1, 2), (3, 4), (5, 6), (7, 8), (9, 10), (11, 12), (13, 14), (15, 16),
                   (20, 21), (22, 23), (24, 25)]
FLIP_PAIRS_HANDS = [(94 + i, 115 + i) for i in range(21)]
_FACE68_FLIP = ([(i, 16 - i) for i in range(8)] + [(17 + i, 26 - i) for i in range(5)]
                + [(31, 35), (32, 34)]
                + [(36, 45), (37, 44), (38, 43), (39, 42), (40, 47), (41, 46)]
                + [(48, 54), (49, 53), (50, 52), (55, 59), (56, 58)]
                + [(60, 64), (61, 63), (65, 67)])
FLIP_PAIRS_FACE = [(26 + a, 26 + b) for a, b in _FACE68_FLIP]
FLIP_PAIRS = FLIP_PAIRS_BODY + FLIP_PAIRS_HANDS + FLIP_PAIRS_FACE


def flip_perm() -> np.ndarray:
    p = np.arange(J)
    for a, b in FLIP_PAIRS:
        p[a], p[b] = b, a
    return p


# --------------------------------------------------------------------------- labels
TASKS = ("dbr", "der", "tcr", "vcr")
TASK_KEY = {"dbr": "driver_behavior_label", "der": "emotion_label",
            "tcr": "scene_centric_context_label", "vcr": "vehicle_based_context_label"}
CLASSES = {
    "dbr": ("Normal Driving", "Looking Around", "Making Phone", "Body Movement",
            "Talking", "Smoking", "Dozing Off"),
    "der": ("Peace", "Anxiety", "Weariness", "Happiness", "Anger"),
    "tcr": ("Smooth Traffic", "Waiting", "Traffic Jam"),
    "vcr": ("Forward Moving", "Parking", "Turning", "Backward Moving", "Changing Lane"),
}
# Raw-string variants observed in the annotation files -> canonical class name.
LABEL_NORMALISE = {
    "happiness": "Happiness",
    "weariness": "Weariness",
    "Forward": "Forward Moving",
}


def normalise_label(task: str, raw: str) -> str:
    s = str(raw).strip()
    s = LABEL_NORMALISE.get(s, s)
    if s in CLASSES[task]:
        return s
    low = {c.lower(): c for c in CLASSES[task]}
    if s.lower() in low:
        return low[s.lower()]
    raise KeyError(f"unknown label {raw!r} for task {task}")


def label_index(task: str, raw: str) -> int:
    return CLASSES[task].index(normalise_label(task, raw))


# --------------------------------------------------------------------------- metrics protocol
PCK_THRESHOLDS = (0.05, 0.10)
# PCK normaliser, LOCKED BY GATE 0: the paper's PCK@0.05 = 85.95 is reproduced (85.948) only
# with threshold = 0.05 * max(W, H) = 96 px; 0.05 * DIAG (110.1 px) gives 87.67. d-nMPJPE,
# in contrast, does use DIAG (100*52.93/2202.9 = 2.403 = paper 2.40). The DIAG-normalised
# PCK is reported as a secondary metric (pck05_diag / pck10_diag).
PCK_NORM = float(max(W, H))


def pck_threshold_px(t: float, norm: str = "max_side") -> float:
    return t * (PCK_NORM if norm == "max_side" else DIAG)
HM_FRACTION = 0.10                             # High-Motion subset = top int(0.1*N) clips
HM_N_TRANSITIONS = 4                           # displacement averaged over the 4 future transitions


def hm_top(n_clips: int) -> int:
    """Size of the High-Motion subset for a split with n_clips clips (test: 60)."""
    return int(HM_FRACTION * n_clips)


assert hm_top(EXPECTED_SPLIT_SIZES["test"]) == 60

# Published Driver-WM Table-1 numbers on the official test split (for Gate 0 / tables).
PAPER = {
    "zero_velocity": {"mpjpe_all": 52.89, "dn_mpjpe": 2.40, "mpjpe_hm": 139.19,
                      "pck05": 85.95, "hm_threshold": 72.05},
    "static_pooling": {"mpjpe_hm": 134.56},
    "driver_wm": {"mpjpe_all": 71.47, "dn_mpjpe": 3.24, "mpjpe_hm": 138.03, "pck05": 71.66,
                  "f1_dbr": 68.07, "f1_der": 72.61, "f1_tcr": 90.15, "f1_vcr": 68.34,
                  "hm_h5": 155.82},
}
GATE0_TOL = {"mpjpe_all": 0.5, "dn_mpjpe": 0.03, "mpjpe_hm": 2.0, "hm_threshold": 2.0, "pck05": 1.0}

# Qwen3-VL feature-extraction prompt (Driver-WM App. B.2, verbatim).
QWEN_PROMPT = ("Extract multimodal features that capture: (A) external traffic cues (vehicles, "
               "signals, road agents), (B) how these cues act as triggers for driver motion "
               "responses, and (C) the short-term evolution (1-2 seconds) of driver behavior in "
               "response to the changing road context.")

Xet Storage Details

Size:: 12.7 kB
Xet hash:: c7f20950cef89c5eae6b25e2cc8c26acc5be17bf6ffcf756d811c2e4605b0e77

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.

	"""AIDE / Driver-WM evaluation protocol constants (arXiv 2605.05092v2, Sec. 4 + App. B).

	Everything that defines what is being measured lives here so that the cache builder,
	the analytic baselines, the trainer and the harness cannot drift apart.

	Coordinate frame: HALPE-136 2-D keypoints of the in-cabin ("incar") view, pixels of a
	1920x1080 image. Keypoint arrays are stored as [..., 136, 3] = (x, y, conf).
	"""
	from __future__ import annotations

	import math
	import os
	from pathlib import Path

	import numpy as np

	# --------------------------------------------------------------------------- paths
	AIDE_ROOT = Path(os.environ.get("AIDE_ROOT", "/home/USER/AIDE"))
	AIDE_ANN_DIR = AIDE_ROOT / "annotation"
	HERE = Path(__file__).resolve().parent
	AIDE_PKG = HERE.parent # cwm3/aide
	RESULTS = AIDE_PKG / "results"
	CACHE_NPZ = RESULTS / "cache" / "aide_kp.npz"
	SPLIT_CSV = {s: HERE / f"splits_{n}.csv" for s, n in
	(("train", "training"), ("val", "validation"), ("test", "testing"))}
	EXPECTED_SPLIT_SIZES = {"train": 1884, "val": 405, "test": 609}
	VIEWS = ("front", "left", "right", "incar")

	# --------------------------------------------------------------------------- image / clip
	W, H = 1920, 1080
	DIAG = math.sqrt(W 2 + H 2) # 2202.907...
	FPS = 15
	N_FRAMES_STD = 45 # 3 s clips; 13 clips are longer
	N_SAMPLE = 10 # frames per clip used by the protocol
	T_OBS = 5
	T_PRED = 5
	assert T_OBS + T_PRED == N_SAMPLE
	J = 136

	# --------------------------------------------------------------------------- sampling grid


	# Protocol grid, LOCKED BY GATE 0 (aide/data/verify_protocol.py, results/cache/verify_protocol.json):
	# 'floor' == np.linspace(0, n-1, 10).astype(int) -> for n=45: 0,4,9,14,19,24,29,34,39,44.
	# This is the only grid that reproduces the paper's zero-velocity All/HM numbers (52.93/139.23
	# vs 52.89/139.19). 'round' (0,5,10,15,20,24,29,34,39,44) gives 53.17/136.17 with the same
	# driver rule and is kept as a sensitivity variant. The future block (24,29,34,39,44) is
	# identical for floor and round, so the HM threshold (72.05) does not depend on this choice.
	SAMPLE_GRID = "floor"


	def sample_idx(n: int, mode: str \| None = None) -> np.ndarray:
	"""Uniformly sample N_SAMPLE frame indices from a clip of n frames (mode=None -> SAMPLE_GRID)."""
	mode = SAMPLE_GRID if mode is None else mode
	g = np.linspace(0, n - 1, N_SAMPLE)
	if mode == "round":
	idx = np.round(g)
	elif mode == "floor":
	idx = np.floor(g)
	elif mode == "ceil":
	idx = np.ceil(g)
	else:
	raise ValueError(mode)
	idx = idx.astype(int)
	assert idx[0] == 0 and idx[-1] == n - 1 and len(idx) == N_SAMPLE
	return idx


	SAMPLE_GRIDS = ("round", "floor", "ceil")

	# --------------------------------------------------------------------------- driver selection
	# AIDE annotation frames often contain several detections (17.5 % of frames; passengers,
	# mirror reflections). bbox is [x, y, w, h] in pixels (verified: max x+w == 1920).
	# Protocol rule, LOCKED BY GATE 0: the driver is the FIRST detection in `result`
	# (`result[0]`; identical to "minimum tracking idx"). On the official test split this
	# reproduces the paper's zero-velocity numbers (All 52.93 / d-n 2.403 / HM 139.23 / HM
	# threshold 72.052 / PCK@0.05 85.95 vs paper 52.89 / 2.40 / 139.19 / 72.05 / 85.95).
	# 'largest_bbox' (the rule used in the earlier all-2,898-clip audit, where its 52.75 was a
	# coincidence) gives 48.64 / 107.9 on the official test split and is kept as the main
	# sensitivity variant (it is arguably the semantically better driver selector).
	# If a frame has zero detections the previous frame's selected detection is carried
	# forward (and the first frames are back-filled from the first frame with a detection).
	DRIVER_RULE = "first"
	DRIVER_RULES = ("first", "largest_bbox", "max_score", "iou_track")
	ALT_DRIVER_RULE, ALT_SAMPLE_GRID = "largest_bbox", "round" # stored in the cache as S10_alt


	def bbox_area(b) -> float:
	return float(b[2]) * float(b[3])


	def bbox_iou_xywh(a, b) -> float:
	ax0, ay0, ax1, ay1 = a[0], a[1], a[0] + a[2], a[1] + a[3]
	bx0, by0, bx1, by1 = b[0], b[1], b[0] + b[2], b[1] + b[3]
	iw = max(0.0, min(ax1, bx1) - max(ax0, bx0))
	ih = max(0.0, min(ay1, by1) - max(ay0, by0))
	inter = iw * ih
	ua = a[2] * a[3] + b[2] * b[3] - inter
	return inter / ua if ua > 0 else 0.0


	def select_driver(frames: list, rule: str = DRIVER_RULE):
	"""frames: pose_list (list over frames of {'imgname', 'result': [det, ...]}).

	Returns (kp [n,136,3] float32, bbox [n,4] float32, n_det [n] int, filled [n] bool).
	`filled` marks frames whose detection was carried over from a neighbour.
	"""
	n = len(frames)
	kp = np.full((n, J, 3), np.nan, dtype=np.float32)
	bb = np.full((n, 4), np.nan, dtype=np.float32)
	ndet = np.zeros(n, dtype=np.int16)
	prev_box = None
	for t, fr in enumerate(frames):
	dets = fr["result"]
	ndet[t] = len(dets)
	if not dets:
	continue
	if rule == "largest_bbox":
	d = max(dets, key=lambda r: bbox_area(r["bbox"]))
	elif rule == "first":
	d = dets[0]
	elif rule == "max_score":
	d = max(dets, key=lambda r: float(r["proposal_score"]))
	elif rule == "iou_track":
	if prev_box is None:
	d = max(dets, key=lambda r: bbox_area(r["bbox"]))
	else:
	d = max(dets, key=lambda r: bbox_iou_xywh(r["bbox"], prev_box))
	else:
	raise ValueError(rule)
	k = np.asarray(d["keypoints"], dtype=np.float32)
	assert k.size == J * 3, k.size
	kp[t] = k.reshape(J, 3)
	bb[t] = np.asarray(d["bbox"], dtype=np.float32)
	prev_box = bb[t]
	filled = ndet == 0
	# forward fill then back fill frames without any detection
	last = None
	for t in range(n):
	if filled[t]:
	if last is not None:
	kp[t], bb[t] = kp[last], bb[last]
	else:
	last = t
	nxt = None
	for t in range(n - 1, -1, -1):
	if filled[t] and np.isnan(kp[t, 0, 0]):
	if nxt is not None:
	kp[t], bb[t] = kp[nxt], bb[nxt]
	elif not filled[t]:
	nxt = t
	return kp, bb, ndet, filled


	# --------------------------------------------------------------------------- HALPE-136 layout
	# 0-16 COCO body, 17 head-top, 18 neck, 19 pelvis, 20-25 feet (L/R big toe, small toe, heel),
	# 26-93 face (68 iBUG landmarks), 94-114 left hand (21), 115-135 right hand (21).
	PARTS = {
	"body": np.arange(0, 26),
	"face": np.arange(26, 94),
	"lhand": np.arange(94, 115),
	"rhand": np.arange(115, 136),
	}
	PART_NAMES = tuple(PARTS)
	assert sum(len(v) for v in PARTS.values()) == J

	# Kinematic edges. Source: AlphaPose `alphapose/utils/vis.py`, `l_pair` for the 136-keypoint
	# Halpe format (github.com/MVIG-SJTU/AlphaPose; Halpe-FullBody repo by Fang-Haoshu).
	EDGES_BODY = [
	(0, 1), (0, 2), (1, 3), (2, 4), # head
	(5, 18), (6, 18), (5, 7), (7, 9), (6, 8), (8, 10), # arms via neck
	(17, 18), (18, 19), (19, 11), (19, 12), # spine / pelvis
	(11, 13), (12, 14), (13, 15), (14, 16), # legs
	(20, 24), (21, 25), (23, 25), (22, 24), (15, 24), (16, 25), # feet
	]
	_HAND_CHAIN = [(0, 1), (1, 2), (2, 3), (3, 4),
	(0, 5), (5, 6), (6, 7), (7, 8),
	(0, 9), (9, 10), (10, 11), (11, 12),
	(0, 13), (13, 14), (14, 15), (15, 16),
	(0, 17), (17, 18), (18, 19), (19, 20)]
	EDGES_LHAND = [(94 + a, 94 + b) for a, b in _HAND_CHAIN]
	EDGES_RHAND = [(115 + a, 115 + b) for a, b in _HAND_CHAIN]
	EDGES_BODY_HANDS = EDGES_BODY + EDGES_LHAND + EDGES_RHAND

	# Face edges derived from the standard iBUG-68 topology (not from AlphaPose): jaw chain,
	# brows, nose bridge + base, two eye loops, outer and inner lip loops. Optional; the
	# geometric losses use EDGES_BODY_HANDS.


	def _chain(lo, hi, close=False):
	e = [(i, i + 1) for i in range(lo, hi)]
	if close:
	e.append((hi, lo))
	return e


	_FACE68 = (_chain(0, 16) + _chain(17, 21) + _chain(22, 26) + _chain(27, 30) + _chain(31, 35)
	+ _chain(36, 41, True) + _chain(42, 47, True) + _chain(48, 59, True) + _chain(60, 67, True))
	EDGES_FACE = [(26 + a, 26 + b) for a, b in _FACE68]
	EDGES_ALL = EDGES_BODY_HANDS + EDGES_FACE

	# Left/right flip pairs (for horizontal-flip augmentation; swap indices then mirror x).
	FLIP_PAIRS_BODY = [(1, 2), (3, 4), (5, 6), (7, 8), (9, 10), (11, 12), (13, 14), (15, 16),
	(20, 21), (22, 23), (24, 25)]
	FLIP_PAIRS_HANDS = [(94 + i, 115 + i) for i in range(21)]
	_FACE68_FLIP = ([(i, 16 - i) for i in range(8)] + [(17 + i, 26 - i) for i in range(5)]
	+ [(31, 35), (32, 34)]
	+ [(36, 45), (37, 44), (38, 43), (39, 42), (40, 47), (41, 46)]
	+ [(48, 54), (49, 53), (50, 52), (55, 59), (56, 58)]
	+ [(60, 64), (61, 63), (65, 67)])
	FLIP_PAIRS_FACE = [(26 + a, 26 + b) for a, b in _FACE68_FLIP]
	FLIP_PAIRS = FLIP_PAIRS_BODY + FLIP_PAIRS_HANDS + FLIP_PAIRS_FACE


	def flip_perm() -> np.ndarray:
	p = np.arange(J)
	for a, b in FLIP_PAIRS:
	p[a], p[b] = b, a
	return p


	# --------------------------------------------------------------------------- labels
	TASKS = ("dbr", "der", "tcr", "vcr")
	TASK_KEY = {"dbr": "driver_behavior_label", "der": "emotion_label",
	"tcr": "scene_centric_context_label", "vcr": "vehicle_based_context_label"}
	CLASSES = {
	"dbr": ("Normal Driving", "Looking Around", "Making Phone", "Body Movement",
	"Talking", "Smoking", "Dozing Off"),
	"der": ("Peace", "Anxiety", "Weariness", "Happiness", "Anger"),
	"tcr": ("Smooth Traffic", "Waiting", "Traffic Jam"),
	"vcr": ("Forward Moving", "Parking", "Turning", "Backward Moving", "Changing Lane"),
	}
	# Raw-string variants observed in the annotation files -> canonical class name.
	LABEL_NORMALISE = {
	"happiness": "Happiness",
	"weariness": "Weariness",
	"Forward": "Forward Moving",
	}


	def normalise_label(task: str, raw: str) -> str:
	s = str(raw).strip()
	s = LABEL_NORMALISE.get(s, s)
	if s in CLASSES[task]:
	return s
	low = {c.lower(): c for c in CLASSES[task]}
	if s.lower() in low:
	return low[s.lower()]
	raise KeyError(f"unknown label {raw!r} for task {task}")


	def label_index(task: str, raw: str) -> int:
	return CLASSES[task].index(normalise_label(task, raw))


	# --------------------------------------------------------------------------- metrics protocol
	PCK_THRESHOLDS = (0.05, 0.10)
	# PCK normaliser, LOCKED BY GATE 0: the paper's PCK@0.05 = 85.95 is reproduced (85.948) only
	# with threshold = 0.05 * max(W, H) = 96 px; 0.05 * DIAG (110.1 px) gives 87.67. d-nMPJPE,
	# in contrast, does use DIAG (100*52.93/2202.9 = 2.403 = paper 2.40). The DIAG-normalised
	# PCK is reported as a secondary metric (pck05_diag / pck10_diag).
	PCK_NORM = float(max(W, H))


	def pck_threshold_px(t: float, norm: str = "max_side") -> float:
	return t * (PCK_NORM if norm == "max_side" else DIAG)
	HM_FRACTION = 0.10 # High-Motion subset = top int(0.1*N) clips
	HM_N_TRANSITIONS = 4 # displacement averaged over the 4 future transitions


	def hm_top(n_clips: int) -> int:
	"""Size of the High-Motion subset for a split with n_clips clips (test: 60)."""
	return int(HM_FRACTION * n_clips)


	assert hm_top(EXPECTED_SPLIT_SIZES["test"]) == 60

	# Published Driver-WM Table-1 numbers on the official test split (for Gate 0 / tables).
	PAPER = {
	"zero_velocity": {"mpjpe_all": 52.89, "dn_mpjpe": 2.40, "mpjpe_hm": 139.19,
	"pck05": 85.95, "hm_threshold": 72.05},
	"static_pooling": {"mpjpe_hm": 134.56},
	"driver_wm": {"mpjpe_all": 71.47, "dn_mpjpe": 3.24, "mpjpe_hm": 138.03, "pck05": 71.66,
	"f1_dbr": 68.07, "f1_der": 72.61, "f1_tcr": 90.15, "f1_vcr": 68.34,
	"hm_h5": 155.82},
	}
	GATE0_TOL = {"mpjpe_all": 0.5, "dn_mpjpe": 0.03, "mpjpe_hm": 2.0, "hm_threshold": 2.0, "pck05": 1.0}

	# Qwen3-VL feature-extraction prompt (Driver-WM App. B.2, verbatim).
	QWEN_PROMPT = ("Extract multimodal features that capture: (A) external traffic cues (vehicles, "
	"signals, road agents), (B) how these cues act as triggers for driver motion "
	"responses, and (C) the short-term evolution (1-2 seconds) of driver behavior in "
	"response to the changing road context.")