File size: 8,480 Bytes
e2d95ed | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 | """
Temporal sequence construction and augmentation.
Everything that changes the time axis lives here so that training, evaluation and
the robustness experiments apply *exactly the same* operations.
Public API
----------
resample_sequence(seq, length, method) fixed-length resampling (no leakage)
pad_or_crop(seq, length, mode) deterministic padding / centre crop
make_windows(seq, length, stride) sliding windows for long clips
temporal_crop(seq, frac, rng) augmentation: random sub-clip
frame_drop(seq, prob, rng) augmentation *and* robustness op
landmark_dropout(seq, prob, rng) augmentation *and* robustness op
gaussian_jitter(seq, sigma, rng) augmentation *and* robustness op
frame_skip(seq, keep_ratio) deterministic frame dropping (robustness)
"""
from __future__ import annotations
from typing import List, Tuple
import numpy as np
import config
COORDS_PER_FRAME = config.NUM_LANDMARKS * config.COORDS_PER_LANDMARK
def _as_2d(seq: np.ndarray) -> Tuple[np.ndarray, int, int]:
"""Return (seq_2d (T, D), T, D). Accepts (T, 75, 3) or (T, D).
A batched array ``(N, T, D)`` is rejected loudly: these operators work on one
clip at a time, and silently reshaping a batch would corrupt the time axis.
"""
seq = np.asarray(seq, dtype=np.float32)
if seq.ndim == 1:
seq = seq[None, :]
elif seq.ndim == 3:
if seq.shape[1] == config.NUM_LANDMARKS and seq.shape[2] == config.COORDS_PER_LANDMARK:
seq = seq.reshape(seq.shape[0], -1) # (T, 75, 3) -> (T, 225)
else:
raise ValueError(
f"sequence operators take a single clip shaped (T, {COORDS_PER_FRAME}) or "
f"(T, {config.NUM_LANDMARKS}, {config.COORDS_PER_LANDMARK}); got {seq.shape}. "
"Loop over the batch instead."
)
elif seq.ndim > 3:
raise ValueError(f"unsupported sequence shape {seq.shape}")
if seq.ndim != 2:
raise ValueError(f"unsupported sequence shape {seq.shape}")
return seq, seq.shape[0], seq.shape[1]
def resample_sequence(seq: np.ndarray, length: int | None = None, method: str | None = None) -> np.ndarray:
"""Resample a clip to exactly ``length`` frames.
``linear`` : linear interpolation over normalised time (default, preserves
gesture dynamics; used for both training and evaluation).
``pad`` : deterministic edge padding / centre crop.
``uniform`` : pick ``length`` evenly spaced existing frames (no interpolation).
"""
length = int(length or config.SEQUENCE_LENGTH)
method = (method or config.SEQ_RESAMPLE_METHOD).lower()
seq, t, _ = _as_2d(seq)
if t == 0:
return np.zeros((length, seq.shape[1]), dtype=np.float32)
if t == length:
return seq.astype(np.float32)
src = np.linspace(0.0, 1.0, num=t, dtype=np.float64)
dst = np.linspace(0.0, 1.0, num=length, dtype=np.float64)
if method == "pad":
return pad_or_crop(seq, length, mode=config.SEQ_PAD_MODE)
if method == "uniform":
idx = np.unique(np.round(dst * (t - 1)).astype(int))
if idx.size < length: # duplicate edges when the clip is very short
idx = np.concatenate([idx, np.repeat(idx[-1], length - idx.size)])
return seq[idx].astype(np.float32)
out = np.empty((length, seq.shape[1]), dtype=np.float32)
for c in range(seq.shape[1]):
out[:, c] = np.interp(dst, src, seq[:, c].astype(np.float64)).astype(np.float32)
return out
def pad_or_crop(seq: np.ndarray, length: int | None = None, mode: str = "edge") -> np.ndarray:
"""Deterministic fixed-length conversion: edge/zero pad, or centre crop."""
length = int(length or config.SEQUENCE_LENGTH)
seq, t, d = _as_2d(seq)
if t == length:
return seq.astype(np.float32)
if t > length:
start = (t - length) // 2
return seq[start: start + length].astype(np.float32)
pad = length - t
before, after = pad // 2, pad - pad // 2
if mode == "zero":
filler = np.zeros((1, d), dtype=np.float32)
else: # 'edge'
filler = seq[-1:]
return np.concatenate([filler.repeat(before, axis=0), seq,
filler.repeat(after, axis=0)]).astype(np.float32)
def make_windows(seq: np.ndarray, length: int | None = None, stride: int | None = None) -> List[np.ndarray]:
"""Sliding windows (used by the real-time buffer and long-video batch mode)."""
length = int(length or config.SEQUENCE_LENGTH)
stride = int(stride or length)
seq, t, _ = _as_2d(seq)
if t < length:
return [resample_sequence(seq, length)]
return [seq[i: i + length] for i in range(0, t - length + 1, stride)]
# --------------------------------------------------------------------------
# Augmentations (training) / perturbations (robustness) - same implementations
# --------------------------------------------------------------------------
def temporal_crop(seq: np.ndarray, frac: float, rng: np.random.Generator) -> np.ndarray:
"""Keep a random sub-window (fraction of the clip), then the caller resamples."""
seq, t, _ = _as_2d(seq)
if frac <= 0 or t <= 2:
return seq
keep = max(2, int(round(t * (1.0 - frac))))
start = int(rng.integers(0, t - keep + 1))
return seq[start: start + keep]
def frame_drop(seq: np.ndarray, prob: float, rng: np.random.Generator) -> np.ndarray:
"""Randomly repeat neighbouring frames (simulates dropped/duplicated frames).
The clip keeps its length, so downstream shapes never change.
"""
seq, t, d = _as_2d(seq)
if prob <= 0 or t < 3:
return seq
mask = rng.random(t) < prob
mask[0] = False
out = seq.copy()
idx = np.nonzero(mask)[0]
out[idx] = seq[idx - 1]
return out
def frame_skip(seq: np.ndarray, keep_ratio: float) -> np.ndarray:
"""Deterministic frame dropping: keep every k-th frame (robustness protocol)."""
seq, t, _ = _as_2d(seq)
if keep_ratio >= 1.0 or t < 3:
return seq
step = max(2, int(round(1.0 / max(keep_ratio, 1e-6))))
kept = seq[::step]
if kept.shape[0] < 2:
kept = seq[[0, -1]]
return kept
def landmark_dropout(seq: np.ndarray, prob: float, rng: np.random.Generator,
groups: Tuple[str, ...] = ("left_hand", "right_hand", "pose")) -> np.ndarray:
"""Zero-out whole landmark groups (hand occlusion) with probability ``prob``."""
seq, t, d = _as_2d(seq)
if prob <= 0:
return seq
seq = seq.copy()
for i in range(t):
for g in groups:
if rng.random() < prob:
sl = config.LAYOUT.slice(g)
lo, hi = sl.start * config.COORDS_PER_LANDMARK, sl.stop * config.COORDS_PER_LANDMARK
if hi <= d:
seq[i, lo:hi] = 0.0
return seq
def gaussian_jitter(seq: np.ndarray, sigma: float, rng: np.random.Generator) -> np.ndarray:
"""Additive Gaussian noise on coordinates (mask channels are left untouched)."""
seq, t, d = _as_2d(seq)
if sigma <= 0:
return seq
noise = rng.normal(0.0, sigma, size=(t, config.FEATURE_DIM)).astype(np.float32)
out = seq.copy()
out[:, : config.FEATURE_DIM] += noise
return out
def augment(seq: np.ndarray, rng: np.random.Generator) -> np.ndarray:
"""Full training augmentation policy (identical for LSTM / GRU / GRU+MHA)."""
cfg = config.AUGMENTATION
if not cfg.get("enabled", True):
return seq
out = _as_2d(seq)[0]
if cfg.get("temporal_crop_frac", 0) > 0:
out = temporal_crop(out, float(cfg["temporal_crop_frac"]), rng)
out = resample_sequence(out, config.SEQUENCE_LENGTH)
if cfg.get("frame_drop_prob", 0) > 0:
out = frame_drop(out, float(cfg["frame_drop_prob"]), rng)
if cfg.get("landmark_dropout_prob", 0) > 0:
out = landmark_dropout(out, float(cfg["landmark_dropout_prob"]), rng)
if cfg.get("jitter_sigma", 0) > 0:
out = gaussian_jitter(out, float(cfg["jitter_sigma"]), rng)
if cfg.get("mirror_prob", 0) > 0 and rng.random() < float(cfg["mirror_prob"]):
from .landmarks import mirror_hands
out = out.reshape(out.shape[0], config.NUM_LANDMARKS, config.COORDS_PER_LANDMARK)
out = np.stack([mirror_hands(f) for f in out]).reshape(out.shape[0], -1)
return resample_sequence(out, config.SEQUENCE_LENGTH)
|