infra: full IaC state after hf-pdocker-01 rebuild (2026-03-08)
Replaces initial Nomad/Consul/Terraform skeleton with current Docker-based infrastructure across all three nodes (hf-pdocker-01, hf-pdocker-02, bart). Key changes: - CLAUDE.md: updated service distribution reflecting hf-pdocker-01 restored - docs/plans/2026-03-08-hf-pdocker-01-rebuild.md: new rebuild plan doc - compose/bart/: ai-agents, guacamole, matrix, media, openvas, smarthome, watchtower - compose/hf-pdocker-02/: traefik, monitoring, adguard, gitea, semaphore, wikijs, unifi, vaultwarden, actual-budget, mcpo, netclaw, trivy, watchtower, n8n - compose/hf-pdocker-01/: frigate, frigate-notify, farm-agent - stacks/: all Swarm stack definitions (monitoring, traefik, smarthome, matrix, guacamole, media, ai-agents, frigate, gitea, semaphore, wikijs, unifi, etc.) - .gitignore: exclude .env credential files and acme.json certs - .env.example: placeholder credential files for all services - .claude/agents/: gitea-backup and other infrastructure agent configs - scripts/: frigate media cleanup and NAS sync scripts Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,259 @@
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import signal
|
||||
import sys
|
||||
import time
|
||||
from collections import defaultdict
|
||||
from http.server import HTTPServer, BaseHTTPRequestHandler
|
||||
from threading import Thread
|
||||
|
||||
import aiohttp
|
||||
import anthropic
|
||||
import paho.mqtt.client as mqtt
|
||||
|
||||
from vision import analyze_snapshot
|
||||
from matrix_client import MatrixClient
|
||||
|
||||
# --- Configuration ---
|
||||
MQTT_HOST = os.environ.get("MQTT_HOST", "192.168.86.192")
|
||||
MQTT_PORT = int(os.environ.get("MQTT_PORT", "1883"))
|
||||
FRIGATE_URL = os.environ.get("FRIGATE_URL", "http://192.168.86.192:5000")
|
||||
MATRIX_HOMESERVER = os.environ.get("MATRIX_HOMESERVER", "http://192.168.86.192:8008")
|
||||
MATRIX_ACCESS_TOKEN = os.environ["MATRIX_ACCESS_TOKEN"]
|
||||
MATRIX_ROOM_ID = os.environ["MATRIX_ROOM_ID"]
|
||||
HEALTH_PORT = int(os.environ.get("HEALTH_PORT", "5199"))
|
||||
|
||||
ALLOWED_LABELS = {"person", "car", "dog", "cat", "bird"}
|
||||
MIN_SCORE = 0.7
|
||||
RATE_LIMIT_SECONDS = 60
|
||||
DEDUP_WINDOW_SECONDS = 30
|
||||
|
||||
# --- Logging ---
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='{"time":"%(asctime)s","level":"%(levelname)s","module":"%(module)s","message":"%(message)s"}',
|
||||
stream=sys.stdout,
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# --- State ---
|
||||
recent_events: dict[str, float] = {}
|
||||
last_notify_per_camera: defaultdict[str, float] = defaultdict(float)
|
||||
events_processed = 0
|
||||
last_event_time: str | None = None
|
||||
mqtt_connected = False
|
||||
|
||||
|
||||
# --- Health endpoint ---
|
||||
class HealthHandler(BaseHTTPRequestHandler):
|
||||
protocol_version = "HTTP/1.1"
|
||||
|
||||
def do_GET(self):
|
||||
if self.path == "/health":
|
||||
body = json.dumps(
|
||||
{
|
||||
"status": "ok",
|
||||
"mqtt_connected": mqtt_connected,
|
||||
"events_processed": events_processed,
|
||||
"last_event": last_event_time,
|
||||
}
|
||||
)
|
||||
encoded = body.encode()
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "application/json")
|
||||
self.send_header("Content-Length", str(len(encoded)))
|
||||
self.end_headers()
|
||||
self.wfile.write(encoded)
|
||||
else:
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
|
||||
def log_message(self, format, *args):
|
||||
pass # suppress access logs
|
||||
|
||||
|
||||
def start_health_server():
|
||||
server = HTTPServer(("0.0.0.0", HEALTH_PORT), HealthHandler)
|
||||
thread = Thread(target=server.serve_forever, daemon=True)
|
||||
thread.start()
|
||||
logger.info("Health server started on port %d", HEALTH_PORT)
|
||||
|
||||
|
||||
# --- Snapshot fetching ---
|
||||
async def fetch_snapshot(event_id: str) -> bytes | None:
|
||||
url = f"{FRIGATE_URL}/api/events/{event_id}/snapshot.jpg"
|
||||
try:
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(url) as resp:
|
||||
if resp.status == 200:
|
||||
data = await resp.read()
|
||||
logger.info(
|
||||
"Fetched snapshot for %s: %d bytes", event_id, len(data)
|
||||
)
|
||||
return data
|
||||
logger.warning(
|
||||
"Snapshot fetch failed for %s: HTTP %d", event_id, resp.status
|
||||
)
|
||||
except Exception:
|
||||
logger.exception("Snapshot fetch error for %s", event_id)
|
||||
return None
|
||||
|
||||
|
||||
# --- Event processing ---
|
||||
async def process_event(
|
||||
event_data: dict, claude: anthropic.Anthropic, matrix: MatrixClient
|
||||
):
|
||||
global events_processed, last_event_time
|
||||
|
||||
event_type = event_data.get("type")
|
||||
if event_type != "end":
|
||||
return
|
||||
|
||||
after = event_data.get("after", {})
|
||||
event_id = after.get("id", "")
|
||||
camera = after.get("camera", "unknown")
|
||||
label = after.get("label", "")
|
||||
score = after.get("top_score", after.get("score", 0))
|
||||
has_snapshot = after.get("has_snapshot", False)
|
||||
|
||||
# --- Filtering ---
|
||||
if label not in ALLOWED_LABELS:
|
||||
return
|
||||
if score < MIN_SCORE:
|
||||
return
|
||||
if not has_snapshot:
|
||||
return
|
||||
|
||||
# Dedup
|
||||
now = time.time()
|
||||
if event_id in recent_events and now - recent_events[event_id] < DEDUP_WINDOW_SECONDS:
|
||||
return
|
||||
recent_events[event_id] = now
|
||||
|
||||
# Rate limit per camera
|
||||
if now - last_notify_per_camera[camera] < RATE_LIMIT_SECONDS:
|
||||
logger.info("Rate limited: %s on %s", label, camera)
|
||||
return
|
||||
last_notify_per_camera[camera] = now
|
||||
|
||||
# Cleanup old dedup entries
|
||||
cutoff = now - DEDUP_WINDOW_SECONDS * 2
|
||||
for eid in [k for k, v in recent_events.items() if v < cutoff]:
|
||||
del recent_events[eid]
|
||||
|
||||
logger.info(
|
||||
"Processing event: %s on %s (%s, score=%.2f)", event_id, camera, label, score
|
||||
)
|
||||
|
||||
# --- Fetch snapshot ---
|
||||
snapshot = await fetch_snapshot(event_id)
|
||||
|
||||
# --- AI analysis ---
|
||||
description = None
|
||||
if snapshot:
|
||||
description = analyze_snapshot(claude, snapshot)
|
||||
if not description:
|
||||
description = f"{label.capitalize()} detected on {camera}."
|
||||
|
||||
# --- Format message ---
|
||||
score_pct = int(score * 100)
|
||||
ts = time.strftime("%I:%M:%S %p", time.localtime())
|
||||
date_str = time.strftime("%b %d, %Y", time.localtime())
|
||||
frigate_link = f"{FRIGATE_URL}/review?id={event_id}"
|
||||
|
||||
plain = (
|
||||
f"{camera} - {label.capitalize()} - {score_pct}%\n"
|
||||
f"{description}\n\n"
|
||||
f"View in Frigate: {frigate_link}\n"
|
||||
f"{ts} - {date_str}"
|
||||
)
|
||||
html = (
|
||||
f"<b>{camera}</b> · {label.capitalize()} · {score_pct}%<br/>"
|
||||
f"{description}<br/><br/>"
|
||||
f'<a href="{frigate_link}">View in Frigate</a><br/>'
|
||||
f"<small>{ts} · {date_str}</small>"
|
||||
)
|
||||
|
||||
# --- Send to Matrix ---
|
||||
if snapshot:
|
||||
filename = f"{camera}_{event_id}.jpg"
|
||||
mxc_uri = await matrix.upload_image(snapshot, filename)
|
||||
if mxc_uri:
|
||||
await matrix.send_image(mxc_uri, filename, len(snapshot))
|
||||
|
||||
await matrix.send_text(plain, html)
|
||||
|
||||
events_processed += 1
|
||||
last_event_time = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
|
||||
logger.info("Notification sent for %s on %s", label, camera)
|
||||
|
||||
|
||||
# --- MQTT ---
|
||||
def create_mqtt_client(
|
||||
loop: asyncio.AbstractEventLoop,
|
||||
claude: anthropic.Anthropic,
|
||||
matrix: MatrixClient,
|
||||
):
|
||||
global mqtt_connected
|
||||
|
||||
def on_connect(client, userdata, flags, reason_code, properties):
|
||||
global mqtt_connected
|
||||
mqtt_connected = True
|
||||
client.subscribe("frigate/events")
|
||||
logger.info("MQTT connected, subscribed to frigate/events")
|
||||
|
||||
def on_disconnect(client, userdata, flags, reason_code, properties):
|
||||
global mqtt_connected
|
||||
mqtt_connected = False
|
||||
logger.warning("MQTT disconnected: %s", reason_code)
|
||||
|
||||
def on_message(client, userdata, msg):
|
||||
try:
|
||||
payload = json.loads(msg.payload)
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
process_event(payload, claude, matrix), loop
|
||||
)
|
||||
except json.JSONDecodeError:
|
||||
logger.warning("Invalid JSON on %s", msg.topic)
|
||||
except Exception:
|
||||
logger.exception("Error processing MQTT message")
|
||||
|
||||
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
|
||||
client.on_connect = on_connect
|
||||
client.on_disconnect = on_disconnect
|
||||
client.on_message = on_message
|
||||
return client
|
||||
|
||||
|
||||
# --- Main ---
|
||||
def main():
|
||||
logger.info("Starting frigate-notify")
|
||||
|
||||
claude = anthropic.Anthropic()
|
||||
matrix = MatrixClient(MATRIX_HOMESERVER, MATRIX_ACCESS_TOKEN, MATRIX_ROOM_ID)
|
||||
|
||||
loop = asyncio.new_event_loop()
|
||||
loop_thread = Thread(target=loop.run_forever, daemon=True)
|
||||
loop_thread.start()
|
||||
|
||||
start_health_server()
|
||||
|
||||
mqtt_client = create_mqtt_client(loop, claude, matrix)
|
||||
mqtt_client.connect(MQTT_HOST, MQTT_PORT)
|
||||
|
||||
def shutdown(signum, frame):
|
||||
logger.info("Shutting down...")
|
||||
mqtt_client.disconnect()
|
||||
loop.call_soon_threadsafe(loop.stop)
|
||||
sys.exit(0)
|
||||
|
||||
signal.signal(signal.SIGTERM, shutdown)
|
||||
signal.signal(signal.SIGINT, shutdown)
|
||||
|
||||
mqtt_client.loop_forever()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user