#!/usr/bin/env python3 """ Fall Detection Inference Script INPUT CONTRACT (what goes in — only this much): pandas DataFrame with per row: x_0, y_0, conf_0, ..., x_16, y_16, conf_16 # 51 keypoint features ONLY x1, y1, x2, y2 # person bounding box (pixels) Total input columns: 55 (51 + 4 bbox). The extra 5 features are NOT input. WHAT THIS SCRIPT EXTRACTS (the extra 5 — computed here, never received): aspect_ratio, nose_relative_y, torso_angle, norm_com_y, head_hip_v_dist FLOW: upstream DataFrame (51 features + bbox) # input contract → extract 5 engineered features in run.py # done here → 56-column pandas DataFrame # 51 + 5 → XGBoost.predict_proba(feature_df) # DataFrame, never CSV Upstream (out of scope): CCTV image → YOLO26x-Pose → 17 keypoints → 51 + bbox. Usage: python scripts/run.py features.csv python scripts/run.py features.csv -o predictions.csv """ import argparse import json import math from pathlib import Path import joblib import numpy as np import pandas as pd # Configuration loaded from the project config file PROJECT_ROOT = Path(__file__).resolve().parents[1] CONFIG_PATH = PROJECT_ROOT / 'config.json' with CONFIG_PATH.open('r', encoding='utf-8') as config_file: APP_CONFIG = json.load(config_file) def resolve_project_path(config_value): path = Path(config_value) return path if path.is_absolute() else (PROJECT_ROOT / path).resolve() XGBOOST_PATH = str(resolve_project_path(APP_CONFIG['paths']['classifier_model'])) CONF_THRESH = float(APP_CONFIG['thresholds']['keypoint_confidence']) FALL_PROB_THRESH = float(APP_CONFIG['thresholds']['fall_probability']) KEYPOINT_COUNT = int(APP_CONFIG['feature_schema']['keypoint_count']) MIN_VALID_KEYPOINTS = 5 # --------------------------------------------------------------------------- # INPUT CONTRACT — only these columns are accepted from upstream (55 total) # 51 keypoint features + 4 bbox. The extra 5 are extracted below, not input. # --------------------------------------------------------------------------- KEYPOINT_COLUMNS = [ f'{name}_{i}' for i in range(KEYPOINT_COUNT) for name in ('x', 'y', 'conf') ] BBOX_COLUMNS = ['x1', 'y1', 'x2', 'y2'] INPUT_COLUMNS = KEYPOINT_COLUMNS + BBOX_COLUMNS # 51 + 4 = 55 input columns INPUT_FEATURE_COUNT = len(KEYPOINT_COLUMNS) # 51 — the input contract features # --------------------------------------------------------------------------- # EXTRACTED HERE (not input) — the extra 5, computed by this script # --------------------------------------------------------------------------- ENGINEERED_FEATURE_NAMES = [ 'aspect_ratio', 'nose_relative_y', 'torso_angle', 'norm_com_y', 'head_hip_v_dist', ] ENGINEERED_COUNT = len(ENGINEERED_FEATURE_NAMES) # 5 # Exact column order fed to XGBoost: 51 input + 5 extracted = 56 FEATURE_COLS = KEYPOINT_COLUMNS + ENGINEERED_FEATURE_NAMES MODEL_FEATURE_COUNT = len(FEATURE_COLS) # 56 # Backwards-compatible alias UPSTREAM_COLUMNS = INPUT_COLUMNS def load_model(): """Load the trained XGBoost classifier.""" print(f"Loading XGBoost model: {XGBOOST_PATH}") model = joblib.load(XGBOOST_PATH) print("Model loaded successfully!\n") return model def midpoint_norm(xs, ys, i, j): """NaN-aware midpoint of keypoints i and j in normalized space.""" points = [] for idx in (i, j): if not (np.isnan(xs[idx]) or np.isnan(ys[idx])): points.append((xs[idx], ys[idx])) if not points: return np.nan, np.nan return ( float(np.mean([p[0] for p in points])), float(np.mean([p[1] for p in points])), ) def extract_engineered_features(kp51, x1, y1, x2, y2, conf_threshold=CONF_THRESH): """Extract the extra 5 features from the input contract (51 keypoints + bbox). These 5 are NOT part of the input — they are computed here: aspect_ratio, nose_relative_y, torso_angle, norm_com_y, head_hip_v_dist Returns them in classifier column order (positions 52-56). """ w, h = x2 - x1, y2 - y1 kpts = np.asarray(kp51, dtype=float).reshape(KEYPOINT_COUNT, 3) xs = kpts[:, 0].copy() ys = kpts[:, 1].copy() confs = kpts[:, 2] # Inference-side occlusion mask: low confidence -> NaN coordinates low_conf = confs < conf_threshold xs[low_conf] = np.nan ys[low_conf] = np.nan # COCO references: 0=nose, 5/6=shoulders, 11/12=hips sh_mid = midpoint_norm(xs, ys, 5, 6) hip_mid = midpoint_norm(xs, ys, 11, 12) # 1. aspect_ratio = w / h aspect_ratio = w / h if h > 0 else 0.0 # 2. nose_relative_y = (Y_nose - y1) / h == normalized nose y nose_relative_y = ys[0] # 3. torso_angle = angle(HipMid -> ShoulderMid) vs vertical Y-axis (degrees) if confs[5] >= conf_threshold and confs[11] >= conf_threshold: if not (np.isnan(sh_mid[0]) or np.isnan(hip_mid[0])): dx = (sh_mid[0] - hip_mid[0]) * w dy = (sh_mid[1] - hip_mid[1]) * h torso_angle = math.degrees(math.atan2(abs(dx), abs(dy) + 1e-6)) else: torso_angle = np.nan else: torso_angle = np.nan # 4. norm_com_y = (Sum(Y_i * conf_i) / Sum(conf_i) - y1) / h # == confidence-weighted mean of normalized y over visible keypoints visible = (confs >= conf_threshold) & ~np.isnan(ys) if np.any(visible) and np.sum(confs[visible]) > 0: norm_com_y = float( np.sum(ys[visible] * confs[visible]) / np.sum(confs[visible]) ) else: norm_com_y = np.nan # 5. head_hip_v_dist = (Y_hip_mid - Y_nose) / h == hip_mid_y_norm - nose_y_norm if confs[0] >= conf_threshold and ( confs[11] >= conf_threshold or confs[12] >= conf_threshold ): if not (np.isnan(hip_mid[1]) or np.isnan(ys[0])): head_hip_v_dist = hip_mid[1] - ys[0] else: head_hip_v_dist = np.nan else: head_hip_v_dist = np.nan return [ aspect_ratio, nose_relative_y, torso_angle, norm_com_y, head_hip_v_dist, ] def build_feature_frame(input_df): """Apply the input contract, then extract the extra 5 → 56-column DataFrame. INPUT CONTRACT (only what goes in — 55 columns): 51 keypoint features (x_0,y_0,conf_0,...,x_16,y_16,conf_16) + bbox (x1,y1,x2,y2) The 5 engineered features are NOT accepted as input. EXTRACTED HERE (never received): aspect_ratio, nose_relative_y, torso_angle, norm_com_y, head_hip_v_dist Returns: feature_df: DataFrame (n × 56) = 51 input + 5 extracted → XGBoost skip_rows: input row indices with < MIN_VALID_KEYPOINTS """ missing = [c for c in INPUT_COLUMNS if c not in input_df.columns] if missing: raise ValueError( f"Input contract violated — missing {len(missing)} required column(s): " f"{', '.join(missing[:10])}" f"{'...' if len(missing) > 10 else ''}. " f"Input contract = {INPUT_FEATURE_COUNT} keypoint features " "(x_0,y_0,conf_0,...,x_16,y_16,conf_16) + bbox (x1,y1,x2,y2) only. " "The extra 5 features are extracted by this script, not required as input." ) # Only input-contract columns are read; engineered cols (if present) are ignored kp_values = input_df[KEYPOINT_COLUMNS].to_numpy(dtype=float) bbox = input_df[BBOX_COLUMNS].to_numpy(dtype=float) conf_values = kp_values[:, 2::3] valid_counts = np.sum(np.nan_to_num(conf_values, nan=0.0) > 0, axis=1) rows = [] skip_rows = [] for row_idx, (kp51, box) in enumerate(zip(kp_values, bbox)): if valid_counts[row_idx] < MIN_VALID_KEYPOINTS: skip_rows.append(row_idx) rows.append([np.nan] * MODEL_FEATURE_COUNT) continue x1, y1, x2, y2 = box # Extract the extra 5 here (input stays 51 + bbox) engineered = extract_engineered_features(kp51, x1, y1, x2, y2) rows.append(list(kp51) + engineered) # 51 + 5 = 56 feature_df = pd.DataFrame(rows, columns=FEATURE_COLS, index=input_df.index) return feature_df, skip_rows def predict(model, feature_df, skip_rows=None): """Run XGBoost on the 56-column DataFrame (51 input + 5 extracted here). XGBoost receives a pandas DataFrame (n rows × 56 columns), not a CSV. Returns a result DataFrame with prediction + fall_probability columns. """ result = pd.DataFrame(index=feature_df.index) result['prediction'] = pd.NA result['fall_probability'] = np.nan skip = set(skip_rows or []) scorable_idx = [i for i in feature_df.index if i not in skip] if not scorable_idx: return result # In-memory 56-col DataFrame → XGBoost (never CSV at inference) X = feature_df.loc[scorable_idx, FEATURE_COLS] proba = model.predict_proba(X)[:, 1] predictions = np.where(proba >= FALL_PROB_THRESH, 'Fall', 'Normal') result.loc[scorable_idx, 'fall_probability'] = proba result.loc[scorable_idx, 'prediction'] = predictions return result def main(): parser = argparse.ArgumentParser( description=( 'Fall Detection Inference — input contract: 51 keypoint features + bbox only. ' 'run.py extracts the extra 5 features, builds a 56-column DataFrame, ' 'and passes that DataFrame to XGBoost.' ) ) parser.add_argument( 'input_csv', help=( 'Upstream CSV matching the input contract: ' 'x_0..conf_16 (51) + x1,y1,x2,y2 — the 5 engineered features are NOT required' ), ) parser.add_argument( '--output', '-o', default=None, help='Path to save predictions CSV (default: print only)', ) args = parser.parse_args() input_path = Path(args.input_csv) if not input_path.exists(): print(f"Error: Input CSV not found: {input_path}") return # CSV is only the upstream transport; everything below is in-memory DataFrames input_df = pd.read_csv(input_path) if input_df.empty: print("Input CSV has no rows.") return print(f"Input (contract: {INPUT_FEATURE_COUNT} features + bbox): " f"{input_path} ({len(input_df)} rows, {len(input_df.columns)} columns)") model = load_model() # INPUT: 51 + bbox → EXTRACT 5 here → 56-column DataFrame feature_df, skip_rows = build_feature_frame(input_df) print(f"Extracted {ENGINEERED_COUNT} features in run.py: {ENGINEERED_FEATURE_NAMES}") print(f"Model input DataFrame: {feature_df.shape[0]} rows × " f"{feature_df.shape[1]} columns " f"({INPUT_FEATURE_COUNT} input + {ENGINEERED_COUNT} extracted)") # 56-column DataFrame → XGBoost pred_df = predict(model, feature_df, skip_rows) result = pd.concat([input_df, pred_df], axis=1) for row_idx in skip_rows: print(f" Row {row_idx + 1}: Skipped (insufficient keypoints)") for row_idx in result.index: if row_idx in set(skip_rows): continue print( f" Row {row_idx + 1}: {result.at[row_idx, 'prediction']} " f"(fall_prob={result.at[row_idx, 'fall_probability']:.3f})" ) fall_count = int((result['prediction'] == 'Fall').sum()) normal_count = int((result['prediction'] == 'Normal').sum()) print(f"\nTotal rows: {len(input_df)}") print(f" - Skipped: {len(skip_rows)}") print(f" - Fall: {fall_count}") print(f" - Normal: {normal_count}") scorable = result['fall_probability'].notna() if scorable.any(): max_fall_prob = float(result.loc[scorable, 'fall_probability'].max()) print("\n" + "=" * 50) if fall_count > 0: print("RESULT: FALL DETECTED") print(f"Confidence: {max_fall_prob:.2%}") else: print("RESULT: NO FALL DETECTED") print(f"Max fall probability: {max_fall_prob:.2%}") print("=" * 50) else: print("\nNo scorable rows (all skipped).") if args.output: output_path = Path(args.output) output_path.parent.mkdir(parents=True, exist_ok=True) result.to_csv(output_path, index=False) print(f"\nPredictions saved to: {output_path}") if __name__ == "__main__": main()