@@ -136,6 +173,7 @@ function App() {
@@ -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"
>
diff --git a/vitallink/simulator/README.md b/vitallink/simulator/README.md
new file mode 100644
index 0000000..8784f3a
--- /dev/null
+++ b/vitallink/simulator/README.md
@@ -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!**
diff --git a/vitallink/simulator/config_system.py b/vitallink/simulator/config_system.py
new file mode 100644
index 0000000..845a0d5
--- /dev/null
+++ b/vitallink/simulator/config_system.py
@@ -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()
diff --git a/vitallink/simulator/main_runner.py b/vitallink/simulator/main_runner.py
new file mode 100644
index 0000000..df109dd
--- /dev/null
+++ b/vitallink/simulator/main_runner.py
@@ -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())
diff --git a/vitallink/simulator/wristband_config.yaml b/vitallink/simulator/wristband_config.yaml
new file mode 100644
index 0000000..12332bc
--- /dev/null
+++ b/vitallink/simulator/wristband_config.yaml
@@ -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"
diff --git a/vitallink/simulator/wristband_manager.py b/vitallink/simulator/wristband_manager.py
new file mode 100644
index 0000000..07d9384
--- /dev/null
+++ b/vitallink/simulator/wristband_manager.py
@@ -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("
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())