# Project export: Vigil This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small. ## Project metadata - Hackathon: TreeHacks 2026 - Tagline: End-to-end kidnapping detection and response, all on your Apple Watch - Devpost: https://devpost.com/software/wristguard - GitHub: https://github.com/anirudhmazumder/TreeHacks-2026 - Video: https://www.youtube.com/embed/1i0KplsVs8Y?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 4 GitHub contributor(s) — Naman Jain (11 commits), Cursor (10 commits), Anirudh Mazumder (6 commits), VXXWu (1 commits) ## Devpost submission (written by the team) ### Inspiration In a kidnapping or abduction scenario, the victim often can't reach for their phone, open an app, or press an SOS button. What if your body could call for help on its own? The Apple Watch is already strapped to over 100 million wrists, tracking heart rate, acceleration, rotation and various other metrics 24/7. We built Vigil to turn that passive data into an active lifeline by detecting panic situations autonomously, recording evidence, and immediately contacting emergency response with location and intelligent incident summaries, all without requiring a single tap. ### What it does Vigil runs silently on your Apple Watch, continuously monitoring your heart rate, accelerometer motion, and GPS data. It automatically: Detects a panic event using two independent on-device ML models (both must agree to minimize false alarms). Record 10 seconds of audio from the Watch microphone plus sensor and GPS data as evidence. Uploads the package to a cloud model for heavyweight threat confirmation using audio and GPS data analysis. Dispatches an emergency phone call and email in real time containing your GPS coordinates, an audio recording of the event, and an intelligent incident summary if the threat is confirmed. No buttons. No phone. Just your Watch. ### How we built it Sensors Vigil collects data from CoreMotion (accelerometer + gyroscope at 10Hz), HealthKit (heart rate and heart rate variation), and CoreLocation (GPS) every second. On-device detection We built a two-stage biometric analysis model that runs entirely on the Apple Watch using Swift: SVM: Anomaly detection, trained using unsupervised learning on a dataset of only normal activity that we collected; detects abnormal biometrics. Probabilistic model: Mathematical model of threat probability based on heart rate and movement data. Panic triggers only when both models exceed their thresholds, dramatically reducing false positives. If both models exceed the decision threshold, the watch begins recording 10s of audio through its microphone and sending the sensor stream to the cloud. Cloud inference Our sensor data and the audio recording is converted to base64 and uploaded via a HTTP request to a Modal serverless endpoint. OpenAI’s gpt-4o-audio-preview analyzes the audio and location context to confirm whether the threat is real. We then use GeoPy to run GPS movement analysis, detecting addresses, distance, and speed. Emergency dispatch call On confirmed threat, Vigil uses Twilio for an automated voice call to emergency services containing an AI voiceover with live GPS coordinates and recorded audio of the event. It also uses SMTP for an emergency email containing a continual database of streamed GPS coordinates and an intelligent incident summary. ### Challenges we ran into Apple Watch edge integration Apple products are notoriously difficult to work with, and things are only harder on the Watch. We had to build in Xcode and Swift; with only two Apple Watches, we had to test a lot of code on the watchOS simulator, which often didn’t translate perfectly to physical deployment. Apple Watch edge integration Apple products are notoriously difficult to work with, and things are only harder on the Watch. We had to build in Xcode and Swift; with only two Apple Watches, we had to test a lot of code on the watchOS simulator, which often didn’t translate perfectly to physical deployment. Local inference We had to design the local infrastructure (initial decision threshold models and data sent to cloud stack) to be sufficiently lightweight and efficient to run seamlessly and continuously on Apple Watch hardware. We constantly run lightweight detection locally on the Apple Watch as a gating mechanism for the powerful cloud models, reducing energy and inference costs. Local inference We had to design the local infrastructure (initial decision threshold models and data sent to cloud stack) to be sufficiently lightweight and efficient to run seamlessly and continuously on Apple Watch hardware. We constantly run lightweight detection locally on the Apple Watch as a gating mechanism for the powerful cloud models, reducing energy and inference costs. Data collection We obviously didn’t have any positive samples of sensor readings from kidnappings, so we had to innovate with model design. We collected over 1,000 negative readings (normal activity), and used it for unsupervised training. We experimented with different architectures like an LSTM-autoencoder, but ultimately settled on using an SVM for anomaly detection, which worked well and provided a lightweight method easily deployable on the Apple Watch. Data collection We obviously didn’t have any positive samples of sensor readings from kidnappings, so we had to innovate with model design. We collected over 1,000 negative readings (normal activity), and used it for unsupervised training. We experimented with different architectures like an LSTM-autoencoder, but ultimately settled on using an SVM for anomaly detection, which worked well and provided a lightweight method easily deployable on the Apple Watch. ### Accomplishments we're proud of Single-device stack: Vigil runs entirely through the Apple Watch, without needing to connect to your iPhone or Mac, ensuring it stays active regardless of where you go. We maximize the functionality of the Apple Watch to get tons of data from a lightweight edge device. Full emergency response pipeline: Vigil doesn’t just detect kidnapping: In under 1min, it records audio, processes GPS information, and automates a call to emergency services with the relevant info for authorities to take immediate action. Robust classification We tackle a challenge that lacks training datasets of positive examples and that occurs in a highly variable environment. We built a pipeline of three different threat classification models: the two local biometric models as gatekeepers and the powerful cloud model for detailed contextual analysis. This multi-layer system maximizes recall, which is crucial in this threat detection scenario, while also prioritizing precision so as to minimize both cloud model calls and false kidnapping alerts. ### What we learned We learned that systems are a lot more functional and robust in simulation than on edge hardware. Sometimes, models or features would work perfectly on the computer, but just break when deployed to the Watch. Building on an Apple Watch forced us to build systems that are robust to variable data and lightweight. Furthermore, through this project, we deeply understood how data really moves throughout the internet. We had to figure out how to convert an audio file into a large base64 string, alongside biometric data, push to our modal endpoint, and properly decode it back on the python server. It helped us understand how to bridge the gap between edge devices and the cloud. ### What's next Broader threat detection: Extend the models to detect a broader range of threat events, like collisions and physical assault patterns iOS companion app: A paired iPhone app for event history, contact management, and connectivity backup Emergency contact network: Let users configure trusted contacts who receive real-time GPS tracking and notifications of threat events ## README (from the GitHub repository) # Vigil **Autonomous safety detection for Apple Watch.** Vigil runs silently on your Apple Watch, continuously monitoring your heart rate, accelerometer motion, and GPS data. It automatically: 1. **Detects a panic event** using two independent on-device ML models (both must agree to minimize false alarms). 2. **Records 10 seconds of audio** from the Watch microphone plus sensor and GPS data as evidence. 3. **Uploads the package to a cloud model** for heavyweight threat confirmation using audio and GPS data analysis. 4. **Dispatches an emergency phone call and email in real time** containing your GPS coordinates, an audio recording of the event, and an intelligent incident summary if the threat is confirmed. > **No buttons. No phone. Just your Watch.** --- ## Team Built at **TreeHacks 2026** by Naman, Vince, Anirudh, Rachel — Stanford University. ## Detected evidence (automated analysis) Indexed codebase: 21 recognized source files, 125 KB. - OpenAI (technology) — detected in the code - Python (language) — detected in the code - Swift (language) — detected in the code - PyTorch (technology) — claimed on Devpost, not found in the code - AI coding agent: Cursor — evidence: commit authorship or trailers ## Codebase structure (from repository index) ### Files (31 of 31) ``` .DS_Store .gitignore action_hub/action_hub.py action_hub/ai_service.py action_hub/email_service.py action_hub/location_agent_enhanced.py action_hub/twilio_service.py panic_detector.py panic_scores.csv README.md requirements.txt safety_detection_model_development/svm_params.json safety_detection_model_development/train_svm.py wearable_safety_detection/wearable_safety_detection Watch App/AirtableUploader.swift wearable_safety_detection/wearable_safety_detection Watch App/AnomalyDetector.swift wearable_safety_detection/wearable_safety_detection Watch App/Assets.xcassets/AccentColor.colorset/Contents.json wearable_safety_detection/wearable_safety_detection Watch App/Assets.xcassets/AppIcon.appiconset/Contents.json wearable_safety_detection/wearable_safety_detection Watch App/Assets.xcassets/Contents.json wearable_safety_detection/wearable_safety_detection Watch App/AudioRecorder.swift wearable_safety_detection/wearable_safety_detection Watch App/ContentView.swift wearable_safety_detection/wearable_safety_detection Watch App/ContentViewModal.swift wearable_safety_detection/wearable_safety_detection Watch App/DataCollector.swift wearable_safety_detection/wearable_safety_detection Watch App/DataCollectorModal.swift wearable_safety_detection/wearable_safety_detection Watch App/DetectionPayload.swift wearable_safety_detection/wearable_safety_detection Watch App/MacServerCommunicator.swift wearable_safety_detection/wearable_safety_detection Watch App/ModalServerCommunicator.swift wearable_safety_detection/wearable_safety_detection Watch App/PanicDetector.swift wearable_safety_detection/wearable_safety_detection Watch App/Preview Content/Preview Assets.xcassets/Contents.json wearable_safety_detection/wearable_safety_detection Watch App/svm_params.json wearable_safety_detection/wearable_safety_detection Watch App/ThreatDataModel.swift wearable_safety_detection/wearable_safety_detection Watch App/wearable_safety_detectionApp.swift ``` ### Dependencies - requirements.txt: geopy, matplotlib, modal, numpy, openai, pandas, pydub, requests, twilio ### Recent commits (newest first) - added final functionality - deploying the trained svm on the model - single figure - Update README - Add README - model figures - Handle corrupt m4a files — ffmpeg fallback + OpenAI format fix - Simplify emergency email — compact layout, start+end only - Add interactive Twilio call menu with audio playback - Remove CSV test data from tracking - Add GPS location analysis and enrich alerts with address data - Delete wearable_safety_detection/.DS_Store - Added microphone listening capability - Add .DS_Store to gitignore - Migrate Action Hub to Modal serverless with GPT-4o audio analysis - Delete safety_detection_wearables directory - Adding data collection from the apple watch - Consolidate action_hub deps into root requirements.txt - Add Action Hub — emergency call + email dispatch service - panic detection model ## Key source files (fetched from GitHub, selected and truncated for size) ### requirements.txt ``` pandas numpy matplotlib twilio requests openai pydub modal geopy ``` ### panic_detector.py ```python """ Panic Detection Model — Python Prototype Processes Apple Watch vitals CSV and outputs panic scores + visualization. Tune weights/thresholds here, then port finalized values to Swift watchOS app. """ import numpy as np import pandas as pd import matplotlib.pyplot as plt # ── Configuration ──────────────────────────────────────────────────────────── BASELINE_WINDOW = 30 # seconds for rolling baseline EMA SMOOTH_WINDOW = 10 # seconds for panic score smoothing SIGNALS = { "heart_rate": 0.40, "accel_variance": 0.25, "rotation_magnitude": 0.20, "accel_magnitude": 0.15, } THRESHOLD_ELEVATED = 0.5 THRESHOLD_PANIC = 0.75 TRIGGER_PROBABILITY = 0.60 # must match PanicDetector.swift triggerProbability def load_and_preprocess(csv_path: str) -> pd.DataFrame: """Load CSV, drop sensor warmup rows, normalize timestamps.""" df = pd.read_csv(csv_path) # Drop the unnamed empty column if present df = df.loc[:, ~df.columns.str.startswith("Unnamed")] # Drop warmup rows where heart_rate is 0 df = df[df["heart_rate"] > 0].copy() df.reset_index(drop=True, inplace=True) # Normalize timestamp to relative seconds from start df["time_s"] = df["timestamp"] - df["timestamp"].iloc[0] return df def compute_z_scores(df: pd.DataFrame) -> pd.DataFrame: """Compute rolling z-scores for each signal.""" for signal in SIGNALS: rolling_mean = df[signal].ewm(span=BASELINE_WINDOW, adjust=False).mean() rolling_std = df[signal].ewm(span=BASELINE_WINDOW, adjust=False).std() # Avoid division by zero rolling_std = rolling_std.replace(0, np.nan).ffill().fillna(1.0) z = (df[signal] - rolling_mean) / rolling_std # Clamp negative z-scores to 0 — only elevated signals matter df[f"z_{signal}"] = z.clip(lower=0) return df def compute_panic_score(df: pd.DataFrame) -> pd.DataFrame: """Weighted composite panic score with smoothing.""" raw_score = sum( weight * df[f"z_{signal}"] for signal, weight in SIGNALS.items() ) # Smooth over SMOOTH_WINDOW seconds df["panic_score_raw"] = raw_score df["panic_score"] = raw_score.rolling(window=SMOOTH_WINDOW, min_periods=1).mean() # Sigmoid transform to probability (0–1), centered at THRESHOLD_PANIC SIGMOID_SLOPE = 3.0 # must match PanicDetector.swift sigmoidSlope df["panic_prob"] = 1 / (1 + np.exp(-SIGMOID_SLOPE * (df["panic_score"] - THRESHOLD_PANIC))) # Classification df["status"] = "normal" df.loc[df["panic_score"] >= THRESHOLD_ELEVATED, "status"] = "elevated" df.loc[df["panic_score"] >= THRESHOLD_PANIC, "status"] = "panic" df["is_panic"] = df["panic_prob"] >= TRIGGER_PROBABILITY return df def find_panic_windows(df: pd.DataFrame) -> list[dict]: """Find contiguous panic windows with start/end times.""" windows = [] in_panic = False start = None for _, row in df.iterrows(): if row["is_panic"] and not in_panic: in_panic = True start = row["time_s"] elif not row["is_panic"] and in_panic: in_panic = False windows.append({"start_s": start, "end_s": row["time_s"]}) # Close open window at end if in_panic: windows.append({"start_s": start, "end_s": df["time_s"].iloc[-1]}) return windows def plot_results(df: pd.DataFrame, panic_windows: list[dict], save_path: str): """Multi-panel visualization of signals, z-scores, and panic score.""" fig, axes = plt.subplots(4, 1, figsize=(14, 10), sharex=True) t = df["time_s"] # Panel 1: Raw signals ax = axes[0] ax.plot(t, df["heart_rate"], label="Heart Rate (BPM)", color="red", alpha=0.8) ax.set_ylabel("Heart Rate") ax.legend(loc="upper right") ax.set_title("Panic Detection Model — Watch Vitals Analysis") ax2 = ax.twinx() ax2.plot(t, df["accel_variance"], label="Accel Variance", color="blue", alpha=0.5) ax2.plot(t, df["rotation_magnitude"], label="Rotation Mag", color="green", alpha=0.5) ax2.set_ylabel("Motion") ax2.legend(loc="upper left") # Panel 2: Z-scores ax = axes[1] for signal in SIGNALS: ax.plot(t, df[f"z_{signal}"], label=f"z_{signal}", alpha=0.7) ax.set_ylabel("Z-Score (clamped ≥0)") ax.legend(loc="upper right", fontsize=8) # Panel 3: Panic score ax = axes[2] ax.plot(t, df["panic_score"], label="Panic Score (smoothed)", color="purple", linewidth=2) ax.axhline(y=THRESHOLD_ELEVATED, color="orange", linestyle="--", label=f"Elevated ({THRESHOLD_ELEVATED})") ax.axhline(y=THRESHOLD_PANIC, color="red", linestyle="--", label=f"Panic ({THRESHOLD_PANIC})") ax.set_ylabel("Panic Score") ax.legend(loc="upper right") # Shade panic windows for w in panic_windows: ax.axvspan(w["start_s"], w["end_s"], alpha=0.3, color="red") # Panel 4: Panic probability ax = axes[3] ax.fill_between(t, df["panic_prob"], alpha=0.4, color="red", label="Panic Probability") ax.axhline(y=0.5, color="gray", linestyle=":", alpha=0.5) ax.set_ylabel("P(panic)") ax.set_xlabel("Time (seconds)") ax.set_ylim(0, 1) ax.legend(loc="upper right") # Shade panic windows on probability panel too for w in panic_windows: ax.axvspan(w["start_s"], w["end_s"], alpha=0.2, color="red") plt.tight_layout() plt.savefig(save_path, dpi=150) plt.close() print(f"Plot saved to: {save_path}") def main(): csv_path = "real_safety_data.csv" plot_path = "panic_detection_results.png" print("Loading data...") df = load_and_preprocess(csv_path) print(f" {len(df)} samples after dropping warmup rows") print(f" Time span: {df['time_s'].iloc[-1]:.1f} seconds") print("\nComputing z-scores...") df = compute_z_scores(df) print("Computing panic scores...") df = compute_panic_score(df) # Summary stats print(f"\n── Results ───────────────────────────────── [truncated — 1200 more characters] ``` ### action_hub/twilio_service.py ```python """Twilio voice-call service — interactive emergency call (Modal edition).""" import os import uuid import modal from twilio.rest import Client from twilio.twiml.voice_response import Gather, VoiceResponse # Shared Dict to pass call data (context + audio) between endpoints call_store = modal.Dict.from_name("call-data-store", create_if_missing=True) def _get_env(name: str) -> str | None: """Read an env var at call time (not import time) for Modal secrets.""" return os.environ.get(name) def _validate_env_vars() -> tuple[str | None, dict[str, str]]: """Return (error_msg, env_dict). Error is None when all vars are present.""" keys = [ "TWILIO_ACCOUNT_SID", "TWILIO_AUTH_TOKEN", "TWILIO_PHONE_NUMBER", "TARGET_PHONE_NUMBER", ] env: dict[str, str] = {} missing: list[str] = [] for k in keys: val = _get_env(k) if not val: missing.append(k) else: env[k] = val if missing: return f"Missing Twilio env vars: {', '.join(missing)}", {} return None, env def _get_base_url() -> str: """Build the base Modal endpoint URL from the app/workspace name.""" workspace = os.environ.get("MODAL_WORKSPACE", "namanj") return f"https://{workspace}--emergency-action-hub" def trigger_twilio_call(context: str, audio_path: str | None = None) -> dict: """Initiate an interactive emergency call. Stores call data in modal.Dict so callback endpoints can access it. Returns a dict with ``status`` and either ``call_sid`` or ``error``. """ err, env = _validate_env_vars() if err: print(f"[twilio_service] {err}") return {"status": "error", "error": err} # Store call data for the callback endpoints call_key: str = str(uuid.uuid4()) audio_bytes: bytes | None = None if audio_path and os.path.exists(audio_path): with open(audio_path, "rb") as f: audio_bytes = f.read() call_store[call_key] = { "context": context, "audio_bytes": audio_bytes, } base_url = _get_base_url() menu_url = f"{base_url}-voice-menu.modal.run?key={call_key}" try: client = Client(env["TWILIO_ACCOUNT_SID"], env["TWILIO_AUTH_TOKEN"]) call = client.calls.create( to=env["TARGET_PHONE_NUMBER"], from_=env["TWILIO_PHONE_NUMBER"], url=menu_url, method="GET", ) print(f"[twilio_service] Call initiated — SID: {call.sid}, key: {call_key}") return {"status": "initiated", "call_sid": call.sid} except Exception as exc: print(f"[twilio_service] Call failed: {exc}") return {"status": "error", "error": str(exc)} ``` ### action_hub/ai_service.py ```python """AI service — analyzes raw audio for emergency/kidnapping indicators.""" import base64 import json import os from pathlib import Path from openai import OpenAI SYSTEM_PROMPT: str = ( "You are an emergency audio analyst. You are listening directly to raw audio — " "NOT a transcription. Analyze non-verbal cues: screams, cries for help, sounds of " "a physical struggle, heavy or panicked breathing, aggressive demands, muffled voices, " "or any indicators of a kidnapping or physical emergency.\n\n" "You MUST respond with ONLY a JSON object in this exact schema:\n" '{"is_kidnapping": bool, "confidence": int (0-100), "context": "2 sentence summary of sounds and voices"}\n\n' "Do not include any text outside the JSON object." ) # OpenAI audio API only supports wav and mp3 FORMAT_MAP: dict[str, str] = { ".wav": "wav", ".mp3": "mp3", } def _read_and_encode(file_path: str) -> tuple[str, str]: """Read an audio file and return (base64_data, format_string).""" path = Path(file_path) suffix: str = path.suffix.lower() audio_format: str = FORMAT_MAP.get(suffix, "wav") with open(path, "rb") as f: raw_bytes = f.read() encoded: str = base64.b64encode(raw_bytes).decode("utf-8") return encoded, audio_format def _parse_response(raw_text: str) -> dict: """Parse the model's JSON response, with a fallback for malformed output.""" # Strip markdown fences if present cleaned = raw_text.strip() if cleaned.startswith("```"): cleaned = cleaned.split("\n", 1)[-1] cleaned = cleaned.rsplit("```", 1)[0].strip() try: return json.loads(cleaned) except json.JSONDecodeError: return { "is_kidnapping": False, "confidence": 0, "context": f"Failed to parse model response: {raw_text[:200]}", } def analyze_audio_threat(file_path: str) -> dict: """Send raw audio to GPT-4o-audio-preview for threat analysis. Returns ``{"is_kidnapping": bool, "confidence": int, "context": str}``. """ path = Path(file_path) if not path.exists(): return { "is_kidnapping": False, "confidence": 0, "context": f"Audio file not found: {file_path}", } if path.suffix.lower() not in FORMAT_MAP: return { "is_kidnapping": False, "confidence": 0, "context": f"Unsupported audio format: {path.suffix}. Must be .wav or .mp3.", } api_key: str | None = os.environ.get("OPENAI_API_KEY") if not api_key: return { "is_kidnapping": False, "confidence": 0, "context": "Missing OPENAI_API_KEY env var", } encoded_audio, audio_format = _read_and_encode(file_path) client = OpenAI(api_key=api_key) try: completion = client.chat.completions.create( model="gpt-4o-audio-preview", messages=[ {"role": "system", "content": SYSTEM_PROMPT}, { "role": "user", "content": [ { "type": "input_audio", "input_audio": { "data": encoded_audio, "format": audio_format, }, }, ], }, ], ) raw_text: str = completion.choices[0].message.content or "" print(f"[ai_service] Raw model response: {raw_text}") return _parse_response(raw_text) except Exception as exc: print(f"[ai_service] OpenAI call failed: {exc}") return { "is_kidnapping": False, "confidence": 0, "context": f"OpenAI API error: {str(exc)}", } ``` ### action_hub/email_service.py ```python """Email service — sends an emergency report with audio + location data (Modal edition).""" import os import smtplib from email.mime.audio import MIMEAudio from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from pathlib import Path SMTP_HOST: str = "smtp.gmail.com" SMTP_PORT: int = 587 def _get_env(name: str) -> str | None: """Read an env var at call time (not import time) for Modal secrets.""" return os.environ.get(name) def _validate_env_vars() -> tuple[str | None, dict[str, str]]: """Return (error_msg, env_dict). Error is None when all vars are present.""" keys = ["SENDER_EMAIL", "SENDER_APP_PASSWORD", "TARGET_EMAIL"] env: dict[str, str] = {} missing: list[str] = [] for k in keys: val = _get_env(k) if not val: missing.append(k) else: env[k] = val if missing: return f"Missing email env vars: {', '.join(missing)}", {} return None, env def _build_maps_url(lat: float, lon: float) -> str: """Google Maps link for a coordinate pair.""" return f"https://www.google.com/maps?q={lat},{lon}" def _compose_body(context: str, location_data: dict | None = None) -> str: """Build a plain-text email body with alert context and location.""" lines: list[str] = [ "EMERGENCY ALERT", "", context, ] if location_data: last_lat = location_data.get("last_lat", 0) last_lon = location_data.get("last_lon", 0) first_lat = location_data.get("first_lat", 0) first_lon = location_data.get("first_lon", 0) distance = location_data.get("distance_m", 0) speed = location_data.get("max_speed_kmh", 0) lines.extend([ "", f"START: {location_data.get('first_address', 'Unknown')}", f"{_build_maps_url(first_lat, first_lon)}", "", f"LAST KNOWN: {location_data.get('last_address', 'Unknown')}", f"{_build_maps_url(last_lat, last_lon)}", "", f"Distance: {distance} m | Max Speed: {speed} km/h", ]) lines.extend([ "", "Audio recording attached.", ]) return "\n".join(lines) def _attach_audio(msg: MIMEMultipart, audio_path: str) -> bool: """Attach an audio file to *msg*. Returns True on success.""" path = Path(audio_path) if not path.exists(): print(f"[email_service] Audio file not found: {audio_path}") return False subtype_map: dict[str, str] = {".wav": "wav", ".mp3": "mpeg", ".m4a": "mp4", ".ogg": "ogg"} subtype: str = subtype_map.get(path.suffix.lower(), "wav") try: with open(path, "rb") as f: audio_data = f.read() attachment = MIMEAudio(audio_data, _subtype=subtype) attachment.add_header("Content-Disposition", "attachment", filename=path.name) msg.attach(attachment) return True except Exception as exc: print(f"[email_service] Failed to attach audio: {exc}") return False def send_emergency_email( context: str, file_path: str, location_data: dict | None = None, ) -> dict: """Send an emergency email with context, location data, and audio attachment. Returns a dict with ``status`` and optionally ``error``. """ err, env = _validate_env_vars() if err: print(f"[email_service] {err}") return {"status": "error", "error": err} msg = MIMEMultipart() msg["From"] = env["SENDER_EMAIL"] msg["To"] = env["TARGET_EMAIL"] msg["Subject"] = "🚨 Emergency Alert — Immediate Attention Required" body: str = _compose_body(context, location_data) msg.attach(MIMEText(body, "plain")) audio_attached: bool = _attach_audio(msg, file_path) try: with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as server: server.starttls() server.login(env["SENDER_EMAIL"], env["SENDER_APP_PASSWORD"]) server.send_message(msg) print("[email_service] Email sent successfully") return {"status": "sent", "audio_attached": audio_attached} except Exception as exc: print(f"[email_service] Email failed: {exc}") return {"status": "error", "error": str(exc)} ``` ### safety_detection_model_development/train_svm.py ```python #!/usr/bin/env python3 """ Train a One-Class SVM on normal Apple Watch sensor data and export for Swift. Usage: python3 -m venv .venv && source .venv/bin/activate pip install pandas numpy scikit-learn python train_svm.py Outputs: svm_params.json — model weights for hardcoded Swift inference (also prints Swift-pasteable arrays for AnomalyDetector.swift) """ import json import pandas as pd import numpy as np from sklearn.svm import OneClassSVM from sklearn.preprocessing import StandardScaler # ── Config ─────────────────────────────────────────────────────────────────── FEATURES = [ 'accel_x', 'accel_y', 'accel_z', 'accel_magnitude', 'accel_variance', 'gyro_x', 'gyro_y', 'gyro_z', 'rotation_magnitude', 'roll', 'pitch', 'yaw', 'heart_rate', ] WINDOW_SIZE = 10 # seconds (at 1Hz) — must match AnomalyDetector.swift windowSize NU = 0.05 # expected anomaly fraction KERNEL = 'rbf' GAMMA = 0.001 # small gamma → wider kernel → better score spread DECISION_THRESHOLD = -20.0 # score below this → anomaly (must match AnomalyDetector.swift) # ── Load data ──────────────────────────────────────────────────────────────── df = pd.read_csv('normal_training_sensor_data.csv') data = df[FEATURES].values print(f"Loaded {data.shape[0]} rows, {data.shape[1]} channels") # ── Sliding window feature extraction ──────────────────────────────────────── def extract_windows(data, window_size=WINDOW_SIZE): """Compute mean/std/max/min per channel over sliding windows.""" X = [] for i in range(len(data) - window_size + 1): w = data[i : i + window_size] feats = np.concatenate([w.mean(axis=0), w.std(axis=0), w.max(axis=0), w.min(axis=0)]) X.append(feats) return np.array(X) X = extract_windows(data) print(f"Feature matrix: {X.shape} (windows x features)") # ── Train ──────────────────────────────────────────────────────────────────── scaler = StandardScaler() X_scaled = scaler.fit_transform(X) svm = OneClassSVM(kernel=KERNEL, nu=NU, gamma=GAMMA) svm.fit(X_scaled) # ── Evaluate on training data ──────────────────────────────────────────────── scores = svm.decision_function(X_scaled) predictions = svm.predict(X_scaled) normal_rate = (predictions == 1).mean() print(f"\n── Training evaluation ──") print(f"Normal classified as normal: {normal_rate:.1%}") print(f"Score range: [{scores.min():.3f}, {scores.max():.3f}]") print(f"Score mean: {scores.mean():.3f}, std: {scores.std():.3f}") print(f"Support vectors: {svm.support_vectors_.shape[0]}") # ── Test on real_safety_data.csv if available ──────────────────────────────── try: df_test = pd.read_csv('real_safety_data.csv') test_data = df_test[FEATURES].values X_test = extract_windows(test_data) X_test_scaled = scaler.transform(X_test) test_scores = svm.decision_function(X_test_scaled) test_preds = svm.predict(X_test_scaled) print(f"\n── Test on real_safety_data.csv ──") print(f"Test windows: {len(X_test)}") print(f"Score range: [{test_scores.min():.3f}, {test_scores.max():.3f}]") print(f"Score mean: {test_scores.mean():.3f}") print(f"Anomaly rate (threshold {DECISION_THRESHOLD}): {(test_scores < DECISION_THRESHOLD).mean():.1%}") except Exception as e: print(f"\n(Skipping real_safety_data.csv test: {e})") # ── Export JSON params ─────────────────────────────────────────────────────── model_params = { 'scaler_mean': scaler.mean_.tolist(), 'scaler_scale': scaler.scale_.tolist(), 'support_vectors': svm.support_vectors_.tolist(), 'dual_coef': svm.dual_coef_[0].tolist(), # flatten from [[...]] to [...] 'intercept': svm.intercept_[0], 'gamma': float(svm._gamma), 'n_features': 52, 'n_support_vectors': svm.support_vectors_.shape[0], 'window_size': WINDOW_SIZE, 'decision_threshold': DECISION_THRESHOLD, 'channels': FEATURES, } with open('svm_params.json', 'w') as f: json.dump(model_params, f, indent=2) print(f"\nSaved svm_params.json") # ── Generate Swift code snippet ────────────────────────────────────────────── def fmt_array(arr, per_line=6): """Format array as Swift literal.""" lines = [] for i in range(0, len(arr), per_line): chunk = ', '.join(f'{v}' for v in arr[i:i+per_line]) lines.append(f' {chunk},') return '\n'.join(lines) print(f"\n── Swift snippet (paste into AnomalyDetector.swift) ──") print(f" static let gamma: Double = {model_params['gamma']}") print(f" static let intercept: Double = {model_params['intercept']}") print(f" static let nSupportVectors: Int = {model_params['n_support_vectors']}") print(f" // scaler mean: {len(model_params['scaler_mean'])} values") print(f" // scaler scale: {len(model_params['scaler_scale'])} values") print(f" // support vectors: {model_params['n_support_vectors']} x 52") print(f" // dual coef: {model_params['n_support_vectors']} values") print(f"\nModel params saved to svm_params.json — load at runtime or embed in Swift.") ``` ### action_hub/location_agent_enhanced.py ```python # location_agent_enhanced.py — GPS movement analysis with reverse geocoding import csv from math import radians, sin, cos, sqrt, atan2 from geopy.geocoders import Nominatim from geopy.extra.rate_limiter import RateLimiter # ── Geocoder setup ─────────────────────────────── geolocator = Nominatim(user_agent="emergency-action-hub", timeout=10) reverse = RateLimiter(geolocator.reverse, min_delay_seconds=1) def haversine(lat1: float, lon1: float, lat2: float, lon2: float) -> float: """Distance between two GPS points in meters.""" R = 6371000 dlat = radians(lat2 - lat1) dlon = radians(lon2 - lon1) a = sin(dlat / 2) ** 2 + cos(radians(lat1)) * cos(radians(lat2)) * sin(dlon / 2) ** 2 c = 2 * atan2(sqrt(a), sqrt(1 - a)) return R * c def bearing(lat1: float, lon1: float, lat2: float, lon2: float) -> tuple[float, str]: """Compass bearing between two GPS points → (degrees, direction_str).""" lat1, lon1, lat2, lon2 = map(radians, [lat1, lon1, lat2, lon2]) dlon = lon2 - lon1 y = sin(dlon) * cos(lat2) x = cos(lat1) * sin(lat2) - sin(lat1) * cos(lat2) * cos(dlon) brng = (atan2(y, x) * 180 / 3.1415926535 + 360) % 360 directions = [ "N", "NNE", "NE", "ENE", "E", "ESE", "SE", "SSE", "S", "SSW", "SW", "WSW", "W", "WNW", "NW", "NNW", ] idx = int((brng + 11.25) / 22.5) % 16 return brng, directions[idx] def calculate_speed(distance_m: float, time_diff_s: float) -> tuple[float, float]: """Return (speed_m/s, speed_km/h).""" if time_diff_s <= 0: return 0.0, 0.0 speed_ms = distance_m / time_diff_s return speed_ms, speed_ms * 3.6 def get_exact_address(lat: float, lon: float) -> dict: """Reverse-geocode GPS coords into a detailed address dict.""" if lat == 0 or lon == 0: return {"full_address": "No valid GPS coordinates", "street": None, "city": None, "state": None, "country": None, "postcode": None} try: location = reverse((lat, lon), exactly_one=True, language="en", zoom=18) if not location: return {"full_address": f"No address found for ({lat:.6f}, {lon:.6f})", "street": None, "city": None, "state": None, "country": None, "postcode": None} addr = location.raw.get("address", {}) house = addr.get("house_number", "") road = addr.get("road") or addr.get("pedestrian") or addr.get("path") or addr.get("footway", "") city = addr.get("city") or addr.get("town") or addr.get("village") or addr.get("suburb", "") state = addr.get("state") or addr.get("region") or addr.get("province", "") country = addr.get("country", "") postcode = addr.get("postcode", "") street = f"{house} {road}".strip() if house or road else "Unknown street" parts = [p for p in [street if street != "Unknown street" else None, city, state, postcode, country] if p] full_address = ", ".join(parts) if parts else location.address return { "full_address": full_address, "street": street, "city": city, "state": state, "country": country, "postcode": postcode, } except Exception as exc: print(f"[location_agent] Geocoding error at ({lat:.6f}, {lon:.6f}): {exc}") return {"full_address": f"Geocoding error: {exc}", "street": None, "city": None, "state": None, "country": None, "postcode": None} def get_latest_address(gps_points: list[dict]) -> dict | None: """Return address dict for the most recent GPS point.""" if not gps_points: return None latest = gps_points[-1] return get_exact_address(latest["lat"], latest["lon"]) def parse_csv_to_gps_points(csv_path: str) -> list[dict]: """Read the panic CSV and extract GPS points with panic flags.""" points: list[dict] = [] with open(csv_path, "r") as f: reader = csv.DictReader(f) for row in reader: lat = float(row.get("latitude", 0)) lon = float(row.get("longitude", 0)) if lat == 0 and lon == 0: continue points.append({ "lat": lat, "lon": lon, "timestamp": float(row.get("timestamp", 0)), "panic": row.get("is_panic", "False").strip() == "True", "heart_rate": float(row.get("heart_rate", 0)), "speed": float(row.get("speed", 0)), }) points.sort(key=lambda p: p["timestamp"]) return points def analyze_movement(gps_points: list[dict]) -> dict: """Analyze a list of GPS points for kidnapping indicators. Returns movement intelligence: distance, speed spikes, address transitions, risk score. """ if len(gps_points) < 2: return {"risk": 0, "reason": "not enough movement data"} total_distance: float = 0 speed_spikes: int = 0 panic_during_motion: bool = False address_transitions: list[dict] = [] last_address: str | None = None max_speed_kmh: float = 0 for i in range(1, len(gps_points)): p1 = gps_points[i - 1] p2 = gps_points[i] dist = haversine(p1["lat"], p1["lon"], p2["lat"], p2["lon"]) total_distance += dist # Prefer sensor speed from CSV; fall back to GPS-derived speed sensor_speed = p2.get("speed", 0) if sensor_speed > 0: speed_kmh = sensor_speed * 3.6 # sensor speed is m/s else: time_diff = p2["timestamp"] - p1["timestamp"] if time_diff <= 0: continue _, speed_kmh = calculate_speed(dist, time_diff) max_speed_kmh = max(max_speed_kmh, speed_kmh) if speed_kmh > 40: speed_spikes += 1 if p2.get("panic"): panic_during_motion = True # Address change detection (only check every few points to avoid rate limiting) if i % 3 == 0 or i == len(gps_points) - 1: addr_data = get_exact_address(p2["lat"], [truncated — 1414 more characters] ``` ### action_hub/action_hub.py ```python """Action Hub — Modal serverless entrypoint for the emergency response system.""" import os import tempfile from pathlib import Path import modal from fastapi import File, Query, Request, UploadFile from fastapi.responses import Response # ── Modal setup ────────────────────────────────── app = modal.App("emergency-action-hub") image = ( modal.Image.debian_slim() .apt_install("ffmpeg") .pip_install( "openai", "twilio", "fastapi[standard]", "python-multipart", "pydub", "geopy", "pandas", ) .add_local_python_source("ai_service") .add_local_python_source("twilio_service") .add_local_python_source("email_service") .add_local_python_source("location_agent_enhanced") ) # Shared Dict for passing call data between endpoints call_store = modal.Dict.from_name("call-data-store", create_if_missing=True) def _convert_to_wav(input_path: str) -> str | None: """Try to convert audio to .wav (or .mp3 fallback). Returns new path or None.""" import subprocess # Try pydub first (handles most formats) try: from pydub import AudioSegment wav_path: str = input_path.rsplit(".", 1)[0] + ".wav" audio = AudioSegment.from_file(input_path) audio.export(wav_path, format="wav") print(f"[action_hub] Converted to wav: {wav_path}") return wav_path except Exception as exc: print(f"[action_hub] Pydub conversion failed: {exc}") # Fallback: try ffmpeg directly with error recovery flags for fmt, ext in [("wav", ".wav"), ("mp3", ".mp3")]: out_path = input_path.rsplit(".", 1)[0] + ext try: result = subprocess.run( ["ffmpeg", "-y", "-err_detect", "ignore_err", "-i", input_path, out_path], capture_output=True, timeout=30, ) if result.returncode == 0: print(f"[action_hub] Converted to {fmt} via ffmpeg fallback") return out_path except Exception as exc: print(f"[action_hub] ffmpeg {fmt} fallback failed: {exc}") print("[action_hub] All conversion attempts failed") return None def _save_upload(upload_file: bytes, suffix: str) -> str: """Write raw bytes to a temp file and return the path.""" with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp: tmp.write(upload_file) return tmp.name def _get_base_url() -> str: """Build the base Modal endpoint URL.""" workspace = os.environ.get("MODAL_WORKSPACE", "namanj") return f"https://{workspace}--emergency-action-hub" # ────────────────────────────────────────────────── # POST /process_audio — main entry point # ────────────────────────────────────────────────── @app.function( image=image, secrets=[modal.Secret.from_name("custom-secret")], timeout=300, ) @modal.fastapi_endpoint(method="POST") async def process_audio( file: UploadFile = File(...), csv_file: UploadFile | None = File(None), ) -> dict: """Receive audio + optional CSV, analyze threats, and dispatch alerts.""" from ai_service import analyze_audio_threat from twilio_service import trigger_twilio_call from email_service import send_emergency_email from location_agent_enhanced import parse_csv_to_gps_points, analyze_movement if file.filename is None: return {"status": "error", "error": "No audio file uploaded"} tmp_paths: list[str] = [] try: # --- 1. Save audio and try to convert to wav --- audio_bytes: bytes = await file.read() audio_suffix: str = os.path.splitext(file.filename)[1] or ".wav" audio_tmp: str = _save_upload(audio_bytes, audio_suffix) tmp_paths.append(audio_tmp) wav_result: str | None = _convert_to_wav(audio_tmp) if wav_result: tmp_paths.append(wav_result) # Use wav if conversion succeeded, otherwise use original file audio_for_analysis: str = wav_result or audio_tmp # --- 2. Analyze audio for threats --- analysis: dict = analyze_audio_threat(audio_for_analysis) print(f"[action_hub] Audio analysis: {analysis}") # --- 3. Analyze CSV location data if provided --- location_data: dict | None = None csv_tmp: str = "" if csv_file and csv_file.filename: csv_bytes: bytes = await csv_file.read() csv_tmp = _save_upload(csv_bytes, ".csv") tmp_paths.append(csv_tmp) gps_points = parse_csv_to_gps_points(csv_tmp) if len(gps_points) >= 2: location_data = analyze_movement(gps_points) print(f"[action_hub] Location analysis: risk={location_data['risk']}, " f"distance={location_data['distance_m']}m") # --- 4. If kidnapping detected, dispatch alerts --- call_result: dict | None = None email_result: dict | None = None if analysis.get("is_kidnapping"): context: str = analysis.get("context", "Emergency detected from audio.") # Enrich context with location if available call_context: str = context if location_data: call_context = ( f"{context} " f"Last known location: {location_data.get('last_address', 'unknown')}. " f"Traveled {location_data.get('distance_m', 0)} meters. " f"Max speed {location_data.get('max_speed_kmh', 0)} kilometers per hour." ) call_result = trigger_twilio_call(call_context, audio_path=audio_for_analysis) email_result = send_emergency_email( context=context, file_path=audio_for_analysis, location_data=location_data, ) return { "status": "processed", "analysis": analysis, "location": location_data, "call": [truncated — 4492 more characters] ``` ### wearable_safety_detection/wearable_safety_detection Watch App/wearable_safety_detectionApp.swift ```swift import SwiftUI @main struct wearable_safety_detection_Watch_AppApp: App { var body: some Scene { WindowGroup { ContentView() } } } ``` ### wearable_safety_detection/wearable_safety_detection Watch App/DetectionPayload.swift ```swift import Foundation struct DetectionPayload: Codable { var dataPoints: [ThreatData] var timestamp: Double var anomalyScore: Double var anomalyPanic: Bool var mathScore: Double var mathPanic: Bool var combinedPanic: Bool } ``` [11 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]