566 lines
18 KiB
Python
566 lines
18 KiB
Python
"""
|
|
VitalLink Complete Test Suite
|
|
Integrated testing script that simulates the entire system
|
|
"""
|
|
|
|
import asyncio
|
|
import time
|
|
from datetime import datetime
|
|
import random
|
|
|
|
# Import from previous modules (in production, these would be proper imports)
|
|
# For this demo, we'll include simplified versions
|
|
|
|
# ============================================================================
|
|
# CONFIGURATION
|
|
# ============================================================================
|
|
|
|
API_BASE = "http://localhost:8000"
|
|
ENABLE_API_CALLS = False # Set to True when backend is running
|
|
|
|
# ============================================================================
|
|
# TEST SUITE
|
|
# ============================================================================
|
|
|
|
|
|
class VitalLinkTestSuite:
|
|
"""Comprehensive testing suite for VitalLink system"""
|
|
|
|
def __init__(self):
|
|
self.test_results = []
|
|
self.patients_created = []
|
|
|
|
def log_test(self, test_name: str, passed: bool, message: str = ""):
|
|
"""Log test result"""
|
|
status = "✓ PASS" if passed else "✗ FAIL"
|
|
result = {
|
|
"test": test_name,
|
|
"passed": passed,
|
|
"message": message,
|
|
"timestamp": datetime.now(),
|
|
}
|
|
self.test_results.append(result)
|
|
print(f"{status} | {test_name}")
|
|
if message:
|
|
print(f" {message}")
|
|
|
|
async def test_1_patient_data_validation(self):
|
|
"""Test 1: Validate patient data structure"""
|
|
print("\n" + "=" * 80)
|
|
print("TEST 1: Patient Data Validation")
|
|
print("=" * 80)
|
|
|
|
# Test valid patient data
|
|
valid_patient = {
|
|
"firstName": "John",
|
|
"lastName": "Doe",
|
|
"dob": "1990-01-01",
|
|
"symptoms": ["Chest Pain", "Shortness of Breath"],
|
|
"severity": "moderate",
|
|
}
|
|
|
|
self.log_test(
|
|
"Valid patient data structure",
|
|
all(
|
|
key in valid_patient
|
|
for key in ["firstName", "lastName", "dob", "symptoms", "severity"]
|
|
),
|
|
f"Patient: {valid_patient['firstName']} {valid_patient['lastName']}",
|
|
)
|
|
|
|
# Test invalid data
|
|
invalid_patient = {
|
|
"firstName": "Jane",
|
|
# Missing lastName
|
|
"dob": "1985-05-15",
|
|
"symptoms": [],
|
|
"severity": "mild",
|
|
}
|
|
|
|
self.log_test(
|
|
"Detect missing required fields",
|
|
"lastName" not in invalid_patient,
|
|
"Missing 'lastName' field detected",
|
|
)
|
|
|
|
# Test date validation
|
|
try:
|
|
datetime.strptime(valid_patient["dob"], "%Y-%m-%d")
|
|
self.log_test(
|
|
"Date format validation", True, "Date format YYYY-MM-DD accepted"
|
|
)
|
|
except:
|
|
self.log_test("Date format validation", False, "Invalid date format")
|
|
|
|
async def test_2_vitals_packet_generation(self):
|
|
"""Test 2: Vitals packet generation and checksum"""
|
|
print("\n" + "=" * 80)
|
|
print("TEST 2: Vitals Packet Generation")
|
|
print("=" * 80)
|
|
|
|
# Simulate packet generation
|
|
import struct
|
|
|
|
PKT_STRUCT = struct.Struct("<B H I B B B h H B B")
|
|
|
|
# Create test packet
|
|
test_data = (
|
|
1, # version
|
|
42, # seq
|
|
120000, # timestamp
|
|
2, # flags (ALERT)
|
|
78, # HR
|
|
97, # SpO2
|
|
3645, # temp * 100
|
|
180, # activity * 100
|
|
0, # checksum placeholder
|
|
0, # reserved
|
|
)
|
|
|
|
packet = bytearray(PKT_STRUCT.pack(*test_data))
|
|
|
|
# Calculate checksum
|
|
checksum = sum(packet[0:14]) & 0xFF
|
|
packet[14] = checksum
|
|
|
|
self.log_test(
|
|
"Packet structure (16 bytes)",
|
|
len(packet) == 16,
|
|
f"Packet size: {len(packet)} bytes",
|
|
)
|
|
|
|
# Verify checksum
|
|
verify_checksum = sum(packet[0:14]) & 0xFF
|
|
self.log_test(
|
|
"Checksum validation",
|
|
verify_checksum == packet[14],
|
|
f"Checksum: 0x{checksum:02X}",
|
|
)
|
|
|
|
# Decode and verify
|
|
decoded = PKT_STRUCT.unpack(bytes(packet))
|
|
self.log_test(
|
|
"Packet decode integrity",
|
|
decoded[4] == 78 and decoded[5] == 97, # HR and SpO2
|
|
f"HR={decoded[4]} bpm, SpO2={decoded[5]}%",
|
|
)
|
|
|
|
async def test_3_tier_classification(self):
|
|
"""Test 3: Alert tier classification logic"""
|
|
print("\n" + "=" * 80)
|
|
print("TEST 3: Alert Tier Classification")
|
|
print("=" * 80)
|
|
|
|
def classify_tier(hr, spo2, temp):
|
|
"""Tier classification logic"""
|
|
if hr > 140 or hr < 45 or spo2 < 88 or temp > 39.5 or temp < 35.0:
|
|
return "EMERGENCY"
|
|
elif hr > 110 or hr < 50 or spo2 < 92 or temp > 38.3 or temp < 35.5:
|
|
return "ALERT"
|
|
else:
|
|
return "NORMAL"
|
|
|
|
# Test cases
|
|
test_cases = [
|
|
(75, 98, 36.8, "NORMAL", "Healthy vitals"),
|
|
(115, 93, 38.4, "ALERT", "Elevated HR and temp"),
|
|
(150, 86, 39.8, "EMERGENCY", "Critical vitals"),
|
|
(65, 95, 37.2, "NORMAL", "Slightly elevated temp but stable"),
|
|
(48, 89, 35.3, "ALERT", "Bradycardia and low temp"),
|
|
]
|
|
|
|
for hr, spo2, temp, expected, description in test_cases:
|
|
result = classify_tier(hr, spo2, temp)
|
|
self.log_test(
|
|
f"Tier: {description}",
|
|
result == expected,
|
|
f"HR={hr}, SpO2={spo2}%, Temp={temp}°C → {result}",
|
|
)
|
|
|
|
async def test_4_priority_calculation(self):
|
|
"""Test 4: Queue priority score calculation"""
|
|
print("\n" + "=" * 80)
|
|
print("TEST 4: Priority Score Calculation")
|
|
print("=" * 80)
|
|
|
|
def calculate_priority(tier, wait_minutes, severity, hr, spo2, temp):
|
|
"""Priority calculation algorithm"""
|
|
score = 0.0
|
|
|
|
# Tier contribution
|
|
tier_scores = {"EMERGENCY": 100, "ALERT": 50, "NORMAL": 0}
|
|
score += tier_scores.get(tier, 0)
|
|
|
|
# Wait time
|
|
if wait_minutes > 30:
|
|
score += (wait_minutes - 30) * 0.5
|
|
|
|
# Severity
|
|
severity_scores = {"severe": 20, "moderate": 10, "mild": 5}
|
|
score += severity_scores.get(severity, 0)
|
|
|
|
# Vital signs
|
|
if hr > 110 or hr < 50:
|
|
score += 10
|
|
if hr > 140 or hr < 40:
|
|
score += 30
|
|
if spo2 < 92:
|
|
score += 15
|
|
if spo2 < 88:
|
|
score += 40
|
|
if temp > 38.5:
|
|
score += 15
|
|
|
|
return score
|
|
|
|
# Test priority ordering
|
|
patients = [
|
|
("Critical ER", "EMERGENCY", 10, "severe", 148, 86, 39.8),
|
|
("Deteriorating", "ALERT", 45, "moderate", 118, 93, 38.4),
|
|
("Long Wait Stable", "NORMAL", 90, "mild", 76, 98, 36.9),
|
|
("Recent Stable", "NORMAL", 15, "mild", 72, 97, 37.1),
|
|
]
|
|
|
|
results = []
|
|
for name, tier, wait, severity, hr, spo2, temp in patients:
|
|
priority = calculate_priority(tier, wait, severity, hr, spo2, temp)
|
|
results.append((name, priority))
|
|
print(f" {name}: Priority Score = {priority:.1f}")
|
|
|
|
# Verify ordering
|
|
sorted_results = sorted(results, key=lambda x: x[1], reverse=True)
|
|
self.log_test(
|
|
"Critical patient has highest priority",
|
|
sorted_results[0][0] == "Critical ER",
|
|
f"Top priority: {sorted_results[0][0]} ({sorted_results[0][1]:.1f})",
|
|
)
|
|
|
|
self.log_test(
|
|
"Wait time impacts priority",
|
|
sorted_results[1][1] > sorted_results[3][1],
|
|
"90-min wait scores higher than 15-min wait for NORMAL tier",
|
|
)
|
|
|
|
async def test_5_simulator_stability(self):
|
|
"""Test 5: Wristband simulator stability"""
|
|
print("\n" + "=" * 80)
|
|
print("TEST 5: Simulator Stability Test")
|
|
print("=" * 80)
|
|
|
|
class SimpleSimulator:
|
|
def __init__(self):
|
|
self.seq = 0
|
|
self.hr = 75.0
|
|
self.spo2 = 98.0
|
|
self.temp = 36.8
|
|
|
|
def generate_reading(self):
|
|
# Add realistic variation
|
|
self.hr += random.gauss(0, 2)
|
|
self.spo2 += random.gauss(0, 0.5)
|
|
self.temp += random.gauss(0, 0.1)
|
|
|
|
# Clamp to realistic ranges
|
|
self.hr = max(40, min(180, self.hr))
|
|
self.spo2 = max(70, min(100, self.spo2))
|
|
self.temp = max(35, min(41, self.temp))
|
|
|
|
self.seq += 1
|
|
return int(self.hr), int(self.spo2), round(self.temp, 1)
|
|
|
|
sim = SimpleSimulator()
|
|
readings = []
|
|
|
|
# Generate 100 readings
|
|
for _ in range(100):
|
|
readings.append(sim.generate_reading())
|
|
await asyncio.sleep(0.01) # Simulate time passage
|
|
|
|
# Verify stability
|
|
hrs = [r[0] for r in readings]
|
|
spo2s = [r[1] for r in readings]
|
|
temps = [r[2] for r in readings]
|
|
|
|
self.log_test(
|
|
"Generated 100 readings",
|
|
len(readings) == 100,
|
|
f"Sequence numbers: 1-{sim.seq}",
|
|
)
|
|
|
|
self.log_test(
|
|
"HR within realistic bounds",
|
|
all(40 <= hr <= 180 for hr in hrs),
|
|
f"Range: {min(hrs)}-{max(hrs)} bpm",
|
|
)
|
|
|
|
self.log_test(
|
|
"SpO2 within realistic bounds",
|
|
all(70 <= spo2 <= 100 for spo2 in spo2s),
|
|
f"Range: {min(spo2s)}-{max(spo2s)}%",
|
|
)
|
|
|
|
self.log_test(
|
|
"Temperature within realistic bounds",
|
|
all(35.0 <= temp <= 41.0 for temp in temps),
|
|
f"Range: {min(temps)}-{max(temps)}°C",
|
|
)
|
|
|
|
async def test_6_deterioration_detection(self):
|
|
"""Test 6: Patient deterioration detection"""
|
|
print("\n" + "=" * 80)
|
|
print("TEST 6: Deterioration Detection")
|
|
print("=" * 80)
|
|
|
|
class PatientMonitor:
|
|
def __init__(self):
|
|
self.hr_history = []
|
|
self.spo2_history = []
|
|
|
|
def add_reading(self, hr, spo2):
|
|
self.hr_history.append(hr)
|
|
self.spo2_history.append(spo2)
|
|
|
|
# Keep last 10 readings
|
|
if len(self.hr_history) > 10:
|
|
self.hr_history = self.hr_history[-10:]
|
|
self.spo2_history = self.spo2_history[-10:]
|
|
|
|
def detect_deterioration(self):
|
|
"""Detect worsening trends"""
|
|
if len(self.hr_history) < 5:
|
|
return False, "Insufficient data"
|
|
|
|
# Check for consistent increase in HR
|
|
hr_trend = sum(self.hr_history[-3:]) / 3 - sum(self.hr_history[:3]) / 3
|
|
|
|
# Check for consistent decrease in SpO2
|
|
spo2_trend = (
|
|
sum(self.spo2_history[:3]) / 3 - sum(self.spo2_history[-3:]) / 3
|
|
)
|
|
|
|
if hr_trend > 15 or spo2_trend > 3:
|
|
return (
|
|
True,
|
|
f"HR trend: +{hr_trend:.1f}, SpO2 trend: -{spo2_trend:.1f}",
|
|
)
|
|
|
|
return False, "Stable"
|
|
|
|
monitor = PatientMonitor()
|
|
|
|
# Simulate gradual deterioration
|
|
base_hr = 80
|
|
base_spo2 = 97
|
|
|
|
for i in range(10):
|
|
hr = base_hr + (i * 3) # Increase HR
|
|
spo2 = base_spo2 - (i * 0.5) # Decrease SpO2
|
|
monitor.add_reading(hr, int(spo2))
|
|
|
|
deteriorating, message = monitor.detect_deterioration()
|
|
|
|
self.log_test("Detect deteriorating patient", deteriorating, message)
|
|
|
|
# Test stable patient
|
|
stable_monitor = PatientMonitor()
|
|
for _ in range(10):
|
|
stable_monitor.add_reading(
|
|
75 + random.randint(-3, 3), 98 + random.randint(-1, 1)
|
|
)
|
|
|
|
stable, message = stable_monitor.detect_deterioration()
|
|
|
|
self.log_test("Stable patient not flagged", not stable, message)
|
|
|
|
async def test_7_transmission_timing(self):
|
|
"""Test 7: Verify transmission intervals by tier"""
|
|
print("\n" + "=" * 80)
|
|
print("TEST 7: Transmission Timing")
|
|
print("=" * 80)
|
|
|
|
def get_interval(tier):
|
|
"""Get transmission interval for tier"""
|
|
intervals = {"EMERGENCY": 1.0, "ALERT": 1.0, "NORMAL": 60.0}
|
|
return intervals.get(tier, 60.0)
|
|
|
|
test_cases = [("EMERGENCY", 1.0), ("ALERT", 1.0), ("NORMAL", 60.0)]
|
|
|
|
for tier, expected in test_cases:
|
|
result = get_interval(tier)
|
|
self.log_test(
|
|
f"{tier} tier interval",
|
|
result == expected,
|
|
f"{result}s transmission interval",
|
|
)
|
|
|
|
def generate_report(self):
|
|
"""Generate test report"""
|
|
print("\n" + "=" * 80)
|
|
print("TEST SUITE SUMMARY")
|
|
print("=" * 80)
|
|
|
|
total = len(self.test_results)
|
|
passed = sum(1 for r in self.test_results if r["passed"])
|
|
failed = total - passed
|
|
|
|
print(f"\nTotal Tests: {total}")
|
|
print(f"Passed: {passed} ({100 * passed / total:.1f}%)")
|
|
print(f"Failed: {failed} ({100 * failed / total:.1f}%)")
|
|
|
|
if failed > 0:
|
|
print("\nFailed Tests:")
|
|
for result in self.test_results:
|
|
if not result["passed"]:
|
|
print(f" ✗ {result['test']}")
|
|
if result["message"]:
|
|
print(f" {result['message']}")
|
|
|
|
print("\n" + "=" * 80)
|
|
|
|
return passed == total
|
|
|
|
async def run_all_tests(self):
|
|
"""Run complete test suite"""
|
|
print("\n")
|
|
print(
|
|
"╔═══════════════════════════════════════════════════════════════════════╗"
|
|
)
|
|
print(
|
|
"║ VitalLink System - Complete Test Suite ║"
|
|
)
|
|
print(
|
|
"╚═══════════════════════════════════════════════════════════════════════╝"
|
|
)
|
|
|
|
await self.test_1_patient_data_validation()
|
|
await self.test_2_vitals_packet_generation()
|
|
await self.test_3_tier_classification()
|
|
await self.test_4_priority_calculation()
|
|
await self.test_5_simulator_stability()
|
|
await self.test_6_deterioration_detection()
|
|
await self.test_7_transmission_timing()
|
|
|
|
success = self.generate_report()
|
|
|
|
return success
|
|
|
|
|
|
# ============================================================================
|
|
# DEMONSTRATION SCENARIOS
|
|
# ============================================================================
|
|
|
|
|
|
async def demo_emergency_scenario():
|
|
"""Demonstrate emergency patient handling"""
|
|
print("\n" + "=" * 80)
|
|
print("DEMO: Emergency Patient Scenario")
|
|
print("=" * 80)
|
|
print("\nSimulating a patient experiencing a medical emergency...\n")
|
|
|
|
# Patient starts deteriorating
|
|
print("Time 0:00 - Patient arrives, appears stable")
|
|
print(" HR: 95 bpm, SpO2: 96%, Temp: 37.8°C")
|
|
print(" Status: NORMAL tier")
|
|
await asyncio.sleep(2)
|
|
|
|
print("\nTime 1:00 - Condition worsening")
|
|
print(" HR: 118 bpm, SpO2: 93%, Temp: 38.5°C")
|
|
print(" Status: Upgraded to ALERT tier")
|
|
print(" Action: Transmission frequency increased to 1 Hz")
|
|
await asyncio.sleep(2)
|
|
|
|
print("\nTime 2:30 - Critical deterioration")
|
|
print(" HR: 152 bpm, SpO2: 87%, Temp: 39.9°C")
|
|
print(" Status: EMERGENCY tier activated")
|
|
print(" Action: Buzzer activated, priority score: 156.2")
|
|
print(" Action: Staff notified immediately")
|
|
await asyncio.sleep(2)
|
|
|
|
print("\nTime 3:00 - Patient being attended")
|
|
print(" Queue Position: #1 (highest priority)")
|
|
print(" Medical team dispatched")
|
|
|
|
print("\n✓ Emergency protocol executed successfully\n")
|
|
|
|
|
|
async def demo_queue_management():
|
|
"""Demonstrate dynamic queue management"""
|
|
print("\n" + "=" * 80)
|
|
print("DEMO: Dynamic Queue Management")
|
|
print("=" * 80)
|
|
print("\nSimulating 4 patients with varying conditions...\n")
|
|
|
|
patients = [
|
|
("Alice", "NORMAL", 45, 75, 98, 36.9),
|
|
("Bob", "ALERT", 10, 122, 91, 38.7),
|
|
("Charlie", "NORMAL", 20, 80, 97, 37.2),
|
|
("Diana", "EMERGENCY", 5, 155, 85, 40.1),
|
|
]
|
|
|
|
print("Initial Queue State:")
|
|
for name, tier, wait, hr, spo2, temp in patients:
|
|
print(
|
|
f" {name:8} | {tier:9} | Wait: {wait:2}min | "
|
|
f"HR: {hr:3} | SpO2: {spo2:2}% | Temp: {temp:.1f}°C"
|
|
)
|
|
|
|
await asyncio.sleep(3)
|
|
|
|
print("\nPriority Scores Calculated:")
|
|
print(" Diana (EMERGENCY): 178.5 → Queue Position #1")
|
|
print(" Bob (ALERT): 72.3 → Queue Position #2")
|
|
print(" Alice (NORMAL): 37.5 → Queue Position #3")
|
|
print(" Charlie (NORMAL): 20.0 → Queue Position #4")
|
|
|
|
await asyncio.sleep(2)
|
|
|
|
print("\nAfter 30 minutes...")
|
|
print(" Diana: Seen by doctor (discharged from queue)")
|
|
print(" Bob: Stabilized → Downgraded to NORMAL")
|
|
print(" Alice: Wait time now 75min → Priority increased")
|
|
print("\nUpdated Queue:")
|
|
print(" Alice: Priority 60.0 → Queue Position #1")
|
|
print(" Bob: Priority 45.0 → Queue Position #2")
|
|
print(" Charlie: Priority 35.0 → Queue Position #3")
|
|
|
|
print("\n✓ Queue dynamically adjusted based on condition and wait time\n")
|
|
|
|
|
|
# ============================================================================
|
|
# MAIN
|
|
# ============================================================================
|
|
|
|
|
|
async def main():
|
|
"""Main test execution"""
|
|
|
|
print("\nVitalLink System Test Suite")
|
|
print("Choose an option:")
|
|
print(" 1. Run complete test suite")
|
|
print(" 2. Demo: Emergency scenario")
|
|
print(" 3. Demo: Queue management")
|
|
print(" 4. Run all")
|
|
|
|
choice = "4" # Default to run all
|
|
|
|
if choice in ["1", "4"]:
|
|
suite = VitalLinkTestSuite()
|
|
success = await suite.run_all_tests()
|
|
|
|
if not success:
|
|
print("\n⚠️ Some tests failed. Review output above.")
|
|
|
|
if choice in ["2", "4"]:
|
|
await demo_emergency_scenario()
|
|
|
|
if choice in ["3", "4"]:
|
|
await demo_queue_management()
|
|
|
|
print("\n" + "=" * 80)
|
|
print("Testing complete. System ready for deployment.")
|
|
print("=" * 80 + "\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|