clio/clio-pi/bin/gen_recording_status.py

#!/usr/bin/env python3
import argparse
import datetime as dt
import html
import os
import re
import shutil
import statistics
from pathlib import Path
from typing import List, Tuple, Optional

FNAME_RE = re.compile(r"^stationary-(\d{4}-\d{2}-\d{2})_(\d{2}-\d{2}-\d{2})\.opus$")

def parse_start(path: Path, tz) -> Optional[dt.datetime]:
    m = FNAME_RE.match(path.name)
    if not m:
        return None
    date_str, time_str = m.groups()
    start = dt.datetime.strptime(f"{date_str} {time_str}", "%Y-%m-%d %H-%M-%S")
    return start.replace(tzinfo=tz)

def get_duration_seconds(path: Path) -> Optional[float]:
    try:
        from mutagen.oggopus import OggOpus  # optional
        audio = OggOpus(path)
        return float(getattr(audio.info, "length", 0.0)) or None
    except Exception:
        return None

def human_delta(delta: dt.timedelta) -> str:
    secs = int(abs(delta).total_seconds())
    d, rem = divmod(secs, 86400)
    h, rem = divmod(rem, 3600)
    m, s = divmod(rem, 60)
    parts = []
    if d: parts.append(f"{d}d")
    if h: parts.append(f"{h}h")
    if m: parts.append(f"{m}m")
    if not parts: parts.append(f"{s}s")
    return " ".join(parts)

def merge_intervals(intervals: List[Tuple[dt.datetime, dt.datetime]]) -> List[Tuple[dt.datetime, dt.datetime]]:
    if not intervals: return []
    intervals.sort(key=lambda x: x[0])
    merged = [intervals[0]]
    for s, e in intervals[1:]:
        ls, le = merged[-1]
        if s <= le:
            merged[-1] = (ls, max(le, e))
        else:
            merged.append((s, e))
    return merged

def day_bounds(d: dt.date, tz):
    start = dt.datetime(d.year, d.month, d.day, tzinfo=tz)
    return start, start + dt.timedelta(days=1)

def intersect(a: Tuple[dt.datetime, dt.datetime], b: Tuple[dt.datetime, dt.datetime]) -> Optional[Tuple[dt.datetime, dt.datetime]]:
    s1, e1 = a; s2, e2 = b
    s = max(s1, s2); e = min(e1, e2)
    return (s, e) if s < e else None

def pct(x, total):
    return 0 if total <= 0 else max(0, min(100, (x/total)*100))

def fmt_ampm(t: dt.datetime) -> str:
    s = t.strftime("%I:%M%p")  # 01:05PM
    return s.lstrip("0")

def build(args):
    tz = dt.datetime.now().astimezone().tzinfo
    rec_dir = Path(args.dir).expanduser().resolve()
    files = [p for p in rec_dir.glob("*.opus") if p.is_file()]
    files.sort()

    # Gather metadata
    items = []  # (path, start_dt, size_bytes, duration_opt)
    for p in files:
        s = parse_start(p, tz)
        if not s:
            continue
        dur = get_duration_seconds(p) if args.read_durations else None
        try:
            size = p.stat().st_size
        except Exception:
            size = 0
        items.append((p, s, size, dur))
    items.sort(key=lambda x: x[1])

    starts = [s for _, s, _, _ in items]
    deltas = [(b - a).total_seconds() for a, b in zip(starts, starts[1:])]

    # Duration policy (defaults: NO inference, assume 2h)
    if args.infer:
        sane = [d for d in deltas if args.infer_min_sec <= d <= args.infer_max_sec]
        inferred_default = (statistics.median(sane) if sane else args.assume_duration_sec or args.infer_default_sec)
        inferred_default = max(args.infer_min_sec, min(args.infer_max_sec, inferred_default))
    else:
        inferred_default = args.assume_duration_sec or args.infer_default_sec

    # Build intervals
    intervals: List[Tuple[dt.datetime, dt.datetime]] = []
    for idx, (p, s, _size, dur) in enumerate(items):
        if dur is None or dur <= 0:
            dur_use = float(inferred_default)
        else:
            dur_use = dur
        if idx < len(items)-1 and args.guard_gap_sec > 0:
            to_next = (items[idx+1][1] - s).total_seconds()
            dur_use = min(dur_use, max(1.0, to_next - args.guard_gap_sec))
        e = s + dt.timedelta(seconds=max(1.0, dur_use))
        intervals.append((s, e))

    gen_time = dt.datetime.now(tz)
    last_start = starts[-1] if starts else None

    # FS usage for recordings partition
    try:
        total_b, used_b, free_b = shutil.disk_usage(str(rec_dir))
    except Exception:
        total_b = used_b = free_b = 0

    # Window
    today = gen_time.date()
    start_date = today - dt.timedelta(days=args.days - 1)

    big_gap = dt.timedelta(minutes=args.gap_yellow_min)
    green_gap = dt.timedelta(minutes=args.gap_green_min)

    rows = []
    window_total_bytes = 0
    for i in range(args.days):
        d = start_date + dt.timedelta(days=i)
        d_start, d_end = day_bounds(d, tz)

        # Per-day file count and bytes (count files that START this day)
        day_files = [(p, s, size) for (p, s, size, _du) in [(p, s, size, du) for (p, s, size, du) in items] if d_start <= s < d_end]
        day_bytes = sum(size for (_p, _s, size) in day_files)
        window_total_bytes += day_bytes

        # Intervals overlapping this day
        day_intervals = []
        for (s, e) in intervals:
            inter = intersect((s, e), (d_start, d_end))
            if inter:
                day_intervals.append(inter)

        merged = merge_intervals(day_intervals)
        covered = sum((e - s).total_seconds() for s, e in merged)
        day_seconds = 86400.0
        coverage_pct = pct(covered, day_seconds)

        # Max gap includes edges
        if not merged:
            max_gap = dt.timedelta(days=1)
        else:
            gaps = [merged[0][0] - d_start]
            gaps += [b[0] - a[1] for a, b in zip(merged, merged[1:])]
            gaps.append(d_end - merged[-1][1])
            max_gap = max(gaps) if gaps else dt.timedelta(0)

        # Status
        count = len(day_files)
        if count == 0:
            status = "red"
        elif max_gap <= green_gap and coverage_pct >= args.green_cover_pct:
            status = "green"
        elif max_gap >= big_gap or coverage_pct < args.yellow_cover_pct:
            status = "yellow"
        else:
            status = "yellow"

        # Mini timeline segments
        segs = []
        for s, e in merged:
            left = pct((s - d_start).total_seconds(), day_seconds)
            width = pct((e - s).total_seconds(), day_seconds)
            if width > 0:
                segs.append((left, width))

        # Details list: show ALL spans; AM/PM times
        details_txt = ", ".join(f"{fmt_ampm(s)}–{fmt_ampm(e)}" for s, e in merged)

        rows.append({
            "date": d.isoformat(),
            "count": count,
            "bytes": day_bytes,
            "max_gap": max_gap,
            "coverage_pct": coverage_pct,
            "status": status,
            "segments": segs,
            "details": details_txt or "—",
        })

    # HTML
    def hhmm(td: dt.timedelta) -> str:
        total = int(td.total_seconds())
        h = total // 3600
        m = (total % 3600) // 60
        return f"{h:02d}:{m:02d}"

    def fmt_bytes(n: int) -> str:
        units = ("B", "KiB", "MiB", "GiB", "TiB", "PiB")
        x = float(n)
        for u in units:
            if x < 1024.0 or u == units[-1]:
                return f"{int(x)} {u}" if u == "B" else f"{x:.1f} {u}"
            x /= 1024.0


    title = args.title
    last_str = "—"
    if last_start:
        last_str = f"{last_start.strftime('%Y-%m-%d %I:%M:%S %p %Z')} ({human_delta(gen_time - last_start)} ago)"

    # FS usage strings
    total_str = fmt_bytes(total_b)
    used_str  = fmt_bytes(used_b)
    free_str  = fmt_bytes(free_b)
    window_bytes_str = fmt_bytes(window_total_bytes)

    html_out = f"""<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta http-equiv="refresh" content="300">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{html.escape(title)}</title>
<style>
  :root {{
    --green:#30a46c; --yellow:#f5d90a; --red:#e5484d;
    --bg-green:#e9f7f0; --bg-yellow:#fff8cc; --bg-red:#fdeaea;
  }}
  body {{ font-family: system-ui, -apple-system, Segoe UI, Roboto, sans-serif; margin: 1.5rem; }}
  h1 {{ margin: 0 0 .4rem 0; }}
  .muted {{ color: #666; font-size: .9rem; }}
  table {{ border-collapse: collapse; width: 100%; max-width: 1200px; }}
  th, td {{ padding: .5rem .6rem; border-bottom: 1px solid #eee; text-align: left; vertical-align: middle; }}
  tr:hover td {{ background: #fafafa; }}
  .right {{ text-align: right; }}
  .row-green td {{ background: var(--bg-green); }}
  .row-yellow td {{ background: var(--bg-yellow); }}
  .row-red td {{ background: var(--bg-red); }}
  .status-dot {{ display:inline-block;width:10px;height:10px;border-radius:50%;vertical-align:middle;margin-right:.4rem;}}
  .green {{ background: var(--green); }}
  .yellow {{ background: var(--yellow); }}
  .red {{ background: var(--red); }}
  .bar {{ position: relative; height: 12px; background: #eee; border-radius: 6px; overflow: hidden; }}
  .bar .seg {{ position: absolute; top:0; bottom:0; background: var(--green); opacity:.9; }}
  .cover-label {{ font-size:.85rem; white-space:nowrap; }}
  .nowrap {{ white-space:nowrap; }}
  .hdr {{ margin-top:.6rem; margin-bottom:.8rem; }}
  .kvs span {{ margin-right: 1rem; }}
</style>
</head>
<body>
  <h1>{html.escape(title)}</h1>
  <div class="muted">Generated at {gen_time.strftime('%Y-%m-%d %I:%M:%S %p %Z')}</div>
  <div class="hdr kvs muted">
    <span><strong>Last recording started:</strong> {html.escape(last_str)}</span>
    <span><strong>Window:</strong> {args.days} days</span>
    <span><strong>Dir:</strong> {html.escape(str(rec_dir))}</span>
  </div>
  <div class="hdr kvs">
    <span><strong>Data in window:</strong> {window_bytes_str}</span>
    <span><strong>Filesystem usage (recordings partition):</strong> used {used_str} • free {free_str} • total {total_str}</span>
  </div>

  <table>
    <thead>
      <tr>
        <th>Date</th>
        <th class="right"># Recs</th>
        <th class="right">Data</th>
        <th class="right">Max gap</th>
        <th>Coverage</th>
        <th>Spans</th>
        <th>Status</th>
      </tr>
    </thead>
    <tbody>
"""
    for r in reversed(rows):
        cls = f"row-{r['status']}"
        cov = int(round(r["coverage_pct"]))
        seg_divs = "".join(
            f'<div class="seg" style="left:{left:.3f}%;width:{width:.3f}%"></div>'
            for left, width in r["segments"]
        )
        data_str = fmt_bytes(r["bytes"])
        html_out += f"""      <tr class="{cls}">
        <td class="nowrap">{r['date']}</td>
        <td class="right">{r['count']}</td>
        <td class="right">{data_str}</td>
        <td class="right">{hhmm(r['max_gap'])}</td>
        <td>
          <div class="bar" title="{cov}% covered">{seg_divs}</div>
          <div class="cover-label">{cov}%</div>
        </td>
        <td><div class="muted">{html.escape(r['details'])}</div></td>
        <td><span class="status-dot {r['status']}"></span>{r['status']}</td>
      </tr>
"""
    html_out += """    </tbody>
  </table>
  <p class="muted">Legend: <span class="status-dot green"></span> green = nearly continuous • <span class="status-dot yellow"></span> yellow = gaps or low coverage • <span class="status-dot red"></span> red = no recordings</p>
</body>
</html>
"""

    out_path = Path(args.out).resolve()
    out_path.parent.mkdir(parents=True, exist_ok=True)
    tmp = out_path.with_suffix(out_path.suffix + ".tmp")
    with open(tmp, "w", encoding="utf-8") as f:
        f.write(html_out)
    os.replace(tmp, out_path)

def main():
    p = argparse.ArgumentParser(description="Generate recordings status HTML")
    p.add_argument("--dir", default=str(Path.home() / "recordings"),
                   help="Directory containing .opus files (default: ~/recordings)")
    p.add_argument("--out", default="/var/www/html/index.html",
                   help="HTML output path (default: /var/www/html/index.html)")
    # Defaults tuned so cron needs no flags:
    p.add_argument("--days", type=int, default=365, help="How many days to show (default: 365)")
    p.add_argument("--gap-yellow-min", type=int, default=120, help="Gap (minutes) that marks a day as yellow (default: 120)")
    p.add_argument("--gap-green-min", type=int, default=15, help="Max gap (minutes) to still be green (default: 15)")
    p.add_argument("--read-durations", action="store_true",
                   help="Use mutagen to read exact file durations (optional; default off).")
    p.add_argument("--green-cover-pct", type=float, default=90.0,
                   help="Min coverage%% for green (default: 90)")
    p.add_argument("--yellow-cover-pct", type=float, default=10.0,
                   help="Coverage%% below which day is yellow (default: 10)")
    # Duration policy: default is NO inference, assume 2h
    p.add_argument("--assume-duration-sec", type=int, default=7200,
                   help="Assumed duration (s) when real duration is unavailable. Default: 7200 (2h).")
    p.add_argument("--infer", action="store_true",
                   help="Enable inferring duration from median start deltas (disabled by default).")
    p.add_argument("--infer-min-sec", type=int, default=60,
                   help="Ignore deltas smaller than this when inferring (default: 60s)")
    p.add_argument("--infer-max-sec", type=int, default=6*3600,
                   help="Ignore deltas larger than this when inferring (default: 6h)")
    p.add_argument("--infer-default-sec", type=int, default=20*60,
                   help="Fallback segment length if nothing else is available (default: 1200s)")
    p.add_argument("--guard-gap-sec", type=int, default=0,
                   help="Don’t let a segment extend within this many seconds of next start (default: 0)")
    p.add_argument("--title", default="Portable Recorder Status", help="Page title")
    args = p.parse_args()
    build(args)

if __name__ == "__main__":
    main()