Connect Home Assistant or Frigate.

#!/usr/bin/env python3
"""Peekaboo bridge for Frigate (Python 3.9+, standard library only).

Polls Frigate's HTTP API for new tracked-object events that have a snapshot, downloads each snapshot and posts it
to the matching camera's Peekaboo push URL, with the object label and the time of the frame. Each event is sent
once, even across restarts, and the first start begins from "now": old history is never replayed.

    FRIGATE_URL=http://frigate:5000 \
    PEEKABOO_CAMERAS="front_door=https://getpeekaboo.pages.dev/api/push/<token>" \
    python3 peekaboo_frigate.py --test

Every option is a flag or an environment variable (run with --help); integrations/README.md has the full guide.
Push URLs are secrets: this script only ever logs the first four characters of a token.
"""
from __future__ import annotations

import argparse
import collections
import http.client
import json
import logging
import os
import re
import signal
import ssl
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from http.cookies import SimpleCookie
from typing import Callable, Dict, NamedTuple, Optional, Tuple

__version__ = "1.0.0"

MAX_FRAME_BYTES = 1_500_000
LOOKBACK_SECONDS = 15 * 60  # Peekaboo only accepts event times from the last 15 minutes
SERVER_LIMIT_PER_HOUR = 120
SNAPSHOT_HEIGHTS = (None, 1080, 720, 480)
BACKOFF_BASE, BACKOFF_CAP = 5.0, 900.0
STATE_MAX_IDS, STATE_MAX_AGE = 2000, 24 * 3600
DEFAULT_LABELS = "person,car,truck,motorcycle,bicycle,bus,package"
USER_AGENT = f"peekaboo-frigate/{__version__}"

VEHICLES = {"car", "truck", "motorcycle", "bus", "boat"}
ANIMALS = {"dog", "cat", "bird", "horse", "sheep", "cow", "bear", "deer", "raccoon", "fox", "squirrel", "rabbit",
           "goat", "skunk"}
TOKEN_RE = re.compile(r"(/api/push/)([A-Za-z0-9_-]+)")
PUSH_PATH_RE = re.compile(r"/api/push/[A-Za-z0-9_-]{20,80}")

log = logging.getLogger("peekaboo_frigate")


def mask(text: object) -> str:
    """Hides all but the first four characters of every push token in `text`."""
    return TOKEN_RE.sub(lambda m: m.group(1) + m.group(2)[:4] + "****", str(text))


class _MaskTokens(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.msg, record.args = mask(record.getMessage()), None
        return True


log.addFilter(_MaskTokens())


def peekaboo_label(frigate_label: str) -> str:
    if frigate_label in ("person", "package"):
        return frigate_label
    if frigate_label in VEHICLES:
        return "car"
    if frigate_label in ANIMALS:
        return "animal"
    return "motion"


def _float(value: object) -> Optional[float]:
    try:
        return float(value) if value is not None and value != "" else None
    except (TypeError, ValueError):
        return None


def event_score(event: dict) -> Optional[float]:
    data = event.get("data") if isinstance(event.get("data"), dict) else {}
    for value in (data.get("top_score"), event.get("top_score"), data.get("score")):
        if _float(value) is not None:
            return _float(value)
    return None


# ---------------------------------------------------------------- configuration

class ConfigError(Exception):
    pass


def parse_camera_map(text: str) -> Dict[str, str]:
    """'front_door=https://...,driveway=https://...' (commas, spaces or newlines between pairs)."""
    cameras = {}
    for number, part in enumerate(p for p in re.split(r"[,\s]+", text.strip()) if p):
        name, sep, url = part.partition("=")
        if not sep or not name.strip() or not url.strip():
            raise ConfigError(f"camera entry {number + 1} should look like name=https://...")
        cameras[name.strip()] = url.strip()
    return cameras


def load_camera_file(path: str) -> Dict[str, str]:
    try:
        with open(path, encoding="utf-8") as f:
            raw = json.load(f)
    except (OSError, ValueError) as e:
        raise ConfigError(f"can't read the cameras file {path}: {e}") from None
    if not isinstance(raw, dict) or not all(isinstance(k, str) and isinstance(v, str) for k, v in raw.items()):
        raise ConfigError(f'{path} should be a JSON object like {{"front_door": "https://..."}}')
    return dict(raw)


def check_push_url(name: str, url: str) -> None:
    parts = urllib.parse.urlsplit(url)
    local = parts.hostname in ("localhost", "127.0.0.1", "::1")
    if parts.scheme != "https" and not (parts.scheme == "http" and local):
        raise ConfigError(f"{name}: the push URL must start with https://")
    if not PUSH_PATH_RE.fullmatch(parts.path) or parts.query:
        raise ConfigError(f"{name}: that doesn't look like a Peekaboo push URL "
                          "(it ends in /api/push/ followed by a long code)")


class Config(NamedTuple):
    frigate_url: str
    cameras: Dict[str, str]
    labels: frozenset
    min_score: float
    poll_seconds: float
    max_per_hour: int
    state_file: str
    frigate_user: str
    frigate_password: str
    frigate_insecure: bool
    once: bool
    test: bool
    verbose: bool


def parse_config(argv: Optional[list], env: Dict[str, str]) -> Config:
    def env_or(key: str, default: str) -> str:
        return env.get(key) or default

    p = argparse.ArgumentParser(
        prog="peekaboo_frigate.py",
        description="Send Frigate event snapshots to Peekaboo push URLs.",
        epilog="Push URLs come from the Peekaboo app (one per camera). Treat them like passwords.",
    )
    p.add_argument("--frigate-url", default=env_or("FRIGATE_URL", "http://127.0.0.1:5000"),
                   help="Frigate address, e.g. http://frigate:5000 (env FRIGATE_URL)")
    p.add_argument("--cameras", default=env.get("PEEKABOO_CAMERAS", ""),
                   help="Frigate camera name = push URL pairs, comma separated (env PEEKABOO_CAMERAS)")
    p.add_argument("--cameras-file", default=env.get("PEEKABOO_CAMERAS_FILE", ""),
                   help='JSON file {"front_door": "https://..."} (env PEEKABOO_CAMERAS_FILE)')
    p.add_argument("--labels", default=env_or("PEEKABOO_LABELS", DEFAULT_LABELS),
                   help=f"Frigate labels to send (env PEEKABOO_LABELS, default {DEFAULT_LABELS})")
    p.add_argument("--min-score", type=float, default=env_or("PEEKABOO_MIN_SCORE", "0.7"),
                   help="skip events whose top score is lower (env PEEKABOO_MIN_SCORE, default 0.7)")
    p.add_argument("--poll-seconds", type=float, default=env_or("PEEKABOO_POLL_SECONDS", "5"),
                   help="how often to ask Frigate for new events (env PEEKABOO_POLL_SECONDS, default 5)")
    p.add_argument("--max-per-hour", type=int, default=env_or("PEEKABOO_MAX_PER_HOUR", "100"),
                   help="frames per camera per hour, at most 120 (env PEEKABOO_MAX_PER_HOUR, default 100)")
    p.add_argument("--state-file", default=env_or("PEEKABOO_STATE_FILE", "peekaboo_frigate_state.json"),
                   help="where to remember sent events (env PEEKABOO_STATE_FILE)")
    p.add_argument("--frigate-user", default=env.get("FRIGATE_USER", ""),
                   help="only for Frigate's login port 8971; the password comes from env FRIGATE_PASSWORD")
    p.add_argument("--frigate-insecure", action="store_true",
                   default=env.get("FRIGATE_INSECURE_TLS", "").lower() in ("1", "true", "yes"),
                   help="accept Frigate's self-signed certificate (env FRIGATE_INSECURE_TLS=1)")
    p.add_argument("--once", action="store_true", help="check Frigate once, send what's new, and exit")
    p.add_argument("--test", action="store_true", help="send one current frame per camera to check the setup")
    p.add_argument("-v", "--verbose", action="store_true", help="log every poll")
    a = p.parse_args(argv)

    try:
        cameras = load_camera_file(a.cameras_file) if a.cameras_file else {}
        cameras.update(parse_camera_map(a.cameras))
        if not cameras:
            raise ConfigError("no cameras: set PEEKABOO_CAMERAS or PEEKABOO_CAMERAS_FILE")
        for name, url in cameras.items():
            check_push_url(name, url)
        labels = frozenset(x.strip() for x in a.labels.split(",") if x.strip())
        if not labels:
            raise ConfigError("the label list is empty")
        if not 0 <= a.min_score <= 1:
            raise ConfigError("--min-score must be between 0 and 1")
        if a.poll_seconds < 1:
            raise ConfigError("--poll-seconds must be at least 1")
        if not 1 <= a.max_per_hour <= SERVER_LIMIT_PER_HOUR:
            raise ConfigError(f"--max-per-hour must be between 1 and {SERVER_LIMIT_PER_HOUR}")
        password = env.get("FRIGATE_PASSWORD", "")
        if a.frigate_user and not password:
            raise ConfigError("FRIGATE_USER is set but FRIGATE_PASSWORD is empty")
    except ConfigError as e:
        p.error(str(e))
    return Config(a.frigate_url, cameras, labels, a.min_score, a.poll_seconds, a.max_per_hour, a.state_file,
                  a.frigate_user, password, a.frigate_insecure, a.once, a.test, a.verbose)


# ---------------------------------------------------------------- Frigate

class FrigateError(Exception):
    def __init__(self, message: str, status: Optional[int] = None):
        super().__init__(message)
        self.status = status


class Frigate:
    """The three Frigate endpoints this bridge uses: /api/events, an event's snapshot.jpg and a camera's latest.jpg."""

    def __init__(self, url: str, user: str = "", password: str = "", insecure: bool = False, timeout: float = 15.0):
        base = url.rstrip("/")
        self.api = (base[:-4] if base.endswith("/api") else base) + "/api"
        self.user, self.password, self.timeout = user, password, timeout
        self.token: Optional[str] = None
        self.context = None
        if insecure:
            self.context = ssl.create_default_context()
            self.context.check_hostname = False
            self.context.verify_mode = ssl.CERT_NONE

    def _get(self, path: str, params: Optional[dict] = None, retry_login: bool = True):
        url = self.api + path + ("?" + urllib.parse.urlencode(params) if params else "")
        headers = {"User-Agent": USER_AGENT}
        if self.token:
            headers["Authorization"] = "Bearer " + self.token
        try:
            with urllib.request.urlopen(urllib.request.Request(url, headers=headers), timeout=self.timeout,
                                        context=self.context) as r:
                return r.read(), r.headers
        except urllib.error.HTTPError as e:
            e.close()
            if e.code == 401 and self.user and retry_login:
                self.login()
                return self._get(path, params, retry_login=False)
            if e.code == 401:
                raise FrigateError("Frigate asked for a login: use port 5000 from inside Docker, "
                                   "or set FRIGATE_USER and FRIGATE_PASSWORD", 401) from None
            raise FrigateError(f"Frigate answered {e.code} for {path}", e.code) from None
        except (urllib.error.URLError, OSError, http.client.HTTPException) as e:
            raise FrigateError(f"can't reach Frigate at {self.api} ({getattr(e, 'reason', e)})") from None

    def login(self) -> None:
        body = json.dumps({"user": self.user, "password": self.password}).encode()
        req = urllib.request.Request(self.api + "/login", data=body,
                                     headers={"Content-Type": "application/json", "User-Agent": USER_AGENT})
        try:
            with urllib.request.urlopen(req, timeout=self.timeout, context=self.context) as r:
                cookies = r.headers.get_all("Set-Cookie") or []
        except urllib.error.HTTPError as e:
            e.close()
            raise FrigateError(f"Frigate login failed ({e.code}); check FRIGATE_USER and FRIGATE_PASSWORD",
                               e.code) from None
        except (urllib.error.URLError, OSError, http.client.HTTPException) as e:
            raise FrigateError(f"can't reach Frigate at {self.api} ({getattr(e, 'reason', e)})") from None
        for header in cookies:
            jar = SimpleCookie()
            jar.load(header)
            if "frigate_token" in jar:
                self.token = jar["frigate_token"].value
                return
        raise FrigateError("Frigate's login answer had no frigate_token cookie")

    def events(self, after: float, cameras, labels, limit: int = 100) -> list:
        params = {
            "cameras": ",".join(sorted(cameras)),
            "labels": ",".join(sorted(labels)),
            "after": f"{after:.3f}",
            "has_snapshot": 1,
            "include_thumbnails": 0,
            "sort": "date_desc",
            "limit": limit,
        }
        raw, _ = self._get("/events", params)
        try:
            events = json.loads(raw)
        except ValueError:
            raise FrigateError("Frigate's /events answer wasn't JSON") from None
        if not isinstance(events, list):
            raise FrigateError("Frigate's /events answer wasn't a list")
        return [e for e in events if isinstance(e, dict)]

    def snapshot(self, event_id: str, height: Optional[int] = None) -> Tuple[bytes, Optional[float]]:
        params = {"bbox": 0, "timestamp": 0, "crop": 0}
        if height:
            params["height"] = height
        raw, headers = self._get(f"/events/{urllib.parse.quote(event_id, safe='')}/snapshot.jpg", params)
        return raw, _float(headers.get("X-Frame-Time"))

    def latest(self, camera: str, height: Optional[int] = None) -> Tuple[bytes, bool]:
        raw, headers = self._get(f"/{urllib.parse.quote(camera, safe='')}/latest.jpg",
                                 {"height": height} if height else None)
        return raw, headers.get("X-Frigate-Offline", "").lower() == "true"


class FrameTooLarge(Exception):
    pass


def fetch_within_limit(fetch: Callable[[Optional[int]], tuple]) -> tuple:
    """Calls fetch(height) at full size, then smaller heights, until the JPEG fits Peekaboo's 1.5 MB limit."""
    for height in SNAPSHOT_HEIGHTS:
        image, extra = fetch(height)
        if len(image) <= MAX_FRAME_BYTES:
            return image, extra
    raise FrameTooLarge("the frame is over 1.5 MB even at 480 pixels high")


# ---------------------------------------------------------------- Peekaboo

class PushResult(NamedTuple):
    status: int  # 0 when Peekaboo couldn't be reached
    data: dict
    retry_after: Optional[float] = None
    error: str = ""


def _json_dict(raw: bytes) -> dict:
    try:
        value = json.loads(raw or b"{}")
    except ValueError:
        return {}
    return value if isinstance(value, dict) else {}


def push_frame(url: str, image: bytes, label: Optional[str] = None, event_time: Optional[float] = None,
               timeout: float = 30.0, event_id: Optional[str] = None) -> PushResult:
    headers = {"Content-Type": "image/jpeg", "User-Agent": USER_AGENT}
    if label:
        headers["X-Peekaboo-Label"] = label
    if event_time:
        headers["X-Peekaboo-Event-Time"] = str(int(event_time))
    if event_id:
        # Lets Peekaboo drop a retry of a push it already took (e.g. after a timeout).
        headers["X-Peekaboo-Event-Id"] = event_id[:80]
    req = urllib.request.Request(url, data=image, headers=headers, method="POST")
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return PushResult(r.status, _json_dict(r.read()))
    except urllib.error.HTTPError as e:
        try:
            body = e.read()
        except (OSError, http.client.HTTPException):
            body = b""
        finally:
            e.close()
        return PushResult(e.code, _json_dict(body), _float(e.headers.get("Retry-After")))
    except (urllib.error.URLError, OSError, http.client.HTTPException) as e:
        return PushResult(0, {}, None, str(getattr(e, "reason", e)))


def explain(result: PushResult) -> str:
    server = result.data.get("message") or ""
    if result.status == 0:
        return f"couldn't reach Peekaboo ({result.error})"
    if result.status == 404:
        return "Peekaboo doesn't recognize this push URL, or push cameras aren't switched on for your account"
    if result.status == 413:
        return "the frame is larger than Peekaboo accepts (1.5 MB)"
    if result.status == 429:
        return "too many frames this hour for this camera"
    if result.status >= 500:
        return f"Peekaboo had a problem (HTTP {result.status}); it will be retried"
    return f"Peekaboo answered {result.status}" + (f": {server}" if server else "")


def _short(text: object, limit: int = 140) -> str:
    text = " ".join(str(text or "").split())
    return text if len(text) <= limit else text[: limit - 1] + "…"


# ---------------------------------------------------------------- bridge

class Lane:
    """One Frigate camera and its push URL, with its own hourly budget and backoff."""

    def __init__(self, name: str, url: str, max_per_hour: int):
        self.name, self.url, self.max_per_hour = name, url, max_per_hour
        self.sent: collections.deque = collections.deque()
        self.blocked_until = 0.0
        self.failures = 0

    def has_room(self, now: float) -> bool:
        while self.sent and self.sent[0] <= now - 3600:
            self.sent.popleft()
        return len(self.sent) < self.max_per_hour

    def back_off(self, now: float, at_least: float = 0.0) -> float:
        self.failures += 1
        delay = max(min(BACKOFF_CAP, BACKOFF_BASE * 2 ** (self.failures - 1)), at_least)
        self.blocked_until = now + delay
        return delay


class State:
    """Sent (or permanently skipped) Frigate event ids, and the moment of the first start."""

    def __init__(self, path: str):
        self.path = path
        self.since: Optional[float] = None
        self.done: Dict[str, float] = {}

    def load(self) -> bool:
        try:
            with open(self.path, encoding="utf-8") as f:
                raw = json.load(f)
            self.since = float(raw["since"])
            self.done = {str(k): float(v) for k, v in raw.get("done", {}).items()}
            return True
        except FileNotFoundError:
            return False
        except (OSError, ValueError, KeyError, TypeError, AttributeError) as e:
            log.warning("State file %s is unreadable (%s); starting again from now.", self.path, e)
            self.since, self.done = None, {}
            return False

    def mark(self, event_id: str, now: float) -> None:
        self.done[event_id] = now

    def save(self, now: float) -> None:
        cutoff = now - STATE_MAX_AGE
        newest = sorted(((t, k) for k, t in self.done.items() if t >= cutoff), reverse=True)[:STATE_MAX_IDS]
        self.done = {k: t for t, k in newest}
        folder = os.path.dirname(os.path.abspath(self.path))
        os.makedirs(folder, exist_ok=True)
        tmp = self.path + ".tmp"
        with open(tmp, "w", encoding="utf-8") as f:
            json.dump({"version": 1, "since": self.since, "done": self.done}, f)
        os.replace(tmp, self.path)


class Bridge:
    def __init__(self, frigate: Frigate, lanes: Dict[str, Lane], labels, min_score: float, state: State,
                 clock: Callable[[], float] = time.time):
        self.frigate, self.lanes, self.labels, self.min_score = frigate, lanes, frozenset(labels), min_score
        self.state, self.clock = state, clock
        self.stop = threading.Event()
        self._noted: Dict[str, str] = {}

    def start(self) -> None:
        now = self.clock()
        if self.state.load():
            log.info("Resuming: %d events remembered; looking back at most %d minutes.",
                     len(self.state.done), LOOKBACK_SECONDS // 60)
        else:
            self.state.since = now
            self.state.save(now)
            log.info("First start: only events from now on will be sent.")

    def poll(self) -> int:
        """Asks Frigate for recent events and sends the ones that are due. Returns how many were sent."""
        now = self.clock()
        after = max(self.state.since or now, now - LOOKBACK_SECONDS)
        events = self.frigate.events(after, self.lanes.keys(), self.labels)
        log.debug("Frigate returned %d events after %.0f.", len(events), after)
        sent = 0
        for event in sorted(events, key=lambda e: _float(e.get("start_time")) or 0.0):
            if self.stop.is_set():
                break
            if self._due(event, after) and self._send(event):
                sent += 1
        return sent

    def _due(self, event: dict, after: float) -> bool:
        event_id = event.get("id")
        if not isinstance(event_id, str) or event_id in self.state.done:
            return False
        if event.get("camera") not in self.lanes or event.get("label") not in self.labels:
            return False
        if not event.get("has_snapshot") or event.get("false_positive"):
            return False
        if (_float(event.get("start_time")) or 0.0) <= after:
            return False
        score = event_score(event)
        return score is None or score >= self.min_score

    def _note(self, event_id: str, message: str) -> None:
        """Logs why an event is waiting, once per event and reason."""
        if self._noted.get(event_id) != message:
            if len(self._noted) > 500:
                self._noted.clear()
            self._noted[event_id] = message
            log.info(message)

    def _send(self, event: dict) -> bool:
        lane, event_id, now = self.lanes[event["camera"]], event["id"], self.clock()
        if now < lane.blocked_until:
            return False
        if not lane.has_room(now):
            self._note(event_id, f"{lane.name}: holding event {event_id}; this camera has sent its "
                                 f"{lane.max_per_hour} frames for the last hour.")
            return False
        try:
            image, frame_time = fetch_within_limit(lambda h: self.frigate.snapshot(event_id, h))
        except FrameTooLarge as e:
            log.warning("%s: event %s skipped: %s.", lane.name, event_id, e)
            self.state.mark(event_id, now)
            return False
        except FrigateError as e:
            self._note(event_id, f"{lane.name}: no snapshot yet for event {event_id} ({e}); will try again.")
            return False

        label = peekaboo_label(event["label"])
        data = event.get("data") if isinstance(event.get("data"), dict) else {}
        when = frame_time or _float(data.get("snapshot_frame_time")) or _float(event.get("start_time")) or now
        result = push_frame(lane.url, image, label, min(when, now), event_id=f"frigate:{event_id}")
        if result.status not in (0, 404):
            lane.sent.append(now)

        if 200 <= result.status < 300:
            lane.failures = 0
            self.state.mark(event_id, now)
            self.state.save(now)
            alerts = result.data.get("alerts")
            log.info("%s: sent %s (event %s). Peekaboo saw: %s%s", lane.name, label, event_id,
                     _short(result.data.get("description")) or "(no description)",
                     f" [{alerts} heads-up]" if alerts else "")
            return True
        if result.status == 429:
            delay = lane.back_off(now, at_least=max(60.0, min(result.retry_after or 0.0, 3600.0)))
            log.warning("%s: %s; pausing this camera for %d s.", lane.name, explain(result), delay)
        elif result.status == 404:
            lane.blocked_until = now + 600
            log.error("%s: %s (%s). Trying again in 10 minutes.", lane.name, explain(result), mask(lane.url))
        elif result.status == 0 or result.status >= 500:
            delay = lane.back_off(now)
            log.warning("%s: event %s not delivered: %s; trying again in %d s.", lane.name, event_id,
                        explain(result), delay)
        else:  # 400, 413 and other answers that won't change on a retry
            self.state.mark(event_id, now)
            log.warning("%s: event %s skipped: %s.", lane.name, event_id, explain(result))
        return False


def run(bridge: Bridge, poll_seconds: float, once: bool = False) -> int:
    bridge.start()
    failures = 0
    while not bridge.stop.is_set():
        try:
            bridge.poll()
            failures, wait = 0, poll_seconds
        except FrigateError as e:
            failures += 1
            wait = min(300.0, poll_seconds * 2 ** failures)
            log.warning("%s; trying again in %d s.", e, wait)
        if once:
            break
        bridge.stop.wait(wait)
    bridge.state.save(bridge.clock())
    log.info("Stopped.")
    return 1 if once and failures else 0


def run_test(frigate: Frigate, lanes: Dict[str, Lane], clock: Callable[[], float] = time.time) -> int:
    """Sends one current frame per camera (Frigate's latest.jpg) and reports what Peekaboo answered."""
    failed = 0
    for lane in lanes.values():
        try:
            image, offline = fetch_within_limit(lambda h: frigate.latest(lane.name, h))
        except (FrigateError, FrameTooLarge) as e:
            log.error("%s: couldn't get a frame from Frigate: %s", lane.name, e)
            failed += 1
            continue
        if offline:
            log.warning("%s: Frigate says this camera is offline, so this is its last saved frame.", lane.name)
        result = push_frame(lane.url, image, None, clock())
        if 200 <= result.status < 300:
            log.info("%s: OK. Peekaboo saw: %s", lane.name,
                     _short(result.data.get("description")) or "(no description)")
        else:
            failed += 1
            log.error("%s: %s.", lane.name, explain(result))
    log.info("Test finished: %d of %d cameras OK.", len(lanes) - failed, len(lanes))
    return 1 if failed else 0


def main(argv: Optional[list] = None, env: Optional[Dict[str, str]] = None) -> int:
    cfg = parse_config(argv, dict(os.environ) if env is None else env)
    if not log.handlers:
        handler = logging.StreamHandler()
        handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
        log.addHandler(handler)
        log.propagate = False
    log.setLevel(logging.DEBUG if cfg.verbose else logging.INFO)

    frigate = Frigate(cfg.frigate_url, cfg.frigate_user, cfg.frigate_password, cfg.frigate_insecure)
    lanes = {name: Lane(name, url, cfg.max_per_hour) for name, url in cfg.cameras.items()}
    log.info("Peekaboo bridge %s: Frigate %s; cameras %s; labels %s; min score %.2f.", __version__, frigate.api,
             ", ".join(sorted(lanes)), ",".join(sorted(cfg.labels)), cfg.min_score)
    if cfg.test:
        return run_test(frigate, lanes)

    bridge = Bridge(frigate, lanes, cfg.labels, cfg.min_score, State(cfg.state_file))
    if threading.current_thread() is threading.main_thread():
        for sig in (signal.SIGTERM, signal.SIGINT):
            signal.signal(sig, lambda *_: bridge.stop.set())
    return run(bridge, cfg.poll_seconds, cfg.once)


if __name__ == "__main__":
    sys.exit(main())