# Project export: TED.ai 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: UC Berkeley AI Hackathon 2025 - Tagline: TED uses AI to detect fires, crashes, and medical emergencies from video feeds in real time, alerting responders instantly. It's fast, automated, and built for public and smart environments. - Devpost: https://devpost.com/software/ted-ai - GitHub: https://github.com/richardwei6/berkeley-ai-hackathon-2025 - Video: https://www.youtube.com/embed/vYSkPUi4GnE?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 4 GitHub contributor(s) — Richard Wei (20 commits), Anthony Miceli (14 commits), AnthonyM27 (8 commits), Jeffrey Aaron Jeyasingh (5 commits) ## Devpost submission (written by the team) ### Inspiration Emergencies often go unnoticed or unreported until it's too late — seconds can make the difference between life and death. Whether it's a car crash in a busy intersection, someone collapsing due to a heart attack, or a fire starting in a crowded space, timely response is critical. With the increasing presence of surveillance and personal video streams, we saw an opportunity to leverage artificial intelligence to automate emergency detection and dramatically reduce emergency response times. This inspired the creation of TED.ai — Total Emergency Detection AI. ### What it does TED.ai is an AI-powered emergency detection system that analyzes live or remote video feeds to identify critical incidents like fires, dangerous activity, car crashes, and medical emergencies in real time. By combining computer vision with intelligent alerting, it can automatically notify first responders to reduce response time and potentially save lives. It's designed to be fast, scalable, and deployable in public spaces, vehicles, or smart surveillance systems. When an emergency is detected, TED.ai automatically triggers alerts to the appropriate first responders using configurable mechanisms such as: Text messages Email Webhooks to dispatch systems or third-party apps Or, most importantly, through our integrated dashboard! The best part about TED.ai is that it works with any pre-existing camera system. All you need is a video feed and a simple server for TED.ai to work. That means traffic light cameras, security cameras, and even your personal phone can be part of the TED.ai network! ### How we built it We built TED.ai as a multi-component pipeline: Multi-Modal Backend Pipeline: Developed in Python using Flask, the primary backend server ingests video stream data from multiple camera feeds and distributes it to specialized processing servers. Each server runs a dedicated model—such as for fire detection or medical emergencies—allowing simultaneous and efficient event detection. Computer Vision Models: In each distributed server, we fine-tuned and leveraged object detection and action recognition models (e.g., YOLOv5, Google MobileNet SSD, and PoseNet) to identify visual patterns associated with emergencies. We ran some models using GPU-accelerated inference (with Torch/ONNX) to maintain high performance under real-time constraints. We also chose specific models like Google's MobileNet SSD model due to the efficiency and portability that comes with a single shot detection model. Temporal Analysis: To prioritize urgent scenarios like medical emergencies, we implemented a split processing pipeline. This ensures efficient resource allocation, as medical events require immediate detection, whereas incidents like fires or car crashes are less time-sensitive and can be processed with slight delays. Alert System: In addition to the dashboard that first responders use, we also have configurable endpoints to allow integration with SMS, email, or other emergency management platforms. ### Challenges we ran into False Positives: Emergency-like motions (e.g., sitting quickly) sometimes triggered alerts. We had to fine-tune thresholds and add temporal filtering. Model Latency: Real-time processing at scale required optimization — we experimented with batching and frame sampling strategies as mentioned above in the temporal analysis section. Limited Training Data: Some edge cases (e.g., seizures) had scarce publicly available data, which made generalization harder. Alert Fatigue: Balancing sensitivity with relevance was tricky — we had to ensure the system is actionable without being annoying. ### Accomplishments we're proud of Built a functioning prototype that can detect a wide range of emergency scenarios in real time Leveraged multiple ML models in a coherent and efficient pipeline Created an efficient distributed server architecture and pipeline Developed a responsive alert system that works with minimal configuration Successfully simulated real-life emergencies and verified accurate detection ### What we learned Being able to detect certain emergencies is hard especially when the frames you feed into the model isn't high enough quality Having a good distributed server architecture is important to not overwhelm any particular server and to ensure that the load is distributed evenly. Building systems for real-world emergencies means prioritizing reliability of the product, not just performance due to the nature of these emergencies. Integrating multi-modal detection—such as motion, object, and fire recognition—is challenging due to overlapping signals that can interfere with one another. ### What's next Expand training datasets to improve accuracy across diverse environments and cultures Integrate audio detection for screams, alarms, or explosions Build a mobile SDK to allow integration with phone or dashcam apps Work with companies to integrate with security cameras or other camera feeds Add explainable AI components to clarify why an emergency was flagged Partner with public safety organizations for pilot testing and real-world validation Improve privacy features like anonymized person detection and local-only processing ## README (from the GitHub repository) README CalHacks AI 2025 # Total Emergency Detection AI This project leverages AI and computer vision to **detect real-world emergencies from streamed video data**, including: - Fires - Car crashes - People collapsing (e.g., fainting, seizures, medical emergencies) When an emergency is detected, the system can **automatically notify the appropriate first responders** via configurable alert mechanisms --- ## Project Goal To develop an AI-powered pipeline that monitors live or remote video feeds, **analyzes frames in real time**, and triggers emergency response workflows with high confidence — reducing response time and improving public safety. --- ## Features - **Live camera or video file input** - **Remote image fetching via HTTP server** - **Image extraction at fixed intervals (e.g., every 0.5 seconds)** - **AI-based emergency classification using open-source vision models** - **Supports fire, crash, and collapse detection** - **Auto-saving of emergency frames** - **Auto-alert system for responders** --- ## Setup ### 1. Clone the Repo ```bash git clone https://github.com/your-username/emergency-ai.git cd emergency-ai ``` ### 2. Install Dependencies ```bash pip install -r requirements.txt ``` BLIP is downloaded automatically using transformers. No API key required. --- ## How it Works ### A. Local Video Stream ```bash python extract_from_video.py --source path/to/video.mp4 ``` This script: Extracts frames every 0.5 seconds Saves them to extracted_images/ Classifies each image and prints result ### B. Remote Image Server (Optional) To fetch images from a remote endpoint (e.g., /screenshot_full on an ngrok server): ```bash python classify_remote.py ``` This script: Sends a GET request Decodes base64 image response Classifies the image (e.g., "fire" or "none") Saves it if it contains an emergency ## Detected evidence (automated analysis) Indexed codebase: 14 recognized source files, 43 KB. - Flask (technology) — detected in the code - Python (language) — detected in the code - PyTorch (technology) — detected in the code - TensorFlow (technology) — detected in the code - Google Gemini (technology) — claimed on Devpost, not found in the code - Next.js (technology) — claimed on Devpost, not found in the code ## Codebase structure (from repository index) ### Files (27 of 27) ``` .gitattributes .gitignore camera_input/camera_input.py event-detection/classify.py event-detection/main.py event-detection/server_connection.py event-detection/test_inference.py event-detection/test_server_connection.py event-detection/vision_inference.py person_detection/deploy.prototxt person_detection/mobilenet_iter_73000.caffemodel person_detection/people_cropper.py person_detection/people-cropper-simple.py pose-detection/collapse_detector.py pose-detection/collapseModel.h5 pose-detection/collapseModel.ipynb pose-detection/package.json pose-detection/pose_image.py pose-detection/pose_labels.csv pose-detection/pose_labelsTieShoe.csv pose-detection/poseModel.ipynb README.md requirements.txt server.py weapon_detection/best.onnx weapon_detection/weapon_detector.py weapon_detection/yolov5s.pt ``` ### Dependencies - pose-detection/package.json: @tensorflow-models/posenet@^2.2.2 - requirements.txt: caffe, flask, imageio, matplotlib, msgpack, numpy, onnxruntime, opencv-python, pandas, seaborn, tensorflow, tensorflow_hub, torch ### Recent commits (newest first) - Merge branch 'main' of https://github.com/richardwei6/berkeley-ai-hackathon-2025 - Updated error messages - Merge branch 'main' of ssh://github.com/richardwei6/berkeley-ai-hackathon-2025 - Add weapon detector - Location data updated - WORKING fire & crash detection - Update README.md - Update README.md - Update README.md - Update README.md - Update README.md - Update README.md - Update README.md - Merge branch 'main' of https://github.com/richardwei6/berkeley-ai-hackathon-2025 - server connected - Remove rate limiting and add more error handling - Merge branch 'main' of ssh://github.com/richardwei6/berkeley-ai-hackathon-2025 - Update server with file limits - bug - fire/cc classification ## Key source files (fetched from GitHub, selected and truncated for size) ### requirements.txt ``` tensorflow opencv-python imageio numpy matplotlib tensorflow_hub caffe flask msgpack onnxruntime torch pandas seaborn ``` ### pose-detection/package.json ``` { "dependencies": { "@tensorflow-models/posenet": "^2.2.2" } } ``` ### server.py ```python import os from camera_input.camera_input import CameraInput from person_detection.people_cropper import PeopleCropper from flask import Flask, send_file import base64 import msgpack from datetime import datetime, timedelta from weapon_detection.weapon_detector import WeaponDetector test_loc = "37.8688956,-122.2600617" shared_screenshots_dir = "./shared/screenshots" shared_people_dir = "./shared/people" shared_weapon_dir = "./shared/weapon" if not os.path.exists(shared_screenshots_dir): os.makedirs(shared_screenshots_dir) if not os.path.exists(shared_people_dir): os.makedirs(shared_people_dir) if not os.path.exists(shared_weapon_dir): os.makedirs(shared_weapon_dir) # Test camera input camera = CameraInput(output_dir=shared_screenshots_dir) # Detection models people_cropper = PeopleCropper(output_dir=shared_people_dir) weapon_detector = WeaponDetector(output_dir=shared_weapon_dir) app = Flask(__name__) @app.route('/screenshot_people', methods=['GET']) def screenshot_people(): print("Taking screenshot and detecting people") # Take screenshot using camera filename = camera.take_screenshot() if filename and os.path.exists(filename): remove_old_screenshots() # Process the image to detect people people_filenames = people_cropper.detect(filename) # Convert each cropped image to base64 if (people_filenames is None or len(people_filenames) == 0): return "No people detected", 400 encoded_images = [] for person_file in people_filenames: with open(person_file, 'rb') as f: img_bytes = f.read() img_b64 = base64.b64encode(img_bytes).decode('utf-8') encoded_images.append(img_b64) # Pack into messagepack format response = { 'people_images': encoded_images, "loc": test_loc, } packed_response = msgpack.packb(response) return packed_response, 200, {'Content-Type': 'application/x-msgpack'} else: return "Failed to take screenshot", 500 @app.route('/screenshot_weapons', methods=['GET']) def screenshot_weapons(): print("Taking screenshot and detecting weapons") # Take screenshot using camera filename = camera.take_screenshot() if filename and os.path.exists(filename): remove_old_screenshots() # Process the image to detect weapons weapon_filenames = weapon_detector.detect(filename) # Convert each cropped image to base64 if (weapon_filenames is None or len(weapon_filenames) == 0): return "No weapons detected", 400 encoded_images = [] for weapon_file in weapon_filenames: with open(weapon_file, 'rb') as f: img_bytes = f.read() img_b64 = base64.b64encode(img_bytes).decode('utf-8') encoded_images.append(img_b64) # Pack into messagepack format response = { 'weapon_images': encoded_images, "loc": test_loc, } packed_response = msgpack.packb(response) return packed_response, 200, {'Content-Type': 'application/x-msgpack'} else: return "Failed to take screenshot", 500 @app.route('/screenshot_full', methods=['GET']) def screenshot_full(): print("Taking screenshot") # Take screenshot using camera filename = camera.take_screenshot() if filename and os.path.exists(filename): remove_old_screenshots() # Convert the image to base64 with open(filename, 'rb') as f: img_bytes = f.read() img_b64 = base64.b64encode(img_bytes).decode('utf-8') # Pack into messagepack format response = { 'image': img_b64, "loc": test_loc, } packed_response = msgpack.packb(response) return packed_response, 200, {'Content-Type': 'application/x-msgpack'} else: return "Failed to take screenshot", 500 def remove_old_screenshots(): # Get all png files in screenshots dir sorted by creation time screenshot_files = [f for f in os.listdir(shared_screenshots_dir) if f.endswith('.png')] screenshot_files.sort(key=lambda x: os.path.getctime(os.path.join(shared_screenshots_dir, x))) # Remove all but the 5 most recent png files if len(screenshot_files) > 5: for file in screenshot_files[:-5]: os.remove(os.path.join(shared_screenshots_dir, file)) print(f"Removed {file}") # Get all folders in people dir sorted by creation time people_folders = [f for f in os.listdir(shared_people_dir) if os.path.isdir(os.path.join(shared_people_dir, f))] people_folders.sort(key=lambda x: os.path.getctime(os.path.join(shared_people_dir, x))) # Remove all but the 5 most recent folders if len(people_folders) > 5: for folder in people_folders[:-5]: folder_path = os.path.join(shared_people_dir, folder) # Remove all files in the folder first for file in os.listdir(folder_path): os.remove(os.path.join(folder_path, file)) # Remove the empty folder os.rmdir(folder_path) print(f"Removed folder {folder}") # Get all folders in weapon dir sorted by creation time weapon_folders = [f for f in os.listdir(shared_weapon_dir) if os.path.isdir(os.path.join(shared_weapon_dir, f))] weapon_folders.sort(key=lambda x: os.path.getctime(os.path.join(shared_weapon_dir, x))) # Remove all but the 5 most recent folders if len(weapon_folders) > 5: for folder in weapon_folders[:-5]: folder_path = os.path.join(shared_weapon_dir, folder) # Remove all files in the folder first for file in os.listdir(folder_path): os.remove(os.path.join(folder_path, file)) # Remove the empty folder os.rmdir(folder_path) print(f"Removed folder [truncated — 80 more characters] ``` ### event-detection/main.py ```python from classify import classify_remote_image import time def main(): while (True): url = "https://6285-2607-f140-400-68-c90d-840b-94de-89bc.ngrok-free.app/screenshot_full" classify_remote_image(url) time.sleep(1) main() ``` ### event-detection/test_inference.py ```python from vision_inference import extract_and_classify sources = ["video_files/fire0.mp4","video_files/fire1.mp4", "video_files/fire2.mp4", "video_files/fire3.mp4", "video_files/fire4.mp4", "video_files/cc0.mp4", "video_files/cc1.mp4", "video_files/cc2.mp4", "video_files/cc3.mp4", "video_files/cc4.mp4", "video_files/none0.mp4", "video_files/none1.mp4", "video_files/none2.mp4", "video_files/none3.mp4", "video_files/none4.mp4"] labels = ["fire", "fire", "fire", "fire", "fire", "cc", "cc", "cc", "cc", "cc", "none", "none", "none", "none", "none"] correct = 0 for idx in range(len(sources)): returned = extract_and_classify(source=sources[idx]) if (labels[idx] in returned): correct += 1 print("yup!") else: print("nope!") total = 15 print("Testing complete Accuracy: " + str(correct/total)) ``` ### event-detection/test_server_connection.py ```python import requests import base64 from io import BytesIO from PIL import Image url = "https://fb3d-2607-f140-400-68-3006-f365-d41f-e5a6.ngrok-free.app/screenshot_full" try: print("[SERVER] Sending GET request...") response = requests.get(url, timeout=10) response.raise_for_status() # Attempt to decode raw byte content as UTF-8, ignoring errors raw_text = response.content.decode("utf-8", errors="ignore") start_index = raw_text.find("iVBOR") if start_index == -1: raise ValueError("No valid base64 image data found in response") base64_data = raw_text[start_index:].strip() # Decode and open image print("[SERVER] Decoding base64 image data...") image_bytes = base64.b64decode(base64_data) image = Image.open(BytesIO(image_bytes)).convert("RGB") image.save("test_server_image.jpg") print("[SERVER] Image saved as test_server_image.jpg") except requests.RequestException as req_err: print(f"[SERVER] Request failed: {req_err}") except Exception as e: print(f"[SERVER] Error decoding or saving image: {e}") ``` ### event-detection/classify.py ```python from transformers import BlipProcessor, BlipForConditionalGeneration from PIL import Image import torch processor = BlipProcessor.from_pretrained("Salesforce/blip-image-captioning-base") model = BlipForConditionalGeneration.from_pretrained("Salesforce/blip-image-captioning-base") device = "cuda" if torch.cuda.is_available() else "cpu" model.to(device) def classify_pil_image(image: Image.Image): try: inputs = processor(image.convert("RGB"), return_tensors="pt").to(device) out = model.generate(**inputs) caption = processor.decode(out[0], skip_special_tokens=True).lower() print(f"[BLIP] Caption: {caption}") if "fire" in caption: return "fire" elif "crash" in caption or "accident" in caption or "wreck" in caption: return "crash" else: return "none" except Exception as e: print(f"[ERROR] BLIP classification failed: {e}") return "error" from server_connection import fetch_image_from_server def classify_remote_image(url): print('------------------------------------------------------------------------------') image = fetch_image_from_server(url) if image is None: print("[ERROR] Could not fetch image.") return "error" label = classify_pil_image(image) print(f"[CLASSIFY] Remote image classified as: {label}") return label ``` ### event-detection/vision_inference.py ```python import openai from PIL import Image import base64 import os import cv2 import time from classify import classify_image_with_blip def alert_emergency(label, image_path): print(f"[ALERT] Detected {label.upper()} in frame: {image_path}") def extract_and_classify(source=0, output_dir='extracted_images', interval=0.5): output = [] os.makedirs(output_dir, exist_ok=True) cap = cv2.VideoCapture(source) if not cap.isOpened(): print(f"Error: Could not open video source {source}") return fps = cap.get(cv2.CAP_PROP_FPS) if fps == 0 or fps != fps: fps = 30 frame_interval = int(fps * interval) frame_count = 0 saved_count = 0 print(f"[INFO] Starting stream at {fps:.1f} FPS, analyzing every {interval}s...") while True: ret, frame = cap.read() if not ret: print("[INFO] Video finished or failed to read frame.") break if frame_count % frame_interval == 0: filename = os.path.join(output_dir, f"frame_{saved_count:04d}.jpg") cv2.imwrite(filename, frame) print(f"[INFO] Saved {filename}") label = classify_image_with_blip(filename) print(f"[CLASSIFY] Frame {saved_count:04d}: {label}") if label in ['fire', 'crash']: alert_emergency(label, filename) output.append(label) saved_count += 1 frame_id = cap.get(cv2.CAP_PROP_POS_FRAMES) total_frames = cap.get(cv2.CAP_PROP_FRAME_COUNT) print(f"[INFO] Frame {frame_id} / {total_frames}") time.sleep(0.5) frame_count += 1 cap.release() print("[INFO] Stream ended.") return output ``` ### camera_input/camera_input.py ```python import cv2 import os from datetime import datetime, timedelta class CameraInput: def __init__(self, camera_index=0, output_dir='screenshots'): self.cap = cv2.VideoCapture(camera_index) self.output_dir = output_dir self.running = False if not os.path.exists(output_dir): os.makedirs(output_dir) def start_feed(self): self.running = True print("Starting webcam feed. Press 'q' to quit.") while self.running: tries = 0 while tries < 3: ret, frame = self.cap.read() if not ret: print("Failed to grab frame.") tries += 1 continue break if not ret: print("Failed to grab frame after 3 tries.") exit() cv2.imshow('Webcam Feed', frame) key = cv2.waitKey(1) if key & 0xFF == ord('q'): self.running = False elif key & 0xFF == ord('k'): self.take_screenshot() self.cap.release() cv2.destroyAllWindows() def take_screenshot(self): if not self.cap.isOpened(): print("Webcam is not open.") return tries = 0 while tries < 3: ret, frame = self.cap.read() if not ret: print("Failed to grab frame - %i", tries) tries += 1 continue break if ret: timestamp = datetime.now().strftime("%m:%d~%H:%M:%S:%f") filename = os.path.join(self.output_dir, f"screenshot {timestamp}.png") cv2.imwrite(filename, frame) print(f"Screenshot saved as {filename}") return filename else: print("!! Failed to take screenshot after 3 tries !!") ``` ### event-detection/server_connection.py ```python import requests import base64 import json from io import BytesIO from PIL import Image import cv2 import numpy as np from classify import classify_pil_image import requests import msgpack import base64 import os def decode_screenshot_response(response_bytes): unpacked = msgpack.unpackb(response_bytes, raw=False) img_b64 = unpacked.get("image") img_bytes = base64.b64decode(img_b64) image_path = "decoded_screenshot.jpg" with open(image_path, "wb") as f: f.write(img_bytes) loc = unpacked.get("loc") print("Location:", loc) print(f"Image saved to {image_path}") return image_path, loc def image_to_base64(image: Image.Image, format: str = 'JPEG') -> str: """ Converts a PIL Image to a base64 string. Args: image (Image.Image): The image to convert. format (str): The format to save the image in (e.g., 'JPEG', 'PNG'). Returns: str: Base64-encoded string of the image. """ buffer = BytesIO() image.save(buffer, format=format) buffer.seek(0) image_bytes = buffer.read() base64_str = base64.b64encode(image_bytes).decode('utf-8') return base64_str def fetch_image_from_server(url): try: print("[SERVER] Sending GET request...") response = requests.get(url, timeout=10) response.raise_for_status() if response.status_code == 200: imgpath, loc = decode_screenshot_response(response.content) else: print(f"[ERROR] Failed to get screenshot: {response.status_code}") cv_image = cv2.imread(imgpath) cv2.imshow("Live Emergency Feed", cv_image) cv2.waitKey(1) img = Image.open(imgpath) label = classify_pil_image(img) if label in ["fire", "crash"]: send_url = "https://42f2-2607-f140-400-49-75cb-ca8-db44-2ce3.ngrok-free.app/api/emergency-detection-base64" try: response = requests.post(send_url, data=image_to_base64(img), timeout=10) if response.status_code == 200: print(f"[SERVER] Emergency '{label}' data sent successfully.") else: print(f"[SERVER] POST failed with status {response.status_code}: {response.text}") except Exception as e: print(f"[ERROR] Error sending emergency POST: {e}") else: print(f"[INFO] No emergency detected (label: {label}) — not sending.") return img except requests.RequestException as req_err: print(f"[SERVER] Request failed: {req_err}") return None except Exception as e: print(f"[SERVER] Error decoding or saving image: {e}") return None ``` [5 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]