"""Today API — Aggregate calendar, weather, and health for dashboard."""
import os
import json
import httpx
import asyncio
from datetime import datetime, timezone
from typing import Optional
Radicale CalDAV config
RADICALE_URL = os.getenv("RADICALE_URL", "http://localhost:5232")
RADICALE_USER = os.getenv("RADICALE_USER", "assistant")
RADICALE_PASS = os.getenv("RADICALE_PASS", "family-assistant-2026")
Weather config
WEATHER_LOCATION = os.getenv("WEATHER_LOCATION", "Green Bay, WI")
async def fetch_calendar_today() -> dict:
"""Fetch today's events from Radicale CalDAV."""
try:
import caldav
client = caldav.DAVClient(
url=f"{RADICALE_URL}/",
username=RADICALE_USER,
password=RADICALE_PASS
)
principal = client.principal()
calendars = principal.calendars()
if not calendars:
return {"status": "error", "error": "No calendars found"}
# Get events for today
today = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0)
tomorrow = today.replace(day=today.day + 1)
events = []
for cal in calendars:
try:
cal_events = cal.date_search(start=today, end=tomorrow)
for event in cal_events:
vevent = event.instance.vobject_instance.vevent
events.append({
"summary": str(vevent.summary.value) if hasattr(vevent, 'summary') else "Untitled",
"start": vevent.dtstart.value.isoformat() if hasattr(vevent, 'dtstart') else today.isoformat(),
"end": vevent.dtend.value.isoformat() if hasattr(vevent, 'dtend') else None,
"location": str(vevent.location.value) if hasattr(vevent, 'location') else None,
"is_all_day": not hasattr(vevent.dtstart.value, 'hour')
})
except Exception:
continue
# Sort by start time
events.sort(key=lambda x: x["start"])
return {"status": "ok", "events": events}
except ImportError:
return {"status": "error", "error": "caldav not installed"}
except Exception as e:
return {"status": "error", "error": str(e)}
async def fetch_weather() -> dict:
"""Fetch weather from wttr.in (no API key)."""
try:
location = WEATHER_LOCATION.replace(" ", "+")
async with httpx.AsyncClient() as client:
response = await client.get(
f"https://wttr.in/{location}?format=j1",
timeout=10.0
)
data = response.json()
current = data.get("current_condition", [{}])[0]
# Get next 6 hours from hourly forecast
hourly = data.get("weather", [{}])[0].get("hourly", [])[:6]
forecast = []
for hour in hourly:
# Fix hour formatting (wttr.in returns 0, 300, 600, etc.)
time_raw = hour.get("time", "0")
if len(time_raw) >= 3:
hour_str = time_raw[:-2] if len(time_raw) > 2 else "0"
else:
hour_str = time_raw
try:
hour_int = int(hour_str)
hour_display = f"{hour_int % 24}:00"
except ValueError:
hour_display = time_raw
forecast.append({
"hour": hour_display,
"temperature_f": float(hour.get("tempF", 0)),
"precipitation_chance_pct": int(hour.get("chanceofrain", 0))
})
return {
"status": "ok",
"current": {
"temperature_f": float(current.get("temp_F", 0)),
"feels_like_f": float(current.get("FeelsLikeF", 0)),
"condition": current.get("weatherDesc", [{"value": "Unknown"}])[0].get("value", "Unknown"),
"humidity_pct": int(current.get("humidity", 0)),
"wind_mph": float(current.get("windspeedMiles", 0))
},
"forecast": forecast
}
except Exception as e:
return {"status": "error", "error": str(e)}
async def fetch_health() -> dict:
"""Ping system components and return health status."""
components = {}
# OpenClaw Gateway
try:
async with httpx.AsyncClient() as client:
r = await client.get("http://localhost:18789/health", timeout=5.0)
components["openclaw_gateway"] = {
"status": "ok" if r.status_code == 200 else "degraded",
"latency_ms": int(r.elapsed.total_seconds() * 1000)
}
except Exception:
components["openclaw_gateway"] = {"status": "critical", "latency_ms": 9999}
# Radicale CalDAV
try:
async with httpx.AsyncClient() as client:
r = await client.get(f"{RADICALE_URL}/.well-known/caldav", timeout=5.0)
components["radicale_caldav"] = {
"status": "ok" if r.status_code in [200, 301, 308] else "degraded",
"latency_ms": int(r.elapsed.total_seconds() * 1000)
}
except Exception:
components["radicale_caldav"] = {"status": "critical", "latency_ms": 9999}
# Ollama Local
try:
async with httpx.AsyncClient() as client:
r = await client.get("http://localhost:11434/api/tags", timeout=5.0)
components["ollama_local"] = {
"status": "ok" if r.status_code == 200 else "degraded",
"latency_ms": int(r.elapsed.total_seconds() * 1000)
}
except Exception:
components["ollama_local"] = {"status": "critical", "latency_ms": 9999}
# Cloudflare Tunnel
try:
async with httpx.AsyncClient() as client:
r = await client.get("https://hoffdesk.com/health", timeout=10.0)
components["cloudflare_tunnel"] = {
"status": "ok" if r.status_code == 200 else "degraded",
"latency_ms": int(r.elapsed.total_seconds() * 1000)
}
except Exception:
components["cloudflare_tunnel"] = {"status": "critical", "latency_ms": 9999}
# Disk usage
try:
import shutil
total, used, free = shutil.disk_usage("/")
disk_usage = {
"total_gb": round(total / (1024**3), 1),
"used_gb": round(used / (1024**3), 1),
"usage_pct": round((used / total) * 100, 1)
}
except Exception:
disk_usage = {"total_gb": 0, "used_gb": 0, "usage_pct": 0}
# Overall status
statuses = [c["status"] for c in components.values()]
if any(s == "critical" for s in statuses):
overall = "critical"
elif any(s == "degraded" for s in statuses):
overall = "degraded"
else:
overall = "healthy"
return {
"overall_status": overall,
"components": components,
"disk_usage": disk_usage
}
async def get_today_data() -> dict:
"""Aggregate all today data."""
calendar, weather, health = await asyncio.gather(
fetch_calendar_today(),
fetch_weather(),
fetch_health()
)
return {
"calendar": calendar,
"weather": weather,
"health": health,
"generated_at": datetime.now(timezone.utc).isoformat()
}