from __future__ import annotations from typing import ClassVar from .base import BaseAgent, AgentInput, AgentOutput from .manifest import AgentManifest, InferredParam from .inference.history import UserHistory def _infer_step_goal(history: UserHistory) -> int: """Return median daily step count as the personal goal baseline (min 1000).""" if not history.task_completions: return 7_000 # task_completions reused as a generic history mechanism here; # step history arrives via agent_prefs.step_history when available. return 7_000 MANIFEST = AgentManifest( id="health-vitals", version="1.0.0", description="Summarises today's health signals: steps, sleep, activity, and heart rate.", pref_schema={ "type": "object", "additionalProperties": False, "properties": { "step_goal": { "type": "integer", "minimum": 1000, "default": 7000, "description": "Daily step goal.", }, "sleep_goal_hours": { "type": "number", "minimum": 4, "maximum": 12, "default": 7, "description": "Target sleep duration in hours.", }, }, }, context_schema=["google-health.steps", "google-health.sleep", "google-health.activity", "google-health.heart_rate"], required_consents=["data:core", "data:google-health", "agent:health-vitals"], output_contract={"type": "snippet", "format": "free_text"}, ttl_sec=1800, # refresh every 30 min — health data changes during the day silenced_in_contexts=[], inferred_params=[ InferredParam( key="step_goal", ttl_sec=7 * 86_400, cold_start_default=7000, min_history=0, infer=lambda h: 7000, # static default; override via user pref ), ], ) class HealthVitalsAgent(BaseAgent): """Summarises today's health signals into an orchestrator prompt snippet.""" agent_id: ClassVar[str] = MANIFEST.id ttl_seconds: ClassVar[int] = MANIFEST.ttl_sec version: ClassVar[str] = MANIFEST.version def compute(self, inp: AgentInput) -> AgentOutput: step_goal = int(inp.agent_prefs.get("step_goal", 7000)) sleep_goal = float(inp.agent_prefs.get("sleep_goal_hours", 7.0)) health = [t for t in inp.tasks if t.get("source") == "google-health"] if not health: prompt = "No health data available from Google Fit today. (Always write the tip in English.)" return self._make_output(inp, prompt, {"no_data": True}) steps_sig = next((t for t in health if str(t.get("id", "")).endswith(":steps")), None) sleep_sig = next((t for t in health if str(t.get("id", "")).endswith(":sleep")), None) activity_sig = next((t for t in health if str(t.get("id", "")).endswith(":activity")), None) hr_sig = next((t for t in health if str(t.get("id", "")).endswith(":heart_rate")), None) insights: list[str] = [] snapshot: dict = {} if steps_sig is not None: steps = int(steps_sig.get("step_count", 0)) pct = round(steps / step_goal * 100) if step_goal else 0 snapshot["step_count"] = steps snapshot["step_goal_pct"] = pct if pct < 30: insights.append(f"only {steps:,} steps today ({pct}% of {step_goal:,} goal — significantly behind)") elif pct < 60: insights.append(f"{steps:,} steps today ({pct}% of {step_goal:,} goal)") elif pct >= 100: insights.append(f"{steps:,} steps today (daily goal reached!)") else: insights.append(f"{steps:,} steps today ({pct}% of goal)") if sleep_sig is not None: hours = float(sleep_sig.get("sleep_hours", 0)) deficit = max(0.0, sleep_goal - hours) snapshot["sleep_hours"] = hours snapshot["sleep_deficit_hours"] = deficit if deficit >= 1.5: insights.append(f"only {hours:.1f}h sleep last night ({deficit:.1f}h below the {sleep_goal:.0f}h goal)") elif deficit > 0: insights.append(f"{hours:.1f}h sleep last night (slightly below {sleep_goal:.0f}h goal)") else: insights.append(f"{hours:.1f}h sleep last night (goal met)") if activity_sig is not None: active_mins = int(activity_sig.get("active_minutes", 0)) calories = int(activity_sig.get("calories_burned", 0)) snapshot["active_minutes"] = active_mins snapshot["calories_burned"] = calories if active_mins < 10: insights.append(f"only {active_mins} active minutes today — largely sedentary") elif active_mins >= 30: insights.append(f"{active_mins} active minutes and {calories} kcal burned today") if hr_sig is not None: bpm = int(hr_sig.get("resting_bpm", 0)) snapshot["resting_bpm"] = bpm if bpm > 90: insights.append(f"elevated resting heart rate: {bpm} bpm") elif bpm > 0: insights.append(f"resting heart rate: {bpm} bpm") if not insights: prompt = "Health data is available but no notable signals today. (Always write the tip in English.)" else: body = "; ".join(insights) prompt = f"Health snapshot: {body}. (Always write the tip in English.)" return self._make_output(inp, prompt, snapshot)