alvis
d37801806d
Compose and supporting code for four services that had been running or prototyped without their config tracked here, per the repo convention that agap_git holds the compose + config while application source lives in each service's own Gitea repo. Only placeholder credentials are included: mood/.env.example and moodtracker/.env.example ship dummy values, and overleaf/variables.env carries app name and feature flags only. Real values stay in Vaultwarden. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
238 lines
8.6 KiB
Python
238 lines
8.6 KiB
Python
"""SQLite storage layer: schema init, idempotent upserts, and read queries.
|
|
|
|
Source-agnostic — the Fit REST adapter and the Takeout importer both call the
|
|
same upsert_* functions. Reads (summary/series/sessions/sleep) back the Adolf
|
|
query CLI. All writes are UPSERTs keyed on the row's natural Fit identity, so a
|
|
re-sync over an overlapping window is a no-op rather than a duplicate.
|
|
"""
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
from datetime import datetime, timezone
|
|
|
|
SCHEMA_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), "schema.sql")
|
|
|
|
|
|
def _now_iso():
|
|
return datetime.now(tz=timezone.utc).isoformat()
|
|
|
|
|
|
def connect(db_path):
|
|
os.makedirs(os.path.dirname(os.path.abspath(db_path)), exist_ok=True)
|
|
conn = sqlite3.connect(db_path)
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute("PRAGMA foreign_keys = ON")
|
|
return conn
|
|
|
|
|
|
def init_db(conn):
|
|
with open(SCHEMA_PATH) as f:
|
|
conn.executescript(f.read())
|
|
conn.commit()
|
|
|
|
|
|
# --- writes ---------------------------------------------------------------
|
|
|
|
def upsert_data_points(conn, rows, source="ha"):
|
|
now = _now_iso()
|
|
n = 0
|
|
for r in rows:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO data_points
|
|
(metric, data_type_name, start_ns, end_ns, start_time, end_time,
|
|
value_int, value_float, value_str, unit, data_source_id, source, ingested_at)
|
|
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)
|
|
ON CONFLICT(metric, start_ns, end_ns, data_source_id) DO UPDATE SET
|
|
value_int=excluded.value_int,
|
|
value_float=excluded.value_float,
|
|
value_str=excluded.value_str,
|
|
unit=excluded.unit,
|
|
data_type_name=excluded.data_type_name,
|
|
source=excluded.source,
|
|
ingested_at=excluded.ingested_at
|
|
""",
|
|
(
|
|
r["metric"], r["data_type_name"], r["start_ns"], r["end_ns"],
|
|
r["start_time"], r["end_time"], r.get("value_int"), r.get("value_float"),
|
|
r.get("value_str"), r.get("unit"), r.get("data_source_id", ""), source, now,
|
|
),
|
|
)
|
|
n += 1
|
|
conn.commit()
|
|
return n
|
|
|
|
|
|
def upsert_sleep_segments(conn, rows, source="ha"):
|
|
now = _now_iso()
|
|
n = 0
|
|
for r in rows:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO sleep_segments
|
|
(start_ns, end_ns, start_time, end_time, stage, stage_name,
|
|
data_source_id, source, ingested_at)
|
|
VALUES (?,?,?,?,?,?,?,?,?)
|
|
ON CONFLICT(start_ns, end_ns, data_source_id) DO UPDATE SET
|
|
stage=excluded.stage,
|
|
stage_name=excluded.stage_name,
|
|
source=excluded.source,
|
|
ingested_at=excluded.ingested_at
|
|
""",
|
|
(
|
|
r["start_ns"], r["end_ns"], r["start_time"], r["end_time"],
|
|
r["stage"], r["stage_name"], r.get("data_source_id", ""), source, now,
|
|
),
|
|
)
|
|
n += 1
|
|
conn.commit()
|
|
return n
|
|
|
|
|
|
def upsert_sessions(conn, rows, source="ha"):
|
|
now = _now_iso()
|
|
n = 0
|
|
for r in rows:
|
|
raw = r.get("raw_json")
|
|
if raw is not None and not isinstance(raw, str):
|
|
raw = json.dumps(raw)
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO sessions
|
|
(id, name, description, activity_type, activity_name, start_ns, end_ns,
|
|
start_time, end_time, modified_ns, application, source, raw_json, ingested_at)
|
|
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)
|
|
ON CONFLICT(id) DO UPDATE SET
|
|
name=excluded.name,
|
|
description=excluded.description,
|
|
activity_type=excluded.activity_type,
|
|
activity_name=excluded.activity_name,
|
|
start_ns=excluded.start_ns,
|
|
end_ns=excluded.end_ns,
|
|
start_time=excluded.start_time,
|
|
end_time=excluded.end_time,
|
|
modified_ns=excluded.modified_ns,
|
|
application=excluded.application,
|
|
source=excluded.source,
|
|
raw_json=excluded.raw_json,
|
|
ingested_at=excluded.ingested_at
|
|
""",
|
|
(
|
|
r["id"], r.get("name"), r.get("description"), r.get("activity_type"),
|
|
r.get("activity_name"), r["start_ns"], r["end_ns"], r["start_time"],
|
|
r["end_time"], r.get("modified_ns"), r.get("application"), source, raw, now,
|
|
),
|
|
)
|
|
n += 1
|
|
conn.commit()
|
|
return n
|
|
|
|
|
|
# --- sync cursor & run audit ---------------------------------------------
|
|
|
|
def get_last_synced_ns(conn, stream_key):
|
|
row = conn.execute(
|
|
"SELECT last_synced_ns FROM sync_state WHERE stream_key=?", (stream_key,)
|
|
).fetchone()
|
|
return row["last_synced_ns"] if row else 0
|
|
|
|
|
|
def set_sync_state(conn, stream_key, last_synced_ns, status="ok", error=None):
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO sync_state (stream_key, last_synced_ns, last_run_at, last_status, last_error)
|
|
VALUES (?,?,?,?,?)
|
|
ON CONFLICT(stream_key) DO UPDATE SET
|
|
last_synced_ns=MAX(sync_state.last_synced_ns, excluded.last_synced_ns),
|
|
last_run_at=excluded.last_run_at,
|
|
last_status=excluded.last_status,
|
|
last_error=excluded.last_error
|
|
""",
|
|
(stream_key, last_synced_ns, _now_iso(), status, error),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def start_run(conn):
|
|
cur = conn.execute(
|
|
"INSERT INTO ingest_runs (started_at, status) VALUES (?, 'running')", (_now_iso(),)
|
|
)
|
|
conn.commit()
|
|
return cur.lastrowid
|
|
|
|
|
|
def finish_run(conn, run_id, status, points=0, sessions=0, segments=0, error=None):
|
|
conn.execute(
|
|
"""UPDATE ingest_runs SET finished_at=?, status=?, points_upserted=?,
|
|
sessions_upserted=?, segments_upserted=?, error=? WHERE id=?""",
|
|
(_now_iso(), status, points, sessions, segments, error, run_id),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
# --- reads (Adolf query tool) --------------------------------------------
|
|
|
|
def daily_metric(conn, metric, days=14):
|
|
return [dict(r) for r in conn.execute(
|
|
"""SELECT date(start_time) AS day,
|
|
SUM(COALESCE(value_int, value_float)) AS total,
|
|
MAX(unit) AS unit
|
|
FROM data_points WHERE metric=?
|
|
AND start_time >= datetime('now', ?)
|
|
GROUP BY day ORDER BY day DESC""",
|
|
(metric, f"-{int(days)} days"),
|
|
).fetchall()]
|
|
|
|
|
|
def latest_metric(conn, metric, limit=20):
|
|
return [dict(r) for r in conn.execute(
|
|
"""SELECT start_time, end_time, value_int, value_float, unit
|
|
FROM data_points WHERE metric=? ORDER BY start_ns DESC LIMIT ?""",
|
|
(metric, int(limit)),
|
|
).fetchall()]
|
|
|
|
|
|
def recent_sessions(conn, days=30, limit=50):
|
|
return [dict(r) for r in conn.execute(
|
|
"""SELECT id, name, activity_type, activity_name, start_time, end_time,
|
|
(end_ns - start_ns)/60000000000.0 AS duration_min
|
|
FROM sessions WHERE start_time >= datetime('now', ?)
|
|
ORDER BY start_ns DESC LIMIT ?""",
|
|
(f"-{int(days)} days", int(limit)),
|
|
).fetchall()]
|
|
|
|
|
|
def sleep_by_night(conn, days=14):
|
|
"""Total minutes per stage grouped by the calendar day the sleep segment ends
|
|
(a night that crosses midnight is attributed to the wake day)."""
|
|
return [dict(r) for r in conn.execute(
|
|
"""SELECT date(end_time) AS night, stage_name,
|
|
SUM((end_ns - start_ns)/60000000000.0) AS minutes
|
|
FROM sleep_segments WHERE end_time >= datetime('now', ?)
|
|
GROUP BY night, stage_name ORDER BY night DESC""",
|
|
(f"-{int(days)} days",),
|
|
).fetchall()]
|
|
|
|
|
|
def summary(conn):
|
|
"""Compact health snapshot for Adolf: row counts, coverage, freshness."""
|
|
out = {}
|
|
for name, q in (
|
|
("data_points", "SELECT COUNT(*) c FROM data_points"),
|
|
("sessions", "SELECT COUNT(*) c FROM sessions"),
|
|
("sleep_segments", "SELECT COUNT(*) c FROM sleep_segments"),
|
|
):
|
|
out[name] = conn.execute(q).fetchone()["c"]
|
|
out["metrics"] = [r["metric"] for r in conn.execute(
|
|
"SELECT DISTINCT metric FROM data_points ORDER BY metric").fetchall()]
|
|
span = conn.execute(
|
|
"SELECT MIN(start_time) a, MAX(start_time) b FROM data_points").fetchone()
|
|
out["coverage"] = {"earliest": span["a"], "latest": span["b"]}
|
|
out["sync_state"] = [dict(r) for r in conn.execute(
|
|
"SELECT stream_key, last_run_at, last_status FROM sync_state").fetchall()]
|
|
last = conn.execute(
|
|
"SELECT started_at, finished_at, status, points_upserted FROM ingest_runs "
|
|
"ORDER BY id DESC LIMIT 1").fetchone()
|
|
out["last_run"] = dict(last) if last else None
|
|
return out
|