mai/vitallink-BS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

aadded simulation bands config with real bands for easy addition + inventory management

Browse Source
This commit is contained in:
Raika Furude 2025-10-18 17:21:13 -04:00
parent aa789b3431
commit 99a275f443
7 changed files with 1647 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

83
vitallink/frontend/dashboard/src/App.jsx
View File

@ -1,6 +1,8 @@
import React, { useState, useEffect } from 'react';
import { Activity, AlertCircle, Clock, Users, Bell, Heart, Thermometer, Wind, CheckCircle, UserX } from 'lucide-react';
const API_BASE = 'http://localhost:8000';
function App() {
const [patients, setPatients] = useState([]);
const [stats, setStats] = useState({
@ -12,6 +14,40 @@ function App() {
const [filter, setFilter] = useState('all');
useEffect(() => {
// Fetch data from backend
const fetchData = async () => {
try {
const queueResponse = await fetch(`${API_BASE}/api/queue`);
const queueData = await queueResponse.json();
const statsResponse = await fetch(`${API_BASE}/api/stats`);
const statsData = await statsResponse.json();
// Set patients from backend
setPatients(queueData.map(p => ({
patient_id: p.patient_id,
band_id: p.band_id,
name: p.name,
tier: p.tier,
priority_score: p.priority_score,
wait_time_minutes: p.wait_time_minutes,
last_hr: p.last_hr,
last_spo2: p.last_spo2,
last_temp: p.last_temp,
symptoms: [] // Backend doesn't send symptoms in queue endpoint
})));
setStats(statsData);
console.log(`✓ Fetched ${queueData.length} patients from backend`);
} catch (error) {
console.error('Failed to fetch from backend:', error);
console.log('⚠️ Using mock data as fallback');
generateMockPatients();
}
};
// Mock data fallback function
const generateMockPatients = () => {
const mockPatients = [
{
@ -85,18 +121,12 @@ function App() {
});
};
generateMockPatients();
// Initial fetch
fetchData();
// Poll every 3 seconds for updates
const interval = setInterval(fetchData, 3000);
const interval = setInterval(() => {
setPatients(prev => prev.map(p => ({
...p,
wait_time_minutes: p.wait_time_minutes + 1,
last_hr: Math.max(40, Math.min(180, p.last_hr + Math.floor(Math.random() * 5) - 2)),
last_spo2: Math.max(70, Math.min(100, p.last_spo2 + Math.floor(Math.random() * 3) - 1)),
last_temp: Math.max(35, Math.min(41, p.last_temp + (Math.random() * 0.2 - 0.1)))
})));
}, 3000);
return () => clearInterval(interval);
}, []);
@ -137,8 +167,19 @@ function App() {
return 'text-gray-700';
};
const handleDischarge = (patientId) => {
setPatients(prev => prev.filter(p => p.patient_id !== patientId));
const handleDischarge = async (patientId) => {
try {
await fetch(`${API_BASE}/api/patients/${patientId}/discharge`, {
method: 'POST',
});
console.log(`✓ Discharged patient ${patientId}`);
// Remove from local state
setPatients(prev => prev.filter(p => p.patient_id !== patientId));
} catch (error) {
console.error('Failed to discharge patient:', error);
// Still remove from UI even if backend fails
setPatients(prev => prev.filter(p => p.patient_id !== patientId));
}
};
const filteredPatients = patients
@ -275,13 +316,15 @@ function App() {
<span>•</span>
<span className="font-mono">{patient.band_id}</span>
</div>
<div className="flex flex-wrap gap-2 mt-2">
{patient.symptoms.map(symptom => (
<span key={symptom} className="bg-gray-100 text-gray-700 px-3 py-1 rounded-full text-xs font-medium">
{symptom}
</span>
))}
</div>
{patient.symptoms && patient.symptoms.length > 0 && (
<div className="flex flex-wrap gap-2 mt-2">
{patient.symptoms.map(symptom => (
<span key={symptom} className="bg-gray-100 text-gray-700 px-3 py-1 rounded-full text-xs font-medium">
{symptom}
</span>
))}
</div>
)}
</div>
</div>

68
vitallink/frontend/kiosk/src/App.jsx
View File

@ -1,6 +1,8 @@
import React, { useState } from 'react';
import { AlertCircle, CheckCircle, Clock, User } from 'lucide-react';
const API_BASE = 'http://localhost:8000';
function App() {
const [step, setStep] = useState('welcome');
const [formData, setFormData] = useState({
@ -11,6 +13,8 @@ function App() {
severity: 'moderate'
});
const [assignedBand, setAssignedBand] = useState(null);
const [isSubmitting, setIsSubmitting] = useState(false);
const [error, setError] = useState(null);
const symptoms = [
'Chest Pain', 'Difficulty Breathing', 'Severe Headache',
@ -28,17 +32,41 @@ function App() {
};
const handleSubmit = async () => {
// Simulate API call to backend
const patientId = `P${Date.now().toString().slice(-6)}`;
const bandId = `VitalLink-${Math.floor(Math.random() * 65536).toString(16).toUpperCase().padStart(4, '0')}`;
setAssignedBand({
patientId,
bandId,
station: Math.floor(Math.random() * 8) + 1
});
setStep('complete');
setIsSubmitting(true);
setError(null);
try {
console.log('Submitting check-in data:', formData);
const response = await fetch(`${API_BASE}/api/checkin`, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify(formData),
});
if (!response.ok) {
throw new Error(`Server returned ${response.status}: ${response.statusText}`);
}
const data = await response.json();
console.log('Check-in successful:', data);
setAssignedBand({
patientId: data.patient_id,
bandId: data.band_id,
station: Math.floor(Math.random() * 8) + 1
});
setStep('complete');
} catch (error) {
console.error('Check-in failed:', error);
setError(error.message);
alert(`Failed to check in: ${error.message}\n\nMake sure the backend is running at ${API_BASE}`);
} finally {
setIsSubmitting(false);
}
};
if (step === 'welcome') {
@ -87,6 +115,15 @@ function App() {
<div className="bg-white rounded-2xl shadow-2xl p-8">
<h2 className="text-3xl font-bold text-gray-800 mb-6">Patient Information</h2>
{error && (
<div className="bg-red-50 border-2 border-red-300 text-red-800 p-4 rounded-lg mb-6">
<div className="flex items-center gap-2">
<AlertCircle className="w-5 h-5" />
<p className="font-semibold">Error: {error}</p>
</div>
</div>
)}
<div className="space-y-6">
<div className="grid grid-cols-2 gap-4">
<div>
@ -136,6 +173,7 @@ function App() {
<button
key={symptom}
onClick={() => handleSymptomToggle(symptom)}
type="button"
className={`px-4 py-3 rounded-lg border-2 transition-all text-left font-medium ${
formData.symptoms.includes(symptom)
? 'bg-blue-100 border-blue-500 text-blue-700'
@ -157,6 +195,7 @@ function App() {
<button
key={level}
onClick={() => setFormData({...formData, severity: level})}
type="button"
className={`px-6 py-4 rounded-lg border-2 transition-all font-semibold capitalize ${
formData.severity === level
? level === 'severe'
@ -177,16 +216,18 @@ function App() {
<div className="flex gap-4 mt-8">
<button
onClick={() => setStep('welcome')}
type="button"
className="flex-1 px-6 py-4 border-2 border-gray-300 text-gray-700 rounded-xl font-semibold hover:bg-gray-50 transition-colors"
>
Back
</button>
<button
onClick={handleSubmit}
disabled={!formData.firstName || !formData.lastName || !formData.dob || formData.symptoms.length === 0}
disabled={!formData.firstName || !formData.lastName || !formData.dob || formData.symptoms.length === 0 || isSubmitting}
type="button"
className="flex-1 px-6 py-4 bg-blue-600 text-white rounded-xl font-semibold hover:bg-blue-700 transition-colors disabled:bg-gray-300 disabled:cursor-not-allowed"
>
Complete Check-In
{isSubmitting ? 'Checking In...' : 'Complete Check-In'}
</button>
</div>
</div>
@ -248,6 +289,7 @@ function App() {
setStep('welcome');
setFormData({ firstName: '', lastName: '', dob: '', symptoms: [], severity: 'moderate' });
setAssignedBand(null);
setError(null);
}}
className="text-blue-600 font-semibold hover:underline"
>

466
vitallink/simulator/README.md Normal file
View File

@ -0,0 +1,466 @@
# VitalLink Hardware Integration Guide
## 🎯 Overview
This system allows you to seamlessly switch between:
- **Simulated wristbands** - For testing without hardware
- **Real wristbands** - Physical BLE devices
- **Mixed mode** - Use both simultaneously!
## 📁 File Structure
```
vitallink/
├── simulator/
│ ├── wristband_simulator.py # Original simulator (still works)
│ ├── wristband_manager.py # NEW: Unified wristband manager
│ ├── config_system.py # NEW: Configuration management
│ └── main_runner.py # NEW: Main system runner
├── wristband_config.yaml # Configuration file (auto-created)
└── backend/
└── server.py # Backend API
```
## 🚀 Quick Start
### 1. Install Dependencies
```bash
cd ~/documents/school/capstone/vitallink-BS/vitallink
source .venv/bin/activate.fish # or .venv/bin/activate
# Install BLE support (for real hardware)
uv pip install bleak pyyaml
# Or with pip:
# pip install bleak pyyaml
```
### 2. Create Configuration
```bash
# Create the new files
cd simulator
nano wristband_manager.py # Copy code from artifact 1
nano config_system.py # Copy code from artifact 2
nano main_runner.py # Copy code from artifact 3
```
### 3. Run Configuration Tool
```bash
python config_system.py --inventory
```
This creates `wristband_config.yaml` with default simulated bands.
### 4. Start the System
```bash
# Start backend first
python backend/server.py
# In another terminal, start the wristband system
python simulator/main_runner.py
```
That's it! The system will:
1. ✅ Auto-check in 3 demo patients
2. ✅ Assign wristbands from config
3. ✅ Start monitoring and sending data
4. ✅ Update the dashboard in real-time
---
## 🔧 Configuration File
Edit `wristband_config.yaml`:
```yaml
# VitalLink Wristband Configuration
backend_url: "http://localhost:8000"
# Auto-scan for real wristbands on startup
auto_scan_ble: false
scan_timeout: 10.0
# Simulated Wristbands (for testing)
simulated_bands:
- band_id: "VitalLink-SIM1"
profile: "stable"
- band_id: "VitalLink-SIM2"
profile: "mild_anxiety"
- band_id: "VitalLink-SIM3"
profile: "deteriorating"
# Real Wristbands (Hardware)
real_bands:
# Add your real wristbands here:
# - band_id: "VitalLink-A3B2"
# ble_address: "D7:91:3F:9A:12:34"
# Prefer real bands over simulated when available
prefer_real_bands: false
```
---
## 🔍 Finding Real Wristbands
### Scan for BLE Devices
```bash
python config_system.py --scan
```
This will:
1. Scan for 15 seconds
2. Find wristbands advertising the VitalLink service UUID
3. Offer to add them to your config
### Manual BLE Address Discovery
If scanning doesn't work, find addresses manually:
```bash
# On Linux
bluetoothctl
scan on
# Look for devices starting with "VitalLink-"
# Note the MAC address
```
Then add to `wristband_config.yaml`:
```yaml
real_bands:
- band_id: "VitalLink-A3B2"
ble_address: "AA:BB:CC:DD:EE:FF" # Your device's MAC
```
---
## 🎮 Usage Modes
### Mode 1: Automatic (Default)
```bash
python simulator/main_runner.py
```
Automatically:
- Loads wristbands from config
- Checks in demo patients
- Assigns bands
- Starts monitoring
### Mode 2: Interactive
```bash
python simulator/main_runner.py --interactive
```
Menu-driven:
- Manually assign bands
- Scan for new devices
- Release bands
- View inventory
### Mode 3: Configuration Management
```bash
# View inventory
python config_system.py --inventory
# Scan for devices
python config_system.py --scan
# Add simulated band
python config_system.py --add-simulated
```
---
## 🔄 Switching Between Real and Simulated
### Using Only Simulated Bands
```yaml
# wristband_config.yaml
simulated_bands:
- band_id: "VitalLink-SIM1"
profile: "stable"
- band_id: "VitalLink-SIM2"
profile: "deteriorating"
real_bands: [] # Empty
prefer_real_bands: false
```
### Using Only Real Bands
```yaml
simulated_bands: [] # Empty
real_bands:
- band_id: "VitalLink-REAL1"
ble_address: "D7:91:3F:9A:12:34"
- band_id: "VitalLink-REAL2"
ble_address: "E1:84:7B:42:56:78"
prefer_real_bands: true
```
### Mixed Mode (1 Real + 2 Simulated)
```yaml
simulated_bands:
- band_id: "VitalLink-SIM1"
profile: "stable"
- band_id: "VitalLink-SIM2"
profile: "critical"
real_bands:
- band_id: "VitalLink-REAL1"
ble_address: "D7:91:3F:9A:12:34"
prefer_real_bands: true # Use real band first
```
---
## 🏥 Real Hardware Integration
### When You Get Physical Wristbands
1. **Power on the wristbands** (remove from charger)
2. **Scan for them:**
```bash
python config_system.py --scan
```
3. **Add to config** when prompted
4. **Set preference:**
```yaml
prefer_real_bands: true
```
5. **Run system:**
```bash
python simulator/main_runner.py
```
The system will:
- ✅ Automatically connect to real wristbands
- ✅ Subscribe to BLE notifications
- ✅ Decode packets according to your spec
- ✅ Send data to backend
- ✅ Fall back to simulated if real bands unavailable
### Packet Decoding
The `PacketDecoder` class handles your exact packet format:
```python
# Your 16-byte packet structure:
# Byte 0: version
# Byte 1-2: sequence number
# Byte 3-6: timestamp (ms)
# Byte 7: flags
# Byte 8: HR (bpm)
# Byte 9: SpO2 (%)
# Byte 10-11: Temperature * 100
# Byte 12-13: Activity * 100
# Byte 14: Checksum
# Byte 15: Reserved
```
No code changes needed - it's already implemented!
---
## 📊 Inventory Management
### View Current Status
```bash
python config_system.py --inventory
```
Output:
```
================================================================================
CONFIGURED WRISTBAND INVENTORY
================================================================================
Simulated Wristbands:
🟢 VitalLink-SIM1 | Profile: stable
🟢 VitalLink-SIM2 | Profile: deteriorating
Real Wristbands (Hardware):
🔵 VitalLink-REAL1 | BLE: D7:91:3F:9A:12:34
================================================================================
Total: 3 wristbands
================================================================================
```
### Add Wristbands
```bash
# Add simulated
python config_system.py --add-simulated
# Add real (after scanning)
python config_system.py --scan
```
### Remove Wristbands
Edit `wristband_config.yaml` and delete the entry.
---
## 🧪 Testing
### Test 1: Simulated Only
```bash
# Use default config (3 simulated bands)
python simulator/main_runner.py
```
Should see 3 patients with updating vitals.
### Test 2: Real Hardware
```bash
# Edit config to add real band
nano wristband_config.yaml
# Run
python simulator/main_runner.py
```
Watch for:
- "🔵 Connecting to real wristband..."
- "✓ Connected to VitalLink-REAL1"
- "✓ Subscribed to notifications"
### Test 3: Mixed Mode
```yaml
# 1 real + 2 simulated
simulated_bands:
- band_id: "VitalLink-SIM1"
profile: "stable"
- band_id: "VitalLink-SIM2"
profile: "critical"
real_bands:
- band_id: "VitalLink-REAL1"
ble_address: "AA:BB:CC:DD:EE:FF"
prefer_real_bands: true
```
First patient gets real band, others get simulated.
---
## 🔧 Troubleshooting
### "Bleak not installed"
```bash
uv pip install bleak
# or
pip install bleak
```
### "No wristbands found during scan"
1. Ensure wristbands are powered (off charger)
2. Check Bluetooth is enabled
3. Try increasing scan timeout in config:
```yaml
scan_timeout: 30.0
```
### "Failed to connect to wristband"
1. Check BLE address is correct
2. Ensure wristband is in range
3. Try re-pairing in system Bluetooth settings
4. Check wristband battery
### "Checksum failed"
The real wristband's packet format doesn't match. Check:
1. Byte order (little-endian)
2. Field sizes match your spec
3. Checksum calculation (sum of bytes 0-13 mod 256)
---
## 📝 Next Steps
1. **Test with simulated bands** ✅ (you've done this)
2. **Get physical wristbands**
3. **Scan and add to config**
4. **Test with 1 real wristband**
5. **Add more real wristbands**
6. **Deploy with full real hardware**
The system is designed to work seamlessly at every stage!
---
## 🎓 For Your Capstone Demo
### Demo Scenario 1: All Simulated (Safe)
- Use 5 simulated wristbands with different profiles
- Show various patient conditions
- Demonstrate deterioration in real-time
### Demo Scenario 2: Mixed (Impressive)
- 1-2 real wristbands you wear
- 3 simulated wristbands
- Show that system handles both seamlessly
### Demo Scenario 3: All Real (Ultimate)
- All physical wristbands
- Live patient simulation
- Production-ready demonstration
---
## 📚 Code Architecture
```
BaseWristband (Abstract)
├── RealWristband (BLE Hardware)
│ ├── Uses Bleak library
│ ├── Connects via BLE
│ ├── Receives notifications
│ └── Decodes packets
│
└── SimulatedWristband (Testing)
├── Uses existing simulator
├── Generates realistic vitals
└── Same interface as real
WristbandManager
├── Manages inventory
├── Assigns to patients
├── Starts/stops monitoring
└── Handles both types uniformly
```
The key: **Both types implement the same interface**, so the rest of your system doesn't care which it's using!
---
**Questions? Check the code comments or run with `--help` flag!**

256
vitallink/simulator/config_system.py Normal file
View File

@ -0,0 +1,256 @@
"""
VitalLink Configuration System
Easy configuration for managing wristband inventory
"""
import yaml
import json
from pathlib import Path
# ============================================================================
# CONFIGURATION FILE EXAMPLE
# ============================================================================
EXAMPLE_CONFIG = """
# VitalLink Wristband Configuration
# Edit this file to manage your wristband inventory
# Backend API URL
backend_url: "http://localhost:8000"
# Auto-scan for real wristbands on startup
auto_scan_ble: false
scan_timeout: 10.0
# Simulated Wristbands
# Add as many as you need for testing
simulated_bands:
- band_id: "VitalLink-SIM1"
profile: "stable"
- band_id: "VitalLink-SIM2"
profile: "mild_anxiety"
- band_id: "VitalLink-SIM3"
profile: "deteriorating"
# Real Wristbands (Hardware)
# Add BLE addresses of your physical wristbands
# You can find these by running: python -m wristband_manager --scan
real_bands:
# Example (uncomment and edit when you have real hardware):
# - band_id: "VitalLink-A3B2"
# ble_address: "D7:91:3F:9A:12:34"
#
# - band_id: "VitalLink-7B42"
# ble_address: "E1:84:7B:42:56:78"
# Default preference when assigning bands
prefer_real_bands: false # Set to true to use real bands first
# Patient profiles for simulated bands
# Options: stable, mild_anxiety, deteriorating, critical, sepsis
default_profile: "stable"
"""
class WristbandConfig:
"""Configuration manager for wristbands"""
def __init__(self, config_path: str = "wristband_config.yaml"):
self.config_path = Path(config_path)
self.config = {}
if not self.config_path.exists():
self.create_default_config()
self.load()
def create_default_config(self):
"""Create default configuration file"""
print(f"Creating default config at {self.config_path}")
with open(self.config_path, "w") as f:
f.write(EXAMPLE_CONFIG)
def load(self):
"""Load configuration from file"""
try:
with open(self.config_path, "r") as f:
self.config = yaml.safe_load(f) or {}
print(f"✓ Loaded configuration from {self.config_path}")
except Exception as e:
print(f"⚠️ Could not load config: {e}")
print("Using default configuration")
self.config = {}
def get(self, key: str, default=None):
"""Get configuration value"""
return self.config.get(key, default)
def get_simulated_bands(self):
"""Get list of simulated bands from config"""
return self.config.get("simulated_bands", [])
def get_real_bands(self):
"""Get list of real bands from config"""
return self.config.get("real_bands", [])
def add_discovered_band(self, band_id: str, ble_address: str):
"""Add a discovered real band to config"""
if "real_bands" not in self.config:
self.config["real_bands"] = []
# Check if already exists
for band in self.config["real_bands"]:
if band.get("ble_address") == ble_address:
return
self.config["real_bands"].append(
{"band_id": band_id, "ble_address": ble_address}
)
self.save()
print(f"✓ Added {band_id} to configuration")
def save(self):
"""Save configuration to file"""
try:
with open(self.config_path, "w") as f:
yaml.safe_dump(
self.config, f, default_flow_style=False, sort_keys=False
)
print(f"✓ Saved configuration to {self.config_path}")
except Exception as e:
print(f"❌ Failed to save config: {e}")
# ============================================================================
# COMMAND LINE INTERFACE
# ============================================================================
async def cli_scan():
"""CLI: Scan for real wristbands"""
from wristband_manager import WristbandManager
print("Scanning for VitalLink wristbands...")
manager = WristbandManager()
config = WristbandConfig()
found = await manager.scan_for_real_bands(timeout=15.0)
if found:
print(f"\nFound {len(found)} wristband(s):")
for band_id in found:
band = manager.inventory[band_id]
if hasattr(band, "ble_address"):
print(f" {band_id}: {band.ble_address}")
# Ask to add to config
response = input(f"Add {band_id} to config? (y/n): ")
if response.lower() == "y":
config.add_discovered_band(band_id, band.ble_address)
else:
print("No wristbands found")
print("\nTroubleshooting:")
print(" 1. Make sure wristbands are powered on")
print(" 2. Remove wristbands from chargers")
print(" 3. Ensure Bluetooth is enabled on this computer")
def cli_inventory():
"""CLI: Show current inventory"""
config = WristbandConfig()
print("\n" + "=" * 80)
print("CONFIGURED WRISTBAND INVENTORY")
print("=" * 80)
print("\nSimulated Wristbands:")
simulated = config.get_simulated_bands()
if simulated:
for band in simulated:
print(f" 🟢 {band['band_id']:20} | Profile: {band['profile']}")
else:
print(" (none configured)")
print("\nReal Wristbands (Hardware):")
real = config.get_real_bands()
if real:
for band in real:
print(f" 🔵 {band['band_id']:20} | BLE: {band['ble_address']}")
else:
print(" (none configured)")
print("\n" + "=" * 80)
print(f"Total: {len(simulated) + len(real)} wristbands")
print("=" * 80 + "\n")
def cli_add_simulated():
"""CLI: Add a simulated wristband"""
config = WristbandConfig()
print("\nAdd Simulated Wristband")
print("-" * 40)
band_id = input("Band ID (e.g., VitalLink-SIM4): ")
print("\nAvailable profiles:")
print(" 1. stable - Normal vitals")
print(" 2. mild_anxiety - Elevated HR")
print(" 3. deteriorating - Gradually worsening")
print(" 4. critical - Severe vitals")
print(" 5. sepsis - Rapid deterioration")
profile_choice = input("\nSelect profile (1-5): ")
profiles = ["stable", "mild_anxiety", "deteriorating", "critical", "sepsis"]
profile = (
profiles[int(profile_choice) - 1]
if profile_choice.isdigit() and 1 <= int(profile_choice) <= 5
else "stable"
)
if "simulated_bands" not in config.config:
config.config["simulated_bands"] = []
config.config["simulated_bands"].append({"band_id": band_id, "profile": profile})
config.save()
print(f"\n✓ Added {band_id} ({profile})")
if __name__ == "__main__":
import argparse
import sys
# Check if yaml is installed
try:
import yaml
except ImportError:
print("❌ PyYAML not installed. Install with: pip install pyyaml")
sys.exit(1)
parser = argparse.ArgumentParser(
description="VitalLink Wristband Configuration Tool"
)
parser.add_argument("--scan", action="store_true", help="Scan for real wristbands")
parser.add_argument(
"--inventory", action="store_true", help="Show configured inventory"
)
parser.add_argument(
"--add-simulated", action="store_true", help="Add a simulated wristband"
)
args = parser.parse_args()
if args.scan:
import asyncio
asyncio.run(cli_scan())
elif args.inventory:
cli_inventory()
elif args.add_simulated:
cli_add_simulated()
else:
parser.print_help()

244
vitallink/simulator/main_runner.py Normal file
View File

@ -0,0 +1,244 @@
"""
VitalLink Main System Runner
Runs the complete system with real and/or simulated wristbands
"""
import asyncio
import aiohttp
from wristband_manager import WristbandManager
from config_system import WristbandConfig
# ============================================================================
# MAIN SYSTEM
# ============================================================================
class VitalLinkSystem:
"""Main system orchestrator"""
def __init__(self):
self.manager = WristbandManager()
self.config = WristbandConfig()
self.backend_url = self.config.get("backend_url", "http://localhost:8000")
self.running = False
async def initialize(self):
"""Initialize the system"""
print("\n" + "=" * 80)
print("VitalLink System Initialization")
print("=" * 80 + "\n")
# Check backend availability
await self.check_backend()
# Scan for real wristbands if configured
if self.config.get("auto_scan_ble", False):
timeout = self.config.get("scan_timeout", 10.0)
await self.manager.scan_for_real_bands(timeout)
# Load configured real wristbands
for band_config in self.config.get_real_bands():
self.manager.add_real_band(
band_config["band_id"], band_config["ble_address"]
)
# Load configured simulated wristbands
for band_config in self.config.get_simulated_bands():
self.manager.add_simulated_band(
band_config["band_id"], band_config.get("profile", "stable")
)
# Show inventory
self.manager.print_inventory()
async def check_backend(self):
"""Check if backend is running"""
try:
async with aiohttp.ClientSession() as session:
async with session.get(f"{self.backend_url}/") as resp:
if resp.status == 200:
print(f"✓ Backend is running at {self.backend_url}")
return True
except Exception as e:
print(f"❌ Backend not reachable at {self.backend_url}")
print(f" Error: {e}")
print("\n⚠️ Start backend with: python backend/server.py")
return False
async def auto_checkin_and_assign(self):
"""Automatically check in patients and assign available bands"""
# Mock patients for demo
demo_patients = [
{
"firstName": "John",
"lastName": "Smith",
"dob": "1985-03-15",
"symptoms": ["Chest Pain"],
"severity": "mild",
},
{
"firstName": "Sarah",
"lastName": "Johnson",
"dob": "1990-07-22",
"symptoms": ["Fever", "Difficulty Breathing"],
"severity": "moderate",
},
{
"firstName": "Michael",
"lastName": "Chen",
"dob": "1978-11-05",
"symptoms": ["Severe Headache"],
"severity": "severe",
},
]
print("\nAuto check-in patients...")
prefer_real = self.config.get("prefer_real_bands", False)
for patient_data in demo_patients:
try:
# Check in patient via API
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.backend_url}/api/checkin", json=patient_data
) as resp:
if resp.status == 200:
data = await resp.json()
patient_id = data["patient_id"]
assigned_band_id = data["band_id"]
print(
f"✓ {patient_data['firstName']} {patient_data['lastName']} → {patient_id}"
)
# Find and assign a physical/simulated band
band = self.manager.assign_band(
patient_id, prefer_real=prefer_real
)
if band:
# Start monitoring
await self.manager.start_monitoring(band.band_id)
except Exception as e:
print(f"❌ Failed to check in {patient_data['firstName']}: {e}")
print()
async def run(self):
"""Run the main system"""
self.running = True
await self.initialize()
await self.auto_checkin_and_assign()
print("=" * 80)
print("VitalLink System Running")
print("=" * 80)
print("\nMonitoring patients... Press Ctrl+C to stop\n")
try:
# Keep running until interrupted
while self.running:
await asyncio.sleep(10)
# Periodic status update
status = self.manager.get_status()
print(
f"[{asyncio.get_event_loop().time():.0f}s] Active: {status['active_monitoring']} | "
f"Available: {status['status_breakdown']['available']}"
)
except KeyboardInterrupt:
print("\n\nShutting down...")
await self.shutdown()
async def shutdown(self):
"""Clean shutdown"""
self.running = False
# Stop all monitoring
for band_id in list(self.manager.active_monitoring.keys()):
await self.manager.stop_monitoring(band_id)
print("\n✓ VitalLink system stopped")
self.manager.print_inventory()
# ============================================================================
# CLI MODES
# ============================================================================
async def interactive_mode():
"""Interactive mode with menu"""
system = VitalLinkSystem()
await system.initialize()
while True:
print("\n" + "=" * 80)
print("VITALLINK INTERACTIVE MODE")
print("=" * 80)
print("\n1. Show inventory")
print("2. Scan for real wristbands")
print("3. Assign band to patient")
print("4. Release band")
print("5. Start auto-demo")
print("6. Exit")
choice = input("\nSelect option: ")
if choice == "1":
system.manager.print_inventory()
elif choice == "2":
await system.manager.scan_for_real_bands(timeout=10.0)
elif choice == "3":
patient_id = input("Patient ID: ")
prefer_real = input("Prefer real band? (y/n): ").lower() == "y"
band = system.manager.assign_band(patient_id, prefer_real=prefer_real)
if band:
await system.manager.start_monitoring(band.band_id)
elif choice == "4":
band_id = input("Band ID to release: ")
await system.manager.release_band(band_id)
elif choice == "5":
await system.run()
break
elif choice == "6":
print("Goodbye!")
break
# ============================================================================
# MAIN ENTRY POINT
# ============================================================================
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="VitalLink System")
parser.add_argument(
"--interactive", "-i", action="store_true", help="Run in interactive mode"
)
parser.add_argument(
"--config",
"-c",
default="wristband_config.yaml",
help="Configuration file path",
)
args = parser.parse_args()
if args.interactive:
asyncio.run(interactive_mode())
else:
# Normal automatic mode
system = VitalLinkSystem()
asyncio.run(system.run())

40
vitallink/simulator/wristband_config.yaml Normal file
View File

@ -0,0 +1,40 @@
# VitalLink Wristband Configuration
# Edit this file to manage your wristband inventory
# Backend API URL
backend_url: "http://localhost:8000"
# Auto-scan for real wristbands on startup
auto_scan_ble: false
scan_timeout: 10.0
# Simulated Wristbands
# Add as many as you need for testing
simulated_bands:
- band_id: "VitalLink-SIM1"
profile: "stable"
- band_id: "VitalLink-SIM2"
profile: "mild_anxiety"
- band_id: "VitalLink-SIM3"
profile: "deteriorating"
# Real Wristbands (Hardware)
# Add BLE addresses of your physical wristbands
# You can find these by running: python -m wristband_manager --scan
real_bands:
# Example (uncomment and edit when you have real hardware):
# - band_id: "VitalLink-A3B2"
# ble_address: "D7:91:3F:9A:12:34"
#
# - band_id: "VitalLink-7B42"
# ble_address: "E1:84:7B:42:56:78"
# Default preference when assigning bands
prefer_real_bands: false # Set to true to use real bands first
# Patient profiles for simulated bands
# Options: stable, mild_anxiety, deteriorating, critical, sepsis
default_profile: "stable"

523
vitallink/simulator/wristband_manager.py Normal file
View File

@ -0,0 +1,523 @@
"""
VitalLink Wristband Management System
Unified system for managing real and simulated wristbands
"""
import asyncio
import struct
import time
import aiohttp
from typing import Dict, List, Optional, Union
from dataclasses import dataclass
from enum import Enum
from abc import ABC, abstractmethod
# Try to import BLE library (for real hardware)
try:
from bleak import BleakScanner, BleakClient
BLE_AVAILABLE = True
except ImportError:
BLE_AVAILABLE = False
print(
"⚠️ Bleak not installed. Real wristbands disabled. Install with: pip install bleak"
)
# ============================================================================
# CONFIGURATION
# ============================================================================
SERVICE_UUID = "8f5a84f1-22a8-4a4b-9b5f-3fe1d8b2a3a1"
CHAR_UUID = "d3e2c4b7-39b2-4b2a-8d5a-7d2a5e3f1199"
BACKEND_URL = "http://localhost:8000"
PKT_STRUCT = struct.Struct("<B H I B B B h H B B")
class WristbandType(Enum):
REAL = "real"
SIMULATED = "simulated"
class WristbandStatus(Enum):
AVAILABLE = "available" # Charged and ready
ASSIGNED = "assigned" # Checked out to patient
IN_USE = "in_use" # Actively sending data
CHARGING = "charging" # On charger
MAINTENANCE = "maintenance" # Needs attention
# ============================================================================
# PACKET DECODER (Works with real hardware packets)
# ============================================================================
class PacketDecoder:
"""Decodes BLE packets from real wristbands according to spec"""
@staticmethod
def decode(data: bytes) -> Optional[dict]:
"""Decode 16-byte packet from wristband"""
if len(data) != 16:
return None
# Verify checksum
checksum_calc = sum(data[0:14]) & 0xFF
if checksum_calc != data[14]:
print(f"⚠️ Checksum failed: expected {data[14]}, got {checksum_calc}")
return None
# Unpack according to spec
(
ver,
seq,
ts_ms,
flags,
hr_bpm,
spo2,
skin_c_x100,
act_rms_x100,
checksum,
rfu,
) = PKT_STRUCT.unpack(data)
# Determine tier from flags
tier = "NORMAL"
if flags & (1 << 4): # EMERGENCY bit
tier = "EMERGENCY"
elif flags & (1 << 3): # ALERT bit
tier = "ALERT"
return {
"ver": ver,
"seq": seq,
"ts_ms": ts_ms,
"tier": tier,
"hr_bpm": hr_bpm,
"spo2": spo2,
"temp_c": skin_c_x100 / 100.0,
"activity": act_rms_x100 / 100.0,
"flags": flags,
"checksum": f"0x{checksum:02X}",
}
# ============================================================================
# ABSTRACT WRISTBAND INTERFACE
# ============================================================================
class BaseWristband(ABC):
"""Abstract base class for all wristbands (real and simulated)"""
def __init__(self, band_id: str, wristband_type: WristbandType):
self.band_id = band_id
self.type = wristband_type
self.status = WristbandStatus.AVAILABLE
self.patient_id: Optional[str] = None
self.last_packet: Optional[dict] = None
self.packet_count = 0
@abstractmethod
async def start_monitoring(self):
"""Start receiving data from wristband"""
pass
@abstractmethod
async def stop_monitoring(self):
"""Stop receiving data"""
pass
def assign_to_patient(self, patient_id: str):
"""Assign wristband to a patient"""
self.patient_id = patient_id
self.status = WristbandStatus.ASSIGNED
print(f"✓ {self.band_id} assigned to patient {patient_id}")
def release(self):
"""Release wristband (return to inventory)"""
self.patient_id = None
self.status = WristbandStatus.AVAILABLE
self.packet_count = 0
print(f"✓ {self.band_id} released and available")
# ============================================================================
# REAL WRISTBAND (Hardware BLE)
# ============================================================================
class RealWristband(BaseWristband):
"""Real physical wristband using BLE"""
def __init__(self, band_id: str, ble_address: str):
super().__init__(band_id, WristbandType.REAL)
self.ble_address = ble_address
self.client: Optional[BleakClient] = None
self.decoder = PacketDecoder()
async def start_monitoring(self):
"""Connect to real wristband via BLE and start receiving data"""
if not BLE_AVAILABLE:
raise RuntimeError("Bleak library not available")
self.status = WristbandStatus.IN_USE
print(
f"🔵 Connecting to real wristband {self.band_id} at {self.ble_address}..."
)
try:
self.client = BleakClient(self.ble_address)
await self.client.connect()
print(f"✓ Connected to {self.band_id}")
# Subscribe to notifications
await self.client.start_notify(CHAR_UUID, self._notification_handler)
print(f"✓ Subscribed to notifications from {self.band_id}")
except Exception as e:
print(f"❌ Failed to connect to {self.band_id}: {e}")
self.status = WristbandStatus.MAINTENANCE
def _notification_handler(self, sender, data: bytearray):
"""Handle incoming BLE notifications"""
decoded = self.decoder.decode(bytes(data))
if decoded:
self.last_packet = decoded
self.packet_count += 1
# Send to backend
asyncio.create_task(self._send_to_backend(decoded))
async def _send_to_backend(self, decoded: dict):
"""Send decoded data to backend"""
if not self.patient_id:
return
payload = {
"band_id": self.band_id,
"patient_id": self.patient_id,
"timestamp": time.time(),
"ver": decoded["ver"],
"seq": decoded["seq"],
"ts_ms": decoded["ts_ms"],
"tier": decoded["tier"],
"flags": [], # Convert flags to list if needed
"hr_bpm": decoded["hr_bpm"],
"spo2": decoded["spo2"],
"temp_c": decoded["temp_c"],
"activity": decoded["activity"],
}
try:
async with aiohttp.ClientSession() as session:
await session.post(f"{BACKEND_URL}/api/vitals", json=payload)
except:
pass # Silently fail if backend unavailable
async def stop_monitoring(self):
"""Disconnect from wristband"""
if self.client and self.client.is_connected:
await self.client.stop_notify(CHAR_UUID)
await self.client.disconnect()
print(f"✓ Disconnected from {self.band_id}")
self.status = WristbandStatus.AVAILABLE
# ============================================================================
# SIMULATED WRISTBAND (For testing without hardware)
# ============================================================================
class SimulatedWristband(BaseWristband):
"""Simulated wristband for testing"""
def __init__(self, band_id: str, profile: str = "stable"):
super().__init__(band_id, WristbandType.SIMULATED)
self.profile = profile
self.seq = 0
self.running = False
# Import simulator profiles
from simulator.wristband_simulator import PATIENT_PROFILES, WristbandSimulator
self.simulator = WristbandSimulator(
band_id, PATIENT_PROFILES.get(profile, PATIENT_PROFILES["stable"]), None
)
async def start_monitoring(self):
"""Start simulated data generation"""
self.status = WristbandStatus.IN_USE
self.running = True
self.simulator.patient_id = self.patient_id
print(f"🟢 Starting simulated wristband {self.band_id} ({self.profile})")
while self.running:
# Generate packet
packet = self.simulator.generate_packet()
# Decode it
decoder = PacketDecoder()
decoded = decoder.decode(packet)
if decoded:
self.last_packet = decoded
self.packet_count += 1
# Send to backend
await self._send_to_backend(decoded)
# Wait based on tier
tier = self.simulator.tier
interval = 1.0 if tier in ["ALERT", "EMERGENCY"] else 60.0
await asyncio.sleep(interval)
async def _send_to_backend(self, decoded: dict):
"""Send to backend"""
if not self.patient_id:
return
payload = {
"band_id": self.band_id,
"patient_id": self.patient_id,
"timestamp": time.time(),
"ver": decoded["ver"],
"seq": decoded["seq"],
"ts_ms": decoded["ts_ms"],
"tier": decoded["tier"],
"flags": [],
"hr_bpm": decoded["hr_bpm"],
"spo2": decoded["spo2"],
"temp_c": decoded["temp_c"],
"activity": decoded["activity"],
}
try:
async with aiohttp.ClientSession() as session:
await session.post(f"{BACKEND_URL}/api/vitals", json=payload)
except:
pass
async def stop_monitoring(self):
"""Stop simulation"""
self.running = False
self.status = WristbandStatus.AVAILABLE
print(f"✓ Stopped simulated wristband {self.band_id}")
# ============================================================================
# WRISTBAND MANAGER
# ============================================================================
class WristbandManager:
"""Central manager for all wristbands (real and simulated)"""
def __init__(self):
self.inventory: Dict[str, BaseWristband] = {}
self.active_monitoring: Dict[str, asyncio.Task] = {}
def add_simulated_band(self, band_id: str, profile: str = "stable"):
"""Add a simulated wristband to inventory"""
band = SimulatedWristband(band_id, profile)
self.inventory[band_id] = band
print(f"➕ Added simulated band {band_id} ({profile})")
return band
def add_real_band(self, band_id: str, ble_address: str):
"""Add a real wristband to inventory"""
if not BLE_AVAILABLE:
print("❌ Cannot add real band: Bleak not installed")
return None
band = RealWristband(band_id, ble_address)
self.inventory[band_id] = band
print(f"➕ Added real band {band_id} (BLE: {ble_address})")
return band
async def scan_for_real_bands(self, timeout: float = 10.0):
"""Scan for real wristbands and add them to inventory"""
if not BLE_AVAILABLE:
print("❌ BLE scanning not available: Install bleak")
return []
print(f"🔍 Scanning for VitalLink wristbands ({timeout}s)...")
devices = await BleakScanner.discover(timeout=timeout)
found = []
for device in devices:
# Check if device advertises our service UUID
uuids = device.metadata.get("uuids", [])
if any(uuid.lower() == SERVICE_UUID.lower() for uuid in uuids):
band_id = (
device.name or f"VitalLink-{device.address[-5:].replace(':', '')}"
)
self.add_real_band(band_id, device.address)
found.append(band_id)
print(f"✓ Found {len(found)} real wristband(s)")
return found
def get_available_bands(self) -> List[BaseWristband]:
"""Get list of available (not assigned) wristbands"""
return [
b for b in self.inventory.values() if b.status == WristbandStatus.AVAILABLE
]
def assign_band(
self, patient_id: str, prefer_real: bool = False
) -> Optional[BaseWristband]:
"""Assign an available band to a patient"""
available = self.get_available_bands()
if not available:
print("❌ No wristbands available")
return None
# Prefer real bands if requested and available
if prefer_real:
real_bands = [b for b in available if b.type == WristbandType.REAL]
if real_bands:
band = real_bands[0]
else:
band = available[0]
else:
band = available[0]
band.assign_to_patient(patient_id)
return band
async def start_monitoring(self, band_id: str):
"""Start monitoring a wristband"""
if band_id not in self.inventory:
print(f"❌ Band {band_id} not in inventory")
return
band = self.inventory[band_id]
if band_id in self.active_monitoring:
print(f"⚠️ {band_id} already being monitored")
return
# Create monitoring task
task = asyncio.create_task(band.start_monitoring())
self.active_monitoring[band_id] = task
async def stop_monitoring(self, band_id: str):
"""Stop monitoring a wristband"""
if band_id in self.active_monitoring:
task = self.active_monitoring[band_id]
task.cancel()
del self.active_monitoring[band_id]
if band_id in self.inventory:
await self.inventory[band_id].stop_monitoring()
async def release_band(self, band_id: str):
"""Release band back to inventory"""
await self.stop_monitoring(band_id)
if band_id in self.inventory:
self.inventory[band_id].release()
def get_status(self) -> dict:
"""Get overall status"""
status_counts = {}
for status in WristbandStatus:
status_counts[status.value] = sum(
1 for b in self.inventory.values() if b.status == status
)
return {
"total_bands": len(self.inventory),
"real_bands": sum(
1 for b in self.inventory.values() if b.type == WristbandType.REAL
),
"simulated_bands": sum(
1 for b in self.inventory.values() if b.type == WristbandType.SIMULATED
),
"status_breakdown": status_counts,
"active_monitoring": len(self.active_monitoring),
}
def print_inventory(self):
"""Print current inventory"""
print("\n" + "=" * 80)
print("WRISTBAND INVENTORY")
print("=" * 80)
for band_id, band in self.inventory.items():
type_symbol = "🔵" if band.type == WristbandType.REAL else "🟢"
status_str = band.status.value.upper()
patient_str = f"(Patient: {band.patient_id})" if band.patient_id else ""
packets_str = (
f"[{band.packet_count} packets]" if band.packet_count > 0 else ""
)
print(
f"{type_symbol} {band_id:20} | {status_str:15} {patient_str:20} {packets_str}"
)
print("=" * 80)
status = self.get_status()
print(
f"Total: {status['total_bands']} | "
f"Real: {status['real_bands']} | "
f"Simulated: {status['simulated_bands']} | "
f"Active: {status['active_monitoring']}"
)
print("=" * 80 + "\n")
# ============================================================================
# EXAMPLE USAGE
# ============================================================================
async def main():
"""Example usage of wristband manager"""
manager = WristbandManager()
print("VitalLink Wristband Management System")
print("=" * 80)
# Option 1: Scan for real wristbands
# await manager.scan_for_real_bands(timeout=10.0)
# Option 2: Add simulated wristbands
manager.add_simulated_band("VitalLink-SIM1", "stable")
manager.add_simulated_band("VitalLink-SIM2", "mild_anxiety")
manager.add_simulated_band("VitalLink-SIM3", "deteriorating")
# Option 3: Manually add real wristband if you know the address
# manager.add_real_band("VitalLink-REAL1", "D7:91:3F:9A:12:34")
# Show inventory
manager.print_inventory()
# Assign bands to patients
band1 = manager.assign_band("P100001")
band2 = manager.assign_band("P100002")
# Start monitoring
if band1:
await manager.start_monitoring(band1.band_id)
if band2:
await manager.start_monitoring(band2.band_id)
# Monitor for 30 seconds
print("\nMonitoring for 30 seconds...")
await asyncio.sleep(30)
# Stop and release
if band1:
await manager.release_band(band1.band_id)
if band2:
await manager.release_band(band2.band_id)
manager.print_inventory()
if __name__ == "__main__":
asyncio.run(main())