From 841a52c0d2a25b9c6cdeefb7a6dbafb8a67db6be Mon Sep 17 00:00:00 2001 From: Homelab Cursor Date: Sat, 1 Aug 2026 19:13:58 +0200 Subject: [PATCH] hermes: Anwesenheitssimulation Skript + Cronjob-Snapshot sichern (CT151) --- .../scripts/anwesenheit_sim.cronjob.json | 43 +++++ infra/hermes/scripts/anwesenheit_sim.py | 151 ++++++++++++++++++ 2 files changed, 194 insertions(+) create mode 100644 infra/hermes/scripts/anwesenheit_sim.cronjob.json create mode 100644 infra/hermes/scripts/anwesenheit_sim.py diff --git a/infra/hermes/scripts/anwesenheit_sim.cronjob.json b/infra/hermes/scripts/anwesenheit_sim.cronjob.json new file mode 100644 index 000000000..b7fa5cfb1 --- /dev/null +++ b/infra/hermes/scripts/anwesenheit_sim.cronjob.json @@ -0,0 +1,43 @@ +{ + "id": "3cd6f07721ec", + "name": "anwesenheit-simulation", + "prompt": "", + "skills": [], + "skill": null, + "model": null, + "provider": null, + "base_url": null, + "script": "anwesenheit_sim.py", + "no_agent": true, + "context_from": null, + "schedule": { + "kind": "cron", + "expr": "0 3 * * *", + "display": "0 3 * * *" + }, + "schedule_display": "0 3 * * *", + "repeat": { + "times": null, + "completed": 0 + }, + "enabled": true, + "state": "scheduled", + "paused_at": null, + "paused_reason": null, + "created_at": "2026-08-01T17:08:10.570084+00:00", + "next_run_at": "2026-08-02T03:00:00+00:00", + "last_run_at": null, + "last_status": null, + "last_error": null, + "last_delivery_error": null, + "deliver": "origin", + "origin": { + "platform": "telegram", + "chat_id": "674951792", + "chat_name": "Michael", + "thread_id": null + }, + "enabled_toolsets": null, + "workdir": null, + "profile": null +} diff --git a/infra/hermes/scripts/anwesenheit_sim.py b/infra/hermes/scripts/anwesenheit_sim.py new file mode 100644 index 000000000..9356fde32 --- /dev/null +++ b/infra/hermes/scripts/anwesenheit_sim.py @@ -0,0 +1,151 @@ +#!/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()) \ No newline at end of file