352 lines
13 KiB
Python
Executable file
352 lines
13 KiB
Python
Executable file
#!/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()
|