Perfecto, vamos a expandir el sistema con esas características: Sistema de Salud Automatizado - Versión Completa 1. Estructura de Archivos ~/health-system/ ├── config/ │ ├── health_config.yaml # Configuración base │ ├── calendar_rules.md # Reglas de adaptación de calendario │ └── xiaomi_config.yaml # Credenciales y endpoints ├── logic/ │ ├── escalation_rules.md # Sistema de escalation │ ├── decision_tree.md # Árbol de decisiones │ └── correlations.md # Patrones detectados ├── data/ │ ├── daily_logs.json # Logs diarios │ ├── weekly_reports.json # Reportes semanales │ └── xiaomi_data.json # Data del reloj └── scripts/ ├── health_agent.py # Agente principal ├── calendar_adapter.py # Integración calendario ├── xiaomi_sync.py # Sync con Xiaomi └── decision_engine.py # Motor de decisiones 2. calendar_rules.md - Reglas de Adaptación # Reglas de Adaptación de Calendario ## Principio Base Los recordatorios de salud se adaptan al contexto del calendario, pero NUNCA se cancelan completamente. Se postponen o modifican. ## Reglas por Tipo de Evento ### MEETINGS (Reuniones) Detección: - Evento con >1 participante - Título contiene: "meeting", "reunión", "call", "sync", "1:1" Adaptaciones: - ❌ NO enviar recordatorio de "movement break" durante el meeting - ✅ SIEMPRE enviar "Elongá 2 min" al terminar el meeting - ✅ Si meeting >1h → recordatorio extra de agua al terminar - ⚠️ Si meeting cerca de horario comida → adelantar recordatorio 15 min Ejemplo: ```yaml Event: "Team Sync" 10:00-11:00 Adaptación: - Cancelar reminder 10:00 (movement break) - Agendar reminder 11:01 "🚶 Meeting terminado. Elongá 2 min + vaso de agua" FOCUS TIME / DEEP WORK Detección: ∙ Evento marcado como “ocupado” sin otros participantes ∙ Título contiene: “focus”, “deep work”, “coding”, “writing” Adaptaciones: ∙ Reducir frecuencia de recordatorios (cada 90 min en vez de 60 min) ∙ Modo silencioso en Telegram (solo vibración) ∙ Permitir postponer breaks con /snooze 30 WORKOUT / CALISTENIA Detección: ∙ Título contiene: “gym”, “workout”, “calistenia”, “ejercicio” ∙ Duración: 60-120 min Adaptaciones: ∙ ❌ Cancelar ALL movement breaks durante ∙ ✅ Reminder POST-workout: “💧 2 vasos de agua + proteína” ∙ ✅ Marcar automáticamente como “sesión de calistenia cumplida” OUT OF OFFICE / VIAJES Detección: ∙ Evento todo el día ∙ Título contiene: “vacaciones”, “viaje”, “off” Adaptaciones: ∙ Reducir recordatorios al mínimo esencial: ∙ Solo: desayuno proteico, agua al despertar, pantallas antes de dormir ∙ No enviar movement breaks (asume movilidad natural) ∙ Mantener tracking de síntomas COMIDAS AGENDADAS Detección: ∙ Título contiene: “almuerzo”, “cena”, “desayuno”, “lunch”, “dinner” ∙ Duración: 30-90 min Adaptaciones: ∙ Reminder 15 min antes: “💧 Toma agua antes de comer” ∙ Reminder 30 min después: “¿Fatiga post-comida? (Sí/No/Leve)” ∙ Si respuesta = Sí → Log con detalles de qué comió Prioridades (Orden de Importancia) 1. NUNCA CANCELAR: ∙ Desayuno con proteína ∙ Screen cutoff (22:30) ∙ Tracking de síntomas 2. ADAPTAR TIMING: ∙ Movement breaks (postponer si hay meeting) ∙ Agua (concentrar en ventanas libres) 3. REDUCIR FRECUENCIA: ∙ En deep work ∙ En días de viaje Lógica de Decisión IF meeting_in_progress: POSTPONE movement_break TO meeting_end + 1min ADD extra_reminder("Elongá + agua") IF focus_time AND time_since_last_break > 90min: SEND gentle_reminder("Break sugerido cuando termines") IF workout_detected: MARK calisthenics_session = True SCHEDULE post_workout_nutrition_reminder IF all_day_event: SWITCH_TO minimal_mode Calendario de Ejemplo Adaptado 08:00 - 09:00 | Deep Work → 08:00 Movement break POSTPONED to 09:01 10:00 - 11:00 | Team Meeting → 10:00 Movement break CANCELLED → 11:01 NEW: "Meeting done. Elongá 2 min + agua" 13:00 - 14:00 | Almuerzo con cliente → 12:45 "💧 Agua antes de almorzar" → 14:30 "¿Fatiga post-comida?"17:00 - 18:30 | Calistenia → All movement breaks CANCELLED → 18:31 "💪 Sesión completa! 2 vasos agua + proteína" --- ### 3. escalation_rules.md - Sistema de Escalation ```markdown # Sistema de Escalation y Alertas ## Niveles de Severidad ### 🟢 INFO (Informativo) Trigger: Desvío leve del objetivo Acción: Notificación simple Ejemplos: ```yaml Condición: water_intake = 8/10 vasos Mensaje: "💧 Vas 8/10 vasos. 2 más y cumplís!" Acción: Ninguna adicional Condición: movement_breaks = 6/8 Mensaje: "🚶 6/8 breaks. Buen día!" Acción: Ninguna adicional 🟡 WARNING (Advertencia) Trigger: Desvío significativo o patrón preocupante Acción: Notificación + sugerencia Ejemplos: Condición: water_intake < 6 vasos AND hora > 16:00 Mensaje: "⚠️ Solo {count} vasos y son las 16hs" "Tomá 2 ahora mismo. Esto afecta tu fatiga post-comida" Acción: Enviar reminder cada 30 min hasta llegar a 8 Condición: breakfast_protein_missed = 2 consecutive days Mensaje: "⚠️ 2do día sin proteína en desayuno" "Recordá: esto explica tu fatiga. ¿Necesitás recetas rápidas?" Acción: Ofrecer quick_buttons(["Recetas", "Posponer", "Ignoring"]) Condición: movement_breaks < 4 AND knee_pain_logged_today Mensaje: "⚠️ Pocas pausas + dolor de rodilla" "Las tendinitis empeoran con inmovilidad" Acción: Aumentar frecuencia de breaks a cada 45 min 🔴 ALERT (Alerta Crítica) Trigger: Desvío crítico o riesgo de salud Acción: Notificación urgente + intervención Ejemplos: Condición: screen_time_violations >= 4 this_week Mensaje: "🚨 4ta violación de pantallas esta semana" "Tu FC puede estar subiendo (fue 62→71 en dic por menos)" "¿Qué está pasando?" Acción: - Enviar cuestionario de diagnóstico - Bloquear "Snooze" option - Reporte especial al final del día Condición: resting_hr > 70 for 3 consecutive days Mensaje: "🚨 FC en reposo: {hr} por 3er día seguido" "Tu baseline es 62. Posibles causas:" "- Mala recuperación" "- Sobreentrenamiento" "- Enfermedad incubando" "- Estrés elevado" Acción: - Sugerir día de descanso completo - Aumentar tracking de síntomas - Si continúa 2 días más → sugerir consulta médica Condición: knee_pain OR shoulder_pain logged 3+ times this_week Mensaje: "🚨 Dolor de {location} reportado {count} veces" "Esto indica empeoramiento de tendinitis" "ACCIÓN REQUERIDA: Reducir volumen de calistenia" Acción: - Bloquear próxima sesión de calistenia - Enviar "Hacé solo movilidad suave hoy" - Reminder: "Agendá fisio/traumatólogo" 🔥 CRITICAL (Crítico - Requiere Acción Inmediata) Trigger: Situación de riesgo inmediato Acción: Intervención directa + bloqueo de conductas Ejemplos: Condición: resting_hr > 80 for 5 consecutive days Mensaje: "🔥 CRÍTICO: FC {hr} por 5to día" "Esto es 30% sobre tu baseline (62)" "Causas posibles:" "- Sobre-entrenamiento severo" "- Infección/enfermedad" "- Estrés crónico extremo" "" "STOP a todo ejercicio intenso" "Consultá médico HOY" Acción: - CANCELAR todas las sesiones de calistenia programadas - Solo permitir caminatas suaves - Enviar reminder cada 6hs hasta normalizar Condición: water_intake < 3 vasos for 2 consecutive days Mensaje: "🔥 DESHIDRATACIÓN SEVERA" "Solo {count} vasos en 2 días" "Esto afecta: riñones, cognición, rendimiento, recuperación" "TOMA 3 VASOS AHORA" Acción: - Reminder cada 15 min hasta tomar 3 vasos - No permitir postponer - Escalar a contacto de emergencia si no responde en 2hs Correlaciones Automáticas El sistema detecta patrones y ajusta alertas: Pattern_Detected: "water < 7 → fatigue_post_meal = 80% probability" Adaptation: - Cuando water < 7 a las 12:00 - Enviar: "💡 PATRÓN: Baja agua = alta fatiga post-comida en vos" - "Tomá 2 vasos antes de almorzar" Pattern_Detected: "screen_time_late → resting_hr + 3-5 bpm next_day" Adaptation: - Aumentar urgencia de screen cutoff reminder - Mostrar: "Ayer pantallas hasta tarde → hoy HR = {hr} (+4 vs baseline)"Pattern_Detected: "no_mobility_routine → knee_pain within 48hs" Adaptation: - Si 2 días sin movilidad - Enviar: "⚠️ Sin movilidad 2 días. Históricamente esto → dolor rodilla" - "Hacé 10 min AHORA para prevenir" Decisión Tree para Síntomas knee_pain OR shoulder_pain logged: ├─ First occurrence this week │ └─ Send: "Anotado. ¿Intensidad 1-10?" │ └─ Log intensity and timestamp │ ├─ Second occurrence (within 7 days) │ └─ Send: "2da vez esta semana. ¿Mismo tipo de dolor?" │ └─ If same location: 🟡 WARNING level │ "Considerá reducir volumen próxima sesión" │ ├─ Third occurrence (within 7 days) │ └─ 🔴 ALERT level │ "3 veces en 7 días = patrón de empeoramiento" │ "REDUCE calistenia a 1h (no 1.5h)" │ "¿Ya agendaste fisio?" │ └─ Offer quick_buttons(["Agendado", "Necesito ayuda", "Ignorar"]) │ └─ Fourth occurrence (within 7 days) └─ 🔥 CRITICAL level "STOP ejercicio intenso hasta ver especialista" "Solo movilidad suave permitida" └─ BLOCK next calisthenics session Prioridades de Acción (Orden) 1. Salud Inmediata (dolor, FC anormal) 2. Fundamentos Diarios (agua, desayuno proteico) 3. Prevención (movilidad, breaks) 4. Optimización (sueño, timing) Timing de Escalations INFO: Real-time (inmediato) WARNING: - Primera vez: Inmediato - Repetición: Al final del día (resumen) ALERT: - Inmediato + follow-up cada 6hs hasta resolver CRITICAL: - Inmediato - Repetir cada 3hs - Escalar a contacto de emergencia si no hay respuesta en 24hs Override Manual Usuario puede: /override [rule_name] [duration] Ejemplo: /override screen_cutoff 1d → Desactiva alertas de pantallas por 1 día (ej: viaje, evento especial) /emergency_mode → Solo alertas CRITICAL → Para situaciones de crisis/estrés extremo Aprendizaje Continuo El sistema actualiza correlations.md cada semana: ## Correlación Detectada - Semana 2025-W07 Pattern: breakfast_no_protein → fatigue_post_lunch Confidence: 85% (6/7 días) Action: Aumentar prioridad de desayuno proteico Escalation: Si 3 días seguidos sin proteína → WARNING (antes era INFO) --- ### 4. xiaomi_config.yaml + Integración ```yaml # xiaomi_config.yaml device: model: "Mi Band 7 / Mi Watch" # Actualizar con tu modelo específico mac_address: "" # Se obtiene del app Mi Fit auth: xiaomi_account: "tu_email@gmail.com" # Las credenciales se manejan via OAuth con Mi Fit app sync_settings: frequency: "every_30_min" # Matching clawdbot cycle retry_on_fail: 3 timeout: 30 # seconds metrics_to_pull: critical: # Pull siempre - heart_rate_resting # FC en reposo - heart_rate_current # FC actual - sleep_duration # Horas de sueño - sleep_deep # Sueño profundo (min) - sleep_light # Sueño ligero (min) - sleep_rem # REM (min) important: # Pull si está disponible - hrv # Variabilidad cardíaca - stress_level # Nivel de estrés (0-100) - steps # Pasos diarios - calories_burned # Calorías - spo2 # Saturación de oxígeno optional: # Nice to have - workout_sessions # Detecta entrenamientos auto - standing_time # Tiempo de pie - distance_walked # Distancia caminada data_validation: hr_resting: min: 40 # Por debajo → error de lectura max: 100 # Por encima → alerta hr_current: min: 40 max: 200 sleep_duration: min: 3 # horas - menos es sospechoso max: 12 # horas - más es sospechoso alert_thresholds: hr_resting_increase: 10 # bpm sobre baseline → warning hr_resting_critical: 15 # bpm sobre baseline → alert sleep_duration_low: 6 # horas - menos → warning hrv_decrease: 20 # % bajo baseline → warning Script de Sync: xiaomi_sync.py """ Xiaomi Mi Band/Watch Integration Usa la librería 'python-miband' o 'huami-token' para auth """ import json from datetime import datetime, timedelta import requests# OPCIÓN 1: Via Mi Fit API (requiere reverse engineering, complicado) # OPCIÓN 2: Via Gadgetbridge (open source, más confiable) # OPCIÓN 3: Via Zepp API (oficial pero limitado) # RECOMENDACIÓN: Gadgetbridge approach class XiaomiSync: def __init__(self, config_path="config/xiaomi_config.yaml"): self.config = self.load_config(config_path) self.last_sync = None def pull_health_data(self): """ Pull desde Gadgetbridge SQLite DB o Zepp API Gadgetbridge guarda todo en: /data/gadgetbridge.db """ try: data = { "timestamp": datetime.now().isoformat(), "hr_resting": self.get_resting_hr(), "hr_current": self.get_current_hr(), "sleep_last_night": self.get_sleep_data(), "steps_today": self.get_steps(), "hrv": self.get_hrv() if self.config['metrics_to_pull']['important']['hrv'] else None } # Validación if not self.validate_data(data): return {"error": "Invalid data from device"} # Guardar self.save_to_file(data, "data/xiaomi_data.json") return data except Exception as e: return {"error": str(e)} def get_resting_hr(self): """ Obtiene FC en reposo del día Xiaomi calcula esto automáticamente en la madrugada """ # Ejemplo con Gadgetbridge DB query = """ SELECT AVG(heart_rate) FROM mi_band_activity_sample WHERE timestamp >= datetime('now', '-8 hours') AND heart_rate BETWEEN 40 AND 80 AND activity_kind = 0 -- 0 = sleep/rest """ # Ejecutar query contra Gadgetbridge DB # O parsear desde JSON export de Zepp return 62 # Placeholder def detect_workout_auto(self): """ Xiaomi detecta workouts automáticamente Marcar como sesión de calistenia si detecta """ query = """ SELECT * FROM workouts WHERE date = TODAY AND (type = 'strength' OR type = 'other') AND duration_minutes BETWEEN 60 AND 120 """ # Si encuentra → auto-marcar daily_goals.calisthenics_session pass def validate_data(self, data): """Valida según xiaomi_config.yaml thresholds""" hr = data.get('hr_resting') if hr and (hr < 40 or hr > 100): return False return True Instalación de Gadgetbridge (en Raspberry Pi): # Gadgetbridge es Android app, pero puedes: # 1. Correr Gadgetbridge en Android con auto-export a Raspberry # 2. O usar Zepp API directamente (más fácil) # OPCIÓN RECOMENDADA: Zepp API (oficial) pip install zepp-python # Si existe una librería # O hacer requests directos a Zepp Cloud API Integración en health_agent.py: from xiaomi_sync import XiaomiSync def main_health_loop(): xiaomi = XiaomiSync() # Cada 30 min (matching clawdbot) while True: # 1. Pull data de Xiaomi xiaomi_data = xiaomi.pull_health_data() if 'error' not in xiaomi_data: # 2. Actualizar baseline update_baseline_hr(xiaomi_data['hr_resting']) # 3. Check escalations basado en data real check_hr_escalation(xiaomi_data['hr_resting']) check_sleep_quality(xiaomi_data['sleep_last_night']) # 4. Auto-mark workout si detectado if xiaomi.detect_workout_auto(): mark_calisthenics_completed() time.sleep(1800) # 30 min 5. decision_engine.py - Motor de Decisiones """ Motor de decisiones que lee los .md files y ejecuta lógica """ import re from datetime import datetime import yaml class DecisionEngine: def __init__(self): self.escalation_rules = self.load_markdown("logic/escalation_rules.md") self.calendar_rules = self.load_markdown("logic/calendar_rules.md") self.correlations = self.load_markdown("logic/correlations.md") def load_markdown(self, filepath): """Parse markdown y extrae reglas en formato estructurado""" with open(filepath, 'r') as f: content = f.read() # Extrae bloques YAML de los code fences yaml_blocks = re.findall(r'```yaml (.*?) ```', content, re.DOTALL) rules = [yaml.safe_load(block) for block in yaml_blocks] return { "raw_markdown": content, "parsed_rules": rules } def evaluate_escalation(self, metric, value, context={}): """ Lee escalation_rules.md y decide qué acción tomar """ for rule in self.escalation_rules['parsed_rules']: condition = rule.get('Condición', rule.get('Condition')) if self.matches_condition(condition, metric, value, context): return { "level": self.extract_level(rule), "message": rule.get('Mensaje', rule.get('Mensaje')), "action": rule.get('Acción', rule.get('Action')) } return None # No escalation needed def matches_condition(self, condition, metric, value, context): """ Evalúa si condición del .md se cumple Ejemplo: "water_intake < 6 vasos AND hora > 16:00" """ # Parse natural language condition condition = condition.replace('{count}', str(value)) # Eval con contexto # NOTA: En producción usar parser seguro, no eval() try: return eval(condition, context) except: return False def extract_level(self, rule): """Extrae nivel de severidad del mensaje""" msg = rule.get('Mensaje', '') if '🔥' in msg or 'CRÍTICO' in msg: return 'CRITICAL' elif '🚨' in msg or 'ALERT' in msg.upper(): return 'ALERT' elif '⚠️' in msg or 'WARNING' in msg.upper(): return 'WARNING' else: return 'INFO' def adapt_to_calendar(self, event, reminder_type): """ Lee calendar_rules.md y adapta recordatorio """ # Detecta tipo de evento event_type = self.classify_event(event) # Busca regla matching en calendar_rules.md for rule_block in self.calendar_rules['raw_markdown'].split('###'): if event_type.upper() in rule_block: # Extrae adaptaciones if 'CANCELAR' in rule_block and reminder_type in rule_block: return {'action': 'cancel'} elif 'POSTPONE' in rule_block or 'POSTPONER' in rule_block: return {'action': 'postpone', 'to': 'event_end + 1min'} return {'action': 'send_normal'} # Default def classify_event(self, event): """Clasifica evento según calendar_rules.md""" title = event.get('title', '').lower() keywords = { 'MEETING': ['meeting', 'reunión', 'call', 'sync', '1:1'], 'FOCUS': ['focus', 'deep work', 'coding', 'writing'], 'WORKOUT': ['gym', 'workout', 'calistenia', 'ejercicio'], 'MEAL': ['almuerzo', 'cena', 'desayuno', 'lunch', 'dinner'] } for event_type, keywords_list in keywords.items(): if any(kw in title for kw in keywords_list): return event_type return 'OTHER' def update_correlations(self, new_pattern): """ Agrega nueva correlación detectada a correlations.md """ with open("logic/correlations.md", 'a') as f: f.write(f" ## Correlación Detectada - {datetime.now().strftime('%Y-W%W')} ") f.write(f"**Pattern:** {new_pattern['pattern']} ") f.write(f"**Confidence:** {new_pattern['confidence']}% ") f.write(f"**Action:** {new_pattern['action']} ") 6. Ejemplo Completo de Flujo # health_agent.py (main loop en clawdbot) from decision_engine import DecisionEngine from xiaomi_sync import XiaomiSync from google_calendar import GoogleCalendar import telegram_bot def run_health_check(): engine = DecisionEngine() xiaomi = XiaomiSync() calendar = GoogleCalendar() # 1. Pull data de Xiaomi health_data = xiaomi.pull_health_data() hr_resting = health_data['hr_resting'] # 2. Check escalation según escalation_rules.md escalation = engine.evaluate_escalation( metric='resting_hr', value=hr_resting, context={'baseline': 62, 'current_day': 3} ) if escalation and escalation['level'] == 'ALERT': telegram_bot.send_urgent(escalation['message']) execute_action(escalation['action']) # 3. Check calendar para adaptar próximo reminder current_time = datetime.now() events = calendar.get_events_in_next_hour() if events: for event in events: adaptation = engine.adapt_to_calendar(event, 'movement_break') if adaptation['action'] == 'postpone': schedule_reminder_after_event(event) elif adaptation['action'] == 'cancel': cancel_reminder(event.start_time) # 4. Send recordatorios normales (si no hay adaptación) send_scheduled_reminders(current_time) primero documenta el plan verbatim