Davide Scaini
94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
"""Pre-generate OG track images for all activities.
|
||
|
||
Writes 400×400 PNGs to {www_root}/og-image/{user}/{activity_id}.png.
|
||
Skips activities that already have an up-to-date image (mtime check).
|
||
Safe to run repeatedly — only processes new/changed activities.
|
||
|
||
Usage:
|
||
uv run scripts/generate_og_images.py [--data-dir /var/bincio/data] [--www-root /var/www/activity]
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
|
||
def generate_all(data_dir: Path, www_root: Path) -> None:
|
||
out_root = www_root / "og-image"
|
||
out_root.mkdir(parents=True, exist_ok=True)
|
||
|
||
from bincio.render.ogimage import generate
|
||
|
||
total = generated = skipped = errors = 0
|
||
|
||
users = sorted(
|
||
d.name for d in data_dir.iterdir()
|
||
if d.is_dir() and not d.name.startswith("_") and d.name != "segments"
|
||
)
|
||
|
||
for handle in users:
|
||
user_dir = data_dir / handle
|
||
acts_dir = user_dir / "activities"
|
||
img_dir = out_root / handle
|
||
if not acts_dir.exists():
|
||
continue
|
||
img_dir.mkdir(exist_ok=True)
|
||
u_gen = u_skip = u_err = 0
|
||
|
||
for ts_path in sorted(acts_dir.glob("*.timeseries.json")):
|
||
activity_id = ts_path.name.replace(".timeseries.json", "")
|
||
out_path = img_dir / f"{activity_id}.png"
|
||
total += 1
|
||
|
||
# Skip if image is newer than timeseries
|
||
if out_path.exists() and out_path.stat().st_mtime >= ts_path.stat().st_mtime:
|
||
skipped += 1
|
||
u_skip += 1
|
||
continue
|
||
|
||
try:
|
||
ts = json.loads(ts_path.read_text(encoding="utf-8"))
|
||
lat_arr = ts.get("lat") or []
|
||
lon_arr = ts.get("lon") or []
|
||
ele_arr = ts.get("elevation_m") or []
|
||
png = generate(lat_arr, lon_arr, ele_arr)
|
||
out_path.write_bytes(png)
|
||
generated += 1
|
||
u_gen += 1
|
||
time.sleep(0.05)
|
||
except Exception as exc:
|
||
errors += 1
|
||
u_err += 1
|
||
print(f" ERROR {handle}/{activity_id}: {exc}", file=sys.stderr)
|
||
|
||
if u_gen or u_err:
|
||
print(f"{handle:<25} generated={u_gen:4d} skipped={u_skip:4d} errors={u_err}")
|
||
else:
|
||
print(f"{handle:<25} skipped={u_skip:4d} (all up to date)")
|
||
|
||
print(f"\nDone — {generated} generated, {skipped} skipped, {errors} errors (total {total})")
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser(description="Pre-generate OG track images")
|
||
ap.add_argument("--data-dir", default="/var/bincio/data", type=Path)
|
||
ap.add_argument("--www-root", default="/var/www/activity", type=Path)
|
||
args = ap.parse_args()
|
||
|
||
if not args.data_dir.exists():
|
||
print(f"ERROR: data dir not found: {args.data_dir}", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
print(f"data-dir : {args.data_dir}")
|
||
print(f"www-root : {args.www_root}")
|
||
print(f"output : {args.www_root}/og-image/\n")
|
||
generate_all(args.data_dir, args.www_root)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|