# Project export: CapyGuardian 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 2025 - Tagline: CapyGuardian: A wearable health system using an agentic AI to analyze health data, video & kinematics. It can detect falls and other abnormalities, instantly alerting others for assistance. - Devpost: https://devpost.com/software/capy-guardian - GitHub: https://github.com/PatP15/TreeHacks2025 - Video: https://www.youtube.com/embed/SaFCoakr5V8?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 1 GitHub contributor(s) — Patrick Puma (1 commits) ## Devpost submission (written by the team) ### Inspiration Imagine a world where care is both personalized and immediate—where every moment counts, and every individual’s unique needs are met instantly. CapyGuardian harnesses the power of agentic AI to blend wearable sensors with real-time health analysis delivering instant, tailored health insights. This approach transforms reactive care into proactive, individualized support that acts precisely when needed. Whether it's alerting a caregiver at the first sign of trouble or ensuring that a soldier in a high-stakes environment is safe, CapyGuardian is set to redefine emergency response. This dynamic system is a game-changer for older adults living independently, providing them with critical peace of mind, and for military platoons, where split-second decisions can make all the difference. What It Does CapyGuardian is a wearable health monitoring system that continuously tracks vital metrics such as heart rate, user kinematics, location, and even provides a waist-level video feed. All this data is presented on an intuitive dashboard where an integrated agentic AI analyzes the information in real-time. For example, if the AI abnormal heart rate patterns, it promptly issues an alert for immediate intervention—vital for both the vulnerable and large, dynamic groups. The dashboard has 4 primary sections: 1) The feed from the camera at waist-level 2) The kinematics of the user displayed on a skeleton (the skeleton moves with the user) 3) Heart rate data 4) Location on a map with friendlies (green dots) and hostiles (red dots) displayed Example Output from the CapyGuardian: ####################### I've submitted an urgent alert through the logs system. Based on my analysis: Critical Concerns: Person appears to have fallen (ground-level camera perspective) Heartbeat signal shows AFib indicators (uneven peaks) Person is alone in their location (map shows no nearby green dots) Person is alone in their location (map shows no nearby green dots) Environmental Assessment: Environmental Assessment: Location: Office setting Location: Office setting Lighting: Adequate for visibility Lighting: Adequate for visibility Surroundings: Office furniture visible Surroundings: Office furniture visible No immediate environmental hazards visible No immediate environmental hazards visible Medical Status: Medical Status: Possible cardiac event (AFib) Possible cardiac event (AFib) Potential fall-related injuries Potential fall-related injuries Unconscious or limited movement (based on camera angle) Unconscious or limited movement (based on camera angle) Location Safety: Location Safety: Person (green dot) is in a building Person (green dot) is in a building Several other people (red dots) are in nearby areas Several other people (red dots) are in nearby areas Location appears to be a monitored facility Location appears to be a monitored facility IMMEDIATE ACTION IS REQUIRED. I've triggered the alert system and recommended immediate medical response team dispatch. The combination of a possible fall and AFib indication makes this a high-priority medical emergency. ####################### How We Built It I built Capy Guardian using an Nvidia Jetson Nano 2GB Developer Kit for robust edge computing. Data is collected from 6 Adafruit IMU sensors (providing 3D acceleration and gyroscopic data) and a USB camera for a continuous chest-level video feed. Heart rate information is sourced via the Terra API, while real-time kinematics are processed from the IMU data using inverse kinematics with OpenSim. A Flask-hosted dashboard on the Jetson displays all these data streams, which are then analyzed by our agentic AI (via Scrapybara). Due to challenges in acquiring a static IP, to allow the agentic AI to interact with the dashboard, I upload prerecorded video feed and kinematic data to the agentic AI's ubuntu environment and host a demo dashboard there for it to analyze. The real time kinematics was inspired by this paper link where it was implemented on a raspberry pi but here wanted to see if the upgrade to a jetson would be helpful. Challenges We Ran Into Compiling the OpenSim package on the Jetson Nano was particularly taxing—taking 4-5 hours and crashing multiple times, leading to initial instability in real-time kinematics due to package conflicts. Additionally, integrating multiple data streams while ensuring performance posed quite a challenge. Accomplishments That We're Proud Of I was actually super excited to see that Capy Guardian to detect critical health events, such as falls, from just some camera data and kinematic data! The integration of agentic AI to interact with and act upon real-time health data is extremely promising! What We Learned This project deepened my understanding of the challenges of compiling and optimizing large software packages on edge computing hardware. I also learned the importance of designing systems that are both user-centric and robust, ensuring accurate data collection and processing in real-time. What's Next for Capy Guardian Moving forward, I plan to integrate real-time streaming of location and heart rate data directly from devices like the Apple Watch, eliminating the need for the dummy data from Terra API. Also, a static IP for the Jetson Nano would allow us to test the agentic AI’s capabilities to handle more complex health scenarios. The IMUs instead of being taped onto my clothing, would be embedded into the clothes. Beyond elder care, there is significant potential for applications in sports and military environments, where continuous and timely health monitoring would be helpful! ## README (from the GitHub repository) No README available. ## Detected evidence (automated analysis) Indexed codebase: 17 recognized source files, 94 KB. - Flask (technology) — detected in the code - HTML (language) — detected in the code - Python (language) — detected in the code - C++ (language) — claimed on Devpost, not found in the code ## Codebase structure (from repository index) ### Files (115 of 115) ``` .gitignore afib_2.csv app_demo.py app.py calibrated_Rajagopal_2015.osim calibration/gyro_offsets.npy capy_test.py Geometry/capitate_lvs.vtp Geometry/capitate_rvs.vtp Geometry/femur_l.vtp Geometry/femur_r.vtp Geometry/hamate_lvs.vtp Geometry/hamate_rvs.vtp Geometry/hat_jaw.vtp Geometry/hat_ribs_scap.vtp Geometry/hat_skull.vtp Geometry/hat_spine.vtp Geometry/humerus_lv.vtp Geometry/humerus_rv.vtp Geometry/index_distal_lvs.vtp Geometry/index_distal_rvs.vtp Geometry/index_medial_lvs.vtp Geometry/index_medial_rvs.vtp Geometry/index_proximal_lvs.vtp Geometry/index_proximal_rvs.vtp Geometry/l_bofoot.vtp Geometry/l_fibula.vtp Geometry/l_foot.vtp Geometry/l_patella.vtp Geometry/l_pelvis.vtp Geometry/l_talus.vtp Geometry/little_distal_lvs.vtp Geometry/little_distal_rvs.vtp Geometry/little_medial_lvs.vtp Geometry/little_medial_rvs.vtp Geometry/little_proximal_lvs.vtp Geometry/little_proximal_rvs.vtp Geometry/lunate_lvs.vtp Geometry/lunate_rvs.vtp Geometry/metacarpal1_lvs.vtp Geometry/metacarpal1_rvs.vtp Geometry/metacarpal2_lvs.vtp Geometry/metacarpal2_rvs.vtp Geometry/metacarpal3_lvs.vtp Geometry/metacarpal3_rvs.vtp Geometry/metacarpal4_lvs.vtp Geometry/metacarpal4_rvs.vtp Geometry/metacarpal5_lvs.vtp Geometry/metacarpal5_rvs.vtp Geometry/middle_distal_lvs.vtp Geometry/middle_distal_rvs.vtp Geometry/middle_medial_lvs.vtp Geometry/middle_medial_rvs.vtp Geometry/middle_proximal_lvs.vtp Geometry/middle_proximal_rvs.vtp Geometry/pisiform_lvs.vtp Geometry/pisiform_rvs.vtp Geometry/r_bofoot.vtp Geometry/r_fibula.vtp Geometry/r_foot.vtp Geometry/r_patella.vtp Geometry/r_pelvis.vtp Geometry/r_talus.vtp Geometry/radius_lv.vtp Geometry/radius_rv.vtp Geometry/ring_distal_lvs.vtp Geometry/ring_distal_rvs.vtp Geometry/ring_medial_lvs.vtp Geometry/ring_medial_rvs.vtp Geometry/ring_proximal_lvs.vtp Geometry/ring_proximal_rvs.vtp Geometry/sacrum.vtp Geometry/scaphoid_lvs.vtp Geometry/scaphoid_rvs.vtp Geometry/thumb_distal_lvs.vtp Geometry/thumb_distal_rvs.vtp Geometry/thumb_proximal_lvs.vtp Geometry/thumb_proximal_rvs.vtp Geometry/tibia_l.vtp Geometry/tibia_r.vtp Geometry/trapezium_lvs.vtp Geometry/trapezium_rvs.vtp Geometry/trapezoid_lvs.vtp Geometry/trapezoid_rvs.vtp Geometry/triquetrum_lvs.vtp Geometry/triquetrum_rvs.vtp Geometry/ulna_lv.vtp Geometry/ulna_rv.vtp helper.py ik_streaming.py load.ipynb opensim-build/opensim-core-linux-build-script.sh opensim-build/opensim-core-linux-build-script2.sh Rajagopal_2015.osim recordings/demo/raw_imu_0.npy recordings/demo/raw_imu_1.npy recordings/demo/raw_imu_2.npy recordings/demo/recording_0.sto recordings/demo/recording_1.sto recordings/demo/timestamp_0.npy recordings/demo/timestamp_1.npy regular_2.csv requirements.txt settings.txt setup_gpio.sh smolVLM.py start_capy_agent.py templates/_macros.html templates/base.html templates/index.html templates/messages.html templates/platoon.html test_guardian.py tiny_file.sto workers.py ``` ### Dependencies - requirements.txt: flask, numpy, opencv-python, pandas, pillow, python-dotenv, scrapybara ### Recent commits (newest first) - final commit - Final demo website - Finished capy guardian and another bug fix - smol change - Finished guardian updated settings for IMU - created requirements.txt - demo vids - Merge branch 'main' of github.com:PatP15/TreeHacks2025 - Added demo and finish capy example - Merge branch 'main' of github.com:PatP15/TreeHacks2025 into main - more trials - Updated settings - Added demo vid - trying to fix camera on jetson - Tried some recordings - added more geometries - Merge branch 'main' of github.com:PatP15/TreeHacks2025 - fix osim model and remove smolvlm - Last bug for IK streaming - Added realtime kin files ## Key source files (fetched from GitHub, selected and truncated for size) ### requirements.txt ``` opencv-python numpy pandas flask scrapybara pillow python-dotenv ``` ### app.py ```python #!/usr/bin/env python3 import cv2 import time import threading from flask import Flask, render_template, request, jsonify, Response, redirect, url_for # import torch # from transformers import AutoProcessor, AutoModelForVision2Seq, BitsAndBytesConfig from PIL import Image from datetime import datetime ########################################## # 1) INITIALIZE FLASK ########################################## app = Flask(__name__) ########################################## # 3) CAMERA SETUP ########################################## cap = cv2.VideoCapture(0) if not cap.isOpened(): print("Error: Cannot open camera.") exit(1) ########################################## # 6) MJPEG STREAM GENERATOR ########################################## def gen_frames(): """ Yields horizontally flipped MJPEG frames for the browser. Updates global 'latest_frame' and 'latest_frame_id' each time. """ global latest_frame, latest_frame_id while True: success, frame = cap.read() if not success: break frame = cv2.flip(frame, 1) ret, buffer = cv2.imencode('.jpg', frame) if not ret: continue yield (b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + buffer.tobytes() + b'\r\n') ########################################## # 7) FLASK ROUTES ########################################## @app.route('/') def home(): return redirect(url_for('platoon')) @app.route('/platoon') def platoon(): """ Load the Platoon page (platoon.html). """ return render_template('platoon.html') @app.route('/messages') def messages(): """ Load the Messages page (messages.html). """ return render_template('messages.html') @app.route('/video_feed') def video_feed(): """ Live camera stream (MJPEG). """ return Response(gen_frames(), mimetype='multipart/x-mixed-replace; boundary=frame') @app.route('/realtimekin_feed') def realtimekin_feed(): """ Streams the RealTime Kin feed (video file). """ return Response(gen_frames(), mimetype='multipart/x-mixed-replace; boundary=frame') import random @app.route('/location.json') def location_json(): """ Returns the main (green) location and five red flashing dots around a home position. """ return jsonify({ "green": { "x": 0.7, "y": 0.6 }, "red": [ {"x": 0.7, "y": 0.8}, # Center position {"x": 0.8, "y": 0.7}, {"x": 0.3, "y": 0.2}, {"x": 0.21, "y": 0.3}, {"x": 0.4, "y": 0.21} ] }) import numpy as np from scipy.signal import find_peaks import pandas as pd def process_ppg_data(file_path): """ Processes the original CSV file to extract and normalize the PPG (IR) signal. Args: file_path (str): Path to the CSV file. Returns: pd.DataFrame: Processed DataFrame with 'timestamp' and 'ppg_signal' (normalized). """ import pandas as pd # Load CSV file df = pd.read_csv(file_path) # Extract IR signal (assuming it's the best for PPG analysis) ppg_signal = df["IR"].values # Normalize PPG signal between 0 and 1 for smoother visualization ppg_signal_normalized = (ppg_signal - ppg_signal.min()) / (ppg_signal.max() - ppg_signal.min()) # Create a cleaned DataFrame ppg_df = pd.DataFrame({ "timestamp": df["sample_index"].values, # Keeping original timestamps "ppg_signal": ppg_signal_normalized # Normalized PPG signal }) return ppg_df import subprocess @app.route('/ppg_data') def ppg_data(): """ Returns either healthy or AFib PPG data based on request. """ file_type = request.args.get("type", "healthy") # Default to healthy file_path = "regular_2.csv" if file_type == "healthy" else "afib_2.csv" # Load the selected PPG dataset ppg_df = process_ppg_data(file_path) return jsonify(ppg_df["ppg_signal"].tolist()) @app.route('/run_script') def run_script(): """ Runs a background Python script when AFib is detected. """ subprocess.Popen(["python3", "test_guardian.py"]) return jsonify({"status": "Script triggered"}) @app.route('/chatbot', methods=['POST']) def chatbot(): """ Fake chatbot: echoes user message. """ data = request.get_json() # msg = data.get('message', '') now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") time.sleep(1) # Simulate processing time return jsonify({"response": f"Received message on {now}"}) ########################################## # 8) START THE APP & THREAD ########################################## if __name__ == '__main__': # threading.Thread(target=caption_loop, daemon=True).start() app.run(host='0.0.0.0', port=5001, debug=True) ``` ### setup_gpio.sh ```shell set -e # Exit immediately if a command fails echo "=== Starting GPIO permission fix ===" # 1) Create gpio group if it doesn't exist if ! getent group gpio >/dev/null; then echo "Creating 'gpio' group..." sudo groupadd gpio else echo "'gpio' group already exists." fi # 2) Add current user to gpio group CURRENT_USER="$(whoami)" echo "Adding user '$CURRENT_USER' to group 'gpio'..." sudo usermod -aG gpio "$CURRENT_USER" # 3) Temporary fix: set the group ownership and permissions on /dev/gpiochip* # This will be overwritten at next reboot, so we also do a udev rule below. if ls /dev/gpiochip* >/dev/null 2>&1; then echo "Updating group ownership and permissions for /dev/gpiochip* ..." sudo chown root:gpio /dev/gpiochip* sudo chmod 660 /dev/gpiochip* else echo "Warning: No /dev/gpiochip* devices found. Are you sure you're on a Jetson or have GPIO enabled?" fi sudo chmod 777 /dev/video0 ``` ### smolVLM.py ```python import torch from transformers import AutoProcessor, AutoModelForVision2Seq import PIL DEVICE = "cuda" if torch.cuda.is_available() else "mps" # Initialize processor and model processor = AutoProcessor.from_pretrained("HuggingFaceTB/SmolVLM-256M-Instruct") model = AutoModelForVision2Seq.from_pretrained( "HuggingFaceTB/SmolVLM-256M-Instruct", torch_dtype=torch.bfloat16, _attn_implementation="flash_attention_2" if DEVICE == "cuda" else "eager", ) print("Loaded model and processor") # Create input messages messages = [ { "role": "user", "content": [ {"type": "image"}, {"type": "text", "text": "Can you describe this image?"} ] }, ] print(messages) image_path = "./pic.jpeg" print("Processing image...") image = PIL.Image.open(image_path) # Preprocess prompt = processor.apply_chat_template(messages, add_generation_prompt=True) inputs = processor(text=prompt, images=[image], return_tensors="pt") print("Generating caption...") # Generate generated_ids = model.generate(**inputs, max_new_tokens=50) generated_texts = processor.batch_decode( generated_ids, skip_special_tokens=True, ) print(generated_texts) ``` ### start_capy_agent.py ```python from scrapybara import Scrapybara from scrapybara.anthropic import Anthropic from scrapybara.tools import BashTool, ComputerTool, EditTool, BrowserTool from scrapybara.prompts import BROWSER_SYSTEM_PROMPT from pydantic import BaseModel from typing import List from dotenv import load_dotenv import os load_dotenv() # Define schemas for structured output class Company(BaseModel): name: str description: str tags: List[str] class Companies(BaseModel): companies: List[Company] class ContactInfo(BaseModel): contact_method: str contact_details: str def handle_step(step): print(f"\nStep output: {step.text}") if step.tool_calls: for call in step.tool_calls: print(f"Tool used: {call.tool_name}") if step.usage: print(f"Tokens used: {step.usage.total_tokens}") def main(): client = Scrapybara( api_key=os.getenv("SCRAPYBARA_API_KEY", "YOUR_API_KEY"), timeout=600, ) # 1. Start instance print("Starting Scrapybara instance...") instance = client.start_browser() # instance.browser.start() # print("Started Scrapybara instance!") try: # Set up tools and model tools = [ BashTool(instance), ComputerTool(instance), EditTool(instance), BrowserTool(instance), ] model = Anthropic() # 2. Scrape W25 companies companies_response = client.act( model=model, tools=tools, system=BROWSER_SYSTEM_PROMPT, prompt="Go to https://ycombinator.com/companies, set batch filter to W25, and scrape all W25 companies, don't evaluate any code, just look at the HTML and return structured data and take a screenshot.", schema=Companies, on_step=handle_step, ) print(f"\nScraped W25 companies: {companies_response.output.companies}") COMPANY_LIMIT = 3 companies = companies_response.output.companies[:COMPANY_LIMIT] # 3. Find best way to contact each company for company in companies: print(f"\nFinding contact info for {company.name}...") contact_response = client.act( model=model, tools=tools, system=BROWSER_SYSTEM_PROMPT, prompt=f"Go to https://ycombinator.com/companies and find the best way to contact YC W25 company {company.name} - {company.description}. Try their website, LinkedIn, and Twitter/X.", schema=ContactInfo, on_step=handle_step, ) print(f"\nFound contact info for {company.name}: {contact_response.output}") # 4. Draft messages for every company client.act( model=model, tools=tools, system=BROWSER_SYSTEM_PROMPT, prompt=f"Open LibreOffice, draft a two sentence message to each of the following YC W25 companies, advertising a capybara zoo in Japan: {companies}", on_step=handle_step, ) print("\nDrafted messages") finally: # 5. Clean up instance.stop() if __name__ == "__main__": main() ``` ### app_demo.py ```python #!/usr/bin/env python3 import cv2 import time import threading import subprocess import random import numpy as np import pandas as pd from flask import Flask, render_template, request, jsonify, Response, redirect, url_for from datetime import datetime from PIL import Image ########################################## # 1) INITIALIZE FLASK ########################################## app = Flask(__name__) ########################################## # 2) VIDEO STREAMING SETUP (Using Video Files) ########################################## video_camera_path = "demo_vids/camera.mp4" video_realtimekin_path = "demo_vids/kin.mov" cap_camera = cv2.VideoCapture(video_camera_path) cap_realtimekin = cv2.VideoCapture(video_realtimekin_path) fps_kin = cap_realtimekin.get(cv2.CAP_PROP_FPS) skip_frames_kin = int(4 * fps_kin) cap_realtimekin.set(cv2.CAP_PROP_POS_FRAMES, skip_frames_kin) total_frames_kin = int(cap_realtimekin.get(cv2.CAP_PROP_FRAME_COUNT)) end_frame_kin = total_frames_kin - int(4 * fps_kin) def gen_frames(cap): fps = cap.get(cv2.CAP_PROP_FPS) if fps <= 0: fps = 30 while True: success, frame = cap.read() if not success: cap.set(cv2.CAP_PROP_POS_FRAMES, 0) continue cv2.waitKey(int(1000 // fps)) ret, buffer = cv2.imencode('.jpg', frame) if not ret: continue yield (b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + buffer.tobytes() + b'\r\n') ########################################## # 3) FLASK ROUTES ########################################## @app.route('/') def home(): return redirect(url_for('platoon')) @app.route('/platoon') def platoon(): """ Load the Platoon page (platoon.html). """ return render_template('platoon.html') @app.route('/messages') def messages(): """ Load the Messages page (messages.html). """ return render_template('messages.html') @app.route('/video_feed') def video_feed(): """ Streams the camera feed (video file). """ return Response(gen_frames(cap_camera), mimetype='multipart/x-mixed-replace; boundary=frame') @app.route('/realtimekin_feed') def realtimekin_feed(): """ Streams the RealTime Kin feed (video file). """ return Response(gen_frames(cap_realtimekin), mimetype='multipart/x-mixed-replace; boundary=frame') ########################################## # 4) MAP TRACKING & PPG PROCESSING ########################################## @app.route('/location.json') def location_json(): """ Returns the main (green) location and five red flashing dots. """ return jsonify({ "green": { "x": 0.7, "y": 0.6 }, "red": [ {"x": 0.7, "y": 0.8}, {"x": 0.8, "y": 0.7}, {"x": 0.3, "y": 0.2}, {"x": 0.21, "y": 0.3}, {"x": 0.4, "y": 0.21} ] }) def process_ppg_data(file_path): """ Extract and normalize PPG signal from CSV file. """ df = pd.read_csv(file_path) ppg_signal = df["IR"].values ppg_signal_normalized = (ppg_signal - ppg_signal.min()) / (ppg_signal.max() - ppg_signal.min()) return pd.DataFrame({"timestamp": df["sample_index"].values, "ppg_signal": ppg_signal_normalized}) @app.route('/ppg_data') def ppg_data(): """ Returns either Healthy or AFib PPG data based on request. """ file_type = request.args.get("type", "healthy") file_path = "regular_2.csv" if file_type == "healthy" else "afib_2.csv" ppg_df = process_ppg_data(file_path) return jsonify(ppg_df["ppg_signal"].tolist()) ########################################## # 5) AFib ALERT SYSTEM ########################################## @app.route('/run_script') def run_script(): """ Runs a background Python script when AFib is detected. """ subprocess.Popen(["python3", "test_guardian.py"]) return jsonify({"status": "Script triggered"}) ########################################## # 6) CHATBOT (FAKE RESPONSE) ########################################## @app.route('/chatbot', methods=['POST']) def chatbot(): """ Fake chatbot: echoes user message with a timestamp. """ data = request.get_json() now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") time.sleep(1) # Simulate processing time return jsonify({"response": f"Received message on {now}"}) ########################################## # 7) START THE APP ########################################## if __name__ == '__main__': app.run(host='0.0.0.0', port=5001, debug=True) ``` ### helper.py ```python import numpy as np import ahrs import time from ahrs.common.orientation import q2R # Function to write data from line_ind to a .sto file def quat2sto_single(sensor_data, header_text, file_dir, t_step, rate, sensor_ind_list = [4,4,2,3,0,1]): num_sensors, _ = sensor_data.shape with open(file_dir, 'w') as f: # initial information to write to file f.write("DataRate={}\n".format(rate)) f.write("DataType=Quaternion\n") f.write("version=3\n") f.write("OpenSimVersion=4.2\n") f.write("endheader\n") f.write(header_text) f.write("{}".format(t_step)) for j in sensor_ind_list: f.write("\t{},{},{},{}".format(sensor_data[j,0],sensor_data[j,1],sensor_data[j,2],sensor_data[j,3])) f.write("\n") # Function to read sto file to load fake real time data in numpy array def sto2quat(file_dir, lines = 3, offset = 6, num_sensors=8): sensor_data = np.zeros((num_sensors,lines,4)) times = np.zeros(lines) with open(file_dir, 'r') as f: for i in range(lines+offset): line = f.readline() if i >= offset: row_ind = i - offset sect_tab = line.split('\t') for j, sect in enumerate(sect_tab): if j == 0: times[row_ind] = float(sect) else: subsect = sect.split(',') for k,val in enumerate(subsect): sensor_data[j-1,row_ind,k] = float(val) return times, sensor_data def compute_quat(all_data, len_sensor_list, quat_cal_offset, rot_inds, num_sensors=5, t_offset=0, signals_per_sensor=6, beta=0.2*np.ones(5), rate=60.0, verbose=False, beta_init=0.4): d2g = ahrs.common.DEG2RAD # Constant to convert degrees to radians Qi = np.zeros((num_sensors,quat_cal_offset,4)) Q = np.zeros((num_sensors,all_data.shape[0]-quat_cal_offset,4)) Qi[:,0,:] = np.array([0.7071, 0.7071, 0.0, 0.0]) z_neg_90 = np.array([[0,1.0,0],[-1.,0,0],[0,0,1.0]]) y_180 = np.array([[-1.0,0,0],[0,1.0,0],[0,0,-1.0]]) z_180 = np.array([[-1.0,0,0],[0,-1.0,0],[0,0,1.0]]) y_neg_90 = np.array([[0,0,-1.0],[0,1.0,0],[1.0,0,0]]) y_pos_90 = np.array([[0,0,1.0],[0,1.0,0],[-1.0,0,0]]) ankle_offset = -100.*d2g x_pos_ankle = np.array([[1.0,0,0],[0,np.cos(ankle_offset),-np.sin(ankle_offset)],[0,np.sin(ankle_offset),np.cos(ankle_offset)]]) hip_rot = np.matmul(y_neg_90,z_180) foot_rot = np.matmul(x_pos_ankle, hip_rot) r_leg_rot = z_neg_90 l_leg_rot = np.matmul(z_neg_90,y_180) rot_mats = np.zeros((len_sensor_list,3,3)) for i in range(len_sensor_list): # define rotation type if rot_inds[i] == 0: # hip, torso, head rot_mats[i,:,:] = hip_rot elif rot_inds[i] == 1: # left side rot_mats[i,:,:] = l_leg_rot elif rot_inds[i] == 2: # right side rot_mats[i,:,:] = r_leg_rot elif rot_inds[i] == 3: # foot rot_mats[i,:,:] = foot_rot for i in range(len_sensor_list): # processing quaternions one sensor at a time s_off = i*signals_per_sensor accel = np.matmul(all_data[:,s_off+t_offset:s_off+t_offset+3],rot_mats[i,:,:]) gyro = np.matmul(all_data[:,s_off+t_offset+3:s_off+t_offset+6],rot_mats[i,:,:]) #mag = np.matmul(all_data[:,s_off+t_offset+6:s_off+t_offset+9],rot_mats[i,:,:]) # calibrating the initial quaternion with a large beta_init value madgwick_i = ahrs.filters.Mahony(frequency=float(rate)) for t in range(1, quat_cal_offset): Qi[i,t,:] = madgwick_i.updateIMU(Qi[i,t-1,:], gyro[0,:], accel[0,:]) quat_ang = np.zeros(num_sensors) for i in range(len_sensor_list): quat_ang[i],_ = orientMat(q2R(np.array(Qi[i,-1,:]))) # multiply rot_mats by the gyro correction angle to correct to same heading as pelvis? Then mean_ang = np.mean(quat_ang) return Qi[:,-1,:], mean_ang, rot_mats def rotateY(th, R): Ry = np.array([[np.cos(th),0,np.sin(th)],[0,1,0],[-np.sin(th),0,np.cos(th)]]) newR = np.matmul(R,Ry) return newR def orientMat(R, num_angles=100, R_ref=np.array([[1.,0.,0. ],[0.,0.,-1.],[0.,1.,0.]])): angles = np.linspace(-np.pi,np.pi,num_angles) best_angle = -np.pi best_distance = np.inf for a in angles: R_mat = rotateY(a,R) norm = np.linalg.norm(R_ref - R_mat) if norm < best_distance: best_distance = norm best_angle = a return best_angle, best_distance ``` ### test_guardian.py ```python from scrapybara import Scrapybara from scrapybara.anthropic import Anthropic from scrapybara.tools import BashTool, BrowserTool, ComputerTool, EditTool from scrapybara.prompts import BROWSER_SYSTEM_PROMPT import os from dotenv import load_dotenv # Load environment variables from a .env file load_dotenv() # Initialize the Scrapybara client client = Scrapybara( api_key=os.getenv("SCRAPYBARA_API_KEY"), timeout=6000, # Set timeout to 600 seconds ) # Start an Ubuntu instance instance = client.start_ubuntu() try: # Initialize tools tools = [ BashTool(instance), ComputerTool(instance), EditTool(instance), BrowserTool(instance), ] # Initialize the Anthropic model model = Anthropic() # Define the repository URL and the script to run repo_url = "https://github.com/PatP15/TreeHacks2025.git" script_name = "app_demo.py" # Clone the GitHub repository output = instance.bash(command=f"git clone {repo_url}") print(f"Cloned repository: {repo_url}") print(output) # Extract the repository name from the URL repo_name = repo_url.split('/')[-1].replace('.git', '') # Navigate to the repository directory and run the Python script output = instance.bash(command=f"cd {repo_name} && pip install -r requirements.txt && python3 {script_name}") print(f"Executed script: {script_name} in {repo_name}") print(output) # Start the browser cdp_url = instance.browser.start().cdp_url print("Browser started.") from playwright.sync_api import sync_playwright # Navigate to the locally hosted web application playwright = sync_playwright().start() browser = playwright.chromium.connect_over_cdp(cdp_url) page = browser.new_page() page.goto("localhost:5001") # Define the prompt for the Anthropic model prompt = """ You are an AI assistant tasked with interacting with the web application that monitors a persons health through the dashboard. Your name is Capy Guardian (like a guardian angel). Your goal is to evaluate what is happening to the person and provide a write up to the user. Here are the steps you should follow: Take multiple pictures of what you see on the dashboard and describe the different sections of the website. From teh camera, try to determine if the person is in a safe environment and if there are any signs of distress. From the heartbeat signal, try to determine if the person is in a healthy state or if there are any abnormalities. From the map streaming, the user is the green dot and the red dots are other people. Try to determine if the person is in a safe location. The real-time kin is a video feed of the person's joint angles. Try to determine if the person is moving normally or if there are any abnormalities. Here is a brief description of the different sections of the website: 1. Dashboard View: - RealTime Kin: Shows real-time video/image feed - Chest Camera: Secondary camera view with caption capabilities - Map Streaming: Displays location data with what appears to be a facility map - Heartbeat Signal (PPG): Presumably for monitoring vital signs 2. Logs Section: - Contains "Capy Guardian Logs" - Has a text input field for logging messages - Includes a "Send" button for submitting log entries Also you may need to make the website full screen to see all the sections. You will have to scroll around the website to view all the sections and gather the necessary information! Please proceed step by step, describing your actions and observations. Once you have gathered all the necessary information, provide a summary of the person's health status by submitting a write-up to the user through the "Send" button in the Logs section. Also determine if the person would require any immediate assistance or medical attention. If so, make it very clear in your write-up. Specificaaly look at the video if it seems it is persepctive from the ground then person may have fallen. The heartbeat signal is red if there is afib. Do not ask for any additional information from the user. Use only the information available on the website. Once you submit the logs leave the browser open. Open a text editor and write a summary of the person's health status based on the information you have gathered. """ # Use the Anthropic model to interact with the website response = client.act( model=model, tools=tools, system=BROWSER_SYSTEM_PROMPT, prompt=prompt, on_step=lambda step: print(f"\nStep output: {step.text}"), ) # Output the final response from the model print("\nFinal response from the model:") print(response.output) finally: print("Stopping the instance...") # Stop the instance to free up resources # instance.stop() # print("Instance stopped.") ``` ### ik_streaming.py ```python #!/usr/bin/python3 # Estimates kinematics from IMU data processed into quaternions using a musculoskeletal model import opensim as osim from opensim import Vec3 import numpy as np from helper import quat2sto_single, sto2quat import helper as h import time import os import sys from multiprocessing import Process, Queue import workers # define the worker functions in this .py file def clear(q): try: while True: q.get_nowait() except: pass # Customize real-time kinematics for use by setting flag and looking at corresponding code below. real_time = True # set to True for using the kinematics in the python script for real-time applications # Parameters for IK solver fake_real_time = False # True to run offline, False to record data and run online log_temp = False # True to log CPU temperature data log_data = True # if true save all IK outputs, else only use those in reporter_list for easier custom coding home_dir = '/home/puma/TreeHacks2025/' # location of the main RealTimeKin folder uncal_model = 'Rajagopal_2015.osim' uncal_model_filename = home_dir + uncal_model model_filename = home_dir+'calibrated_' + uncal_model fake_online_data = home_dir+'recordings/'#test_data.npy'#'test_IMU_data.npy'#'MT_012005D6_009-001_orientations.sto' sto_filename = home_dir+'tiny_file.sto' visualize = True rate = 20.0 # samples hz of IMUs accuracy = 0.001 # value tuned for accurate and fast IK solver constraint_var = 10.0 # value tuned for accurate and fast IK solver init_time = 4.0 # seconds of data to initialize from # Initialize the quaternions signals_per_sensor = 6 file_cnt = 0 save_dir_init = home_dir+ 'recordings/' # appending folder name here save_file = '/recording_' ts_file = '/timestamp_' script_live = True q = Queue() # queue for IMU messages b = Queue() # queue for button messages imuProc = Process(target=workers.readIMU, args=(q, b, fake_online_data, init_time, signals_per_sensor, save_dir_init,home_dir)) imuProc.start() # spawning IMU process sensor_ind_list, rate, header_text, save_folder, save_folder, file_cnt, sim_len, fake_real_time, fake_data_len = b.get() save_dir = save_dir_init+save_folder+'/' # append the folder name here kin_store_size = sim_len + 10.0 sim_steps = int(sim_len*rate) dt = 1/rate while(script_live): while(q.qsize()>0): # clearing the queues that may have old messages q.get() while(b.qsize()>0): b.get() print("Ready to initialize...") init_time, Qi, head_err = q.get() # calibrate model and save quat2sto_single(Qi, header_text, sto_filename, 0., rate, sensor_ind_list) visualize_init = False sensor_to_opensim_rotations = Vec3(-np.pi/2,head_err,0) imuPlacer = osim.IMUPlacer(); imuPlacer.set_model_file(uncal_model_filename); imuPlacer.set_orientation_file_for_calibration(sto_filename); imuPlacer.set_sensor_to_opensim_rotations(sensor_to_opensim_rotations); imuPlacer.run(visualize_init); model = imuPlacer.getCalibratedModel(); model.printToXML(model_filename) # Initialize model rt_samples = int(kin_store_size*rate) #kin_mat = np.zeros((rt_samples, 39)) # 39 is the number of joints stored in the .sto files accessible at each time step time_vec = np.zeros((rt_samples,2)) coordinates = model.getCoordinateSet() ikReporter = osim.TableReporter() ikReporter.setName('ik_reporter') for coord in coordinates: if log_data: ikReporter.addToReport(coord.getOutput('value'),coord.getName()) model.addComponent(ikReporter) model.finalizeConnections # Initialize simulation quatTable = osim.TimeSeriesTableQuaternion(sto_filename) orientationsData = osim.OpenSenseUtilities.convertQuaternionsToRotations(quatTable) oRefs = osim.BufferedOrientationsReference() init_state = model.initSystem() mRefs = osim.MarkersReference() coordinateReferences = osim.SimTKArrayCoordinateReference() # new version # Initialize with something (e.g. time=0, first row): rowVecView = orientationsData.getNearestRow(0) rowVec0 = osim.RowVectorRotation(rowVecView) oRefs.putValues(0, rowVec0) if visualize: model.setUseVisualizer(True) model.initSystem() s0 = init_state ikSolver = osim.InverseKinematicsSolver(model, mRefs, oRefs, coordinateReferences, constraint_var) ikSolver.setAccuracy(accuracy) s0.setTime(0.) ikSolver.assemble(s0) if visualize: # initialize visualization model.getVisualizer().show(s0) model.getVisualizer().getSimbodyVisualizer().setShowSimTime(True) # IK solver loop t = 0 # number of steps st = 0. # timing simulation temp_data = [] add_time = 0. running = True start_sim_time = time.time() q.put(['received']) # tell IMUs to start passing real-time data print("Starting recording...") while(running): if (b.qsize() > 0) or t == sim_steps: # new button press so we should save the data and restart the sim if t == sim_steps: # tell IMUs to reset too b.put(["done"]) if log_data: ik_results = ikReporter.getTable() osim.STOFileAdapter.write(ik_results, save_dir+save_file+str(file_cnt)+'.sto') np.save(save_dir+ts_file+str(file_cnt)+'.npy', time_vec[:t,:]) if log_temp and not fake_real_time: np.save(save_dir+'/tempdata_'+str(file_cnt)+'.npy', temp_data) file_cnt += 1 print("Time used in IK:",st,"Total time:",time.time()-start_sim_time) time.sleep(0.5) if fake_real_time: print("Saved the offline files...") exit() else: break # exit loop and wait until button pressed for reset time_stamp, Qi = q.get() add_time = time.time() time_s = t*dt quat2sto_single(Qi, header_text, sto_filename, time_s, rate, sensor_ind_list) # store ne [truncated — 1491 more characters] ``` ### templates/_macros.html ```html {% macro camera_and_caption(width="100%", max_height="300px", caption_id="captionBox") %}
Camera Feed
{% endmacro %} {% macro ecg_graph_block(prefix="") %}
Live ECG
{% endmacro %} ``` [8 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]