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"{camera} · {label.capitalize()} · {score_pct}%
"
f"{description}
"
f'View in Frigate
'
f"{ts} · {date_str}"
)
# --- 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()