homelab-brain/infra/hermes/scripts/anwesenheit_sim.py

#!/usr/bin/env python3
"""Plant täglich Sonnenuntergang-EIN + Zufalls-AUS (23:30-02:00 MESZ/MEZ) für Shelly Plug.
   Meldet beim Lauf auch, ob der vorherige Tag erfolgreich war."""

import random
import re
import subprocess
import sys
from datetime import datetime, date, timedelta, timezone
from zoneinfo import ZoneInfo
from astral.sun import sun
from astral import LocationInfo

SHELLY_IP = "192.168.178.50"
LAT, LON = 51.67, 12.33         # Muldenstein
TZ = ZoneInfo("Europe/Berlin")  # Lokalzeit (MESZ/MEZ)
UTC = ZoneInfo("UTC")
LOG = "/var/log/anwesenheit_sim.log"

# ---------------------------------------------------------------------------
def log_status_yesterday():
    """Liest Log von gestern und gibt Zusammenfassung."""
    yesterday = date.today() - timedelta(days=1)
    label = yesterday.strftime("%a %d.%m.%Y")
    try:
        with open(LOG) as f:
            lines = f.readlines()
    except FileNotFoundError:
        return f"  📭 {label}: Kein Log (noch nie gelaufen?)"

    hits = []
    for line in lines:
        line = line.strip()
        # Zeilenformat: "Sat Aug  1 19:02:00 UTC 2026: on -> 200"
        m = re.search(r"(\w{3} \w{3} \d+ \d+:\d+:\d+ \w+ \d{4}):\s+(on|off)\s+->\s+(\d+)", line)
        if not m:
            continue
        ts_str = m.group(1)
        action = m.group(2)
        code = m.group(3)
        # Parse datetime grob
        try:
            ts = datetime.strptime(ts_str, "%a %b %d %H:%M:%S %Z %Y")
        except ValueError:
            # fallback: Datum extrahieren
            continue
        if ts.date() == yesterday:
            hits.append((ts, action, code))

    if not hits:
        return f"  ⏳ {label}: Keine Einträge im Log"

    on_ok = any(a == "on" and c == "200" for _, a, c in hits)
    off_ok = any(a == "off" and c == "200" for _, a, c in hits)
    parts = []
    for ts, action, code in sorted(hits):
        lok = ts.astimezone(TZ).strftime("%H:%M")
        icon = "✅" if code == "200" else f"❌({code})"
        label_a = "EIN " if action == "on" else "AUS"
        parts.append(f"{icon} {label_a}{lok}")
    status = "✅ Erfolgreich" if (on_ok and off_ok) else "⚠️ Teilweise"
    return f"  {label}: {' · '.join(parts)}  — {status}"


def shelly_cmd(action: str, when_utc: str):
    """Plant Shelly-Schaltvorgang via at (when_utc = 'HH:MM today|tomorrow')."""
    cmds = {
        "on": f"curl -s -o /dev/null -w '%{{http_code}}' 'http://{SHELLY_IP}/relay/0?turn=on'",
        "off": f"curl -s -o /dev/null -w '%{{http_code}}' 'http://{SHELLY_IP}/relay/0?turn=off'",
    }
    script = f"""#!/bin/sh
echo "$(date): {action} -> $( {cmds[action]} )" >> {LOG}
"""
    proc = subprocess.run(["at", when_utc], input=script,
                          capture_output=True, text=True, timeout=10)
    if proc.returncode != 0:
        print(f"FEHLER at '{when_utc}': {proc.stderr.strip()}", file=sys.stderr)
        return False
    print(f"  ✓ Geplant: {action} um {when_utc} UTC")
    return True


def plan():
    now_local = datetime.now(TZ)
    today_local = now_local.date()

    # --- Status gestern ---
    print("📊 Rückblick:")
    print(log_status_yesterday())
    print()

    # --- Sonnenuntergang (UTC!) ---
    loc = LocationInfo("Muldenstein", "Germany", "Europe/Berlin", LAT, LON)
    sun_utc = sun(loc.observer, date=today_local)["sunset"]
    sunset_utc = sun_utc.replace(tzinfo=UTC)
    sunset_local = sunset_utc.astimezone(TZ)

    # Einschalten: Sonnenuntergang + 3 Minuten (UTC)
    on_time = sunset_utc + timedelta(minutes=3)
    on_at = f"{on_time.hour:02d}:{on_time.minute:02d} today"

    # --- Zufällige Ausschaltzeit (23:30 - 02:00 Lokalzeit) ---
    start_local = 23 * 60 + 30   # 23:30
    end_local   = 2 * 60         # 02:00 (nächster Tag)

    # Liste aller möglichen Lokalzeit-Minuten
    if end_local > start_local:   # beide am selben Tag
        choices = list(range(start_local, end_local + 1))
    else:                         # end_local liegt am nächsten Tag
        today_mins = list(range(start_local, 24 * 60))
        tomorrow_mins = list(range(0, end_local + 1))
        choices = today_mins + [m + 24 * 60 for m in tomorrow_mins]

    chosen = random.choice(choices)

    # chosen in Lokaldatetime umrechnen
    off_local = datetime(today_local.year, today_local.month, today_local.day,
                         tzinfo=TZ) + timedelta(minutes=chosen)
    off_utc = off_local.astimezone(UTC)

    # at-Zeit in UTC
    today_utc = datetime(today_local.year, today_local.month, today_local.day, tzinfo=UTC)
    tomorrow_utc = today_utc + timedelta(days=1)

    if off_utc.date() == today_utc.date():
        off_at = f"{off_utc.hour:02d}:{off_utc.minute:02d} today"
    else:
        off_at = f"{off_utc.hour:02d}:{off_utc.minute:02d} tomorrow"

    # --- Bestehende at-Jobs räumen ---
    result = subprocess.run(["atq"], capture_output=True, text=True, timeout=5)
    for line in result.stdout.strip().splitlines():
        if line:
            job_id = line.split("\t")[0]
            subprocess.run(["atrm", job_id], timeout=5)

    # --- Ausgabe & Planung ---
    print(f"Plane Anwesenheitssimulation für {today_local}:")
    print(f"  Sonnenuntergang: {sunset_local:%H:%M} Lokalzeit")
    print(f"  EIN um {on_time.hour:02d}:{on_time.minute:02d} UTC  →  {on_time.astimezone(TZ):%H:%M} Lokalzeit")
    print(f"  AUS um {off_utc:%H:%M} UTC  →  {off_local:%H:%M} Lokalzeit")
    print()

    ok_on = shelly_cmd("on", on_at)
    ok_off = shelly_cmd("off", off_at)

    return 0 if (ok_on and ok_off) else 1


if __name__ == "__main__":
    sys.exit(plan())