mitai-jinkendo/backend/routers/charts.py
Lars c79cc9eafb
All checks were successful
Deploy Development / deploy (push) Successful in 47s
Build Test / lint-backend (push) Successful in 0s
Build Test / build-frontend (push) Successful in 15s
feat: Phase 0c - Multi-Layer Data Architecture (Proof of Concept)
- Add data_layer/ module structure with utils.py + body_metrics.py
- Migrate 3 functions: weight_trend, body_composition, circumference_summary
- Refactor placeholders to use data layer
- Add charts router with 3 Chart.js endpoints
- Tests: Syntax , Confidence logic 

Phase 0c PoC (3 functions): Foundation for 40+ remaining functions

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-28 18:26:22 +01:00

329 lines
9.7 KiB
Python

"""
Charts Router - Chart.js-compatible Data Endpoints
Provides structured data for frontend charts/diagrams.
All endpoints return Chart.js-compatible JSON format:
{
"chart_type": "line" | "bar" | "scatter" | "pie",
"data": {
"labels": [...],
"datasets": [...]
},
"metadata": {
"confidence": "high" | "medium" | "low" | "insufficient",
"data_points": int,
...
}
}
Phase 0c: Multi-Layer Architecture
Version: 1.0
"""
from fastapi import APIRouter, Depends, HTTPException, Query
from typing import Dict, List, Optional
from datetime import datetime, timedelta
from auth import require_auth
from data_layer.body_metrics import (
get_weight_trend_data,
get_body_composition_data,
get_circumference_summary_data
)
from data_layer.utils import serialize_dates
router = APIRouter()
# ── Body Charts ─────────────────────────────────────────────────────────────
@router.get("/charts/weight-trend")
def get_weight_trend_chart(
days: int = Query(default=90, ge=7, le=365, description="Analysis window in days"),
session: dict = Depends(require_auth)
) -> Dict:
"""
Weight trend chart data.
Returns Chart.js-compatible line chart with:
- Raw weight values
- Trend line (if enough data)
- Confidence indicator
Args:
days: Analysis window (7-365 days, default 90)
session: Auth session (injected)
Returns:
{
"chart_type": "line",
"data": {
"labels": ["2026-01-01", ...],
"datasets": [{
"label": "Gewicht",
"data": [85.0, 84.5, ...],
"borderColor": "#1D9E75",
"backgroundColor": "rgba(29, 158, 117, 0.1)",
"tension": 0.4
}]
},
"metadata": {
"confidence": "high",
"data_points": 60,
"first_value": 92.0,
"last_value": 85.0,
"delta": -7.0,
"direction": "decreasing"
}
}
Metadata fields:
- confidence: Data quality ("high", "medium", "low", "insufficient")
- data_points: Number of weight entries in period
- first_value: First weight in period (kg)
- last_value: Latest weight (kg)
- delta: Weight change (kg, negative = loss)
- direction: "increasing" | "decreasing" | "stable"
"""
profile_id = session['profile_id']
# Get structured data from data layer
trend_data = get_weight_trend_data(profile_id, days)
# Early return if insufficient data
if trend_data['confidence'] == 'insufficient':
return {
"chart_type": "line",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Nicht genug Daten für Trend-Analyse"
}
}
# Get raw data points for chart
from db import get_db, get_cursor
with get_db() as conn:
cur = get_cursor(conn)
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
cur.execute(
"""SELECT date, weight FROM weight_log
WHERE profile_id=%s AND date >= %s
ORDER BY date""",
(profile_id, cutoff)
)
rows = cur.fetchall()
# Format for Chart.js
labels = [row['date'].isoformat() for row in rows]
values = [float(row['weight']) for row in rows]
return {
"chart_type": "line",
"data": {
"labels": labels,
"datasets": [
{
"label": "Gewicht",
"data": values,
"borderColor": "#1D9E75",
"backgroundColor": "rgba(29, 158, 117, 0.1)",
"borderWidth": 2,
"tension": 0.4,
"fill": True,
"pointRadius": 3,
"pointHoverRadius": 5
}
]
},
"metadata": serialize_dates({
"confidence": trend_data['confidence'],
"data_points": trend_data['data_points'],
"first_value": trend_data['first_value'],
"last_value": trend_data['last_value'],
"delta": trend_data['delta'],
"direction": trend_data['direction'],
"first_date": trend_data['first_date'],
"last_date": trend_data['last_date'],
"days_analyzed": trend_data['days_analyzed']
})
}
@router.get("/charts/body-composition")
def get_body_composition_chart(
days: int = Query(default=90, ge=7, le=365),
session: dict = Depends(require_auth)
) -> Dict:
"""
Body composition chart (body fat percentage, lean mass trend).
Returns Chart.js-compatible multi-line chart.
Args:
days: Analysis window (7-365 days, default 90)
session: Auth session (injected)
Returns:
Chart.js format with datasets for:
- Body fat percentage (%)
- Lean mass (kg, if weight available)
"""
profile_id = session['profile_id']
# Get latest body composition
comp_data = get_body_composition_data(profile_id, days)
if comp_data['confidence'] == 'insufficient':
return {
"chart_type": "line",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine Körperfett-Messungen vorhanden"
}
}
# For PoC: Return single data point
# TODO in bulk migration: Fetch historical caliper entries
return {
"chart_type": "line",
"data": {
"labels": [comp_data['date'].isoformat() if comp_data['date'] else ""],
"datasets": [
{
"label": "Körperfett %",
"data": [comp_data['body_fat_pct']],
"borderColor": "#D85A30",
"backgroundColor": "rgba(216, 90, 48, 0.1)",
"borderWidth": 2
}
]
},
"metadata": serialize_dates({
"confidence": comp_data['confidence'],
"data_points": comp_data['data_points'],
"body_fat_pct": comp_data['body_fat_pct'],
"method": comp_data['method'],
"date": comp_data['date']
})
}
@router.get("/charts/circumferences")
def get_circumferences_chart(
max_age_days: int = Query(default=90, ge=7, le=365),
session: dict = Depends(require_auth)
) -> Dict:
"""
Latest circumference measurements as bar chart.
Shows most recent measurement for each body point.
Args:
max_age_days: Maximum age of measurements (default 90)
session: Auth session (injected)
Returns:
Chart.js bar chart with all circumference points
"""
profile_id = session['profile_id']
circ_data = get_circumference_summary_data(profile_id, max_age_days)
if circ_data['confidence'] == 'insufficient':
return {
"chart_type": "bar",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine Umfangsmessungen vorhanden"
}
}
# Sort by value (descending) for better visualization
measurements = sorted(
circ_data['measurements'],
key=lambda m: m['value'],
reverse=True
)
labels = [m['point'] for m in measurements]
values = [m['value'] for m in measurements]
return {
"chart_type": "bar",
"data": {
"labels": labels,
"datasets": [
{
"label": "Umfang (cm)",
"data": values,
"backgroundColor": "#1D9E75",
"borderColor": "#085041",
"borderWidth": 1
}
]
},
"metadata": serialize_dates({
"confidence": circ_data['confidence'],
"data_points": circ_data['data_points'],
"newest_date": circ_data['newest_date'],
"oldest_date": circ_data['oldest_date'],
"measurements": circ_data['measurements'] # Full details
})
}
# ── Health Endpoint ──────────────────────────────────────────────────────────
@router.get("/charts/health")
def health_check() -> Dict:
"""
Health check endpoint for charts API.
Returns:
{
"status": "ok",
"version": "1.0",
"available_charts": [...]
}
"""
return {
"status": "ok",
"version": "1.0",
"phase": "0c",
"available_charts": [
{
"endpoint": "/charts/weight-trend",
"type": "line",
"description": "Weight trend over time"
},
{
"endpoint": "/charts/body-composition",
"type": "line",
"description": "Body fat % and lean mass"
},
{
"endpoint": "/charts/circumferences",
"type": "bar",
"description": "Latest circumference measurements"
}
]
}