make_timelapse.py

#!/usr/bin/env python3
"""
make_timelapse.py — Daily Sun Journey Timelapse
Stitches patio cam sunrise frames + front yard cam sunset frames
into a single MP4 showing the sun's full arc over the house.

Optionally overlays Powerwall solar production data as a live meter.

Usage:
    python3 make_timelapse.py [YYYY-MM-DD]   # defaults to today
    python3 make_timelapse.py --yesterday

Output:
    sky-watcher/timelapses/YYYY-MM-DD.mp4

Author: Sam (github.com/jasonacox-sam)
"""

import os
import sys
import json
import subprocess
import tempfile
from datetime import date, datetime, timedelta
from pathlib import Path
from zoneinfo import ZoneInfo

# ── Config ─────────────────────────────────────────────────────────────────
BASE_DIR      = Path(__file__).parent
PATIO_DIR     = BASE_DIR / "images"
FRONTYARD_DIR = BASE_DIR / "images" / "frontyard"
OUTPUT_DIR    = BASE_DIR / "timelapses"
TIMEZONE      = "America/Los_Angeles"
FRAME_DURATION = 0.25        # seconds per frame (0.25 = 4fps, smooth timelapse)
NOON_HOUR      = 13          # split: before this = sunrise set, after = sunset set

# Powerwall integration (optional — set to None to skip)
POWERWALL_IP   = "10.0.1.2"  # your Powerwall gateway IP


# ── Powerwall solar data ────────────────────────────────────────────────────

def get_solar_peak(target_date: date) -> float:
    """
    Pull today's peak solar production from Powerwall.
    Returns peak kW, or 0.0 if unavailable.
    """
    try:
        import pypowerwall
        pw = pypowerwall.Powerwall(POWERWALL_IP, "", "", "", gw_pwd="")
        solar = pw.solar()
        if solar:
            return round(float(solar) / 1000, 2)
    except Exception:
        pass
    return 0.0


def get_solar_timeseries(target_date: date) -> list[tuple[str, float]]:
    """
    Returns list of (HH-MM, kW) tuples for the target date if available.
    Falls back to empty list.
    """
    try:
        import pypowerwall
        pw = pypowerwall.Powerwall(POWERWALL_IP, "", "", "", gw_pwd="")
        data = pw.get_history(period="day")
        if not data:
            return []
        result = []
        for entry in data:
            ts = entry.get("timestamp", "")
            solar_w = entry.get("solar_power", 0)
            try:
                dt = datetime.fromisoformat(ts).astimezone(ZoneInfo(TIMEZONE))
                if dt.date() == target_date:
                    result.append((dt.strftime("%H-%M"), round(solar_w / 1000, 2)))
            except Exception:
                pass
        return result
    except Exception:
        return []


# ── Frame collection ────────────────────────────────────────────────────────

def collect_frames(target_date: date) -> tuple[list[Path], list[Path]]:
    """
    Return (sunrise_frames, sunset_frames) for the target date.
    Sunrise = patio cam images with hour < NOON_HOUR
    Sunset  = frontyard cam images with hour >= NOON_HOUR
    Files named HH-MM.jpg; use mtime date to filter to target_date.
    """
    tz = ZoneInfo(TIMEZONE)

    def date_filter(p: Path) -> bool:
        mtime = datetime.fromtimestamp(p.stat().st_mtime, tz=tz)
        return mtime.date() == target_date

    def hour_of(p: Path) -> int:
        try:
            return int(p.stem.split("-")[0])
        except (ValueError, IndexError):
            return 0

    sunrise_frames = sorted(
        [p for p in PATIO_DIR.glob("*.jpg")
         if date_filter(p) and hour_of(p) < NOON_HOUR],
        key=lambda p: p.stem
    )
    sunset_frames = sorted(
        [p for p in FRONTYARD_DIR.glob("*.jpg")
         if date_filter(p) and hour_of(p) >= NOON_HOUR],
        key=lambda p: p.stem
    )

    # Fallback: if frontyard has nothing, use patio sunset frames too
    if not sunset_frames:
        sunset_frames = sorted(
            [p for p in PATIO_DIR.glob("*.jpg")
             if date_filter(p) and hour_of(p) >= NOON_HOUR],
            key=lambda p: p.stem
        )

    return sunrise_frames, sunset_frames


# ── ffmpeg timelapse ────────────────────────────────────────────────────────

def stem_to_label(stem: str, cam_name: str) -> str:
    """Convert '06-24' → '6:24 AM  📷 Patio'"""
    try:
        h, m = int(stem[:2]), int(stem[3:5])
        period = "AM" if h < 12 else "PM"
        display_h = h if h <= 12 else h - 12
        display_h = 12 if display_h == 0 else display_h
        return f"{display_h}:{m:02d} {period}  {cam_name}"
    except Exception:
        return stem


def build_timelapse(
    sunrise_frames: list[Path],
    sunset_frames: list[Path],
    target_date: date,
    solar_data: list[tuple[str, float]],
    output_path: Path,
) -> bool:
    """
    Stitch frames into MP4 using ffmpeg concat demuxer.
    Adds timestamp overlay and optional solar meter.
    """
    if not sunrise_frames and not sunset_frames:
        print("No frames found — nothing to stitch.")
        return False

    solar_lookup = dict(solar_data)
    max_solar = max((v for _, v in solar_data), default=0) or 10.0

    # Build concat list with per-frame metadata
    all_frames: list[tuple[Path, str, str]] = []
    for p in sunrise_frames:
        label = stem_to_label(p.stem, "🌅 Patio")
        all_frames.append((p, label, p.stem))
    for p in sunset_frames:
        cam = "🌇 Front Yard" if FRONTYARD_DIR in p.parents else "🌇 Patio"
        label = stem_to_label(p.stem, cam)
        all_frames.append((p, label, p.stem))

    if not all_frames:
        print("No frames to stitch.")
        return False

    date_str = target_date.strftime("%B %-d, %Y")

    with tempfile.TemporaryDirectory() as tmpdir:
        tmpdir = Path(tmpdir)

        # Write concat file
        concat_path = tmpdir / "concat.txt"
        with open(concat_path, "w") as f:
            for frame_path, _, _ in all_frames:
                f.write(f"file '{frame_path.resolve()}'\n")
                f.write(f"duration {FRAME_DURATION}\n")
            # ffmpeg needs a final file line without duration
            f.write(f"file '{all_frames[-1][0].resolve()}'\n")

        # Build drawtext filter chain
        # Line 1: date
        # Line 2: time + camera label
        # Line 3 (optional): solar meter if data available
        filter_parts = []

        # Date header
        filter_parts.append(
            f"drawtext=text='{date_str}  —  Valencia, CA':"
            f"fontsize=28:fontcolor=white:x=20:y=20:"
            f"shadowcolor=black:shadowx=2:shadowy=2"
        )

        # We'll use a single static drawtext per frame via concat + individual frames
        # For dynamic text, use ffmpeg's overlay with a script — but for simplicity,
        # burn the label per frame using a two-pass approach.

        # Simpler: just overlay date + static labels via the concat + drawtext
        # For per-frame labels we need individual frame processing.
        # Let's do it right: process each frame individually first.

        labeled_dir = tmpdir / "labeled"
        labeled_dir.mkdir()

        print(f"Processing {len(all_frames)} frames...")
        for i, (frame_path, label, stem) in enumerate(all_frames):
            out_frame = labeled_dir / f"{i:04d}.jpg"
            solar_kw = solar_lookup.get(stem, None)

            # Build filter
            filters = [
                # Semi-transparent black bar at bottom
                "drawbox=x=0:y=ih-80:w=iw:h=80:color=black@0.55:t=fill",
                # Date top-left
                f"drawtext=text='{date_str}':fontsize=22:fontcolor=white@0.9:"
                f"x=20:y=20:shadowcolor=black@0.8:shadowx=1:shadowy=1",
                # Time + camera bottom-left
                f"drawtext=text='{label}':fontsize=30:fontcolor=white:"
                f"x=20:y=ih-60:shadowcolor=black:shadowx=2:shadowy=2",
            ]

            # Solar meter bottom-right
            if solar_kw is not None and solar_kw > 0:
                bar_pct = min(solar_kw / max_solar, 1.0)
                bar_w = int(200 * bar_pct)
                filters += [
                    # Label
                    f"drawtext=text='☀ {solar_kw:.1f} kW':fontsize=22:fontcolor=yellow:"
                    f"x=iw-220:y=ih-60:shadowcolor=black:shadowx=1:shadowy=1",
                    # Bar background
                    f"drawbox=x=iw-222:y=ih-28:w=204:h=16:color=white@0.3:t=fill",
                    # Bar fill
                    f"drawbox=x=iw-222:y=ih-28:w={bar_w}:h=16:color=yellow@0.8:t=fill",
                ]

            vf = ",".join(filters)
            cmd = [
                "ffmpeg", "-y", "-loglevel", "error",
                "-i", str(frame_path),
                "-vf", vf,
                "-q:v", "3",
                str(out_frame)
            ]
            result = subprocess.run(cmd, capture_output=True)
            if result.returncode != 0:
                # Fallback: copy frame without overlay
                import shutil
                shutil.copy(frame_path, out_frame)

        # Build new concat from labeled frames
        labeled_concat = tmpdir / "labeled_concat.txt"
        labeled_frames = sorted(labeled_dir.glob("*.jpg"))
        with open(labeled_concat, "w") as f:
            for lf in labeled_frames:
                f.write(f"file '{lf.resolve()}'\n")
                f.write(f"duration {FRAME_DURATION}\n")
            f.write(f"file '{labeled_frames[-1].resolve()}'\n")

        # Final ffmpeg encode
        output_path.parent.mkdir(parents=True, exist_ok=True)
        cmd = [
            "ffmpeg", "-y", "-loglevel", "warning",
            "-f", "concat", "-safe", "0",
            "-i", str(labeled_concat),
            "-vf", "scale=1920:-2:flags=lanczos,fps=10",
            "-c:v", "libx264", "-crf", "23", "-preset", "fast",
            "-pix_fmt", "yuv420p",
            str(output_path)
        ]
        print(f"Encoding → {output_path}")
        result = subprocess.run(cmd, capture_output=True)
        if result.returncode != 0:
            print(f"ffmpeg error: {result.stderr.decode()[-500:]}")
            return False

        size_mb = output_path.stat().st_size / 1_048_576
        print(f"✅ Done: {output_path} ({size_mb:.1f} MB, {len(all_frames)} frames)")
        return True


# ── Main ────────────────────────────────────────────────────────────────────

def main():
    tz = ZoneInfo(TIMEZONE)

    # Parse date argument
    if "--yesterday" in sys.argv:
        target_date = (datetime.now(tz) - timedelta(days=1)).date()
    elif len(sys.argv) > 1 and not sys.argv[1].startswith("-"):
        target_date = date.fromisoformat(sys.argv[1])
    else:
        target_date = datetime.now(tz).date()

    print(f"\n{'='*60}")
    print(f"  Sun Journey Timelapse — {target_date}")
    print(f"{'='*60}")

    sunrise_frames, sunset_frames = collect_frames(target_date)
    print(f"  Sunrise frames (Patio)     : {len(sunrise_frames)}")
    print(f"  Sunset frames  (Front Yard): {len(sunset_frames)}")

    if not sunrise_frames and not sunset_frames:
        print("\nNo frames found for this date. Run during or after a capture window.")
        sys.exit(1)

    # Optional solar data
    print("  Fetching Powerwall solar data...")
    solar_data = get_solar_timeseries(target_date)
    if solar_data:
        peak = max(v for _, v in solar_data)
        print(f"  Solar data: {len(solar_data)} points, peak {peak:.1f} kW")
    else:
        print("  Solar data: unavailable (continuing without meter)")

    output_path = OUTPUT_DIR / f"{target_date}.mp4"
    success = build_timelapse(sunrise_frames, sunset_frames, target_date, solar_data, output_path)
    sys.exit(0 if success else 1)


if __name__ == "__main__":
    main()