moritz/mop-skills
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: consolidate datastore into sleep, nutrition, mila databases

Browse Source
This commit is contained in:
Moritz Graf
2026-10-13 09:28:07 +02:00
parent fe4edcc217
commit 7b5d039323
5 changed files with 896 additions and 161 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/datastore/scripts/datastore.py
+424 -143
View File
@@ -154,8 +154,57 @@ class DatastoreEngine:
self._create_inline_schema()
def _create_inline_schema(self):
schema_src = Path("/Users/moritz/src/infrapuzzle/knowledge-wiki/data/_schema")
if not schema_src.is_dir():
schema_src = Path("/home/node/.openclaw/wiki/main/data/_schema")
if schema_src.is_dir():
for f in ["envelope.sql", "entities.sql", "views.sql"]:
sql_file = schema_src / f
if sql_file.is_file():
self.con.execute(sql_file.read_text(encoding="utf-8"))
return
self.con.execute("""
CREATE TABLE IF NOT EXISTS events (
CREATE TABLE IF NOT EXISTS sleep (
id VARCHAR PRIMARY KEY,
subject VARCHAR NOT NULL,
calendarDate DATE NOT NULL,
sleepType VARCHAR NOT NULL,
location VARCHAR,
isOpen BOOLEAN DEFAULT false,
startIso TIMESTAMPTZ NOT NULL,
endIso TIMESTAMPTZ,
sleepStartTimestampGMT BIGINT,
sleepEndTimestampGMT BIGINT,
sleepStartTimestampLocal BIGINT,
sleepEndTimestampLocal BIGINT,
sleepTimeSeconds INTEGER,
napTimeSeconds INTEGER,
deepSleepSeconds INTEGER,
lightSleepSeconds INTEGER,
remSleepSeconds INTEGER,
awakeSleepSeconds INTEGER,
awakeCount INTEGER,
avgSleepStress DOUBLE,
avgHeartRate DOUBLE,
averageSpO2Value DOUBLE,
lowestSpO2Value INTEGER,
highestSpO2Value INTEGER,
averageRespirationValue DOUBLE,
sleepScores JSON,
avgOvernightHrv DOUBLE,
hrvStatus VARCHAR,
bodyBatteryChange INTEGER,
restingHeartRate INTEGER,
restlessMomentsCount INTEGER,
note VARCHAR,
actor VARCHAR NOT NULL,
created_at TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ,
deleted_at TIMESTAMPTZ
);
CREATE TABLE IF NOT EXISTS nutrition (
id VARCHAR PRIMARY KEY,
kind VARCHAR NOT NULL,
subject VARCHAR NOT NULL,
@@ -173,6 +222,30 @@ class DatastoreEngine:
CHECK ("end" IS NULL OR "end" >= start)
);
CREATE TABLE IF NOT EXISTS mila (
id VARCHAR PRIMARY KEY,
kind VARCHAR NOT NULL,
subject VARCHAR NOT NULL DEFAULT 'mila',
actor VARCHAR NOT NULL,
start TIMESTAMPTZ NOT NULL,
"end" TIMESTAMPTZ,
note VARCHAR,
payload JSON NOT NULL,
source_ref VARCHAR,
created_at TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ,
deleted_at TIMESTAMPTZ
);
CREATE OR REPLACE VIEW events AS
SELECT id, kind, subject, actor, start, "end", estimated, meal, note, payload, source_ref, created_at, updated_at, deleted_at FROM nutrition
UNION ALL
SELECT id, kind, subject, actor, start, "end", false AS estimated, NULL AS meal, note, payload, source_ref, created_at, updated_at, deleted_at FROM mila
UNION ALL
SELECT id, 'sleep.session' AS kind, subject, actor, startIso AS start, endIso AS "end", false AS estimated, NULL AS meal, note,
to_json(struct_pack(type := sleepType, location := location, duration_minutes := COALESCE(ROUND(sleepTimeSeconds / 60.0), 0))) AS payload,
NULL AS source_ref, created_at, updated_at, deleted_at FROM sleep;
CREATE TABLE IF NOT EXISTS people (
id VARCHAR PRIMARY KEY,
name VARCHAR NOT NULL,
@@ -229,38 +302,19 @@ class DatastoreEngine:
restingHeartRate INTEGER,
raw JSON
);
CREATE OR REPLACE VIEW v_growth AS
SELECT
id,
subject,
actor,
start,
start AT TIME ZONE 'Europe/Berlin' AS local_time,
CAST(start AT TIME ZONE 'Europe/Berlin' AS DATE) AS log_date,
CAST(json_extract(payload, '$.weight_kg') AS DOUBLE) AS weight_kg,
CAST(json_extract(payload, '$.height_cm') AS DOUBLE) AS height_cm,
CAST(json_extract(payload, '$.head_circ_cm') AS DOUBLE) AS head_circ_cm,
note,
created_at
FROM events
WHERE kind = 'growth.measurement' AND deleted_at IS NULL;
CREATE OR REPLACE VIEW v_growth_daily AS
SELECT
subject,
log_date,
ROUND(AVG(weight_kg), 3) AS weight_kg,
ROUND(AVG(height_cm), 1) AS height_cm,
ROUND(AVG(head_circ_cm), 1) AS head_circ_cm,
COUNT(*) AS measurement_count
FROM v_growth
GROUP BY subject, log_date
ORDER BY log_date DESC;
""")
def _get_target_db(self, kind: str) -> str:
"""Maps an event kind to one of the 3 target databases: 'sleep', 'nutrition', or 'mila'."""
if kind.startswith("sleep."):
return "sleep"
elif kind.startswith("diaper.") or kind.startswith("growth."):
return "mila"
else:
return "nutrition"
def _load_all_data(self):
"""Loads all existing JSONL files from entities/, events/, and garmin/ into DuckDB."""
"""Loads all existing JSONL files from entities/, sleep/, nutrition/, mila/, and garmin/ into DuckDB."""
# 1. Load entities
entities_dir = self.data_dir / "entities"
if entities_dir.is_dir():
@@ -269,18 +323,80 @@ class DatastoreEngine:
if collection in ["people", "foods", "recipes", "places"]:
self._load_jsonl_file(f, collection)
# 2. Load events
# 2. Load the 3 core databases: sleep, nutrition, mila
for db_name in ["sleep", "nutrition", "mila"]:
db_dir = self.data_dir / db_name
if db_dir.is_dir():
for f in db_dir.glob("*.jsonl"):
self._load_jsonl_file(f, db_name)
# 3. Fallback for legacy events/ if present
events_dir = self.data_dir / "events"
if events_dir.is_dir():
for f in events_dir.glob("**/*.jsonl"):
self._load_jsonl_file(f, "events")
domain = f.parent.name
if domain == "sleep":
self._load_legacy_sleep_file(f)
elif domain in ("diaper", "growth"):
self._load_jsonl_file(f, "mila")
else:
self._load_jsonl_file(f, "nutrition")
# 3. Load garmin daily
# 4. Load garmin daily
garmin_daily_dir = self.data_dir / "garmin" / "daily"
if garmin_daily_dir.is_dir():
for f in garmin_daily_dir.glob("*.jsonl"):
self._load_garmin_daily_file(f)
def _load_legacy_sleep_file(self, file_path: Path):
"""Converts legacy events/sleep/*.jsonl records into the unified sleep table."""
if not file_path.is_file() or file_path.stat().st_size == 0:
return
with open(file_path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
r = json.loads(line)
start_str = r.get("start")
start_dt = datetime.datetime.fromisoformat(start_str.replace("Z", "+00:00")).astimezone(BERLIN_TZ)
cal_date = start_dt.strftime("%Y-%m-%d")
end_str = r.get("end")
end_dt = datetime.datetime.fromisoformat(end_str.replace("Z", "+00:00")).astimezone(BERLIN_TZ) if end_str else None
dur_sec = int((end_dt - start_dt).total_seconds()) if end_dt else None
payload = r.get("payload") or {}
stype = payload.get("type") or "nap"
loc = payload.get("location") or "bett"
self.con.execute(
"""
INSERT OR REPLACE INTO sleep (
id, subject, calendarDate, sleepType, location, isOpen,
startIso, endIso, sleepTimeSeconds, napTimeSeconds,
note, actor, created_at, updated_at, deleted_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
[
r["id"],
r.get("subject", "mila"),
cal_date,
stype,
loc,
end_str is None,
start_str,
end_str,
dur_sec,
dur_sec if stype == "nap" else 0,
r.get("note"),
r.get("actor", "agent:hebamme"),
r.get("created_at") or start_str,
r.get("updated_at"),
r.get("deleted_at"),
],
)
except Exception as e:
logger.warning(f"Error loading legacy sleep from {file_path}: {e}")
def _load_garmin_daily_file(self, file_path: Path):
"""Reads daily summaries from JSONL into garmin_daily table."""
if not file_path.is_file() or file_path.stat().st_size == 0:
@@ -356,6 +472,21 @@ class DatastoreEngine:
if not file_path.is_file() or file_path.stat().st_size == 0:
return
valid_cols_map = {
"nutrition": {"id", "kind", "subject", "actor", "start", "end", "estimated", "meal", "note", "payload", "source_ref", "created_at", "updated_at", "deleted_at"},
"mila": {"id", "kind", "subject", "actor", "start", "end", "note", "payload", "source_ref", "created_at", "updated_at", "deleted_at"},
"sleep": {
"id", "subject", "calendarDate", "sleepType", "location", "isOpen", "startIso", "endIso",
"sleepStartTimestampGMT", "sleepEndTimestampGMT", "sleepStartTimestampLocal", "sleepEndTimestampLocal",
"sleepTimeSeconds", "napTimeSeconds", "deepSleepSeconds", "lightSleepSeconds", "remSleepSeconds",
"awakeSleepSeconds", "awakeCount", "avgSleepStress", "avgHeartRate", "averageSpO2Value",
"lowestSpO2Value", "highestSpO2Value", "averageRespirationValue", "sleepScores", "avgOvernightHrv",
"hrvStatus", "bodyBatteryChange", "restingHeartRate", "restlessMomentsCount", "note", "actor",
"created_at", "updated_at", "deleted_at"
},
}
allowed_cols = valid_cols_map.get(table_name)
with open(file_path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
@@ -363,14 +494,23 @@ class DatastoreEngine:
continue
try:
record = json.loads(line)
if table_name == "events":
if table_name in ("nutrition", "mila"):
if "created_at" not in record or not record.get("created_at"):
record["created_at"] = record.get("start") or datetime.datetime.now(datetime.timezone.utc).isoformat()
if "estimated" not in record:
if table_name == "nutrition" and "estimated" not in record:
record["estimated"] = False
if "deleted_at" not in record:
record["deleted_at"] = None
columns = list(record.keys())
elif table_name == "sleep":
if "created_at" not in record or not record.get("created_at"):
record["created_at"] = record.get("startIso") or datetime.datetime.now(datetime.timezone.utc).isoformat()
if "deleted_at" not in record:
record["deleted_at"] = None
if allowed_cols:
columns = [c for c in record.keys() if c in allowed_cols]
else:
columns = list(record.keys())
placeholders = ", ".join(["?"] * len(columns))
col_names = ", ".join([f'"{c}"' for c in columns])
values = [
@@ -388,7 +528,7 @@ class DatastoreEngine:
"""Determines if two event payloads represent the same intake/session."""
if p1 == p2:
return True
if kind.startswith("nutrition."):
if kind.startswith("nutrition.") or kind == "feeding.solid":
food1 = p1.get("food_id")
food2 = p2.get("food_id")
item1 = str(p1.get("item", "")).strip().lower()
@@ -400,8 +540,8 @@ class DatastoreEngine:
or (food2 and item1 and food2 == item1)
)
if same_item:
qty1 = p1.get("qty")
qty2 = p2.get("qty")
qty1 = p1.get("qty") or p1.get("amount_g")
qty2 = p2.get("qty") or p2.get("amount_g")
if qty1 is not None and qty2 is not None:
try:
return abs(float(qty1) - float(qty2)) < 0.01
@@ -417,7 +557,7 @@ class DatastoreEngine:
if p1.get("pee") == p2.get("pee") and p1.get("poo") == p2.get("poo"):
return True
elif kind.startswith("sleep."):
if p1.get("type") == p2.get("type"):
if (p1.get("type") or "nap") == (p2.get("type") or "nap"):
return True
elif kind.startswith("growth."):
w1, w2 = p1.get("weight_kg"), p2.get("weight_kg")
@@ -472,13 +612,13 @@ class DatastoreEngine:
return None
def _get_open_sleep_sessions(self, subject: str, exclude_id: Optional[str] = None) -> List[Dict[str, Any]]:
"""Finds open sleep sessions (end IS NULL) for subject, sorted chronologically."""
"""Finds open sleep sessions (isOpen=true or endIso IS NULL) for subject, sorted chronologically."""
rows = self.con.execute(
"""
SELECT id, start, payload, note
FROM events
WHERE kind = 'sleep.session' AND subject = ? AND "end" IS NULL AND deleted_at IS NULL
ORDER BY start ASC
SELECT id, startIso, sleepType, location, note
FROM sleep
WHERE subject = ? AND (isOpen = true OR endIso IS NULL) AND deleted_at IS NULL
ORDER BY startIso ASC
""",
[subject],
).fetchall()
@@ -489,8 +629,8 @@ class DatastoreEngine:
results.append({
"id": r[0],
"start": str(r[1]),
"payload": json.loads(r[2]) if isinstance(r[2], str) else (r[2] or {}),
"note": r[3],
"payload": {"type": r[2], "location": r[3]},
"note": r[4],
})
return results
@@ -512,7 +652,7 @@ class DatastoreEngine:
source_ref: Optional[str] = None,
allow_duplicate: bool = False,
) -> Dict[str, Any]:
"""Adds a time-series event, saves to JSONL partition, and returns confirmation."""
"""Adds a time-series event into the appropriate database (sleep, nutrition, or mila)."""
event_id = generate_ulid()
now_dt = datetime.datetime.now(datetime.timezone.utc)
now_iso = now_dt.isoformat()
@@ -555,6 +695,7 @@ class DatastoreEngine:
start_iso = start_dt.isoformat()
end_iso = None
end_dt = None
if end:
end_dt = datetime.datetime.fromisoformat(end)
if end_dt.tzinfo is None:
@@ -585,16 +726,141 @@ class DatastoreEngine:
o_start_dt = datetime.datetime.fromisoformat(str(first_open["start"]).replace("Z", "+00:00")).astimezone(BERLIN_TZ)
open_warning = f"Vorheriges Schlaf-Event ({first_open['id']} ab {o_start_dt.strftime('%d.%m. %H:%M Uhr')}) ist noch offen! Bitte Aufwachzeit nachreichen."
# If adding a new open interval (end is None)
if end_iso is None:
# If existing open session is recent (< 8 hours on same calendar day) and not allow_duplicate, prevent accidental duplicate live nap
diff_hours = (start_dt - o_start_dt).total_seconds() / 3600.0
if diff_hours < 8.0 and start_dt.date() == o_start_dt.date() and not allow_duplicate:
raise ValueError(
f"Subject '{subject}' hat bereits ein aktives Schlaf-Event ({first_open['id']} gestartet um {o_start_dt.strftime('%H:%M Uhr')}). Bitte beende es zuerst."
)
record = {
target_db = self._get_target_db(kind)
month_str = start_dt.astimezone(BERLIN_TZ).strftime("%Y-%m")
db_dir = self.data_dir / target_db
db_dir.mkdir(parents=True, exist_ok=True)
partition_file = db_dir / f"{month_str}.jsonl"
if target_db == "sleep":
cal_date = start_dt.astimezone(BERLIN_TZ).strftime("%Y-%m-%d")
sleep_type = payload.get("type") or "nap"
location = payload.get("location") or "bett"
is_open = (end_iso is None)
start_gmt = int(start_dt.timestamp() * 1000)
end_gmt = int(end_dt.timestamp() * 1000) if end_dt else None
start_local = int((start_dt.timestamp() + start_dt.utcoffset().total_seconds()) * 1000)
end_local = int((end_dt.timestamp() + end_dt.utcoffset().total_seconds()) * 1000) if end_dt else None
dur_sec = int((end_dt - start_dt).total_seconds()) if end_dt else None
nap_sec = dur_sec if (sleep_type == "nap" and dur_sec is not None) else 0
disk_record = {
"id": event_id,
"subject": subject,
"calendarDate": cal_date,
"sleepType": sleep_type,
"location": location,
"isOpen": is_open,
"startIso": start_iso,
"endIso": end_iso,
"sleepStartTimestampGMT": start_gmt,
"sleepEndTimestampGMT": end_gmt,
"sleepStartTimestampLocal": start_local,
"sleepEndTimestampLocal": end_local,
"sleepTimeSeconds": dur_sec,
"napTimeSeconds": nap_sec,
"deepSleepSeconds": None,
"lightSleepSeconds": None,
"remSleepSeconds": None,
"awakeSleepSeconds": None,
"awakeCount": None,
"avgSleepStress": None,
"avgHeartRate": None,
"averageSpO2Value": None,
"lowestSpO2Value": None,
"highestSpO2Value": None,
"averageRespirationValue": None,
"sleepScores": None,
"avgOvernightHrv": None,
"hrvStatus": None,
"bodyBatteryChange": None,
"restingHeartRate": None,
"restlessMomentsCount": None,
"note": note,
"actor": actor,
"created_at": now_iso,
"updated_at": None,
"deleted_at": None,
}
self.con.execute(
"""
INSERT INTO sleep (
id, subject, calendarDate, sleepType, location, isOpen,
startIso, endIso, sleepStartTimestampGMT, sleepEndTimestampGMT,
sleepStartTimestampLocal, sleepEndTimestampLocal,
sleepTimeSeconds, napTimeSeconds, note, actor, created_at, updated_at, deleted_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
[
event_id, subject, cal_date, sleep_type, location, is_open,
start_iso, end_iso, start_gmt, end_gmt, start_local, end_local,
dur_sec, nap_sec, note, actor, now_iso, None, None,
],
)
sort_key = lambda x: (x.get("calendarDate", ""), x.get("startIso", ""), x.get("id", ""))
elif target_db == "mila":
disk_record = {
"id": event_id,
"kind": kind,
"subject": subject,
"actor": actor,
"start": start_iso,
"end": end_iso,
"note": note,
"payload": payload,
"source_ref": source_ref,
"created_at": now_iso,
"updated_at": None,
"deleted_at": None,
}
self.con.execute(
"""
INSERT INTO mila (id, kind, subject, actor, start, "end", note, payload, source_ref, created_at, updated_at, deleted_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
[event_id, kind, subject, actor, start_iso, end_iso, note, json.dumps(payload), source_ref, now_iso, None, None],
)
sort_key = lambda x: (x.get("start", ""), x.get("id", ""))
else:
disk_record = {
"id": event_id,
"kind": kind,
"subject": subject,
"actor": actor,
"start": start_iso,
"end": end_iso,
"estimated": estimated,
"meal": meal,
"note": note,
"payload": payload,
"source_ref": source_ref,
"created_at": now_iso,
"updated_at": None,
"deleted_at": None,
}
self.con.execute(
"""
INSERT INTO nutrition (id, kind, subject, actor, start, "end", estimated, meal, note, payload, source_ref, created_at, updated_at, deleted_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
[event_id, kind, subject, actor, start_iso, end_iso, estimated, meal, note, json.dumps(payload), source_ref, now_iso, None, None],
)
sort_key = lambda x: (x.get("start", ""), x.get("id", ""))
with open(partition_file, "a", encoding="utf-8") as f:
f.write(json.dumps(disk_record, ensure_ascii=False) + "\n")
self._sort_jsonl_file(partition_file, key_fn=sort_key)
record_response = {
"id": event_id,
"kind": kind,
"subject": subject,
@@ -611,44 +877,6 @@ class DatastoreEngine:
"deleted_at": None,
}
# Insert into in-memory DuckDB
self.con.execute(
"""
INSERT INTO events (id, kind, subject, actor, start, "end", estimated, meal, note, payload, source_ref, created_at, updated_at, deleted_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
[
event_id,
kind,
subject,
actor,
start_iso,
end_iso,
estimated,
meal,
note,
json.dumps(payload),
source_ref,
now_iso,
None,
None,
],
)
# Write to partitioned JSONL file
# Partition path: events/<domain>/<YYYY-MM>.jsonl
domain = kind.split(".")[0]
month_str = start_dt.astimezone(BERLIN_TZ).strftime("%Y-%m")
domain_dir = self.data_dir / "events" / domain
domain_dir.mkdir(parents=True, exist_ok=True)
partition_file = domain_dir / f"{month_str}.jsonl"
with open(partition_file, "a", encoding="utf-8") as f:
f.write(json.dumps(record, ensure_ascii=False) + "\n")
# Sort partition deterministically by start, id
self._sort_jsonl_file(partition_file, key_fn=lambda x: (x.get("start", ""), x.get("id", "")))
# Fetch computed feedback
feedback = self._get_event_feedback(kind, subject, start_dt.date())
if open_warning:
@@ -663,7 +891,7 @@ class DatastoreEngine:
return {
"status": "success",
"event": record,
"event": record_response,
"feedback": feedback,
}
@@ -680,29 +908,55 @@ class DatastoreEngine:
if since and end_time:
raise ValueError("Cannot specify both 'since' and 'end_time'. Use 'since' for relative times or 'end_time' for absolute times.")
if event_id:
row = self.con.execute(
"""
SELECT id, start FROM events
WHERE id = ? AND kind = ? AND subject = ? AND "end" IS NULL AND deleted_at IS NULL
""",
[event_id, kind, subject],
).fetchone()
if not row:
raise ValueError(f"No open event with id='{event_id}' found for kind='{kind}' and subject='{subject}'")
target_db = self._get_target_db(kind)
if target_db == "sleep":
if event_id:
row = self.con.execute(
"""
SELECT id, startIso, sleepType FROM sleep
WHERE id = ? AND subject = ? AND (isOpen = true OR endIso IS NULL) AND deleted_at IS NULL
""",
[event_id, subject],
).fetchone()
if not row:
raise ValueError(f"No open event with id='{event_id}' found for kind='{kind}' and subject='{subject}'")
else:
row = self.con.execute(
"""
SELECT id, startIso, sleepType FROM sleep
WHERE subject = ? AND (isOpen = true OR endIso IS NULL) AND deleted_at IS NULL
ORDER BY startIso DESC LIMIT 1
""",
[subject],
).fetchone()
if not row:
raise ValueError(f"No open event found for kind='{kind}' and subject='{subject}'")
resolved_event_id, start_str, sleep_type = row[0], str(row[1]), row[2]
else:
row = self.con.execute(
"""
SELECT id, start FROM events
WHERE kind = ? AND subject = ? AND "end" IS NULL AND deleted_at IS NULL
ORDER BY start DESC LIMIT 1
""",
[kind, subject],
).fetchone()
if not row:
raise ValueError(f"No open event found for kind='{kind}' and subject='{subject}'")
if event_id:
row = self.con.execute(
f"""
SELECT id, start FROM "{target_db}"
WHERE id = ? AND kind = ? AND subject = ? AND "end" IS NULL AND deleted_at IS NULL
""",
[event_id, kind, subject],
).fetchone()
if not row:
raise ValueError(f"No open event with id='{event_id}' found for kind='{kind}' and subject='{subject}'")
else:
row = self.con.execute(
f"""
SELECT id, start FROM "{target_db}"
WHERE kind = ? AND subject = ? AND "end" IS NULL AND deleted_at IS NULL
ORDER BY start DESC LIMIT 1
""",
[kind, subject],
).fetchone()
if not row:
raise ValueError(f"No open event found for kind='{kind}' and subject='{subject}'")
resolved_event_id, start_str, sleep_type = row[0], str(row[1]), None
resolved_event_id, start_str = row[0], str(row[1])
now_dt = datetime.datetime.now(datetime.timezone.utc)
if since:
delta = parse_duration(since)
@@ -722,24 +976,41 @@ class DatastoreEngine:
end_iso = end_dt.isoformat()
now_iso = now_dt.isoformat()
duration_seconds = int((end_dt.astimezone(BERLIN_TZ) - start_dt.astimezone(BERLIN_TZ)).total_seconds())
duration_minutes = int(duration_seconds / 60)
# Update in DuckDB
self.con.execute(
"""
UPDATE events
SET "end" = ?, updated_at = ?
WHERE id = ?
""",
[end_iso, now_iso, resolved_event_id],
)
# Update in JSONL
domain = kind.split(".")[0]
self._update_jsonl_record(domain, resolved_event_id, {"end": end_iso, "updated_at": now_iso})
# Calculate duration
start_dt = datetime.datetime.fromisoformat(start_str.replace("Z", "+00:00"))
duration_minutes = int((end_dt.astimezone(BERLIN_TZ) - start_dt.astimezone(BERLIN_TZ)).total_seconds() / 60)
if target_db == "sleep":
end_gmt = int(end_dt.timestamp() * 1000)
end_local = int((end_dt.timestamp() + end_dt.utcoffset().total_seconds()) * 1000)
nap_sec = duration_seconds if sleep_type == "nap" else 0
self.con.execute(
"""
UPDATE sleep
SET endIso = ?, isOpen = false, sleepEndTimestampGMT = ?, sleepEndTimestampLocal = ?,
sleepTimeSeconds = ?, napTimeSeconds = ?, updated_at = ?
WHERE id = ?
""",
[end_iso, end_gmt, end_local, duration_seconds, nap_sec, now_iso, resolved_event_id],
)
self._update_jsonl_record("sleep", resolved_event_id, {
"endIso": end_iso,
"isOpen": False,
"sleepEndTimestampGMT": end_gmt,
"sleepEndTimestampLocal": end_local,
"sleepTimeSeconds": duration_seconds,
"napTimeSeconds": nap_sec,
"updated_at": now_iso,
})
else:
self.con.execute(
f"""
UPDATE "{target_db}"
SET "end" = ?, updated_at = ?
WHERE id = ?
""",
[end_iso, now_iso, resolved_event_id],
)
self._update_jsonl_record(target_db, resolved_event_id, {"end": end_iso, "updated_at": now_iso})
self._trigger_website_update(subject)
self._git_sync(f"chore(data): stop {kind} for {subject}")
@@ -758,7 +1029,7 @@ class DatastoreEngine:
}
def delete_event(self, event_id: str) -> Dict[str, Any]:
"""Soft-deletes an event."""
"""Soft-deletes an event across sleep, nutrition, or mila."""
now_iso = datetime.datetime.now(datetime.timezone.utc).isoformat()
row = self.con.execute("SELECT kind, subject FROM events WHERE id = ?", [event_id]).fetchone()
if not row:
@@ -766,10 +1037,10 @@ class DatastoreEngine:
kind = row[0]
subject = row[1]
domain = kind.split(".")[0]
target_db = self._get_target_db(kind)
self.con.execute("UPDATE events SET deleted_at = ? WHERE id = ?", [now_iso, event_id])
self._update_jsonl_record(domain, event_id, {"deleted_at": now_iso})
self.con.execute(f'UPDATE "{target_db}" SET deleted_at = ? WHERE id = ?', [now_iso, event_id])
self._update_jsonl_record(target_db, event_id, {"deleted_at": now_iso})
self._trigger_website_update(subject)
self._git_sync(f"chore(data): delete event {event_id} ({kind}) for {subject}")
@@ -950,7 +1221,7 @@ class DatastoreEngine:
def _get_event_feedback(self, kind: str, subject: str, target_date: datetime.date) -> Dict[str, Any]:
"""Generates compact running status feedback after an insert."""
date_str = target_date.strftime("%Y-%m-%d")
if kind.startswith("nutrition"):
if kind.startswith("nutrition") and subject != "mila":
daily = self.con.execute(
"SELECT total_protein_g, total_carbs_g, total_sugar_g, total_fat_g, total_kcal FROM v_nutrition_daily WHERE subject = ? AND log_date = ?",
[subject, date_str],
@@ -985,16 +1256,26 @@ class DatastoreEngine:
"wet_count": daily[1],
"dirty_count": daily[2],
}
elif kind.startswith("feeding"):
elif kind.startswith("feeding") or (kind.startswith("nutrition") and subject == "mila"):
daily = self.con.execute(
"SELECT bottle_count, total_consumed_ml FROM v_feeding_daily WHERE subject = ? AND log_date = ?",
"SELECT bottle_count, total_consumed_ml, solid_count, total_solid_g, total_kcal FROM v_feeding_daily WHERE subject = ? AND log_date = ?",
[subject, date_str],
).fetchone()
if daily:
b_cnt, b_ml, s_cnt, s_g, tot_kcal = daily[0], daily[1], daily[2], daily[3], daily[4]
parts = []
if b_cnt > 0 or b_ml > 0:
parts.append(f"{b_cnt} Fläschchen ({int(b_ml)} ml getrunken)")
if s_cnt > 0 or s_g > 0:
parts.append(f"{s_cnt}x Brei/Beikost ({int(s_g)} g)")
summary_str = " · ".join(parts) if parts else f"{b_cnt} Fläschchen ({int(b_ml)} ml getrunken)"
return {
"day_total_formatted": f"Fläschchen ({date_str}): {daily[0]} Fläschchen ({int(daily[1])} ml getrunken)",
"bottle_count": daily[0],
"total_consumed_ml": daily[1],
"day_total_formatted": f"Ernährung Mila ({date_str}): {summary_str} · ~{int(tot_kcal)} kcal",
"bottle_count": b_cnt,
"total_consumed_ml": b_ml,
"solid_count": s_cnt,
"total_solid_g": s_g,
"total_kcal": tot_kcal,
}
elif kind.startswith("growth"):
row = self.con.execute(
@@ -1056,13 +1337,13 @@ class DatastoreEngine:
for r in records:
f.write(json.dumps(r, ensure_ascii=False) + "\n")
def _update_jsonl_record(self, domain: str, event_id: str, updates: Dict[str, Any]):
"""Finds and updates a record in events/<domain>/*.jsonl."""
domain_dir = self.data_dir / "events" / domain
if not domain_dir.is_dir():
def _update_jsonl_record(self, target_db: str, event_id: str, updates: Dict[str, Any]):
"""Finds and updates a record in <target_db>/*.jsonl."""
db_dir = self.data_dir / target_db
if not db_dir.is_dir():
return
for p_file in domain_dir.glob("*.jsonl"):
for p_file in db_dir.glob("*.jsonl"):
records = []
found = False
with open(p_file, "r", encoding="utf-8") as f: