Jeremy Hobbs
6086ed3cc0
AI-powered video alert service that bridges Frigate NVR events to Matrix. Listens to MQTT, uses Claude Haiku vision to analyze snapshots, and posts annotated alerts to a Matrix room. - app/main.py: MQTT subscriber and event orchestration loop - app/vision.py: Claude Haiku vision analysis wrapper - app/matrix_client.py: Matrix HTTP client for sending messages/images - app/requirements.txt: Python dependencies - Dockerfile: container image build definition - docker-compose.yml: deployment configuration (host network, NAS mounts) - .env.example: required environment variable template (copy to .env)
85 lines
3.2 KiB
Python
Executable File
85 lines
3.2 KiB
Python
Executable File
import asyncio
|
|
import logging
|
|
import time
|
|
import aiohttp
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
RETRY_DELAYS = [1, 2, 4]
|
|
|
|
|
|
class MatrixClient:
|
|
def __init__(self, homeserver: str, token: str, room_id: str):
|
|
self.homeserver = homeserver.rstrip("/")
|
|
self.token = token
|
|
self.room_id = room_id
|
|
self.headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
async def _request(self, method: str, url: str, **kwargs) -> dict | None:
|
|
for attempt, delay in enumerate(RETRY_DELAYS):
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.request(
|
|
method, url, headers=self.headers, **kwargs
|
|
) as resp:
|
|
if resp.status < 300:
|
|
return await resp.json()
|
|
body = await resp.text()
|
|
logger.warning(
|
|
"Matrix API %s (attempt %d): %s",
|
|
resp.status,
|
|
attempt + 1,
|
|
body,
|
|
)
|
|
except Exception:
|
|
logger.exception("Matrix request failed (attempt %d)", attempt + 1)
|
|
if attempt < len(RETRY_DELAYS) - 1:
|
|
await asyncio.sleep(delay)
|
|
logger.error("All Matrix retries exhausted for %s", url)
|
|
return None
|
|
|
|
async def upload_image(self, image_bytes: bytes, filename: str) -> str | None:
|
|
url = f"{self.homeserver}/_matrix/media/v3/upload?filename={filename}"
|
|
headers = {**self.headers, "Content-Type": "image/jpeg"}
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(url, headers=headers, data=image_bytes) as resp:
|
|
if resp.status < 300:
|
|
data = await resp.json()
|
|
mxc_uri = data.get("content_uri")
|
|
logger.info("Uploaded image: %s", mxc_uri)
|
|
return mxc_uri
|
|
body = await resp.text()
|
|
logger.error("Image upload failed %s: %s", resp.status, body)
|
|
except Exception:
|
|
logger.exception("Image upload failed")
|
|
return None
|
|
|
|
async def send_image(self, mxc_uri: str, filename: str, size: int) -> dict | None:
|
|
txn_id = str(int(time.time() * 1000))
|
|
url = (
|
|
f"{self.homeserver}/_matrix/client/r0/rooms/"
|
|
f"{self.room_id}/send/m.room.message/{txn_id}"
|
|
)
|
|
body = {
|
|
"msgtype": "m.image",
|
|
"body": filename,
|
|
"url": mxc_uri,
|
|
"info": {"mimetype": "image/jpeg", "size": size},
|
|
}
|
|
return await self._request("PUT", url, json=body)
|
|
|
|
async def send_text(self, plain: str, html: str) -> dict | None:
|
|
txn_id = str(int(time.time() * 1000) + 1)
|
|
url = (
|
|
f"{self.homeserver}/_matrix/client/r0/rooms/"
|
|
f"{self.room_id}/send/m.room.message/{txn_id}"
|
|
)
|
|
body = {
|
|
"msgtype": "m.text",
|
|
"body": plain,
|
|
"format": "org.matrix.custom.html",
|
|
"formatted_body": html,
|
|
}
|
|
return await self._request("PUT", url, json=body)
|