"""Day aggregate derivation per spec §5.4, §3.3. One row per UTC day, derived monotonically from hour rows — the grain at which usage-habit evidence is judged. No absent hour is ever interpolated, estimated, or fabricated (§5.3, FL-3). Coverage: the share of wall-clock seconds inside monitoring periods whose usage-habit classification is known rather than unknown (§5.3). """ import sqlite3 from dataclasses import dataclass from datetime import datetime, timedelta, timezone @dataclass(frozen=True) class DayAggregate: """One UTC day's aggregated stats.""" day: str # ISO 8601 UTC date, e.g. "2026-09-01" seconds_active: int seconds_idle: int seconds_powered_off: int seconds_unknown: int bytes_written_delta: int bytes_read_delta: int sample_count: int coverage: float def _period_wall_clock_for_day(conn: sqlite3.Connection, day: str) -> int: """Total wall-clock seconds inside monitoring periods for a UTC day. Clamps each period to the day boundary [dayT00:00, dayT24:00). """ day_start = datetime.fromisoformat(f"{day}T00:00:00+00:00") day_end = day_start + timedelta(days=1) day_start_str = day_start.isoformat() day_end_str = day_end.isoformat() cursor = conn.execute( "SELECT started_at, ended_at FROM monitoring_periods " "WHERE (ended_at IS NULL OR ended_at > ?) AND started_at < ? " "ORDER BY started_at", (day_start_str, day_end_str), ) total = 0 for row in cursor.fetchall(): period_start = row[0] period_end = row[1] effective_start = max(period_start, day_start_str) if period_end is not None: effective_end = min(period_end, day_end_str) else: effective_end = day_end_str if effective_start < effective_end: s = datetime.fromisoformat(effective_start) e = datetime.fromisoformat(effective_end) total += int((e - s).total_seconds()) return total def _hour_overlaps_period(conn: sqlite3.Connection, hour_iso: str) -> bool: """Check if an hour's wall-clock span overlaps any monitoring period.""" hour_start = datetime.fromisoformat(hour_iso) hour_end = hour_start + timedelta(hours=1) hs = hour_start.isoformat() he = hour_end.isoformat() cursor = conn.execute( "SELECT 1 FROM monitoring_periods " "WHERE started_at < ? AND (ended_at IS NULL OR ended_at > ?) " "LIMIT 1", (he, hs), ) return cursor.fetchone() is not None def derive_day(conn: sqlite3.Connection, day: str) -> DayAggregate | None: """Derive a single day aggregate from its hour rows + monitoring periods. Only hours overlapping a monitoring period contribute to the aggregate. Gap hours inside periods contribute unknown seconds. Hours outside all monitoring periods are excluded entirely (§5.2). Returns None if no hours exist for the day. """ cursor = conn.execute( "SELECT hour, active_seconds, idle_seconds, powered_off_seconds, unknown_seconds, " " bytes_written_delta, bytes_read_delta, sample_count " "FROM hour_observations " "WHERE hour LIKE ? " "ORDER BY hour", (day + "T%",), ) rows = cursor.fetchall() if not rows: return None total_active = 0 total_idle = 0 total_powered_off = 0 total_unknown_from_hours = 0 total_bw = 0 total_br = 0 total_samples = 0 total_hour_wall_clock = 0 for row in rows: # Only count hours overlapping a monitoring period if not _hour_overlaps_period(conn, row[0]): continue total_active += row[1] total_idle += row[2] total_powered_off += row[3] total_unknown_from_hours += row[4] total_bw += row[5] total_br += row[6] total_samples += row[7] total_hour_wall_clock += row[1] + row[2] + row[3] + row[4] # Wall-clock seconds inside monitoring periods for this day period_wc = _period_wall_clock_for_day(conn, day) # Gap seconds = period wall-clock - sum of existing hour wall-clock gap_seconds = max(0, period_wc - total_hour_wall_clock) total_unknown = total_unknown_from_hours + gap_seconds # Coverage: known seconds / period wall-clock (§5.2, §5.3) known_seconds = total_active + total_idle + total_powered_off coverage = known_seconds / period_wc if period_wc > 0 else 0.0 return DayAggregate( day=day, seconds_active=total_active, seconds_idle=total_idle, seconds_powered_off=total_powered_off, seconds_unknown=total_unknown, bytes_written_delta=total_bw, bytes_read_delta=total_br, sample_count=total_samples, coverage=coverage, ) def derive_all_days(conn: sqlite3.Connection) -> list[DayAggregate]: """Derive day aggregates for all days that have hour rows. Returns days sorted by date. """ cursor = conn.execute( "SELECT DISTINCT substr(hour, 1, 10) as day FROM hour_observations ORDER BY day" ) days = [row[0] for row in cursor.fetchall()] results = [] for day in days: agg = derive_day(conn, day) if agg is not None: results.append(agg) return results def persist_day_aggregate(conn: sqlite3.Connection, agg: DayAggregate) -> None: """Upsert a derived day aggregate into the day_aggregates table. Merges attributed bytes from hour observations with any existing unattributed cross-hour evidence already stored for this day. Caller must manage transactions and commits. """ existing = conn.execute( "SELECT id FROM day_aggregates WHERE day = ?", (agg.day,), ).fetchone() if existing is None: conn.execute( "INSERT INTO day_aggregates " "(day, active_seconds, idle_seconds, powered_off_seconds, unknown_seconds, " " bytes_written_delta, bytes_read_delta, sample_count, coverage) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", (agg.day, agg.seconds_active, agg.seconds_idle, agg.seconds_powered_off, agg.seconds_unknown, agg.bytes_written_delta, agg.bytes_read_delta, agg.sample_count, agg.coverage), ) else: conn.execute( "UPDATE day_aggregates " "SET active_seconds = ?, idle_seconds = ?, powered_off_seconds = ?, " " unknown_seconds = ?, bytes_written_delta = ?, bytes_read_delta = ?, " " sample_count = ?, coverage = ? " "WHERE id = ?", (agg.seconds_active, agg.seconds_idle, agg.seconds_powered_off, agg.seconds_unknown, agg.bytes_written_delta, agg.bytes_read_delta, agg.sample_count, agg.coverage, existing[0]), )