api.py

"""
FastAPI backend for the Flow Control v2 dashboard.
Wraps the existing db.py and alerts.py modules as JSON endpoints and serves
the static HTML/CSS/JS frontend on the same port.

Run with:
    uvicorn api:app --host 0.0.0.0 --port 8502

Or install as NSSM service with install_services_v2.bat — runs alongside the
Streamlit dashboard on 8501 without conflict.
"""
import os
import sys
import base64
import logging
from datetime import datetime, date, timedelta
from pathlib import Path

# Silence Streamlit cache warnings — db.py uses @st.cache_data decorators for
# the live dashboard, but when imported from FastAPI there's no Streamlit runtime
# so every cached call logs a harmless warning. We don't need those here.
logging.getLogger("streamlit").setLevel(logging.ERROR)
logging.getLogger("streamlit.runtime.caching.cache_data_api").setLevel(logging.ERROR)

from fastapi import FastAPI, HTTPException
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse, JSONResponse
from pydantic import BaseModel

# Ensure parent directory is on the path so we can import the shared modules
PARENT_DIR = Path(__file__).resolve().parent.parent
if str(PARENT_DIR) not in sys.path:
    sys.path.insert(0, str(PARENT_DIR))

# Import shared modules from the main Flowmeter folder
import db
import alerts
from config_loader import CONFIG, PROJECT_DIR

V2_DIR = Path(__file__).resolve().parent
STATIC_DIR = V2_DIR / "static"
AUDIO_DIR = PROJECT_DIR_AUDIO = Path(PROJECT_DIR) / "audio"

app = FastAPI(title="Flow Control v2", version="2.0")

# Serve static frontend
app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")


def _df_to_records(df):
    """Convert pandas DataFrame to list of dicts with JSON-safe types."""
    if df is None or df.empty:
        return []
    df = df.copy()
    for col in df.columns:
        if str(df[col].dtype).startswith('datetime'):
            df[col] = df[col].dt.strftime('%Y-%m-%dT%H:%M:%S')
    return df.to_dict(orient='records')


# ============================================================
# ROOT / FRONTEND
# ============================================================

@app.get("/")
async def root():
    return FileResponse(str(STATIC_DIR / "index.html"))


@app.get("/audio/{filename}")
async def serve_audio(filename: str):
    """Serve alert sound files from the project-root /audio folder."""
    path = Path(PROJECT_DIR) / "audio" / filename
    if not path.exists() or not path.is_file():
        raise HTTPException(status_code=404, detail="audio not found")
    # Basic sanity: prevent path traversal
    try:
        path.resolve().relative_to((Path(PROJECT_DIR) / "audio").resolve())
    except ValueError:
        raise HTTPException(status_code=400, detail="invalid path")
    return FileResponse(str(path), media_type="audio/mpeg")


@app.get("/brand/{which}")
async def serve_brand(which: str):
    """Serve the hotel logo (image.png) and maker credit (image2.png) from the
    project root, same files the v1 Streamlit dashboard uses. Cached aggressively."""
    mapping = {"logo": "image.png", "credit": "image2.png"}
    fname = mapping.get(which)
    if fname is None:
        raise HTTPException(status_code=404, detail="unknown brand asset")
    path = Path(PROJECT_DIR) / fname
    if not path.exists() or not path.is_file():
        raise HTTPException(status_code=404, detail=f"{fname} not found in project folder")
    return FileResponse(str(path), media_type="image/png", headers={"Cache-Control": "public, max-age=86400"})


# ============================================================
# CONFIG + META
# ============================================================

@app.get("/api/config")
async def get_config():
    """Returns the pieces of config the frontend needs to render meters."""
    return {
        "meters": [
            {
                "id": m['id'],
                "name": m['name'],
                "color": m['color'],
                "max_flow_rate": m.get('max_flow_rate'),
            }
            for m in CONFIG['meters']
        ],
        "dashboard": CONFIG['dashboard'],
        "health": CONFIG['health'],
        "alerts_rules": [
            {
                "id": r['id'],
                "severity": r['severity'],
                "audio_file": r.get('audio_file'),
            }
            for r in CONFIG['alerts']['rules']
        ],
    }


# ============================================================
# LIVE DATA
# ============================================================

@app.get("/api/stats")
async def get_stats():
    """System-wide stats for the top pills."""
    total_readings, latest = db.fetch_system_stats()
    settings_all = db.fetch_all_settings()

    meters_state = []
    trip_totals = {}
    online_count = 0

    for _, row in latest.iterrows():
        mid = int(row['meter_id'])
        offset_matches = settings_all.loc[settings_all['meter_id'] == mid, 'totalizer_offset'].values
        offset = float(offset_matches[0]) if len(offset_matches) > 0 else 0.0
        trip = float(row['total_flow']) - offset
        trip_totals[mid] = trip
        if row['status'] == 'OK':
            online_count += 1

        # Temperature isn't in fetch_system_stats output — grab latest from recent readings
        recent = db.fetch_recent_readings(mid, 1)
        temperature = float(recent.iloc[-1]['temperature']) if not recent.empty else None

        meters_state.append({
            "meter_id": mid,
            "instant_flow": float(row['instant_flow']),
            "total_flow": float(row['total_flow']),
            "trip_total": trip,
            "temperature": temperature,
            "status": row['status'],
            "timestamp": str(row['timestamp']),
        })

    return {
        "total_readings": int(total_readings),
        "online_count": online_count,
        "total_meters": len(CONFIG['meters']),
        "meters": meters_state,
        "trip_totals": trip_totals,
        "return_trip": trip_totals.get(2, 0.0) - trip_totals.get(1, 0.0),
    }


@app.get("/api/meter/{meter_id}/recent")
async def get_recent_readings(meter_id: int, limit: int = 120):
    """Recent readings for charts (live flow, temp)."""
    df = db.fetch_recent_readings(meter_id, limit)
    if df.empty:
        return {"meter_id": meter_id, "readings": []}
    return {
        "meter_id": meter_id,
        "readings": _df_to_records(df),
    }


@app.get("/api/meter/{meter_id}/daily")
async def get_daily_usage(meter_id: int, days: int = 30):
    df = db.fetch_daily_usage(meter_id, days)
    return {"meter_id": meter_id, "days": _df_to_records(df)}


@app.get("/api/daily-all")
async def get_daily_all(days: int = 30):
    """Daily usage for all meters — used for Return Flow card."""
    df = db.fetch_daily_usage_all(days)
    return {"days": _df_to_records(df)}


@app.get("/api/health")
async def get_health():
    per_meter, overall_age = db.fetch_health_status()
    return {
        "per_meter": {
            str(mid): {
                "last_reading": data['last_reading'].strftime('%Y-%m-%dT%H:%M:%S'),
                "age_seconds": float(data['age_seconds']),
            }
            for mid, data in per_meter.items()
        },
        "overall_age_seconds": float(overall_age) if overall_age is not None else None,
        "thresholds": CONFIG['health'],
    }


# ============================================================
# ALERTS
# ============================================================

@app.get("/api/alerts")
async def get_alerts():
    """Evaluate rules and return unresolved alerts."""
    try:
        alerts.evaluate_all()
    except Exception as e:
        print(f"[ALERT EVAL ERROR] {e}")

    df = db.fetch_unresolved_alerts()
    # Map rule_id to audio_file
    rule_audio = {r['id']: r.get('audio_file') for r in CONFIG['alerts']['rules']}

    records = _df_to_records(df)
    for rec in records:
        base_rule = rec['rule_id'].split(':')[0]
        rec['audio_file'] = rule_audio.get(base_rule)

    return {"alerts": records}


class AcknowledgeRequest(BaseModel):
    alert_id: int


@app.post("/api/alerts/acknowledge")
async def acknowledge_alert(req: AcknowledgeRequest):
    db.acknowledge_alert(req.alert_id)
    return {"ok": True, "alert_id": req.alert_id}


@app.post("/api/alerts/reset")
async def reset_alert(req: AcknowledgeRequest):
    """Force-close an acknowledged alert. The rule engine will re-fire it on the
    next evaluation cycle if the condition is still active."""
    affected = db.reset_alert(req.alert_id)
    if affected == 0:
        # Either not found, not acknowledged yet, or already resolved
        raise HTTPException(status_code=400, detail="Alert must be acknowledged before reset")
    return {"ok": True, "alert_id": req.alert_id}


# ============================================================
# OCCUPANCY
# ============================================================

class OccupancyEntry(BaseModel):
    night_date: str   # ISO format YYYY-MM-DD
    occupancy_pct: float


@app.post("/api/occupancy")
async def save_occupancy(entry: OccupancyEntry):
    try:
        night_dt = datetime.strptime(entry.night_date, "%Y-%m-%d").date()
    except ValueError:
        raise HTTPException(status_code=400, detail="invalid date format, expected YYYY-MM-DD")
    if not (0 <= entry.occupancy_pct <= 100):
        raise HTTPException(status_code=400, detail="occupancy_pct must be 0-100")

    previous = db.upsert_occupancy(night_dt, entry.occupancy_pct)
    return {
        "ok": True,
        "night_date": entry.night_date,
        "occupancy_pct": entry.occupancy_pct,
        "previous": float(previous) if previous is not None else None,
    }


@app.get("/api/occupancy/recent")
async def get_recent_occupancy(limit: int = 10):
    df = db.fetch_recent_occupancy(limit)
    return {"entries": _df_to_records(df)}


@app.get("/api/occupancy/correlation")
async def get_correlation(days: int = 30, alignment: str = "next_day"):
    """Return merged daily-water-usage + occupancy data with Pearson r.
    alignment='next_day' shifts occupancy +1 day (night-of X → water on X+1).
    alignment='same_day' uses direct calendar date alignment."""
    import pandas as pd
    daily_all = db.fetch_daily_usage_all(days)
    occupancy_df = db.fetch_occupancy_range(days + 5)

    if daily_all.empty or occupancy_df.empty:
        return {"merged": [], "pearson_r": None, "alignment": alignment}

    makeup = daily_all[daily_all['meter_id'] == 1][['day', 'usage_m3']].copy()
    makeup['day'] = pd.to_datetime(makeup['day'])
    makeup = makeup.rename(columns={'usage_m3': 'water_m3'})

    occ = occupancy_df.copy()
    occ['night_date'] = pd.to_datetime(occ['night_date'])
    if alignment == "next_day":
        occ['day'] = occ['night_date'] + pd.Timedelta(days=1)
    else:
        occ['day'] = occ['night_date']

    merged = pd.merge(makeup, occ[['day', 'occupancy_pct']], on='day', how='inner')
    merged = merged.sort_values('day').reset_index(drop=True)
    merged['label'] = merged['day'].dt.strftime('%b %d')

    r = None
    if len(merged) >= 2:
        try:
            r_val = merged['water_m3'].corr(merged['occupancy_pct'])
            if not pd.isna(r_val):
                r = float(r_val)
        except Exception:
            r = None

    # Convert day to string for JSON
    merged['day'] = merged['day'].dt.strftime('%Y-%m-%d')
    return {
        "merged": merged.to_dict(orient='records'),
        "pearson_r": r,
        "alignment": alignment,
        "matched_days": len(merged),
    }


# ============================================================
# PDAM (municipal water utility) — daily manual readings
# ============================================================

class PdamReading(BaseModel):
    reading_date: str         # ISO date YYYY-MM-DD
    main_total_m3: float
    tower_total_m3: float
    notes: str | None = None


def _pdam_config():
    """Read PDAM config block, with safe defaults if absent."""
    p = (CONFIG.get('pdam') or {})
    return {
        'idr_per_m3': float(p.get('idr_per_m3', 21500)),
    }


@app.post("/api/pdam")
async def upsert_pdam(reading: PdamReading):
    """Save a daily PDAM reading. Replaces any existing entry for that date.
    Returns the previous value if it existed, for UX feedback."""
    conn = db.get_db_connection()
    try:
        cur = conn.cursor()
        cur.execute(
            "SELECT main_total_m3, tower_total_m3 FROM pdam_readings WHERE reading_date = ?",
            (reading.reading_date,)
        )
        prev = cur.fetchone()
        cur.execute(
            """INSERT INTO pdam_readings
               (reading_date, main_total_m3, tower_total_m3, notes, recorded_at)
               VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP)
               ON CONFLICT(reading_date) DO UPDATE SET
                   main_total_m3 = excluded.main_total_m3,
                   tower_total_m3 = excluded.tower_total_m3,
                   notes = excluded.notes,
                   recorded_at = CURRENT_TIMESTAMP""",
            (reading.reading_date, reading.main_total_m3, reading.tower_total_m3, reading.notes)
        )
        conn.commit()
        return {
            "ok": True,
            "reading_date": reading.reading_date,
            "previous": {
                "main_total_m3": prev[0],
                "tower_total_m3": prev[1],
            } if prev else None,
        }
    finally:
        conn.close()


@app.get("/api/pdam/recent")
async def get_recent_pdam(limit: int = 30):
    """Recent daily PDAM entries, most recent first."""
    conn = db.get_db_connection()
    try:
        rows = conn.execute(
            """SELECT reading_date, main_total_m3, tower_total_m3, notes, recorded_at
               FROM pdam_readings ORDER BY reading_date DESC LIMIT ?""",
            (int(limit),)
        ).fetchall()
        cfg = _pdam_config()
        rate = cfg['idr_per_m3']
        return {
            "readings": [
                {
                    "reading_date": r[0],
                    "main_total_m3": r[1],
                    "tower_total_m3": r[2],
                    "total_m3": round(r[1] + r[2], 2),
                    "cost_idr": round((r[1] + r[2]) * rate),
                    "notes": r[3],
                    "recorded_at": r[4],
                }
                for r in rows
            ],
            "idr_per_m3": rate,
        }
    finally:
        conn.close()


@app.get("/api/pdam/summary")
async def get_pdam_summary(days: int = 30):
    """Cost, balance, per-guest efficiency across N days.
    Cold consumption is derived system-wide as (PDAM_total - M1_makeup),
    since M1 measures combined make-up for both buildings."""
    import pandas as pd
    cfg = _pdam_config()
    rate = cfg['idr_per_m3']

    conn = db.get_db_connection()
    try:
        df = pd.read_sql_query(
            """SELECT reading_date, main_total_m3, tower_total_m3
               FROM pdam_readings
               WHERE reading_date >= date('now', 'localtime', ? || ' days')
               ORDER BY reading_date""",
            conn, params=(f'-{int(days)}',)
        )

        if df.empty:
            return {
                "days": days,
                "has_data": False,
                "idr_per_m3": rate,
                "message": "No PDAM readings in this window. Enter daily readings to see analysis.",
            }

        df['total'] = df['main_total_m3'] + df['tower_total_m3']
        total_m3 = float(df['total'].sum())
        main_m3 = float(df['main_total_m3'].sum())
        tower_m3 = float(df['tower_total_m3'].sum())
        total_cost = total_m3 * rate

        # Previous window for trend
        prev_df = pd.read_sql_query(
            """SELECT main_total_m3, tower_total_m3 FROM pdam_readings
               WHERE reading_date >= date('now', 'localtime', ? || ' days')
                 AND reading_date <  date('now', 'localtime', ? || ' days')""",
            conn, params=(f'-{int(days)*2}', f'-{int(days)}')
        )
        prev_total_m3 = None
        pct_change = None
        if not prev_df.empty:
            prev_total_m3 = float((prev_df['main_total_m3'] + prev_df['tower_total_m3']).sum())
            if prev_total_m3 > 0:
                pct_change = ((total_m3 - prev_total_m3) / prev_total_m3) * 100

        # M1 (Make Up Water) measured volume over the same window — defines
        # how much of PDAM total went to the boiler loop.
        # Cold-to-rooms = PDAM_total - M1_makeup (system-wide, can't be split per building).
        m1_df = pd.read_sql_query(
            """SELECT MAX(total_flow) - MIN(total_flow) AS m1_volume
               FROM flow_readings
               WHERE meter_id = 1 AND status = 'OK'
                 AND timestamp >= datetime('now', 'localtime', ? || ' days')""",
            conn, params=(f'-{int(days)}',)
        )
        m1_makeup = float(m1_df['m1_volume'].iloc[0]) if not m1_df.empty and m1_df['m1_volume'].iloc[0] is not None else None

        cold_m3 = None
        cold_pct = None
        makeup_pct = None
        cold_to_makeup_ratio = None
        balance_warning = None
        if m1_makeup is not None and m1_makeup >= 0:
            cold_m3 = total_m3 - m1_makeup
            if total_m3 > 0:
                cold_pct = (cold_m3 / total_m3) * 100
                makeup_pct = (m1_makeup / total_m3) * 100
            if m1_makeup > 0:
                cold_to_makeup_ratio = cold_m3 / m1_makeup
            # Sanity check — if M1 > PDAM total, something is off
            if m1_makeup > total_m3:
                balance_warning = (
                    f"M1 measured {m1_makeup:.1f} m³ of make-up but PDAM only delivered {total_m3:.1f} m³ "
                    f"in this window. Check if M1 totalizer rolled over, was reset, or PDAM data is incomplete."
                )

        # Per-guest efficiency
        occ_df = pd.read_sql_query(
            """SELECT night_date, occupancy_pct FROM occupancy
               WHERE night_date >= date('now', 'localtime', ? || ' days')""",
            conn, params=(f'-{int(days)}',)
        )
        per_room_night_m3 = None
        per_room_night_idr = None
        room_nights_proxy = None
        if not occ_df.empty:
            total_pct = float(occ_df['occupancy_pct'].sum())
            hotel_rooms = (CONFIG.get('hotel') or {}).get('total_rooms')
            if hotel_rooms:
                room_nights_proxy = (total_pct / 100.0) * float(hotel_rooms)
                if room_nights_proxy > 0:
                    per_room_night_m3 = total_m3 / room_nights_proxy
                    per_room_night_idr = total_cost / room_nights_proxy

        return {
            "days": days,
            "has_data": True,
            "idr_per_m3": rate,
            "totals": {
                "total_m3": round(total_m3, 2),
                "cost_idr": round(total_cost),
                "cold_m3": round(cold_m3, 2) if cold_m3 is not None else None,
                "makeup_m3": round(m1_makeup, 2) if m1_makeup is not None else None,
                "cold_pct": round(cold_pct, 1) if cold_pct is not None else None,
                "makeup_pct": round(makeup_pct, 1) if makeup_pct is not None else None,
                "hot_cold_ratio": round(cold_to_makeup_ratio, 1) if cold_to_makeup_ratio is not None else None,
            },
            "buildings": {
                "main_m3": round(main_m3, 2),
                "tower_m3": round(tower_m3, 2),
                "main_cost_idr": round(main_m3 * rate),
                "tower_cost_idr": round(tower_m3 * rate),
                "main_pct": round((main_m3 / total_m3) * 100, 1) if total_m3 else 0,
                "tower_pct": round((tower_m3 / total_m3) * 100, 1) if total_m3 else 0,
            },
            "trend": {
                "prev_total_m3": round(prev_total_m3, 2) if prev_total_m3 is not None else None,
                "pct_change": round(pct_change, 1) if pct_change is not None else None,
            },
            "per_guest": {
                "room_nights": round(room_nights_proxy, 1) if room_nights_proxy else None,
                "per_room_night_m3": round(per_room_night_m3, 2) if per_room_night_m3 else None,
                "per_room_night_idr": round(per_room_night_idr) if per_room_night_idr else None,
                "note": "Set 'hotel.total_rooms' in config.yaml to enable per-room-night metrics." if not room_nights_proxy else None,
            },
            "balance_warning": balance_warning,
            "entries_count": len(df),
        }
    finally:
        conn.close()


@app.get("/explore")
async def explore_page():
    """Serve the data explorer page."""
    return FileResponse(str(STATIC_DIR / "explore.html"))


# ============================================================
# EXPLORE — adaptive-aggregation multi-series queries
# ============================================================

def _pick_bucket_seconds(span_seconds: float) -> int:
    """Decide aggregation granularity based on time span.
    The goal: keep returned points in the 500-2000 range for smooth rendering.
    - <= 2 hours: raw 3s data (no aggregation, ~2400 points/hour)
    - 2h-1d: 30s buckets
    - 1d-7d: 5min buckets (288/day)
    - 7d-30d: 30min buckets (48/day)
    - >30d: 3h buckets
    """
    if span_seconds <= 7200:
        return 0  # no aggregation
    if span_seconds <= 86400:
        return 30
    if span_seconds <= 7 * 86400:
        return 300
    if span_seconds <= 30 * 86400:
        return 1800
    return 10800


@app.get("/api/explore/series")
async def explore_series(start: str, end: str):
    """Return all series (flow 1, flow 2, return, temp 1, temp 2, occupancy)
    for a given [start, end] range, aggregated adaptively. Timestamps are
    ISO strings like '2026-04-07T10:00:00'."""
    import pandas as pd
    try:
        start_dt = pd.to_datetime(start)
        end_dt = pd.to_datetime(end)
    except Exception:
        raise HTTPException(status_code=400, detail="invalid start/end, use ISO format")

    if end_dt <= start_dt:
        raise HTTPException(status_code=400, detail="end must be after start")

    span = (end_dt - start_dt).total_seconds()
    bucket = _pick_bucket_seconds(span)

    conn = db.get_db_connection()
    try:
        start_str = start_dt.strftime('%Y-%m-%d %H:%M:%S')
        end_str = end_dt.strftime('%Y-%m-%d %H:%M:%S')

        if bucket == 0:
            # Raw readings
            df = pd.read_sql_query(
                """SELECT timestamp, meter_id, instant_flow, temperature, total_flow
                   FROM flow_readings
                   WHERE status='OK' AND timestamp BETWEEN ? AND ?
                   ORDER BY timestamp""",
                conn, params=(start_str, end_str)
            )
        else:
            # SQLite bucket trick. Timestamps in flow_readings are stored as
            # LOCAL time strings (the fetcher writes datetime('now','localtime')).
            # SQLite's strftime('%s', X) treats X as UTC and converts to epoch.
            # That works for bucketing because we only need consistent epoch
            # values to group by — the absolute UTC offset cancels out.
            # On the OUTPUT side, we MUST NOT use the 'localtime' modifier:
            # datetime(epoch, 'unixepoch', 'localtime') would add the system TZ
            # offset, breaking the round-trip on any non-UTC server (Jakarta=+7h).
            # Without 'localtime', the output string equals the original local
            # time, which is what the rest of the system expects.
            df = pd.read_sql_query(
                f"""SELECT
                       datetime(CAST(strftime('%s', timestamp) AS INTEGER) / {bucket} * {bucket}, 'unixepoch') as timestamp,
                       meter_id,
                       AVG(instant_flow) as instant_flow,
                       AVG(temperature) as temperature,
                       MAX(total_flow) as total_flow
                   FROM flow_readings
                   WHERE status='OK' AND timestamp BETWEEN ? AND ?
                   GROUP BY CAST(strftime('%s', timestamp) AS INTEGER) / {bucket}, meter_id
                   ORDER BY timestamp""",
                conn, params=(start_str, end_str)
            )

        # Pivot into series arrays
        series = {
            "makeup_flow": [],      # meter 1 instant flow
            "supply_flow": [],      # meter 2 instant flow
            "return_flow": [],      # computed supply - makeup
            "supply_temp": [],      # meter 2 temperature (meter 1 temp excluded - not reliable)
            "occupancy": [],        # per-night occupancy
        }
        if not df.empty:
            df['timestamp'] = pd.to_datetime(df['timestamp'])
            # Build per-meter dicts keyed by timestamp for merging
            m1 = df[df['meter_id'] == 1].set_index('timestamp')
            m2 = df[df['meter_id'] == 2].set_index('timestamp')
            for ts, row in m1.iterrows():
                ts_iso = ts.strftime('%Y-%m-%dT%H:%M:%S')
                series['makeup_flow'].append([ts_iso, round(float(row['instant_flow']), 3)])
            for ts, row in m2.iterrows():
                ts_iso = ts.strftime('%Y-%m-%dT%H:%M:%S')
                series['supply_flow'].append([ts_iso, round(float(row['instant_flow']), 3)])
                series['supply_temp'].append([ts_iso, round(float(row['temperature']), 2)])
            # Return flow = supply - makeup, matched on timestamp
            joined = pd.merge(
                m1[['instant_flow']].rename(columns={'instant_flow': 'm1'}),
                m2[['instant_flow']].rename(columns={'instant_flow': 'm2'}),
                left_index=True, right_index=True, how='inner'
            )
            joined['ret'] = joined['m2'] - joined['m1']
            for ts, row in joined.iterrows():
                ts_iso = ts.strftime('%Y-%m-%dT%H:%M:%S')
                series['return_flow'].append([ts_iso, round(float(row['ret']), 3)])

        # Occupancy covers the range — step line (one value per night)
        occ_df = pd.read_sql_query(
            """SELECT night_date, occupancy_pct FROM occupancy
               WHERE night_date BETWEEN ? AND ?
               ORDER BY night_date""",
            conn, params=(start_dt.date().isoformat(), end_dt.date().isoformat())
        )
        if not occ_df.empty:
            occ_df['night_date'] = pd.to_datetime(occ_df['night_date'])
            for _, row in occ_df.iterrows():
                series['occupancy'].append([
                    row['night_date'].strftime('%Y-%m-%dT%H:%M:%S'),
                    round(float(row['occupancy_pct']), 1),
                ])

        # Compute summary stats per numeric series
        stats = {}
        for key, points in series.items():
            if not points:
                stats[key] = None
                continue
            values = [p[1] for p in points]
            stats[key] = {
                "count": len(values),
                "min": round(min(values), 3),
                "max": round(max(values), 3),
                "mean": round(sum(values) / len(values), 3),
                "first": values[0],
                "last": values[-1],
            }

        # Build operational insights from the data — rule-based, not statistical.
        insights = _compute_insights(series, stats, span)

        return {
            "start": start,
            "end": end,
            "span_seconds": int(span),
            "bucket_seconds": bucket,
            "point_count": sum(len(v) for v in series.values()),
            "series": series,
            "stats": stats,
            "insights": insights,
            "thresholds": OPERATIONAL_RULES,
        }
    finally:
        conn.close()


# ============================================================
# OPERATIONAL RULES
# Domain-specific thresholds for the closed-loop water system at Le Méridien:
#   Boiler → Supply (M2) → Rooms → Return → Make Up (M1) → Boiler
# ============================================================
OPERATIONAL_RULES = {
    # Make Up water (M1) - returning cooled water to boiler
    "makeup_normal_max": 14.0,    # Typical peak (morning showers etc)
    "makeup_elevated_max": 18.0,  # Above this = abnormal consumption
    # Supply hot water (M2) - hot water out to rooms
    "supply_hard_ceiling": 30.0,  # Physical ceiling - above = sensor/data error
    "supply_typical_max": 25.0,   # Normal operational max
    # Temperature
    "temp_min": 40.0,             # Supply + Make Up should never drop below
    # Return (computed Supply - Make Up)
    "return_min_sustained_seconds": 120,  # Negative for >2 min = data integrity issue
}


def _linear_trend(points):
    """Simple least-squares slope on [(ts_iso, value), ...].
    Returns (slope_per_day, r_squared). Returns (None, None) if <2 points.
    Slope is in "value units per day" so it's readable regardless of bucket."""
    if not points or len(points) < 2:
        return None, None
    from datetime import datetime
    # Convert to (seconds_since_first, value)
    first_t = None
    xs, ys = [], []
    for ts, v in points:
        try:
            dt = datetime.fromisoformat(ts)
        except Exception:
            continue
        if first_t is None:
            first_t = dt
        xs.append((dt - first_t).total_seconds() / 86400.0)  # days
        ys.append(float(v))
    if len(xs) < 2:
        return None, None
    n = len(xs)
    mean_x = sum(xs) / n
    mean_y = sum(ys) / n
    num = sum((xs[i] - mean_x) * (ys[i] - mean_y) for i in range(n))
    den_x = sum((x - mean_x) ** 2 for x in xs)
    den_y = sum((y - mean_y) ** 2 for y in ys)
    if den_x == 0 or den_y == 0:
        return 0.0, 0.0
    slope = num / den_x
    r = num / ((den_x * den_y) ** 0.5)
    return round(slope, 4), round(r * r, 3)


def _find_violations(points, predicate, min_consecutive=1):
    """Find timestamps where `predicate(value)` is true for >= min_consecutive
    consecutive points. Returns list of {start, end, peak_value, count}."""
    runs = []
    current = None
    for ts, v in points:
        if predicate(v):
            if current is None:
                current = {"start": ts, "end": ts, "peak_value": v, "count": 1}
            else:
                current["end"] = ts
                current["count"] += 1
                if abs(v) > abs(current["peak_value"]):
                    current["peak_value"] = v
        else:
            if current and current["count"] >= min_consecutive:
                runs.append(current)
            current = None
    if current and current["count"] >= min_consecutive:
        runs.append(current)
    return runs


def _compute_insights(series, stats, span_seconds):
    """Generate a list of insight objects from the data. Each insight has:
    { kind: 'ok'|'info'|'warn'|'critical', title, detail, category,
      timestamp (optional, for jumping to that moment) }
    """
    import pandas as pd
    from datetime import datetime

    insights = []
    rules = OPERATIONAL_RULES

    # ====== RULE 1: Supply hard ceiling breach (30 m³/h) ======
    supply = series.get('supply_flow', [])
    breaches = _find_violations(supply, lambda v: v > rules['supply_hard_ceiling'])
    if breaches:
        total_points = sum(b['count'] for b in breaches)
        peak = max(b['peak_value'] for b in breaches)
        insights.append({
            "kind": "critical",
            "category": "Data integrity",
            "title": f"Supply exceeded {rules['supply_hard_ceiling']:.0f} m³/h ceiling",
            "detail": f"Occurred {len(breaches)} time{'s' if len(breaches) > 1 else ''} across {total_points} readings. Peak reached {peak:.1f} m³/h — physically impossible, suggests sensor error or data corruption.",
            "timestamp": breaches[0]['start'],
        })

    # ====== RULE 2: Make Up > Supply (impossible physics) ======
    # For this we need to compute per timestamp; use return_flow sign
    ret = series.get('return_flow', [])
    # Return < 0 sustained = Make Up > Supply (closed loop can't produce this)
    bucket_sec = 30 if span_seconds > 7200 else 3
    min_pts = max(1, rules['return_min_sustained_seconds'] // max(bucket_sec, 1))
    neg_runs = _find_violations(ret, lambda v: v < -0.5, min_consecutive=min_pts)
    if neg_runs:
        longest = max(neg_runs, key=lambda r: r['count'])
        total_dur = sum(r['count'] for r in neg_runs) * bucket_sec
        mins = total_dur // 60
        insights.append({
            "kind": "critical",
            "category": "Physics violation",
            "title": "Make Up flow exceeded Supply",
            "detail": f"Closed-loop physics violated for ~{mins} min total across {len(neg_runs)} event{'s' if len(neg_runs) > 1 else ''}. Longest: {longest['count'] * bucket_sec // 60} min starting {longest['start'][-8:-3]}. Likely meter calibration drift — Make Up can never exceed Supply in a closed loop.",
            "timestamp": longest['start'],
        })

    # ====== RULE 3: Make Up abnormally elevated (>18 m³/h) ======
    makeup = series.get('makeup_flow', [])
    high_makeup = _find_violations(makeup, lambda v: v > rules['makeup_elevated_max'])
    if high_makeup:
        peak = max(b['peak_value'] for b in high_makeup)
        insights.append({
            "kind": "warn",
            "category": "Consumption",
            "title": f"Make Up exceeded {rules['makeup_elevated_max']:.0f} m³/h",
            "detail": f"Peaked at {peak:.1f} m³/h during {len(high_makeup)} event{'s' if len(high_makeup) > 1 else ''}. Normal peaks stay around {rules['makeup_normal_max']:.0f} m³/h. Could indicate a leak, meter fault, or unusual draw event.",
            "timestamp": high_makeup[0]['start'],
        })

    # ====== RULE 4: Supply temperature below 40°C ======
    # Meter 1 (Make Up) temperature isn't reliable, so we only check supply.
    temps = series.get('supply_temp', [])
    low_temps = _find_violations(temps, lambda v: v < rules['temp_min'])
    if low_temps:
        coldest = min(b['peak_value'] for b in low_temps)
        insights.append({
            "kind": "critical",
            "category": "Temperature",
            "title": f"Supply temperature dropped below {rules['temp_min']:.0f}°C",
            "detail": f"Coldest reading: {coldest:.1f}°C across {len(low_temps)} event{'s' if len(low_temps) > 1 else ''}. Likely boiler issue or heat loss in the supply loop.",
            "timestamp": low_temps[0]['start'],
        })

    # ====== TREND ANALYSIS ======
    # Make Up, Supply, Return — trend over the visible range
    for key, label, unit in [
        ('makeup_flow', 'Make Up', 'm³/h'),
        ('supply_flow', 'Supply', 'm³/h'),
        ('return_flow', 'Return', 'm³/h'),
    ]:
        pts = series.get(key, [])
        slope, r2 = _linear_trend(pts)
        if slope is None or stats[key] is None:
            continue
        mean = stats[key]['mean']
        if mean == 0 or abs(mean) < 0.1:
            continue
        # Express slope as %/day of the mean
        pct_per_day = (slope / mean) * 100
        # Only flag if the trend is meaningful: r² > 0.4 (statistically detectable trend)
        # AND |%/day| > 1 (practically significant)
        is_meaningful = r2 > 0.4 and abs(pct_per_day) > 1
        if is_meaningful and span_seconds > 86400:  # only show trends for ranges > 1 day
            direction = "rising" if slope > 0 else "falling"
            insights.append({
                "kind": "info",
                "category": "Trend",
                "title": f"{label} is trending {direction}",
                "detail": f"{slope:+.2f} {unit} per day ({pct_per_day:+.1f}% daily, R²={r2:.2f}) across the selected range.",
            })

    # ====== PEAK ANALYSIS ======
    # For Make Up, find the peak and note its hour-of-day
    makeup_pts = series.get('makeup_flow', [])
    if makeup_pts:
        peak_ts, peak_val = max(makeup_pts, key=lambda p: p[1])
        try:
            peak_dt = datetime.fromisoformat(peak_ts)
            hour = peak_dt.hour
            time_context = ""
            if 6 <= hour <= 10:
                time_context = " — morning shower window"
            elif 17 <= hour <= 22:
                time_context = " — evening usage window"
            elif 0 <= hour <= 5:
                time_context = " — unusual for this hour, worth investigating"
            within_normal = "within normal range" if peak_val <= rules['makeup_normal_max'] else \
                           ("elevated" if peak_val <= rules['makeup_elevated_max'] else "abnormal")
            insights.append({
                "kind": "ok" if peak_val <= rules['makeup_normal_max'] else "info",
                "category": "Peak analysis",
                "title": f"Make Up peaked at {peak_val:.1f} m³/h",
                "detail": f"Reached at {peak_dt.strftime('%a %d %b, %H:%M')}{time_context}. This is {within_normal} (typical peak around {rules['makeup_normal_max']:.0f}).",
                "timestamp": peak_ts,
            })
        except Exception:
            pass

    # ====== RECOVERY ANALYSIS ======
    # Average recovery rate across the range = mean(return) / mean(supply) * 100
    if stats.get('return_flow') and stats.get('supply_flow') and stats['supply_flow']['mean'] > 0:
        recovery_pct = (stats['return_flow']['mean'] / stats['supply_flow']['mean']) * 100
        if recovery_pct >= 70:
            kind, note = "ok", "Healthy — loop is efficient."
        elif recovery_pct >= 50:
            kind, note = "info", "Moderate — some loss, worth monitoring."
        elif recovery_pct >= 0:
            kind, note = "warn", "Low — significant water loss in the loop."
        else:
            kind, note = "critical", "Negative — Make Up exceeds Supply, data integrity issue."
        insights.append({
            "kind": kind,
            "category": "Loop efficiency",
            "title": f"Average recovery: {recovery_pct:.1f}%",
            "detail": f"Mean Return ({stats['return_flow']['mean']:.2f}) as fraction of mean Supply ({stats['supply_flow']['mean']:.2f}). {note}",
        })

    # ====== NORMAL DAY AFFIRMATION ======
    # If there are no warn/critical insights, add a reassuring one
    has_issues = any(i['kind'] in ('warn', 'critical') for i in insights)
    if not has_issues:
        insights.insert(0, {
            "kind": "ok",
            "category": "Health check",
            "title": "System operating within nominal bounds",
            "detail": "No rule violations detected across the selected range. Supply below ceiling, Make Up below operational max, temperatures above minimum, recovery positive.",
        })
    else:
        # Count violations for headline
        issue_count = sum(1 for i in insights if i['kind'] in ('warn', 'critical'))
        insights.insert(0, {
            "kind": "warn",
            "category": "Health check",
            "title": f"{issue_count} {'issue' if issue_count == 1 else 'issues'} detected in this range",
            "detail": "Review the items below. Critical issues are physics violations or hardware concerns.",
        })

    return insights


# ============================================================
# ANNOTATIONS (explorer notes pinned to the timeline)
# ============================================================

class AnnotationCreate(BaseModel):
    timestamp: str
    note: str
    series: str | None = None


@app.get("/api/annotations")
async def list_annotations(start: str | None = None, end: str | None = None):
    conn = db.get_db_connection()
    try:
        if start and end:
            rows = conn.execute(
                "SELECT id, timestamp, series, note, created_at FROM annotations WHERE timestamp BETWEEN ? AND ? ORDER BY timestamp",
                (start, end)
            ).fetchall()
        else:
            rows = conn.execute(
                "SELECT id, timestamp, series, note, created_at FROM annotations ORDER BY timestamp DESC LIMIT 200"
            ).fetchall()
        return {"annotations": [
            {"id": r[0], "timestamp": r[1], "series": r[2], "note": r[3], "created_at": r[4]}
            for r in rows
        ]}
    finally:
        conn.close()


@app.post("/api/annotations")
async def create_annotation(body: AnnotationCreate):
    if not body.note.strip():
        raise HTTPException(status_code=400, detail="note cannot be empty")
    conn = db.get_db_connection()
    try:
        cur = conn.cursor()
        cur.execute(
            "INSERT INTO annotations (timestamp, series, note) VALUES (?, ?, ?)",
            (body.timestamp, body.series, body.note.strip()[:500])
        )
        conn.commit()
        new_id = cur.lastrowid
        return {"ok": True, "id": new_id}
    finally:
        conn.close()


@app.delete("/api/annotations/{annotation_id}")
async def delete_annotation(annotation_id: int):
    conn = db.get_db_connection()
    try:
        conn.execute("DELETE FROM annotations WHERE id = ?", (annotation_id,))
        conn.commit()
        return {"ok": True}
    finally:
        conn.close()


# ============================================================
# DATA EXPORT
# ============================================================

@app.get("/api/meter/{meter_id}/export")
async def export_readings(meter_id: int, hours: int = 24):
    """CSV-friendly export of recent readings."""
    where = f"meter_id = {int(meter_id)} AND timestamp >= datetime('now', 'localtime', '-{int(hours)} hours')"
    df = db.fetch_export_readings(meter_id, f"export_{meter_id}_{hours}", where)
    return {"meter_id": meter_id, "hours": hours, "readings": _df_to_records(df)}


@app.get("/api/settings/{meter_id}")
async def get_settings(meter_id: int):
    s = db.fetch_settings(meter_id)
    # Convert timestamps to strings
    if s.get('last_reset_time'):
        s['last_reset_time'] = str(s['last_reset_time'])
    return s


class ResetRequest(BaseModel):
    meter_id: int
    current_total: float


@app.post("/api/settings/reset")
async def reset_totalizer(req: ResetRequest):
    db.reset_totalizer(req.meter_id, req.current_total)
    return {"ok": True}


# ============================================================
# HEALTH CHECK
# ============================================================

@app.get("/api/ping")
async def ping():
    return {"pong": True, "ts": datetime.now().isoformat()}