#!/usr/bin/env python3 """Plant täglich Sonnenuntergang-EIN + Zufalls-AUS (23:30-02:00 MESZ/MEZ) für mehrere Shelly Plugs. Jeder Plug bekommt eigene Zufallszeiten (kein Synchron-Schalten, wirkt realistischer). 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 SHELLIES = [ {"ip": "192.168.178.50", "label": "plug_50"}, # Anwesenheit simulieren {"ip": "192.168.178.58", "label": "plug_58"}, # Anwesenheit simulieren 2 ] 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 pro Plug.""" 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: dict[str, list] = {s["label"]: [] for s in SHELLIES} for line in lines: line = line.strip() # Zeilenformat: "Sat Aug 1 19:02:00 UTC 2026: plug_50 on -> 200" m = re.search(r"(\w{3} \w{3} \d+ \d+:\d+:\d+ \w+ \d{4}):\s+(\S+)\s+(on|off)\s+->\s+(\d+)", line) if not m: continue ts_str, plug_label, action, code = m.groups() try: ts = datetime.strptime(ts_str, "%a %b %d %H:%M:%S %Z %Y") except ValueError: continue if ts.date() == yesterday and plug_label in hits: hits[plug_label].append((ts, action, code)) lines_out = [f" {label}:"] for s in SHELLIES: plug_hits = hits.get(s["label"], []) if not plug_hits: lines_out.append(f" {s['label']}: ⏳ keine Einträge") continue on_ok = any(a == "on" and c == "200" for _, a, c in plug_hits) off_ok = any(a == "off" and c == "200" for _, a, c in plug_hits) parts = [] for ts, action, code in sorted(plug_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" lines_out.append(f" {s['label']}: {' · '.join(parts)} — {status}") return "\n".join(lines_out) def shelly_cmd(ip: str, plug_label: str, 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://{ip}/relay/0?turn=on'", "off": f"curl -s -o /dev/null -w '%{{http_code}}' 'http://{ip}/relay/0?turn=off'", } script = f"""#!/bin/sh echo "$(date): {plug_label} {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}' ({plug_label}): {proc.stderr.strip()}", file=sys.stderr) return False print(f" ✓ Geplant: {plug_label} {action} um {when_utc} UTC") return True def _pick_off_time(today_local: date) -> tuple[datetime, str]: """Zufällige Ausschaltzeit 23:30-02:00 Lokalzeit -> (off_utc, at-Ausdruck).""" start_local = 23 * 60 + 30 # 23:30 end_local = 2 * 60 # 02:00 (nächster 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) off_local = datetime(today_local.year, today_local.month, today_local.day, tzinfo=TZ) + timedelta(minutes=chosen) off_utc = off_local.astimezone(UTC) today_utc = datetime(today_local.year, today_local.month, today_local.day, tzinfo=UTC) 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" return off_utc, off_at 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) # --- Bestehende at-Jobs räumen (alle, egal welcher Plug) --- 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) print(f"Plane Anwesenheitssimulation für {today_local}:") print(f" Sonnenuntergang: {sunset_local:%H:%M} Lokalzeit") print() all_ok = True for s in SHELLIES: ip, plug_label = s["ip"], s["label"] # Einschalten: Sonnenuntergang + individueller Zufalls-Jitter 0-15min (nicht synchron) on_time = sunset_utc + timedelta(minutes=random.randint(0, 15)) on_at = f"{on_time.hour:02d}:{on_time.minute:02d} today" # Ausschalten: eigene Zufallszeit 23:30-02:00 Lokalzeit off_utc, off_at = _pick_off_time(today_local) print(f" {plug_label}:") 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_utc.astimezone(TZ):%H:%M} Lokalzeit") ok_on = shelly_cmd(ip, plug_label, "on", on_at) ok_off = shelly_cmd(ip, plug_label, "off", off_at) all_ok = all_ok and ok_on and ok_off print() return 0 if all_ok else 1 if __name__ == "__main__": sys.exit(plan())