#!/usr/bin/env python3 """ analyze_patient_json_single.py - Single Test Automated Analysis (for PHP backend) Reads test data from JSON file, generates automated clinical analysis, and outputs JSON result to stdout for PHP consumption. Usage: python analyze_patient_json_single.py --input /tmp/test_data.json """ import json import sys import argparse from datetime import datetime def analyze_leg_balance(leg_data, leg_name): """Analyze balance data for a single leg.""" if not leg_data: return None findings = { 'leg': leg_name, 'score': leg_data.get('movement_score'), 'test_duration': leg_data.get('test_duration'), 'concerns': [], 'strengths': [] } # Get metrics stability = leg_data.get('stability', {}) windows = leg_data.get('windows', {}) # Check 5-second window if '5_second' in windows and windows['5_second']: q_diff = windows['5_second'].get('q_diff_range') if q_diff: if q_diff <= 0.010: findings['strengths'].append(f"Exceptional stability (Q-Diff: {q_diff:.4f})") elif q_diff <= 0.020: findings['strengths'].append(f"Excellent stability (Q-Diff: {q_diff:.4f})") elif q_diff >= 0.300: findings['concerns'].append(f"High movement variability (Q-Diff: {q_diff:.4f})") # Check stability duration cont_duration = stability.get('continuous_duration', 0) if cont_duration < 2.0: findings['concerns'].append(f"Poor endurance - only {cont_duration:.1f}s continuous stability") elif cont_duration > 5.0: findings['strengths'].append(f"Excellent endurance - {cont_duration:.1f}s continuous stability") # Check percent stable percent_stable = stability.get('percent_stable', 0) if percent_stable < 70: findings['concerns'].append(f"Frequent wobbling - only {percent_stable:.0f}% stable") elif percent_stable > 90: findings['strengths'].append(f"Consistent stability - {percent_stable:.0f}% stable") return findings def compare_legs(left_findings, right_findings): """Compare left and right leg balance.""" if not left_findings or not right_findings: return None comparison = { 'has_asymmetry': False, 'findings': [] } # Compare scores if both available left_score = left_findings.get('score') right_score = right_findings.get('score') if left_score and right_score and abs(left_score - right_score) >= 2: better_leg = 'right' if right_score > left_score else 'left' comparison['has_asymmetry'] = True comparison['findings'].append( f"Score asymmetry: {better_leg.upper()} leg scores {abs(right_score - left_score)} points higher" ) return comparison def generate_recommendations(score, concerns): """Generate clinical recommendations based on score and concerns.""" recommendations = [] if score >= 9: recommendations.append("✓ Excellent balance - no immediate interventions needed") recommendations.append("✓ Continue regular physical activity") recommendations.append("✓ Retest annually or if balance concerns arise") elif score >= 7: recommendations.append("⚠ Moderate balance - consider:") recommendations.append(" • Balance training exercises 2-3x per week") recommendations.append(" • Home safety assessment") recommendations.append(" • Retest in 6 months") elif score >= 5: recommendations.append("⚠️ HIGH RISK - Immediate interventions:") recommendations.append(" • Physical therapy referral") recommendations.append(" • Fall risk assessment") recommendations.append(" • Home modifications required") recommendations.append(" • Retest in 3 months") else: recommendations.append("🚨 CRITICAL RISK - Urgent interventions:") recommendations.append(" • IMMEDIATE physical therapy referral") recommendations.append(" • Comprehensive medical evaluation") recommendations.append(" • Home safety modifications REQUIRED") recommendations.append(" • Close monitoring needed") recommendations.append(" • Retest in 1 month") return recommendations def analyze_test(test_data): """Main analysis function.""" try: # Parse results_json results_json = json.loads(test_data['results_json']) # Analyze each leg left_findings = analyze_leg_balance(results_json.get('lbeo'), 'Left') right_findings = analyze_leg_balance(results_json.get('rbeo'), 'Right') # Compare legs comparison = None if left_findings and right_findings: comparison = compare_legs(left_findings, right_findings) # Collect all concerns all_concerns = [] if left_findings: all_concerns.extend(left_findings['concerns']) if right_findings: all_concerns.extend(right_findings['concerns']) # Generate recommendations score = test_data.get('score', 0) recommendations = generate_recommendations(score, all_concerns) # Build text analysis analysis_parts = [] # Header analysis_parts.append(f"AUTOMATED BALANCE ANALYSIS") analysis_parts.append(f"Patient: {test_data['patient_name']}") analysis_parts.append(f"Age: {test_data['age']}") analysis_parts.append(f"Test Date: {test_data['test_date']}") analysis_parts.append(f"Overall Score: {score}/10") analysis_parts.append("") # Left leg if left_findings: analysis_parts.append("LEFT LEG:") analysis_parts.append(f" Score: {left_findings['score']}/10") if left_findings['strengths']: analysis_parts.append(" Strengths:") for s in left_findings['strengths']: analysis_parts.append(f" • {s}") if left_findings['concerns']: analysis_parts.append(" Concerns:") for c in left_findings['concerns']: analysis_parts.append(f" • {c}") analysis_parts.append("") # Right leg if right_findings: analysis_parts.append("RIGHT LEG:") analysis_parts.append(f" Score: {right_findings['score']}/10") if right_findings['strengths']: analysis_parts.append(" Strengths:") for s in right_findings['strengths']: analysis_parts.append(f" • {s}") if right_findings['concerns']: analysis_parts.append(" Concerns:") for c in right_findings['concerns']: analysis_parts.append(f" • {c}") analysis_parts.append("") # Asymmetry if comparison and comparison['has_asymmetry']: analysis_parts.append("ASYMMETRY DETECTED:") for finding in comparison['findings']: analysis_parts.append(f" • {finding}") analysis_parts.append("") # Recommendations analysis_parts.append("RECOMMENDATIONS:") for rec in recommendations: analysis_parts.append(f" {rec}") analysis_text = "\n".join(analysis_parts) return { 'success': True, 'analysis': analysis_text, 'left_leg': left_findings, 'right_leg': right_findings, 'comparison': comparison, 'recommendations': recommendations } except Exception as e: return { 'success': False, 'error': str(e) } def main(): parser = argparse.ArgumentParser(description='Automated patient balance analysis (single test)') parser.add_argument('--input', required=True, help='Path to input JSON file') args = parser.parse_args() try: # Read input with open(args.input, 'r') as f: test_data = json.load(f) # Analyze result = analyze_test(test_data) # Output JSON to stdout print(json.dumps(result)) except Exception as e: error_result = { 'success': False, 'error': f'Failed to process test: {str(e)}' } print(json.dumps(error_result)) sys.exit(1) if __name__ == '__main__': main()