diff --git a/infra/hermes/scripts/anwesenheit_sim.py b/infra/hermes/scripts/anwesenheit_sim.py index 9356fde32..e46553799 100644 --- a/infra/hermes/scripts/anwesenheit_sim.py +++ b/infra/hermes/scripts/anwesenheit_sim.py @@ -1,5 +1,6 @@ #!/usr/bin/env python3 -"""Plant täglich Sonnenuntergang-EIN + Zufalls-AUS (23:30-02:00 MESZ/MEZ) für Shelly Plug. +"""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 @@ -11,7 +12,10 @@ from zoneinfo import ZoneInfo from astral.sun import sun from astral import LocationInfo -SHELLY_IP = "192.168.178.50" +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") @@ -19,7 +23,7 @@ LOG = "/var/log/anwesenheit_sim.log" # --------------------------------------------------------------------------- def log_status_yesterday(): - """Liest Log von gestern und gibt Zusammenfassung.""" + """Liest Log von gestern und gibt Zusammenfassung pro Plug.""" yesterday = date.today() - timedelta(days=1) label = yesterday.strftime("%a %d.%m.%Y") try: @@ -28,58 +32,79 @@ def log_status_yesterday(): except FileNotFoundError: return f" 📭 {label}: Kein Log (noch nie gelaufen?)" - hits = [] + 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: on -> 200" - m = re.search(r"(\w{3} \w{3} \d+ \d+:\d+:\d+ \w+ \d{4}):\s+(on|off)\s+->\s+(\d+)", line) + # 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 = m.group(1) - action = m.group(2) - code = m.group(3) - # Parse datetime grob + ts_str, plug_label, action, code = m.groups() 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 ts.date() == yesterday and plug_label in hits: + hits[plug_label].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}" + 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(action: str, when_utc: str): +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://{SHELLY_IP}/relay/0?turn=on'", - "off": f"curl -s -o /dev/null -w '%{{http_code}}' 'http://{SHELLY_IP}/relay/0?turn=off'", + "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): {action} -> $( {cmds[action]} )" >> {LOG} +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}': {proc.stderr.strip()}", file=sys.stderr) + print(f"FEHLER at '{when_utc}' ({plug_label}): {proc.stderr.strip()}", file=sys.stderr) return False - print(f" ✓ Geplant: {action} um {when_utc} UTC") + 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() @@ -95,56 +120,38 @@ def plan(): 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 --- + # --- 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) - # --- 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) + all_ok = True + for s in SHELLIES: + ip, plug_label = s["ip"], s["label"] - return 0 if (ok_on and ok_off) else 1 + # 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__": diff --git a/infra/hermes/shelly_mcp.py b/infra/hermes/shelly_mcp.py index ae64de18c..edccdbfce 100755 --- a/infra/hermes/shelly_mcp.py +++ b/infra/hermes/shelly_mcp.py @@ -101,6 +101,25 @@ DEVICES: dict[str, dict] = { "presence", ], }, + "plug_58": { + "ip": "192.168.178.58", + "name": "Anwesenheit simulieren 2", + "gen": 3, + "kind": "switch", + "aliases": [ + "plug_58", + "shelly_58", + "plug58", + "steckdose_58", + "anwesenheit_2", + "anwesenheit_simulieren_2", + "anwesenheitssimulation_2", + "anwesenheit2", + "praesenz_2", + "praesenz2", + "presence_2", + ], + }, } # flatten alias lookup