508 lines
20 KiB
Python
508 lines
20 KiB
Python
import asyncio
|
||
import json
|
||
import logging
|
||
import math
|
||
import os
|
||
import threading
|
||
import uuid
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from xml.etree import ElementTree as ET
|
||
|
||
from fastapi import FastAPI, HTTPException, Query, Request
|
||
from fastapi.responses import StreamingResponse
|
||
from fastapi.staticfiles import StaticFiles
|
||
from garminconnect import Garmin, GarminConnectAuthenticationError
|
||
from openai import OpenAI
|
||
|
||
logging.basicConfig(
|
||
level=logging.DEBUG,
|
||
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
|
||
datefmt="%H:%M:%S",
|
||
)
|
||
log = logging.getLogger("gpxport")
|
||
|
||
GPX_DIR = Path(__file__).parent / "garmin_gpx_exports"
|
||
INSIGHTS_FILE = Path(__file__).parent / "insights.json"
|
||
LLAMACPP_BASE_URL = os.getenv("LLAMACPP_BASE_URL", "http://localhost:8080/v1")
|
||
MODEL_NAME = "local-model" # overwritten at startup by auto-discovery
|
||
|
||
app = FastAPI()
|
||
|
||
_insights_cache: dict[str, str] = {}
|
||
_mfa_sessions: dict[str, dict] = {}
|
||
|
||
|
||
def _save_insights() -> None:
|
||
INSIGHTS_FILE.write_text(json.dumps(_insights_cache, indent=2, ensure_ascii=False))
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# GPX parsing helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
NS = {
|
||
"gpx": "http://www.topografix.com/GPX/1/1",
|
||
"ns3": "http://www.garmin.com/xmlschemas/TrackPointExtension/v1",
|
||
}
|
||
|
||
|
||
def haversine(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
|
||
"""Return distance in metres between two GPS coordinates."""
|
||
R = 6_371_000
|
||
phi1, phi2 = math.radians(lat1), math.radians(lat2)
|
||
dphi = math.radians(lat2 - lat1)
|
||
dlambda = math.radians(lon2 - lon1)
|
||
a = math.sin(dphi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlambda / 2) ** 2
|
||
return 2 * R * math.asin(math.sqrt(a))
|
||
|
||
|
||
def parse_gpx(path: Path) -> dict:
|
||
tree = ET.parse(path)
|
||
root = tree.getroot()
|
||
|
||
name_el = root.find(".//gpx:trk/gpx:name", NS)
|
||
name = name_el.text if name_el is not None else path.stem
|
||
type_el = root.find(".//gpx:trk/gpx:type", NS)
|
||
activity_type = type_el.text if type_el is not None else "cycling"
|
||
|
||
points = []
|
||
for pt in root.findall(".//gpx:trkpt", NS):
|
||
lat = float(pt.get("lat"))
|
||
lon = float(pt.get("lon"))
|
||
ele_el = pt.find("gpx:ele", NS)
|
||
ele = float(ele_el.text) if ele_el is not None else 0.0
|
||
time_el = pt.find("gpx:time", NS)
|
||
ts = None
|
||
if time_el is not None:
|
||
ts = datetime.fromisoformat(time_el.text.replace("Z", "+00:00"))
|
||
hr_el = pt.find(".//ns3:hr", NS)
|
||
hr = int(hr_el.text) if hr_el is not None else None
|
||
points.append({"lat": lat, "lon": lon, "ele": ele, "time": ts, "hr": hr})
|
||
|
||
# Compute stats
|
||
distance_m = 0.0
|
||
ele_gain = 0.0
|
||
ele_loss = 0.0
|
||
for i in range(1, len(points)):
|
||
p0, p1 = points[i - 1], points[i]
|
||
distance_m += haversine(p0["lat"], p0["lon"], p1["lat"], p1["lon"])
|
||
de = p1["ele"] - p0["ele"]
|
||
if de > 0:
|
||
ele_gain += de
|
||
else:
|
||
ele_loss += abs(de)
|
||
|
||
duration_s = 0
|
||
start_time = None
|
||
end_time = None
|
||
if points and points[0]["time"] and points[-1]["time"]:
|
||
start_time = points[0]["time"]
|
||
end_time = points[-1]["time"]
|
||
duration_s = int((end_time - start_time).total_seconds())
|
||
|
||
hrs = [p["hr"] for p in points if p["hr"] is not None]
|
||
avg_hr = round(sum(hrs) / len(hrs)) if hrs else None
|
||
max_hr = max(hrs) if hrs else None
|
||
|
||
avg_speed_kmh = (distance_m / 1000) / (duration_s / 3600) if duration_s > 0 else 0
|
||
|
||
# HR zone distribution (% of points)
|
||
zone_counts = {"z1": 0, "z2": 0, "z3": 0, "z4": 0, "z5": 0}
|
||
for h in hrs:
|
||
if h < 120:
|
||
zone_counts["z1"] += 1
|
||
elif h < 140:
|
||
zone_counts["z2"] += 1
|
||
elif h < 160:
|
||
zone_counts["z3"] += 1
|
||
elif h < 180:
|
||
zone_counts["z4"] += 1
|
||
else:
|
||
zone_counts["z5"] += 1
|
||
total_hr_pts = len(hrs)
|
||
hr_zones = {
|
||
k: round(v / total_hr_pts * 100, 1) if total_hr_pts else 0
|
||
for k, v in zone_counts.items()
|
||
}
|
||
|
||
ride_id = path.stem # use filename without extension as ID
|
||
|
||
return {
|
||
"id": ride_id,
|
||
"name": name,
|
||
"activity_type": activity_type,
|
||
"filename": path.name,
|
||
"start_time": start_time.isoformat() if start_time else None,
|
||
"end_time": end_time.isoformat() if end_time else None,
|
||
"duration_s": duration_s,
|
||
"distance_m": round(distance_m),
|
||
"distance_km": round(distance_m / 1000, 2),
|
||
"avg_speed_kmh": round(avg_speed_kmh, 1),
|
||
"elevation_gain_m": round(ele_gain),
|
||
"elevation_loss_m": round(ele_loss),
|
||
"avg_hr": avg_hr,
|
||
"max_hr": max_hr,
|
||
"hr_zones": hr_zones,
|
||
"has_gps": len(points) > 0 and points[0].get("lat") is not None,
|
||
"points": [
|
||
{
|
||
"lat": p["lat"],
|
||
"lon": p["lon"],
|
||
"ele": round(p["ele"], 1),
|
||
"time": p["time"].isoformat() if p["time"] else None,
|
||
"hr": p["hr"],
|
||
}
|
||
for p in points
|
||
],
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Load all rides at startup
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_rides: dict[str, dict] = {}
|
||
|
||
|
||
@app.on_event("startup")
|
||
def load_rides():
|
||
global MODEL_NAME
|
||
# Auto-discover model name from llamacpp
|
||
try:
|
||
client = OpenAI(base_url=LLAMACPP_BASE_URL, api_key="not-needed")
|
||
models = client.models.list()
|
||
first = next(iter(models), None)
|
||
if first:
|
||
MODEL_NAME = first.id
|
||
log.info("Discovered model: %s", MODEL_NAME)
|
||
else:
|
||
log.warning("No models returned from %s — using fallback '%s'", LLAMACPP_BASE_URL, MODEL_NAME)
|
||
except Exception as exc:
|
||
log.warning("Could not reach llamacpp at %s (%s) — using fallback '%s'", LLAMACPP_BASE_URL, exc, MODEL_NAME)
|
||
|
||
for gpx_file in sorted(GPX_DIR.glob("*.gpx"), reverse=True):
|
||
try:
|
||
ride = parse_gpx(gpx_file)
|
||
_rides[ride["id"]] = ride
|
||
except Exception as exc:
|
||
log.error("Failed to parse %s: %s", gpx_file.name, exc)
|
||
log.info("Loaded %d rides.", len(_rides))
|
||
|
||
if INSIGHTS_FILE.exists():
|
||
try:
|
||
_insights_cache.update(json.loads(INSIGHTS_FILE.read_text()))
|
||
cached = sum(1 for rid in _insights_cache if rid in _rides)
|
||
log.info("Loaded %d cached insights from %s", cached, INSIGHTS_FILE)
|
||
except Exception as exc:
|
||
log.warning("Could not load insights cache: %s", exc)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# API endpoints
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@app.get("/api/rides")
|
||
def list_rides():
|
||
summary_keys = [
|
||
"id", "name", "activity_type", "start_time", "duration_s",
|
||
"distance_km", "avg_speed_kmh", "elevation_gain_m",
|
||
"avg_hr", "max_hr", "has_gps",
|
||
]
|
||
rides = []
|
||
for r in _rides.values():
|
||
rides.append({k: r[k] for k in summary_keys})
|
||
# Sort by start_time descending
|
||
rides.sort(key=lambda x: x["start_time"] or "", reverse=True)
|
||
return rides
|
||
|
||
|
||
@app.get("/api/rides/{ride_id}")
|
||
def get_ride(ride_id: str):
|
||
if ride_id not in _rides:
|
||
raise HTTPException(status_code=404, detail="Ride not found")
|
||
r = _rides[ride_id]
|
||
result = {k: v for k, v in r.items() if k != "points"}
|
||
result["cached_insight"] = _insights_cache.get(ride_id)
|
||
return result
|
||
|
||
|
||
@app.get("/api/rides/{ride_id}/points")
|
||
def get_ride_points(ride_id: str):
|
||
if ride_id not in _rides:
|
||
raise HTTPException(status_code=404, detail="Ride not found")
|
||
return _rides[ride_id]["points"]
|
||
|
||
|
||
def _karvonen_zones(rhr: int, max_hr: int, points: list[dict]) -> dict:
|
||
hrr = max_hr - rhr
|
||
bounds = [rhr + p * hrr for p in (0.50, 0.60, 0.70, 0.80)]
|
||
labels = [
|
||
f"Z1 <{round(bounds[0])}bpm",
|
||
f"Z2 {round(bounds[0])}–{round(bounds[1])}bpm",
|
||
f"Z3 {round(bounds[1])}–{round(bounds[2])}bpm",
|
||
f"Z4 {round(bounds[2])}–{round(bounds[3])}bpm",
|
||
f"Z5 >{round(bounds[3])}bpm",
|
||
]
|
||
hrs = [p["hr"] for p in points if p.get("hr") is not None]
|
||
if not hrs:
|
||
return {l: 0 for l in labels}
|
||
counts = [0] * 5
|
||
for h in hrs:
|
||
if h < bounds[0]: counts[0] += 1
|
||
elif h < bounds[1]: counts[1] += 1
|
||
elif h < bounds[2]: counts[2] += 1
|
||
elif h < bounds[3]: counts[3] += 1
|
||
else: counts[4] += 1
|
||
return {labels[i]: round(counts[i] / len(hrs) * 100, 1) for i in range(5)}
|
||
|
||
|
||
def _llm_client() -> OpenAI:
|
||
return OpenAI(base_url=LLAMACPP_BASE_URL, api_key="not-needed")
|
||
|
||
|
||
def _get_previous_insight(ride: dict) -> str | None:
|
||
"""Return the cached insight for the most recent ride before this one, or None."""
|
||
ride_time = ride["start_time"]
|
||
if not ride_time:
|
||
return None
|
||
gps_rides = [r for r in _rides.values() if r["has_gps"] and r["distance_km"] > 0 and r["start_time"]]
|
||
prior = sorted(
|
||
[r for r in gps_rides if r["start_time"] < ride_time and r["id"] in _insights_cache],
|
||
key=lambda r: r["start_time"],
|
||
)
|
||
if not prior:
|
||
return None
|
||
return _insights_cache[prior[-1]["id"]]
|
||
|
||
|
||
def _build_prompt(ride: dict, rhr: int, global_max_hr: int) -> str:
|
||
ride_time = ride["start_time"]
|
||
gps_rides = [r for r in _rides.values() if r["has_gps"] and r["distance_km"] > 0 and r["start_time"]]
|
||
past_rides = [r for r in gps_rides if r["start_time"] < ride_time]
|
||
|
||
zones = _karvonen_zones(rhr, global_max_hr, ride["points"])
|
||
zone_str = ", ".join(f"{k}={v}%" for k, v in zones.items())
|
||
duration_min = ride["duration_s"] // 60
|
||
ride_date = ride_time[:10] if ride_time else "unknown"
|
||
|
||
if past_rides:
|
||
avg_dist = sum(r["distance_km"] for r in past_rides) / len(past_rides)
|
||
hr_list = [r["avg_hr"] for r in past_rides if r["avg_hr"]]
|
||
avg_hr = sum(hr_list) / len(hr_list) if hr_list else 0
|
||
history_str = f"Before today they've completed {len(past_rides)} rides, averaging {avg_dist:.1f} km and {avg_hr:.0f} bpm."
|
||
else:
|
||
history_str = "This is one of their earliest recorded rides — no prior history to compare against."
|
||
|
||
previous_insight = _get_previous_insight(ride)
|
||
if previous_insight:
|
||
prior_section = f"""
|
||
Your feedback from their last ride:
|
||
---
|
||
{previous_insight}
|
||
---
|
||
|
||
Based on that advice, briefly note (in one sentence woven naturally into your response) whether today's data suggests they followed it or not. Do not lecture about it — just one honest observation."""
|
||
else:
|
||
prior_section = ""
|
||
|
||
return f"""You are a friendly cycling coach giving post-ride feedback. You are writing this on {ride_date}, the day of the ride. Do not plan, outline, or show any reasoning — just write the response directly.
|
||
|
||
Today's ride: {duration_min} min, {ride["distance_km"]} km, {ride["avg_speed_kmh"]} km/h, {ride["elevation_gain_m"]} m elevation. Avg HR {ride["avg_hr"]}, max {ride["max_hr"]}, resting HR {rhr} bpm (lifetime max {global_max_hr}). Time in zones: {zone_str}.
|
||
|
||
Athlete history (as of {ride_date}): {history_str}{prior_section}
|
||
|
||
Your response must begin with exactly one line in this format:
|
||
TITLE: <a funny, punchy title for this specific ride in 4–8 words>
|
||
|
||
Then one blank line, then write exactly 3–4 short paragraphs in a warm, direct "you/your" voice: effort level, what the zones reveal about the session type, how it compares to their history so far, one concrete suggestion. No bullet points, no headers, no preamble. Begin writing now."""
|
||
|
||
|
||
@app.post("/api/rides/{ride_id}/insights")
|
||
def get_insights(ride_id: str, rhr: int = Query(60, ge=30, le=100)):
|
||
if ride_id not in _rides:
|
||
raise HTTPException(status_code=404, detail="Ride not found")
|
||
ride = _rides[ride_id]
|
||
|
||
gps_rides = [r for r in _rides.values() if r["has_gps"] and r["distance_km"] > 0]
|
||
global_max_hr = max((r["max_hr"] or 0 for r in gps_rides), default=200)
|
||
prompt = _build_prompt(ride, rhr, global_max_hr)
|
||
|
||
client = _llm_client()
|
||
log.info("POST /insights ride_id=%s rhr=%d", ride_id, rhr)
|
||
log.debug("Prompt:\n%s", prompt)
|
||
|
||
def stream_response():
|
||
accumulated = []
|
||
try:
|
||
log.info("Opening stream...")
|
||
stream = client.chat.completions.create(
|
||
model=MODEL_NAME,
|
||
messages=[{"role": "user", "content": prompt}],
|
||
stream=True,
|
||
max_tokens=600,
|
||
extra_body={"chat_template_kwargs": {"enable_thinking": False}},
|
||
)
|
||
chunks_received = 0
|
||
for chunk in stream:
|
||
chunks_received += 1
|
||
if chunks_received <= 3:
|
||
log.debug("Chunk #%d raw: %s", chunks_received, chunk.model_dump())
|
||
d = chunk.choices[0].delta.model_dump()
|
||
delta = d.get("content") or d.get("reasoning_content")
|
||
if delta:
|
||
accumulated.append(delta)
|
||
yield delta
|
||
log.info("Stream complete — %d chunks, %d chars", chunks_received, sum(len(c) for c in accumulated))
|
||
_insights_cache[ride_id] = "".join(accumulated)
|
||
_save_insights()
|
||
log.info("Saved insight for %s", ride_id)
|
||
except Exception as exc:
|
||
log.exception("Error streaming insights: %s", exc)
|
||
yield f"\n\n[Error: {exc}]"
|
||
|
||
return StreamingResponse(stream_response(), media_type="text/plain")
|
||
|
||
|
||
@app.post("/api/insights/generate-all")
|
||
def generate_all_insights(rhr: int = Query(60, ge=30, le=100)):
|
||
gps_rides = [r for r in _rides.values() if r["has_gps"] and r["distance_km"] > 0 and r["start_time"]]
|
||
global_max_hr = max((r["max_hr"] or 0 for r in gps_rides), default=200)
|
||
# Chronological order so time-aware context is always correct
|
||
ordered = sorted(gps_rides, key=lambda r: r["start_time"])
|
||
to_generate = [r for r in ordered if r["id"] not in _insights_cache]
|
||
total = len(to_generate)
|
||
log.info("generate-all: %d rides to generate (rhr=%d)", total, rhr)
|
||
|
||
def sse_stream():
|
||
client = _llm_client()
|
||
for idx, ride in enumerate(to_generate, 1):
|
||
ride_id = ride["id"]
|
||
yield f"data: {json.dumps({'index': idx, 'total': total, 'ride_id': ride_id, 'status': 'generating'})}\n\n"
|
||
log.info("Generating insight %d/%d: %s", idx, total, ride_id)
|
||
try:
|
||
prompt = _build_prompt(ride, rhr, global_max_hr)
|
||
response = client.chat.completions.create(
|
||
model=MODEL_NAME,
|
||
messages=[{"role": "user", "content": prompt}],
|
||
stream=False,
|
||
max_tokens=600,
|
||
extra_body={"chat_template_kwargs": {"enable_thinking": False}},
|
||
)
|
||
msg = response.choices[0].message.model_dump()
|
||
text = msg.get("content") or msg.get("reasoning_content") or ""
|
||
_insights_cache[ride_id] = text
|
||
_save_insights()
|
||
yield f"data: {json.dumps({'index': idx, 'total': total, 'ride_id': ride_id, 'status': 'done'})}\n\n"
|
||
except Exception as exc:
|
||
log.exception("Failed insight for %s: %s", ride_id, exc)
|
||
yield f"data: {json.dumps({'index': idx, 'total': total, 'ride_id': ride_id, 'status': 'error', 'error': str(exc)})}\n\n"
|
||
yield f"data: {json.dumps({'status': 'complete', 'total': total})}\n\n"
|
||
|
||
return StreamingResponse(sse_stream(), media_type="text/event-stream")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Garmin Connect sync
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@app.post("/api/fetch-rides")
|
||
async def fetch_rides_from_garmin(request: Request):
|
||
body = await request.json()
|
||
email = body.get("email", "")
|
||
password = body.get("password", "")
|
||
|
||
session_id = uuid.uuid4().hex
|
||
queue: asyncio.Queue = asyncio.Queue()
|
||
loop = asyncio.get_running_loop()
|
||
mfa_event = threading.Event()
|
||
_mfa_sessions[session_id] = {"event": mfa_event, "code": None}
|
||
|
||
def emit(evt):
|
||
loop.call_soon_threadsafe(queue.put_nowait, evt)
|
||
|
||
def prompt_mfa():
|
||
emit({"status": "mfa_required", "session_id": session_id})
|
||
mfa_event.wait(timeout=120)
|
||
return _mfa_sessions.get(session_id, {}).get("code") or ""
|
||
|
||
def garmin_worker():
|
||
try:
|
||
emit({"status": "auth"})
|
||
client = Garmin(email=email, password=password, prompt_mfa=prompt_mfa)
|
||
client.login()
|
||
|
||
emit({"status": "fetching"})
|
||
activities = client.get_activities(0, 50)
|
||
cycling = [
|
||
a for a in activities
|
||
if "cycling" in a.get("activityType", {}).get("typeKey", "")
|
||
]
|
||
|
||
to_download = []
|
||
for act in cycling:
|
||
act_id = act["activityId"]
|
||
start = act.get("startTimeLocal", "")
|
||
name = act.get("activityName", "activity")
|
||
safe_time = start.replace(":", "-").replace(" ", "_")
|
||
safe_name = "".join(c for c in name if c.isalnum() or c in (" ", "_", "-")).strip()
|
||
filename = f"{safe_time}_{safe_name}_{act_id}.gpx"
|
||
if not (GPX_DIR / filename).exists():
|
||
to_download.append((act_id, name, GPX_DIR / filename))
|
||
|
||
skipped = len(cycling) - len(to_download)
|
||
added = 0
|
||
for idx, (act_id, name, filepath) in enumerate(to_download, 1):
|
||
emit({"status": "downloading", "index": idx, "total": len(to_download), "name": name})
|
||
try:
|
||
gpx_data = client.download_activity(act_id, dl_fmt=client.ActivityDownloadFormat.GPX)
|
||
filepath.write_bytes(gpx_data)
|
||
ride = parse_gpx(filepath)
|
||
_rides[ride["id"]] = ride
|
||
added += 1
|
||
except Exception as exc:
|
||
log.error("Failed to download activity %s: %s", act_id, exc)
|
||
|
||
emit({"status": "complete", "added": added, "skipped": skipped})
|
||
except GarminConnectAuthenticationError:
|
||
emit({"status": "error", "message": "Login failed. Check your email and password."})
|
||
except Exception as exc:
|
||
log.exception("fetch-rides error: %s", exc)
|
||
emit({"status": "error", "message": str(exc)})
|
||
finally:
|
||
_mfa_sessions.pop(session_id, None)
|
||
emit(None) # sentinel to end the SSE stream
|
||
|
||
async def generate():
|
||
loop.run_in_executor(None, garmin_worker)
|
||
while True:
|
||
evt = await queue.get()
|
||
if evt is None:
|
||
break
|
||
yield f"data: {json.dumps(evt)}\n\n"
|
||
|
||
return StreamingResponse(generate(), media_type="text/event-stream")
|
||
|
||
|
||
@app.post("/api/mfa")
|
||
async def submit_mfa(request: Request):
|
||
body = await request.json()
|
||
session_id = body.get("session_id", "")
|
||
code = body.get("code", "")
|
||
session = _mfa_sessions.get(session_id)
|
||
if not session:
|
||
raise HTTPException(status_code=404, detail="Session not found or expired")
|
||
session["code"] = code
|
||
session["event"].set()
|
||
return {"ok": True}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Serve frontend
|
||
# ---------------------------------------------------------------------------
|
||
|
||
static_dir = Path(__file__).parent / "static"
|
||
static_dir.mkdir(exist_ok=True)
|
||
app.mount("/", StaticFiles(directory=static_dir, html=True), name="static")
|