import json
import logging
import os
import queue
import re
import signal
import socket
import struct
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path


VERSION = "1.0"
DEFAULT_CONFIG = "/etc/frigate-camera-control-bridge.json"
LOG = logging.getLogger("frigate-camera-control-bridge")
STREAM_URL_RE = re.compile(r"(?:rtsp|rtsps|http|https)://[^\s'\"<>]+", re.IGNORECASE)
HOME_ASSISTANT_ACTION_ICONS = {
    "action.siren": "mdi:bullhorn",
    "action.red_blue": "mdi:alarm-light",
    "action.manual_white": "mdi:lightbulb-on",
    "action.all_off": "mdi:power",
}

from .i18n import DEFAULT_LANGUAGE, available_languages, normalize_language, text_for
from .ui import read_static_file, render_page


def mqtt_encode_string(value):
    # Minimal MQTT v3.1.1 encoder. Avoiding a client dependency keeps the image
    # small and avoids pip installs on offline homelab hosts.
    data = value.encode("utf-8")
    return struct.pack("!H", len(data)) + data


def mqtt_encode_remaining_length(length):
    out = bytearray()
    while True:
        digit = length % 128
        length //= 128
        if length:
            digit |= 0x80
        out.append(digit)
        if not length:
            return bytes(out)


def read_exact(sock, size):
    buf = bytearray()
    while len(buf) < size:
        part = sock.recv(size - len(buf))
        if not part:
            raise EOFError("socket closed")
        buf.extend(part)
    return bytes(buf)


def read_mqtt_packet(sock):
    first = read_exact(sock, 1)[0]
    multiplier = 1
    remaining_length = 0
    while True:
        digit = read_exact(sock, 1)[0]
        remaining_length += (digit & 127) * multiplier
        if not digit & 128:
            break
        multiplier *= 128
        if multiplier > 128 * 128 * 128:
            raise ValueError("malformed MQTT remaining length")
    return first, read_exact(sock, remaining_length)


def send_mqtt_packet(sock, packet_type_with_flags, payload=b""):
    sock.sendall(
        bytes([packet_type_with_flags])
        + mqtt_encode_remaining_length(len(payload))
        + payload
    )


def load_config(path):
    with open(path, "r", encoding="utf-8") as handle:
        return json.load(handle)


def slug(value):
    return "".join(ch if ch.isalnum() else "_" for ch in value.lower()).strip("_")


def env_bool(name, default=True):
    value = os.environ.get(name)
    if value is None:
        return default
    return value.lower() in ("1", "true", "yes", "on")


def env_list(name):
    value = os.environ.get(name, "")
    return [item.strip() for item in value.split(",") if item.strip()]


def env_secret(name):
    # Support both classic env vars and *_FILE secret mounts. Kubernetes users
    # will usually use env from Secret, while Docker/Podman users may prefer a
    # mounted secret file. The direct env var wins when both are present.
    value = os.environ.get(name)
    if value:
        return value
    path = os.environ.get(f"{name}_FILE")
    if not path:
        return None
    try:
        return Path(path).read_text(encoding="utf-8").strip()
    except OSError as exc:
        raise RuntimeError(f"failed to read {name}_FILE={path}: {exc}") from exc


def language_from_accept_header(value):
    # HTTP Accept-Language can contain regional tags and q weights. The GUI
    # ships language catalogs by base code, so pl-PL and pl both resolve to pl.
    if not value:
        return DEFAULT_LANGUAGE
    supported = available_languages()
    matches = []
    for index, raw_item in enumerate(value.split(",")):
        parts = [item.strip() for item in raw_item.split(";") if item.strip()]
        if not parts:
            continue
        tag = parts[0].lower().replace("_", "-")
        if tag == "*":
            continue
        quality = 1.0
        for param in parts[1:]:
            if not param.lower().startswith("q="):
                continue
            try:
                quality = float(param.split("=", 1)[1])
            except ValueError:
                quality = 0.0
        if quality <= 0:
            continue
        base = tag.split("-", 1)[0]
        if base in supported:
            matches.append((-quality, index, base))
    if not matches:
        return DEFAULT_LANGUAGE
    matches.sort()
    return matches[0][2]


def config_from_env():
    # Env mode is the preferred sidecar mode. It needs only a Frigate API URL,
    # MQTT connection parameters, and camera credentials. The old JSON config is
    # still accepted for manual runs and migration debugging.
    frigate_name = os.environ.get("FRIGATE_NAME", "frigate")
    language = normalize_language(os.environ.get("UI_LANGUAGE", DEFAULT_LANGUAGE))
    frigate_url = os.environ.get("FRIGATE_URL")
    if not frigate_url:
        frigate_host = os.environ.get("FRIGATE_HOST")
        if not frigate_host:
            raise RuntimeError("FRIGATE_URL or FRIGATE_HOST is required in env mode")
        frigate_url = f"http://{frigate_host}:5000"

    mqtt_host = os.environ.get("MQTT_HOST")
    if not mqtt_host:
        raise RuntimeError("MQTT_HOST is required in env mode")
    mqtt_user = env_secret("MQTT_USER")
    mqtt_password = env_secret("MQTT_PASSWORD")
    if not mqtt_user:
        raise RuntimeError("MQTT_USER is required in env mode")
    if not mqtt_password:
        raise RuntimeError("MQTT_PASSWORD is required in env mode")
    camera_user = env_secret("DEFAULT_CAMERA_USER")
    camera_password = env_secret("DEFAULT_CAMERA_PASSWORD")
    if not camera_user:
        raise RuntimeError("DEFAULT_CAMERA_USER is required in env mode")
    if not camera_password:
        raise RuntimeError("DEFAULT_CAMERA_PASSWORD is required in env mode")

    mqtt = {
        "host": mqtt_host,
        "port": int(os.environ.get("MQTT_PORT", "1883")),
        "base_topic": os.environ.get("BASE_TOPIC", f"frigate_camera_control/{slug(frigate_name)}"),
        "discovery_prefix": os.environ.get("DISCOVERY_PREFIX", "homeassistant"),
        "client_id": os.environ.get("MQTT_CLIENT_ID", f"frigate-camera-control-{slug(frigate_name)}"),
        "username": mqtt_user,
        "password": mqtt_password,
        "keepalive": int(os.environ.get("MQTT_KEEPALIVE", "60")),
    }

    camera_auth = {
        "username": camera_user,
        "password": camera_password,
    }

    return {
        "mqtt": mqtt,
        "camera_auth": camera_auth,
        "discovery": {
            "enabled": True,
            "state_file": os.environ.get(
                "DISCOVERY_STATE_FILE",
                "/var/lib/frigate-camera-control-bridge/discovery_state.json",
            ),
            "include_hikvision_colorvu_white": env_bool("INCLUDE_HIKVISION_COLORVU_WHITE", True),
            "include_dahua_coaxial_off_probe": env_bool("INCLUDE_DAHUA_COAXIAL_OFF_PROBE", True),
            "wait_for_frigate": env_bool("WAIT_FOR_FRIGATE", True),
            "retry_seconds": float(os.environ.get("FRIGATE_RETRY_SECONDS", "5")),
            "startup_wait_seconds": float(os.environ.get("FRIGATE_STARTUP_WAIT_SECONDS", "0")),
            "dahua_probe_channels": [int(item) for item in env_list("DAHUA_PROBE_CHANNELS") or ["1", "0", "2"]],
            "dahua_siren_probe_channels": [
                int(item) for item in env_list("DAHUA_SIREN_PROBE_CHANNELS") or ["2", "1", "0"]
            ],
            "frigates": [
                {
                    "name": frigate_name,
                    "host": os.environ.get("FRIGATE_HOST", urllib.parse.urlsplit(frigate_url).hostname or frigate_name),
                    "url": frigate_url,
                }
            ],
        },
        "http": {
            "enabled": env_bool("HTTP_ENABLED", True),
            "bind": os.environ.get("HTTP_BIND", "0.0.0.0"),
            "port": int(os.environ.get("HTTP_PORT", "5011")),
            "token": env_secret("HTTP_TOKEN"),
        },
        "ui": {
            "default_language": language,
        },
        "defaults": {
            "http_timeout": float(os.environ.get("CAMERA_HTTP_TIMEOUT", "6")),
            "colorvu_brightness": int(os.environ.get("COLORVU_BRIGHTNESS", "100")),
            "snapshot_enabled": env_bool("SNAPSHOT_ENABLED", True),
            "snapshot_height": int(os.environ.get("SNAPSHOT_HEIGHT", "180")),
            "snapshot_cache_seconds": int(os.environ.get("SNAPSHOT_CACHE_SECONDS", "300")),
            "snapshot_retry_seconds": int(os.environ.get("SNAPSHOT_RETRY_SECONDS", "600")),
            "snapshot_refresh_seconds": int(os.environ.get("SNAPSHOT_REFRESH_SECONDS", "3600")),
            "snapshot_live_on_gui": env_bool("SNAPSHOT_LIVE_ON_GUI", True),
            "frigate_public_url": os.environ.get("FRIGATE_PUBLIC_URL"),
            "frigate_camera_path_template": os.environ.get(
                "FRIGATE_CAMERA_PATH_TEMPLATE",
                "/#{camera_quoted}",
            ),
            "frigate_refresh_seconds": int(os.environ.get("FRIGATE_REFRESH_SECONDS", "300")),
            "auto_off_siren_seconds": int(os.environ.get("AUTO_OFF_SIREN_SECONDS", "180")),
            "auto_off_red_blue_seconds": int(os.environ.get("AUTO_OFF_RED_BLUE_SECONDS", "300")),
            "auto_off_white_light_seconds": int(os.environ.get("AUTO_OFF_WHITE_LIGHT_SECONDS", "600")),
            "auto_off_default_seconds": int(os.environ.get("AUTO_OFF_DEFAULT_SECONDS", "0")),
            "startup_all_off": env_bool("STARTUP_ALL_OFF", True),
        },
        "actions": [
            {
                "id": f"{slug(frigate_name)}_all_off",
                "name": text_for(language, "action.all_off", target=frigate_name),
                "name_key": "action.all_off",
                "frigate_name": frigate_name,
                "platform": "button",
                "kind": "all_off",
                "device": {
                    "identifiers": [f"frigate_camera_control_{slug(frigate_name)}"],
                    "name": f"Frigate camera control {frigate_name}",
                    "manufacturer": "Local",
                    "model": "Sidecar",
                },
            }
        ],
    }


def load_config_or_env(path):
    if path and Path(path).exists():
        return load_config(path)
    return config_from_env()


def response_status_ok(body):
    text = body.decode("utf-8", errors="replace")
    if "Error" in text and "ResponseStatus" not in text:
        return False, text.strip()[:180]
    if "<statusCode>1</statusCode>" in text or "<statusCode>0</statusCode>" in text:
        return True, text.strip()[:180]
    if "OK" in text and "Error" not in text:
        return True, text.strip()[:180]
    if not text.strip():
        return True, ""
    return True, text.strip()[:180]


class DigestHttpClient:
    # Camera APIs in this installation use HTTP Digest. The client also allows
    # Basic Auth because some camera firmwares accept either scheme.
    def __init__(self, username, password, timeout=6):
        self.username = username
        self.password = password
        self.timeout = timeout

    def _opener(self, base_url):
        mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()
        mgr.add_password(None, base_url, self.username, self.password)
        return urllib.request.build_opener(
            urllib.request.HTTPDigestAuthHandler(mgr),
            urllib.request.HTTPBasicAuthHandler(mgr),
        )

    def request(self, url, method="GET", body=None, content_type=None):
        headers = {}
        data = None
        if body is not None:
            data = body.encode("utf-8")
            headers["Content-Type"] = content_type or "application/xml"
        parsed = urllib.parse.urlsplit(url)
        base_url = f"{parsed.scheme}://{parsed.netloc}/"
        req = urllib.request.Request(url, data=data, headers=headers, method=method)
        try:
            with self._opener(base_url).open(req, timeout=self.timeout) as response:
                payload = response.read()
                ok, detail = response_status_ok(payload)
                if not ok:
                    raise RuntimeError(detail)
                return response.status, payload.decode("utf-8", errors="replace")
        except urllib.error.HTTPError as exc:
            payload = exc.read().decode("utf-8", errors="replace")
            raise RuntimeError(f"HTTP {exc.code}: {payload[:180]}") from exc
        except urllib.error.URLError as exc:
            raise RuntimeError(str(exc.reason)) from exc


class Action:
    # Action is the stable internal contract used by MQTT, HTTP GUI, CLI, and
    # Home Assistant discovery. Camera-specific code is kept behind kind-specific
    # runners so new vendors can be added without touching the MQTT layer.
    def __init__(self, cfg, http_client):
        self.cfg = cfg
        self.id = cfg["id"]
        self.name = cfg.get("name", self.id)
        self.platform = cfg.get("platform", "switch")
        self.kind = cfg["kind"]
        self.http = http_client
        self.state = None
        self.auto_off_seconds = int(cfg.get("auto_off_seconds") or 0)
        self.auto_off_at = None
        self.snapshot_version = 0

    def display_name(self, lang):
        names = self.cfg.get("name_i18n") or {}
        normalized = normalize_language(lang)
        if names:
            return names.get(normalized) or names.get(DEFAULT_LANGUAGE) or self.name

        key = self.cfg.get("name_key")
        if not key:
            return self.name

        prefix = self.cfg.get("camera_display_name") or self.cfg.get("camera_name") or self.cfg.get("frigate_name") or ""
        return text_for(normalized, key, target=prefix).strip()

    def command_list(self):
        if self.kind == "all_off":
            return ["press"]
        if self.platform == "button":
            return ["press"]
        return ["on", "off"]

    def safe_off_command(self):
        if self.kind == "all_off":
            return None
        if "safe_off" in self.cfg:
            return "safe_off"
        if self.platform == "switch" and "off" in self.cfg:
            return "off"
        return None

    def run(self, command, bridge=None):
        command = command.lower()
        if self.kind == "all_off":
            if bridge is None:
                raise RuntimeError("all_off requires bridge context")
            return bridge.all_off()
        if self.kind == "hikvision_colorvu":
            self._run_hikvision(command)
        elif self.kind == "dahua_coaxial_io":
            self._run_dahua_coax(command)
        else:
            raise RuntimeError(f"unsupported action kind {self.kind}")

        if self.platform == "switch":
            if command == "on":
                self.state = "ON"
            elif command in ("off", "safe_off"):
                self.state = "OFF"
        return {"ok": True, "action": self.id, "command": command}

    def _run_hikvision(self, command):
        if command == "on":
            self._hikvision_put_supplement(colorvu_xml(self.cfg.get("brightness", 100)))
            self._hikvision_put_ircut(ircut_xml("night"))
            return
        if command in ("off", "safe_off"):
            restore = self.cfg.get("restore", {})
            self._hikvision_put_ircut(ircut_xml(restore.get("ircutFilterType", "auto")))
            self._hikvision_put_supplement(restore_xml(self.cfg.get("brightness", 100), restore))
            return
        raise RuntimeError(f"unsupported command for {self.id}: {command}")

    def _hikvision_put_supplement(self, xml):
        host = self.cfg["host"]
        self.http.request(
            f"http://{host}/ISAPI/Image/channels/1/supplementLight",
            method="PUT",
            body=xml,
        )

    def _hikvision_put_ircut(self, xml):
        host = self.cfg["host"]
        self.http.request(
            f"http://{host}/ISAPI/Image/channels/1/ircutFilter",
            method="PUT",
            body=xml,
        )

    def _run_dahua_coax(self, command):
        params = self.cfg.get(command)
        if params is None and command == "safe_off":
            params = self.cfg.get("off") or self.cfg.get("press")
        if params is None:
            raise RuntimeError(f"unsupported command for {self.id}: {command}")
        host = self.cfg["host"]
        query = urllib.parse.urlencode(
            {
                "action": "control",
                "channel": int(params["channel"]),
                "info[0].Type": int(params["type"]),
                "info[0].IO": int(params["io"]),
            },
            safe="[]",
        )
        self.http.request(f"http://{host}/cgi-bin/coaxialControlIO.cgi?{query}")


def colorvu_xml(brightness):
    # Hikvision ColorVu white light is controlled through supplementLight plus
    # ircutFilter. Turning it off restores the mode captured during discovery.
    return f"""<?xml version="1.0" encoding="UTF-8"?>
<SupplementLight version="2.0" xmlns="http://www.hikvision.com/ver20/XMLSchema">
<supplementLightMode>colorVuWhiteLight</supplementLightMode>
<mixedLightBrightnessRegulatMode>manual</mixedLightBrightnessRegulatMode>
<whiteLightBrightness>{int(brightness)}</whiteLightBrightness>
<irLightBrightness>100</irLightBrightness>
<EventIntelligenceModeCfg>
<brightnessRegulatMode>manual</brightnessRegulatMode>
<whiteLightBrightness>{int(brightness)}</whiteLightBrightness>
<irLightBrightness>100</irLightBrightness>
</EventIntelligenceModeCfg>
</SupplementLight>"""


def restore_xml(brightness, restore):
    associated = restore.get("associatedVMDHuman")
    associated_xml = ""
    if associated is not None:
        associated_xml = f"\n<associatedVMDHuman>{str(bool(associated)).lower()}</associatedVMDHuman>"
    return f"""<?xml version="1.0" encoding="UTF-8"?>
<SupplementLight version="2.0" xmlns="http://www.hikvision.com/ver20/XMLSchema">
<supplementLightMode>{restore.get("supplementLightMode", "eventIntelligence")}</supplementLightMode>
<mixedLightBrightnessRegulatMode>{restore.get("mixedLightBrightnessRegulatMode", "auto")}</mixedLightBrightnessRegulatMode>
<whiteLightBrightness>{int(restore.get("whiteLightBrightness", brightness))}</whiteLightBrightness>
<irLightBrightness>{int(restore.get("irLightBrightness", 100))}</irLightBrightness>
<EventIntelligenceModeCfg>
<brightnessRegulatMode>{restore.get("eventIntelligenceBrightnessRegulatMode", "auto")}</brightnessRegulatMode>
<whiteLightBrightness>{int(restore.get("whiteLightBrightness", brightness))}</whiteLightBrightness>
<irLightBrightness>{int(restore.get("irLightBrightness", 100))}</irLightBrightness>{associated_xml}
</EventIntelligenceModeCfg>
</SupplementLight>"""


def ircut_xml(mode):
    return f"""<?xml version="1.0" encoding="UTF-8"?>
<IrcutFilter version="2.0" xmlns="http://www.hikvision.com/ver20/XMLSchema">
<IrcutFilterType>{mode}</IrcutFilterType>
<nightToDayFilterLevel>4</nightToDayFilterLevel>
<nightToDayFilterTime>5</nightToDayFilterTime>
</IrcutFilter>"""


class MqttBridgeClient:
    # This is a deliberately small MQTT 3.1.1 client. It supports only what the
    # bridge needs: connect, subscribe to command topics, publish retained HA
    # discovery/state messages, and Last Will availability.
    def __init__(self, cfg, messages):
        self.cfg = cfg
        self.messages = messages
        self.sock = None
        self.lock = threading.Lock()
        self.packet_id = 1

    @property
    def base_topic(self):
        return self.cfg.get("base_topic", "frigate_camera_control")

    @property
    def availability_topic(self):
        return f"{self.base_topic}/status"

    def next_packet_id(self):
        self.packet_id += 1
        if self.packet_id > 65535:
            self.packet_id = 1
        return self.packet_id

    def connect(self, username=None, password=None):
        host = self.cfg["host"]
        port = int(self.cfg.get("port", 1883))
        keepalive = int(self.cfg.get("keepalive", 60))
        client_id = self.cfg.get("client_id", "frigate-camera-control-bridge")

        sock = socket.create_connection((host, port), timeout=10)
        sock.settimeout(1)

        flags = 0x02
        payload = mqtt_encode_string(client_id)

        flags |= 0x04 | 0x20
        payload += mqtt_encode_string(self.availability_topic)
        payload += mqtt_encode_string("offline")

        if username is not None:
            flags |= 0x80
        if password is not None:
            flags |= 0x40

        if username is not None:
            payload += mqtt_encode_string(username)
        if password is not None:
            payload += mqtt_encode_string(password)

        variable_header = mqtt_encode_string("MQTT") + bytes([4, flags]) + struct.pack("!H", keepalive)
        send_mqtt_packet(sock, 0x10, variable_header + payload)
        packet_type, body = read_mqtt_packet(sock)
        if packet_type >> 4 != 2 or len(body) != 2 or body[1] != 0:
            code = body[1] if len(body) >= 2 else "unknown"
            raise RuntimeError(f"MQTT CONNACK failed: {code}")

        discovery_prefix = self.cfg.get("discovery_prefix", "homeassistant")
        topics = [f"{self.base_topic}/+/set", f"{discovery_prefix}/status"]
        packet_id = self.next_packet_id()
        subscribe_payload = struct.pack("!H", packet_id)
        for topic in topics:
            subscribe_payload += mqtt_encode_string(topic) + b"\x00"
        send_mqtt_packet(sock, 0x82, subscribe_payload)

        while True:
            packet_type, body = read_mqtt_packet(sock)
            if packet_type >> 4 != 9:
                continue
            if len(body) < 2 or struct.unpack("!H", body[:2])[0] != packet_id:
                continue
            if any(code == 0x80 for code in body[2:]):
                raise RuntimeError(f"MQTT subscribe rejected for {topics}")
            break

        with self.lock:
            self.sock = sock
        return sock

    def publish(self, topic, payload, retain=False):
        if not isinstance(payload, bytes):
            payload = str(payload).encode("utf-8")
        packet = mqtt_encode_string(topic) + payload
        with self.lock:
            if self.sock is None:
                raise RuntimeError("MQTT is not connected")
            send_mqtt_packet(self.sock, 0x31 if retain else 0x30, packet)

    def close(self):
        with self.lock:
            sock = self.sock
            self.sock = None
        if sock is not None:
            try:
                sock.close()
            except OSError:
                pass

    def serve(self, credentials, on_connect, stop_event):
        backoff = 1
        credential_index = 0
        keepalive = int(self.cfg.get("keepalive", 60))

        while not stop_event.is_set():
            credential = credentials[credential_index % len(credentials)]
            credential_index += 1
            sock = None
            try:
                sock = self.connect(credential.get("username"), credential.get("password"))
                LOG.info("MQTT connected to %s:%s", self.cfg["host"], self.cfg.get("port", 1883))
                on_connect()
                backoff = 1
                last_io = time.monotonic()

                while not stop_event.is_set():
                    try:
                        first, body = read_mqtt_packet(sock)
                    except socket.timeout:
                        if time.monotonic() - last_io > max(10, keepalive / 2):
                            self.publish_ping()
                            last_io = time.monotonic()
                        continue

                    last_io = time.monotonic()
                    packet_type = first >> 4
                    flags = first & 0x0F
                    if packet_type == 3:
                        message = self.parse_publish(flags, body)
                        if message:
                            self.messages.put(message)
                    elif packet_type == 13:
                        continue
                    elif packet_type == 14:
                        raise EOFError("broker disconnected")
            except Exception as exc:
                if not stop_event.is_set():
                    LOG.warning("MQTT failed: %s", exc)
                    stop_event.wait(backoff)
                    backoff = min(backoff * 2, 60)
            finally:
                self.close()
                if sock is not None:
                    try:
                        sock.close()
                    except OSError:
                        pass

    def publish_ping(self):
        with self.lock:
            if self.sock is not None:
                send_mqtt_packet(self.sock, 0xC0)

    def disconnect_cleanly(self):
        try:
            self.publish(self.availability_topic, "offline", retain=True)
        except Exception:
            pass
        with self.lock:
            if self.sock is not None:
                try:
                    send_mqtt_packet(self.sock, 0xE0)
                except OSError:
                    pass

    @staticmethod
    def parse_publish(flags, body):
        if len(body) < 2:
            return None
        topic_len = struct.unpack("!H", body[:2])[0]
        if len(body) < 2 + topic_len:
            return None
        topic = body[2 : 2 + topic_len].decode("utf-8", errors="replace")
        index = 2 + topic_len
        qos = (flags >> 1) & 0x03
        packet_id = None
        if qos:
            packet_id = struct.unpack("!H", body[index : index + 2])[0]
            index += 2
        payload = body[index:].decode("utf-8", errors="replace").strip()
        return topic, payload, packet_id


class CameraControlBridge:
    # CameraControlBridge wires together four boundaries:
    # - Frigate API discovery,
    # - camera vendor HTTP commands,
    # - MQTT/Home Assistant discovery,
    # - manual HTTP fallback GUI.
    def __init__(self, config):
        self.config = config
        self.stop_event = threading.Event()
        self.messages = queue.Queue()
        self.action_lock = threading.RLock()
        self.auto_off_timers = {}
        self.credentials = self.load_mqtt_credentials()
        camera_password = self.load_camera_password()
        camera_user = config.get("camera_auth", {}).get("username", "admin")
        self.default_language = normalize_language(
            config.get("ui", {}).get("default_language", DEFAULT_LANGUAGE)
        )
        self.snapshot_cache = {}
        self.http_client = DigestHttpClient(
            camera_user,
            camera_password,
            timeout=float(config.get("defaults", {}).get("http_timeout", 6)),
        )
        self.inventory = []
        self.actions = self.build_actions()
        self.initialize_action_states()
        self.refresh_snapshots()
        self.mqtt = MqttBridgeClient(config["mqtt"], self.messages)
        self.http_server = None

    @property
    def base_topic(self):
        return self.config["mqtt"].get("base_topic", "frigate_camera_control")

    def actions_snapshot(self):
        with self.action_lock:
            return list(self.actions.values())

    def action_count(self):
        with self.action_lock:
            return len(self.actions)

    def action_by_id(self, action_id):
        with self.action_lock:
            return self.actions.get(action_id)

    def inherit_action_runtime_state(self, old_actions, new_actions):
        for action_id, action in new_actions.items():
            old = old_actions.get(action_id)
            if old is None:
                if action.platform == "switch":
                    action.state = "OFF"
                continue
            action.state = old.state
            action.auto_off_at = old.auto_off_at
            action.snapshot_version = old.snapshot_version

    def initialize_action_states(self):
        for action in self.actions.values():
            if action.platform == "switch" and action.state is None:
                action.state = "OFF"

    def load_mqtt_credentials(self):
        mqtt_cfg = self.config["mqtt"]
        username = mqtt_cfg.get("username")
        password = mqtt_cfg.get("password")
        if not username:
            raise RuntimeError("MQTT username is required")
        if not password:
            raise RuntimeError("MQTT password is required")
        return [{"username": username, "password": password, "source": "environment"}]

    def load_camera_password(self):
        auth = self.config.get("camera_auth", {})
        if auth.get("password"):
            return auth["password"]
        raise RuntimeError("default camera password is required")

    def build_actions(self):
        # Keep this bridge self-maintaining: every service start reads Frigate runtime
        # configs, derives camera hosts, creates safe HA MQTT entities, and later
        # removes stale discovery entities for cameras that disappeared.
        self.inventory = self.discover_inventory()
        present_hosts = {item["host"] for item in self.inventory if item.get("host")}
        action_cfgs = []

        for action_cfg in self.config.get("actions", []):
            required_host = action_cfg.get("required_host")
            if required_host and required_host not in present_hosts:
                LOG.info("skip static action %s, host %s not in Frigate inventory", action_cfg["id"], required_host)
                continue
            action_cfgs.append(action_cfg)

        action_cfgs.extend(self.discover_action_configs())
        actions = {}
        for action_cfg in action_cfgs:
            cfg = dict(action_cfg)
            auto_off_seconds = self.auto_off_seconds_for_config(cfg)
            if auto_off_seconds > 0 and cfg.get("platform") == "switch":
                cfg["auto_off_seconds"] = auto_off_seconds
            if cfg["id"] in actions:
                LOG.warning("duplicate action id skipped: %s", cfg["id"])
                continue
            actions[cfg["id"]] = Action(cfg, self.http_client)
        return actions

    def auto_off_seconds_for_config(self, cfg):
        if "auto_off_seconds" in cfg:
            return int(cfg.get("auto_off_seconds") or 0)
        defaults = self.config.get("defaults", {})
        name_key = cfg.get("name_key")
        if name_key == "action.siren":
            return int(defaults.get("auto_off_siren_seconds", 180))
        if name_key == "action.red_blue":
            return int(defaults.get("auto_off_red_blue_seconds", 300))
        if name_key == "action.manual_white":
            return int(defaults.get("auto_off_white_light_seconds", 600))
        return int(defaults.get("auto_off_default_seconds", 0))

    def discover_inventory(self):
        # Frigate may reference cameras either through direct ffmpeg URLs or
        # through local go2rtc gateway URLs. The selected source URLs are then
        # used to discover the camera API host without hardcoded addresses.
        discovery = self.config.get("discovery", {})
        if not discovery.get("enabled", False):
            return []

        inventory = []
        for frigate in discovery.get("frigates", []):
            name = frigate.get("name") or frigate["host"]
            base_url = frigate.get("url") or f"http://{frigate['host']}:5000"
            data = self.load_frigate_config_with_retry(frigate, base_url)
            if data is None:
                continue

            go2rtc = data.get("go2rtc") or {}
            streams = go2rtc.get("streams") or go2rtc
            cameras = data.get("cameras") or {}
            for camera_name, camera_cfg in cameras.items():
                urls = self.camera_candidate_urls(camera_name, camera_cfg, streams)
                endpoints = self.camera_endpoints_from_urls(urls)
                kinds = self.kinds_from_urls(urls)
                camera_display_name = self.camera_display_name(camera_name, camera_cfg)
                for host in endpoints:
                    inventory.append(
                        {
                            "frigate": name,
                            "frigate_url": base_url,
                            "camera": camera_name,
                            "camera_display_name": camera_display_name,
                            "host": host,
                            "kinds": kinds,
                        }
                    )
        LOG.info("discovered %d camera host mapping(s) from Frigate", len(inventory))
        return inventory

    @staticmethod
    def camera_display_name(camera_name, camera_cfg):
        # Frigate camera keys remain the API identifier. friendly_name is only
        # used for human-facing labels when present.
        friendly_name = camera_cfg.get("friendly_name")
        if friendly_name:
            return str(friendly_name)
        return camera_name

    def load_frigate_config_with_retry(self, frigate, base_url):
        discovery = self.config.get("discovery", {})
        wait = bool(discovery.get("wait_for_frigate", True))
        retry_seconds = float(discovery.get("retry_seconds", 5))
        max_wait = float(discovery.get("startup_wait_seconds", 0))
        started = time.monotonic()
        source = frigate.get("config_file") or f"{base_url.rstrip('/')}/api/config"

        while True:
            try:
                if frigate.get("config_file"):
                    return self.fetch_frigate_config_file(frigate["config_file"])
                return self.fetch_json(f"{base_url.rstrip('/')}/api/config")
            except Exception as exc:
                if not wait:
                    LOG.warning("Frigate discovery failed for %s: %s", source, exc)
                    return None
                elapsed = time.monotonic() - started
                if max_wait and elapsed >= max_wait:
                    LOG.warning("Frigate discovery timed out for %s after %.0fs: %s", source, elapsed, exc)
                    return None
                LOG.info("waiting for Frigate API %s: %s", source, exc)
                time.sleep(retry_seconds)

    @staticmethod
    def fetch_json(url):
        with urllib.request.urlopen(url, timeout=8) as response:
            return json.loads(response.read().decode("utf-8", errors="replace"))

    @staticmethod
    def fetch_frigate_config_file(path):
        try:
            import yaml
        except ImportError as exc:
            raise RuntimeError("PyYAML is required for config_file discovery") from exc
        with open(path, "r", encoding="utf-8") as handle:
            return yaml.safe_load(handle) or {}

    @staticmethod
    def as_list(value):
        if value is None:
            return []
        if isinstance(value, list):
            return value
        return [value]

    @staticmethod
    def camera_candidate_urls(camera_name, camera_cfg, streams):
        # Discovery supports two Frigate layouts:
        # 1. Direct camera URL in ffmpeg, e.g.
        #    rtsp://camera-a.example.test/...
        #    If such URL exists, it is the source of truth.
        # 2. go2rtc gateway path in ffmpeg, e.g. rtsp://127.0.0.1:8554/stream-a.
        #    In this case the loopback URL is only a pointer; the real camera
        #    endpoint must be resolved from go2rtc.streams["stream-a"].
        urls = CameraControlBridge.camera_direct_urls(camera_cfg)
        if urls:
            return urls
        for stream_name in CameraControlBridge.camera_go2rtc_stream_names(camera_cfg):
            resolved = CameraControlBridge.resolve_go2rtc_stream_urls(streams, stream_name)
            if not resolved:
                LOG.info("go2rtc stream %s/%s not found in Frigate config", camera_name, stream_name)
            urls.extend(resolved)
        return urls

    @staticmethod
    def camera_go2rtc_stream_names(camera_cfg):
        # go2rtc-backed Frigate inputs usually look like:
        # rtsp://127.0.0.1:8554/<stream_name>. Loopback here is not a camera API
        # host; it is the go2rtc gateway. We keep only the stream name and
        # resolve it through go2rtc.streams.
        names = []
        for item in camera_cfg.get("ffmpeg", {}).get("inputs", []):
            path = str(item.get("path", ""))
            stream_name = CameraControlBridge.go2rtc_stream_name_from_url(path)
            if stream_name:
                names.append(stream_name)
        return list(dict.fromkeys(names))

    @staticmethod
    def camera_direct_urls(camera_cfg):
        # Some installations skip go2rtc and put camera RTSP/HTTP URLs directly
        # in ffmpeg inputs. Those URLs are inspected as-is. go2rtc loopback URLs
        # are intentionally excluded here because they must be resolved through
        # go2rtc.streams first.
        urls = []
        for item in camera_cfg.get("ffmpeg", {}).get("inputs", []):
            path = str(item.get("path", ""))
            if path and not CameraControlBridge.go2rtc_stream_name_from_url(path):
                urls.extend(CameraControlBridge.stream_source_urls(path))
        return urls

    @staticmethod
    def go2rtc_stream_name_from_url(url):
        # go2rtc and Frigate often wrap real stream URLs, for example:
        # ffmpeg:rtsp://127.0.0.1:8554/front-door#video=copy. Normalize that
        # first so loopback gateway detection works for wrapped and raw values.
        for candidate in CameraControlBridge.stream_source_urls(url) or [str(url)]:
            try:
                parsed = urllib.parse.urlsplit(candidate)
            except ValueError:
                continue
            host = parsed.hostname or ""
            if parsed.scheme != "rtsp" or parsed.port != 8554:
                continue
            if host not in ("127.0.0.1", "localhost", "::1"):
                continue
            return parsed.path.lstrip("/").split("/", 1)[0] or None
        return None

    @staticmethod
    def stream_source_urls(value):
        # go2rtc supports source wrappers such as "ffmpeg:rtsp://..." and may
        # append options after a URL fragment. Extract URL tokens instead of
        # treating the wrapper prefix as the network scheme.
        text = str(value or "")
        urls = STREAM_URL_RE.findall(text)
        if urls:
            return urls
        if "://" in text:
            return [text]
        return []

    @staticmethod
    def resolve_go2rtc_stream_urls(streams, stream_name):
        # go2rtc stream values may be a string, a list of strings, or a small
        # mapping depending on how the Frigate config was authored. Flatten only
        # the values that can contain source URLs; non-URL helper values are
        # harmless because host/vendor extraction is conservative.
        value = streams.get(stream_name) if isinstance(streams, dict) else None
        return CameraControlBridge.flatten_go2rtc_value(value)

    @staticmethod
    def flatten_go2rtc_value(value):
        if value is None:
            return []
        if isinstance(value, str):
            urls = CameraControlBridge.stream_source_urls(value)
            return urls or [value]
        if isinstance(value, list):
            out = []
            for item in value:
                out.extend(CameraControlBridge.flatten_go2rtc_value(item))
            return out
        if isinstance(value, dict):
            out = []
            for key in ("streams", "url", "urls", "source", "sources"):
                if key in value:
                    out.extend(CameraControlBridge.flatten_go2rtc_value(value[key]))
            return out
        return [str(value)]

    @staticmethod
    def hosts_from_urls(urls):
        return CameraControlBridge.camera_endpoints_from_urls(urls)

    @staticmethod
    def camera_endpoints_from_urls(urls):
        # Extract the camera API endpoint from URLs without hardcoding camera IPs.
        # IP literals and DNS names are both valid. Loopback addresses should
        # already have been resolved through go2rtc.streams, so skip them if a
        # config still leaks one through.
        endpoints = []
        for url in urls:
            try:
                parsed = urllib.parse.urlsplit(str(url))
            except ValueError:
                continue
            host = parsed.hostname
            if not host:
                continue
            normalized = host.strip("[]").lower()
            if (
                normalized in ("localhost", "::1", "0.0.0.0", "255.255.255.255")
                or normalized.startswith("127.")
            ):
                continue
            endpoints.append(normalized)
        return list(dict.fromkeys(endpoints))

    @staticmethod
    def kinds_from_urls(urls):
        # Vendor detection is based on URL shape. This decides which safe
        # capability probes may be attempted for each discovered camera.
        kinds = set()
        for url in urls:
            if "Streaming/Channels" in url:
                kinds.add("hikvision")
            elif "cam/realmonitor" in url:
                kinds.add("dahua")
            elif "/stream1" in url or "/stream2" in url:
                kinds.add("tapo")
            elif "camera/stream" in url:
                kinds.add("tablet")
            else:
                kinds.add("other")
        return sorted(kinds)

    def discover_action_configs(self):
        discovery = self.config.get("discovery", {})
        if not discovery.get("enabled", False):
            return []
        actions = []
        for item in self.primary_inventory_items():
            host = item["host"]
            if discovery.get("include_hikvision_colorvu_white", True):
                action = self.discover_hikvision_colorvu_action(item)
                if action:
                    actions.append(action)
            if discovery.get("include_dahua_coaxial_off_probe", True):
                actions.extend(self.discover_dahua_coaxial_actions(item))
        LOG.info("auto-created %d action(s) from camera capabilities", len(actions))
        return actions

    def primary_inventory_items(self):
        grouped = {}
        for item in self.inventory:
            host = item["host"]
            current = grouped.get(host)
            if current is None:
                grouped[host] = item
                continue
            if "ptz" in item["camera"].lower() and "ptz" not in current["camera"].lower():
                grouped[host] = item
        return list(grouped.values())

    def discover_hikvision_colorvu_action(self, item):
        if "hikvision" not in item.get("kinds", []):
            return None
        host = item["host"]
        try:
            _, device_info = self.http_client.request(f"http://{host}/ISAPI/System/deviceInfo")
            model = self.xml_text(device_info, "model") or "Hikvision"
            _, caps = self.http_client.request(f"http://{host}/ISAPI/Image/channels/1/supplementLight/capabilities")
            modes = self.xml_attr(caps, "supplementLightMode", "opt") or ""
            if "colorVuWhiteLight" not in modes.split(","):
                return None
            _, current = self.http_client.request(f"http://{host}/ISAPI/Image/channels/1/supplementLight")
        except Exception as exc:
            LOG.info("skip Hikvision discovery %s/%s: %s", item["camera"], host, exc)
            return None

        restore = {
            "supplementLightMode": self.xml_text(current, "supplementLightMode") or "eventIntelligence",
            "mixedLightBrightnessRegulatMode": self.xml_text(current, "mixedLightBrightnessRegulatMode") or "auto",
            "eventIntelligenceBrightnessRegulatMode": self.xml_text(current, "brightnessRegulatMode") or "auto",
            "whiteLightBrightness": int(self.xml_text(current, "whiteLightBrightness") or 100),
            "irLightBrightness": int(self.xml_text(current, "irLightBrightness") or 100),
            "ircutFilterType": "auto",
        }
        associated = self.xml_text(current, "associatedVMDHuman")
        if associated is not None:
            restore["associatedVMDHuman"] = associated.lower() == "true"

        action_id = f"{slug(item['frigate'])}_{slug(item['camera'])}_colorvu_white"
        lang = self.default_language
        return {
            "id": action_id,
            "name": text_for(lang, "action.manual_white", target=item["camera"]),
            "name_key": "action.manual_white",
            "camera_name": item["camera"],
            "camera_display_name": item.get("camera_display_name", item["camera"]),
            "platform": "switch",
            "kind": "hikvision_colorvu",
            "host": host,
            "brightness": int(self.config.get("defaults", {}).get("colorvu_brightness", 100)),
            "restore": restore,
            "snapshot": self.snapshot_config(item),
            "device": {
                "identifiers": [f"camera_{host.replace('.', '_')}"],
                "name": item.get("camera_display_name", item["camera"]),
                "manufacturer": "Hikvision",
                "model": model,
                "configuration_url": item["frigate_url"],
            },
        }

    def discover_dahua_coaxial_actions(self, item):
        # Dahua red-blue lights and sirens are discovered with OFF-only probes.
        # A successful OFF endpoint means the command exists without starting
        # the alarm/light during bridge startup.
        if "dahua" not in item.get("kinds", []):
            return []
        host = item["host"]
        try:
            _, info = self.http_client.request(f"http://{host}/cgi-bin/magicBox.cgi?action=getSystemInfo")
        except Exception as exc:
            LOG.info("skip Dahua discovery %s/%s: %s", item["camera"], host, exc)
            return []

        model = self.dahua_value(info, "deviceType") or "Dahua"
        actions = []
        red_blue_channel = self.dahua_off_probe(host, action_type=1)
        lang = self.default_language
        if red_blue_channel is not None:
            actions.append(
                {
                    "id": f"{slug(item['frigate'])}_{slug(item['camera'])}_red_blue",
                    "name": text_for(lang, "action.red_blue", target=item["camera"]),
                    "name_key": "action.red_blue",
                    "camera_name": item["camera"],
                    "camera_display_name": item.get("camera_display_name", item["camera"]),
                    "platform": "switch",
                    "kind": "dahua_coaxial_io",
                    "host": host,
                    "on": {"channel": red_blue_channel, "type": 1, "io": 1},
                    "off": {"channel": red_blue_channel, "type": 1, "io": 2},
                    "snapshot": self.snapshot_config(item),
                    "device": self.discovered_device(item, "Dahua", model),
                }
            )

        siren_channel = self.dahua_off_probe(host, action_type=2)
        if siren_channel is not None:
            actions.append(
                {
                    "id": f"{slug(item['frigate'])}_{slug(item['camera'])}_siren",
                    "name": text_for(lang, "action.siren", target=item["camera"]),
                    "name_key": "action.siren",
                    "camera_name": item["camera"],
                    "camera_display_name": item.get("camera_display_name", item["camera"]),
                    "platform": "switch",
                    "kind": "dahua_coaxial_io",
                    "host": host,
                    "on": {"channel": siren_channel, "type": 2, "io": 1},
                    "off": {"channel": siren_channel, "type": 2, "io": 2},
                    "safe_off": {"channel": siren_channel, "type": 2, "io": 2},
                    "requires_ack": True,
                    "snapshot": self.snapshot_config(item),
                    "device": self.discovered_device(item, "Dahua", model),
                }
            )
        return actions

    def dahua_off_probe(self, host, action_type):
        channels = self.config.get("discovery", {}).get("dahua_probe_channels", [1, 2, 0])
        if action_type == 2:
            channels = self.config.get("discovery", {}).get("dahua_siren_probe_channels", [2, 1, 0])
        for channel in channels:
            query = urllib.parse.urlencode(
                {
                    "action": "control",
                    "channel": int(channel),
                    "info[0].Type": int(action_type),
                    "info[0].IO": 2,
                },
                safe="[]",
            )
            try:
                self.http_client.request(f"http://{host}/cgi-bin/coaxialControlIO.cgi?{query}")
                LOG.info("Dahua OFF probe OK host=%s type=%s channel=%s", host, action_type, channel)
                return int(channel)
            except Exception:
                continue
        return None

    @staticmethod
    def dahua_value(text, key):
        match = re.search(rf"^{re.escape(key)}=(.+)$", text, re.MULTILINE)
        return match.group(1).strip().strip("\r") if match else None

    @staticmethod
    def discovered_device(item, manufacturer, model):
        return {
            "identifiers": [f"camera_{item['host'].replace('.', '_')}"],
            "name": item.get("camera_display_name", item["camera"]),
            "manufacturer": manufacturer,
            "model": model,
            "configuration_url": item["frigate_url"],
        }

    def snapshot_config(self, item):
        config = {
            "frigate_url": item["frigate_url"],
            "camera": item["camera"],
            "camera_display_name": item.get("camera_display_name", item["camera"]),
        }
        defaults = self.config.get("defaults", {})
        if defaults.get("frigate_public_url"):
            config["public_url"] = defaults["frigate_public_url"]
        if defaults.get("frigate_camera_path_template"):
            config["camera_path_template"] = defaults["frigate_camera_path_template"]
        return config

    @staticmethod
    def xml_text(xml, tag):
        match = re.search(rf"<(?:[^:>]+:)?{re.escape(tag)}[^>]*>([^<]+)</(?:[^:>]+:)?{re.escape(tag)}>", xml)
        return match.group(1).strip() if match else None

    @staticmethod
    def xml_attr(xml, tag, attr):
        match = re.search(rf"<(?:[^:>]+:)?{re.escape(tag)}[^>]*\s{re.escape(attr)}=\"([^\"]*)\"", xml)
        return match.group(1).strip() if match else None

    def list_actions(self):
        for action in self.actions_snapshot():
            commands = ",".join(action.command_list())
            print(f"{action.id}\t{action.platform}\t{commands}\t{action.display_name(self.default_language)}")

    def check(self):
        ok = True
        for credential in self.credentials:
            try:
                test = MqttBridgeClient(self.config["mqtt"], queue.Queue())
                sock = test.connect(credential.get("username"), credential.get("password"))
                sock.close()
                print(f"MQTT OK: {self.config['mqtt']['host']} as {credential['source']}")
                break
            except Exception as exc:
                ok = False
                print(f"MQTT ERROR using {credential['source']}: {exc}")
        for action in self.actions_snapshot():
            if action.kind == "hikvision_colorvu":
                print(f"CHECK: {action.id} Hikvision host={action.cfg['host']} configured")
            elif action.kind == "dahua_coaxial_io":
                print(f"CHECK: {action.id} Dahua host={action.cfg['host']} configured")
            elif action.kind == "all_off":
                print(f"CHECK: {action.id} all-off button configured")
        return 0 if ok else 1

    def refresh_snapshots(self):
        # Snapshot refresh is best-effort. It improves the manual GUI but should
        # never prevent MQTT controls from starting.
        if not self.config.get("defaults", {}).get("snapshot_enabled", True):
            return True
        seen = set()
        ok = True
        for action in self.actions_snapshot():
            url = self.snapshot_url(action)
            if not url or url in seen:
                continue
            seen.add(url)
            try:
                self.fetch_snapshot(url, force=True)
            except Exception as exc:
                ok = False
                LOG.info("snapshot unavailable for %s: %s", action.id, exc)
        return ok

    @staticmethod
    def should_refresh_snapshot_after_command(action, command):
        if command.lower() not in ("on", "off", "safe_off"):
            return False
        return action.cfg.get("name_key") in ("action.manual_white", "action.red_blue")

    def invalidate_action_snapshot(self, action):
        url = self.snapshot_url(action)
        if not url:
            return
        with self.action_lock:
            self.snapshot_cache.pop(url, None)
            version = int(time.time() * 1000)
            for item in self.actions.values():
                if self.snapshot_url(item) == url:
                    item.snapshot_version = version

    def snapshot_url(self, action):
        snapshot = action.cfg.get("snapshot")
        if not snapshot:
            return None
        camera = urllib.parse.quote(snapshot["camera"], safe="")
        height = int(self.config.get("defaults", {}).get("snapshot_height", 180))
        return f"{snapshot['frigate_url'].rstrip('/')}/api/{camera}/latest.jpg?h={height}"

    def fetch_snapshot(self, url, force=False):
        now = time.monotonic()
        max_age = int(self.config.get("defaults", {}).get("snapshot_cache_seconds", 300))
        cached = self.snapshot_cache.get(url)
        if cached and not force and now - cached["time"] < max_age:
            return cached
        with urllib.request.urlopen(url, timeout=8) as response:
            data = response.read()
            content_type = response.headers.get("Content-Type", "image/jpeg").split(";", 1)[0]
        if not data:
            raise RuntimeError("empty snapshot")
        cached = {"data": data, "content_type": content_type, "time": now}
        self.snapshot_cache[url] = cached
        return cached

    def snapshot_for_action(self, action_id, live=False):
        action = self.action_by_id(action_id)
        if not action:
            return None
        url = self.snapshot_url(action)
        if not url:
            return None
        try:
            return self.fetch_snapshot(url, force=live)
        except Exception:
            cached = self.snapshot_cache.get(url)
            if cached:
                return cached
            raise

    def publish_discovery(self):
        # Home Assistant MQTT discovery topics are retained. The bridge remembers
        # what it published previously and clears stale entities when Frigate no
        # longer exposes a camera/action.
        discovery_prefix = self.config["mqtt"].get("discovery_prefix", "homeassistant")
        availability = f"{self.base_topic}/status"
        self.mqtt.publish(availability, "online", retain=True)
        previous = self.load_discovery_state()
        current = {}
        for action in self.actions_snapshot():
            payload = self.discovery_payload(action, availability)
            topic = f"{discovery_prefix}/{action.platform}/frigate_camera_control/{action.id}/config"
            self.mqtt.publish(topic, json.dumps(payload, separators=(",", ":")), retain=True)
            self.publish_state(action, retain=True)
            current[action.id] = {"platform": action.platform, "topic": topic}
        for action_id, old in previous.items():
            if action_id not in current:
                topic = old.get("topic") or f"{discovery_prefix}/{old.get('platform', 'switch')}/frigate_camera_control/{action_id}/config"
                self.mqtt.publish(topic, "", retain=True)
                LOG.info("removed stale MQTT discovery entity %s", action_id)
        self.save_discovery_state(current)
        LOG.info("published MQTT discovery for %d actions", len(current))

    def publish_state(self, action, retain=True):
        if action.platform == "switch" and action.state in ("ON", "OFF"):
            self.mqtt.publish(f"{self.base_topic}/{action.id}/state", action.state, retain=retain)

    def discovery_state_file(self):
        return Path(
            self.config.get("discovery", {}).get(
                "state_file",
                "/var/lib/frigate-camera-control-bridge/discovery_state.json",
            )
        )

    def load_discovery_state(self):
        path = self.discovery_state_file()
        try:
            return json.loads(path.read_text(encoding="utf-8"))
        except FileNotFoundError:
            return {}
        except Exception as exc:
            LOG.warning("failed to read discovery state %s: %s", path, exc)
            return {}

    def save_discovery_state(self, state):
        path = self.discovery_state_file()
        try:
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_text(json.dumps(state, indent=2), encoding="utf-8")
        except Exception as exc:
            LOG.warning("failed to write discovery state %s: %s", path, exc)

    def discovery_payload(self, action, availability_topic):
        base = f"{self.base_topic}/{action.id}"
        device_cfg = action.cfg.get("device") or {}
        device = {
            "identifiers": device_cfg.get("identifiers", ["frigate_camera_control"]),
            "name": device_cfg.get("name", "Frigate Camera Control"),
        }
        for key in ("manufacturer", "model", "sw_version", "configuration_url"):
            if device_cfg.get(key):
                device[key] = device_cfg[key]
        payload = {
            "name": action.display_name(self.default_language),
            "unique_id": f"frigate_camera_control_{action.id}",
            "availability_topic": availability_topic,
            "payload_available": "online",
            "payload_not_available": "offline",
            "device": device,
            "origin": {
                "name": "frigate-camera-control-bridge",
                "sw_version": VERSION,
            },
            "qos": 0,
        }
        icon = self.home_assistant_icon(action)
        if icon:
            payload["icon"] = icon
        if action.platform == "switch":
            payload.update(
                {
                    "command_topic": f"{base}/set",
                    "state_topic": f"{base}/state",
                    "payload_on": "ON",
                    "payload_off": "OFF",
                    "state_on": "ON",
                    "state_off": "OFF",
                    "optimistic": False,
                    "retain": False,
                }
            )
        elif action.platform == "button":
            payload.update(
                {
                    "command_topic": f"{base}/set",
                    "payload_press": "PRESS",
                    "retain": False,
                }
            )
        return payload

    @staticmethod
    def home_assistant_icon(action):
        # Home Assistant MQTT discovery accepts MDI icons on switches and
        # buttons. Keep icons semantic and overrideable by static action config.
        configured = action.cfg.get("icon")
        if configured:
            return configured
        name_key = action.cfg.get("name_key")
        if name_key in HOME_ASSISTANT_ACTION_ICONS:
            return HOME_ASSISTANT_ACTION_ICONS[name_key]
        if action.kind == "all_off":
            return HOME_ASSISTANT_ACTION_ICONS["action.all_off"]
        return None

    def execute_action(self, action_id, command, publish=True):
        action = self.action_by_id(action_id)
        if action is None:
            raise RuntimeError(f"unknown action {action_id}")
        if command == "on" and self.auto_off_running(action):
            raise RuntimeError("action is already active until automatic turn-off")
        result = action.run(command, bridge=self)
        if action.platform == "switch":
            if command == "on":
                self.schedule_auto_off(action)
            elif command in ("off", "safe_off"):
                self.cancel_auto_off(action.id)
        if self.should_refresh_snapshot_after_command(action, command):
            self.invalidate_action_snapshot(action)
        if publish:
            try:
                self.publish_result(action, command, result)
            except Exception as exc:
                LOG.warning("action %s %s worked, but MQTT publish failed: %s", action_id, command, exc)
        return result

    @staticmethod
    def auto_off_running(action):
        return bool(getattr(action, "auto_off_at", None))

    def schedule_auto_off(self, action):
        seconds = int(action.auto_off_seconds or 0)
        if seconds <= 0:
            self.cancel_auto_off(action.id)
            return
        self.cancel_auto_off(action.id)
        with self.action_lock:
            action.auto_off_at = time.time() + seconds
            auto_off_at = action.auto_off_at
        timer = threading.Timer(seconds, self.auto_off_action, args=(action.id,))
        timer.daemon = True
        with self.action_lock:
            self.auto_off_timers[action.id] = timer
        timer.start()
        LOG.info("scheduled auto-off for %s in %ss at %s", action.id, seconds, int(auto_off_at))

    def cancel_auto_off(self, action_id):
        with self.action_lock:
            timer = self.auto_off_timers.pop(action_id, None)
        if timer is not None:
            timer.cancel()
        action = self.action_by_id(action_id)
        if action is not None:
            with self.action_lock:
                action.auto_off_at = None

    def auto_off_action(self, action_id):
        with self.action_lock:
            self.auto_off_timers.pop(action_id, None)
            action = self.actions.get(action_id)
        if action is None:
            return
        command = action.safe_off_command()
        if not command:
            return
        try:
            LOG.info("auto-off executing %s %s", action_id, command)
            action.run(command, bridge=self)
            action.auto_off_at = None
            self.publish_result(action, command, {"ok": True, "auto_off": True})
        except Exception as exc:
            LOG.error("auto-off failed for %s: %s", action_id, exc)
            self.publish_error(action_id, command, exc)

    def publish_result(self, action, command, result):
        base = f"{self.base_topic}/{action.id}"
        self.publish_state(action, retain=True)
        self.mqtt.publish(
            f"{base}/result",
            json.dumps({"ok": True, "command": command, "time": int(time.time())}),
            retain=False,
        )

    def publish_error(self, action_id, command, exc):
        topic = f"{self.base_topic}/{action_id}/result"
        payload = {"ok": False, "command": command, "error": str(exc), "time": int(time.time())}
        try:
            self.mqtt.publish(topic, json.dumps(payload), retain=False)
        except Exception:
            pass

    def all_off(self):
        results = []
        for action in self.actions_snapshot():
            command = action.safe_off_command()
            if not command:
                continue
            try:
                action.run(command, bridge=self)
                self.cancel_auto_off(action.id)
                results.append({"action": action.id, "ok": True})
                if action.platform == "switch":
                    try:
                        self.publish_state(action, retain=True)
                    except Exception:
                        pass
            except Exception as exc:
                results.append({"action": action.id, "ok": False, "error": str(exc)})
                LOG.error("all_off failed for %s: %s", action.id, exc)
        return {"ok": all(item["ok"] for item in results), "results": results}

    def startup_all_off(self):
        if not self.config.get("defaults", {}).get("startup_all_off", True):
            return
        LOG.info("running startup all-off")
        result = self.all_off()
        if not result["ok"]:
            LOG.warning("startup all-off completed with errors")

    def refresh_frigate_inventory(self):
        LOG.info("refreshing Frigate camera inventory")
        with self.action_lock:
            old_actions = dict(self.actions)
        new_actions = self.build_actions()
        with self.action_lock:
            removed_ids = set(old_actions) - set(new_actions)
            for action_id in removed_ids:
                timer = self.auto_off_timers.pop(action_id, None)
                if timer is not None:
                    timer.cancel()
            self.inherit_action_runtime_state(old_actions, new_actions)
            self.actions = new_actions
        self.refresh_snapshots()
        old_ids = set(old_actions)
        new_ids = set(new_actions)
        old_cameras = self.camera_keys_from_actions(old_actions.values())
        new_cameras = self.camera_keys_from_actions(new_actions.values())
        result = {
            "ok": True,
            "added_actions": len(new_ids - old_ids),
            "removed_actions": len(old_ids - new_ids),
            "added_cameras": len(new_cameras - old_cameras),
            "removed_cameras": len(old_cameras - new_cameras),
            "total_actions": len(new_ids),
            "total_cameras": len(new_cameras),
            "mqtt_published": True,
        }
        result["changed"] = any(
            result[key] > 0
            for key in ("added_actions", "removed_actions", "added_cameras", "removed_cameras")
        )
        try:
            self.publish_discovery()
        except Exception as exc:
            result["mqtt_published"] = False
            LOG.warning("Frigate refresh could not publish MQTT discovery: %s", exc)
        return result

    @staticmethod
    def camera_keys_from_actions(actions):
        keys = set()
        for action in actions:
            if action.kind == "all_off":
                continue
            snapshot = action.cfg.get("snapshot") or {}
            camera = snapshot.get("camera") or action.cfg.get("camera_name")
            if camera:
                keys.add((snapshot.get("frigate_url"), camera))
        return keys

    @staticmethod
    def refresh_message(lang, result):
        if result.get("added_cameras") or result.get("removed_cameras"):
            return text_for(
                lang,
                "refresh.cameras_changed",
                added=result.get("added_cameras", 0),
                removed=result.get("removed_cameras", 0),
            )
        if result.get("added_actions") or result.get("removed_actions"):
            return text_for(
                lang,
                "refresh.controls_changed",
                added=result.get("added_actions", 0),
                removed=result.get("removed_actions", 0),
            )
        return text_for(lang, "refresh.no_changes")

    def refresh_payload(self, lang, result, redirect_url):
        payload = dict(result)
        payload["message"] = self.refresh_message(lang, result)
        payload["redirect_url"] = redirect_url
        return payload

    def frigate_refresh_loop(self):
        interval = int(self.config.get("defaults", {}).get("frigate_refresh_seconds", 300))
        if interval <= 0:
            LOG.info("Frigate periodic refresh disabled")
            return
        interval = max(60, interval)
        while not self.stop_event.wait(interval):
            try:
                self.refresh_frigate_inventory()
            except Exception as exc:
                LOG.warning("Frigate periodic refresh failed: %s", exc)

    def handle_mqtt_message(self, topic, payload):
        discovery_prefix = self.config["mqtt"].get("discovery_prefix", "homeassistant")
        if topic == f"{discovery_prefix}/status" and payload.lower() == "online":
            self.publish_discovery()
            return
        prefix = f"{self.base_topic}/"
        if not topic.startswith(prefix) or not topic.endswith("/set"):
            return
        action_id = topic[len(prefix) : -len("/set")]
        action = self.action_by_id(action_id)
        if not action:
            LOG.warning("MQTT command for unknown action %s", action_id)
            return
        value = payload.upper()
        if action.platform == "button":
            command = "press"
        elif value == "ON":
            command = "on"
        elif value == "OFF":
            command = "off"
        else:
            LOG.warning("unsupported MQTT payload for %s: %s", action_id, payload)
            return
        try:
            LOG.info("MQTT command %s %s", action_id, command)
            self.execute_action(action_id, command, publish=True)
        except Exception as exc:
            LOG.error("MQTT action failed %s %s: %s", action_id, command, exc)
            self.publish_error(action_id, command, exc)

    def serve(self):
        signal.signal(signal.SIGTERM, self._handle_signal)
        signal.signal(signal.SIGINT, self._handle_signal)
        self.startup_all_off()
        mqtt_thread = threading.Thread(
            target=self.mqtt.serve,
            args=(self.credentials, self.publish_discovery, self.stop_event),
            daemon=True,
        )
        mqtt_thread.start()
        snapshot_thread = None
        if self.config.get("defaults", {}).get("snapshot_enabled", True):
            snapshot_thread = threading.Thread(target=self.snapshot_refresh_loop, daemon=True)
            snapshot_thread.start()
        refresh_thread = threading.Thread(target=self.frigate_refresh_loop, daemon=True)
        refresh_thread.start()
        http_cfg = self.config.get("http", {})
        if http_cfg.get("enabled", True):
            self.start_http(http_cfg.get("bind", "0.0.0.0"), int(http_cfg.get("port", 5011)))
        LOG.info("serving %d actions", self.action_count())

        while not self.stop_event.is_set():
            try:
                topic, payload, packet_id = self.messages.get(timeout=1)
                self.handle_mqtt_message(topic, payload)
            except queue.Empty:
                pass

        self.mqtt.disconnect_cleanly()
        if self.http_server is not None:
            self.http_server.shutdown()
        mqtt_thread.join(timeout=2)
        if snapshot_thread is not None:
            snapshot_thread.join(timeout=2)
        refresh_thread.join(timeout=2)

    def snapshot_refresh_loop(self):
        retry_interval = max(60, int(self.config.get("defaults", {}).get("snapshot_retry_seconds", 600)))
        refresh_interval = max(retry_interval, int(self.config.get("defaults", {}).get("snapshot_refresh_seconds", 3600)))
        interval = retry_interval if self.missing_snapshots() else refresh_interval
        while not self.stop_event.wait(interval):
            all_ok = self.refresh_snapshots()
            interval = refresh_interval if all_ok and not self.missing_snapshots() else retry_interval

    def missing_snapshots(self):
        for action in self.actions_snapshot():
            url = self.snapshot_url(action)
            if url and url not in self.snapshot_cache:
                return True
        return False

    def http_token(self):
        return self.config.get("http", {}).get("token")

    def http_authorized(self, headers, query):
        token = self.http_token()
        if not token:
            return True
        values = urllib.parse.parse_qs(query)
        supplied = values.get("token", [None])[0] or headers.get("X-Bridge-Token")
        return supplied == token

    def start_http(self, bind, port):
        bridge = self

        class Handler(BaseHTTPRequestHandler):
            def do_GET(self):
                parsed = urllib.parse.urlsplit(self.path)
                lang = bridge.request_language(parsed.query, self.headers)
                if parsed.path in ("/healthz", "/readyz"):
                    self.send_json({"ok": True, "actions": bridge.action_count()})
                    return
                if parsed.path.startswith("/static/"):
                    name = parsed.path.rsplit("/", 1)[-1]
                    payload = read_static_file(name)
                    if payload is None:
                        self.send_response(404)
                        self.end_headers()
                        return
                    content_type = "text/css" if name.endswith(".css") else "application/javascript"
                    self.send_bytes(payload, content_type, cache_seconds=3600)
                    return
                if not bridge.http_authorized(self.headers, parsed.query):
                    self.send_response(401)
                    self.end_headers()
                    return
                if parsed.path.startswith("/snapshot/"):
                    action_id = urllib.parse.unquote(parsed.path.rsplit("/", 1)[-1]).removesuffix(".jpg")
                    try:
                        values = urllib.parse.parse_qs(parsed.query)
                        refresh = values.get("refresh", ["0"])[0].lower() in ("1", "true", "yes", "on")
                        live = refresh or bool(bridge.config.get("defaults", {}).get("snapshot_live_on_gui", True))
                        snapshot = bridge.snapshot_for_action(action_id, live=live)
                    except Exception as exc:
                        LOG.info("snapshot request failed for %s: %s", action_id, exc)
                        snapshot = None
                    if not snapshot:
                        self.send_response(404)
                        self.end_headers()
                        return
                    self.send_bytes(snapshot["data"], snapshot["content_type"])
                    return
                if parsed.path == "/api/actions":
                    self.send_json(bridge.api_actions())
                    return
                token_param = urllib.parse.parse_qs(parsed.query).get("token", [None])[0]
                self.send_html(bridge.html_page(lang, token_param))

            def do_POST(self):
                parsed = urllib.parse.urlsplit(self.path)
                lang = bridge.request_language(parsed.query, self.headers)
                if not bridge.http_authorized(self.headers, parsed.query):
                    self.send_response(401)
                    self.end_headers()
                    return
                if parsed.path == "/refresh":
                    token_param = urllib.parse.parse_qs(parsed.query).get("token", [None])[0]
                    token_query = f"&token={urllib.parse.quote(token_param)}" if token_param else ""
                    redirect_url = f"/?lang={urllib.parse.quote(lang)}{token_query}"
                    wants_json = "application/json" in self.headers.get("Accept", "").lower()
                    try:
                        result = bridge.refresh_frigate_inventory()
                        if wants_json:
                            self.send_json(bridge.refresh_payload(lang, result, redirect_url))
                        else:
                            self.send_response(303)
                            self.send_header("Location", redirect_url)
                            self.end_headers()
                    except Exception as exc:
                        if wants_json:
                            self.send_json(
                                {
                                    "ok": False,
                                    "changed": False,
                                    "message": text_for(lang, "refresh.failed", error=str(exc)),
                                    "redirect_url": redirect_url,
                                },
                                status=500,
                            )
                        else:
                            self.send_response(500)
                            self.end_headers()
                            self.wfile.write(str(exc).encode("utf-8"))
                    return
                parts = parsed.path.strip("/").split("/")
                if len(parts) == 3 and parts[0] == "action":
                    action_id = urllib.parse.unquote(parts[1])
                    command = urllib.parse.unquote(parts[2]).lower()
                    try:
                        result = bridge.execute_action(action_id, command, publish=True)
                        token_param = urllib.parse.parse_qs(parsed.query).get("token", [None])[0]
                        token_query = f"&token={urllib.parse.quote(token_param)}" if token_param else ""
                        self.send_response(303)
                        self.send_header("Location", f"/?lang={urllib.parse.quote(lang)}{token_query}")
                        self.end_headers()
                    except Exception as exc:
                        self.send_response(500)
                        self.end_headers()
                        self.wfile.write(str(exc).encode("utf-8"))
                    return
                self.send_response(404)
                self.end_headers()

            def send_json(self, payload, status=200):
                data = json.dumps(payload, indent=2).encode("utf-8")
                self.send_response(status)
                self.send_header("Content-Type", "application/json")
                self.send_header("Content-Length", str(len(data)))
                self.end_headers()
                self.wfile.write(data)

            def send_html(self, body):
                data = body.encode("utf-8")
                self.send_response(200)
                self.send_header("Content-Type", "text/html; charset=utf-8")
                self.send_header("Content-Length", str(len(data)))
                self.end_headers()
                self.wfile.write(data)

            def send_bytes(self, data, content_type, cache_seconds=300):
                self.send_response(200)
                self.send_header("Content-Type", content_type)
                self.send_header("Cache-Control", f"private, max-age={cache_seconds}")
                self.send_header("Content-Length", str(len(data)))
                self.end_headers()
                self.wfile.write(data)

            def log_message(self, fmt, *args):
                LOG.info("http %s - %s", self.address_string(), fmt % args)

        self.http_server = ThreadingHTTPServer((bind, port), Handler)
        thread = threading.Thread(target=self.http_server.serve_forever, daemon=True)
        thread.start()
        LOG.info("HTTP fallback GUI listening on %s:%s", bind, port)

    def api_actions(self):
        return [
            {
                "id": action.id,
                "name": action.display_name(self.default_language),
                "platform": action.platform,
                "commands": action.command_list(),
                "state": action.state,
                "snapshot": bool(action.cfg.get("snapshot")),
                "auto_off_seconds": action.auto_off_seconds,
                "auto_off_at": int(action.auto_off_at) if action.auto_off_at else None,
                "requires_ack": bool(action.cfg.get("requires_ack")),
            }
            for action in self.actions_snapshot()
        ]

    def html_page(self, lang=None, token_param=None):
        with self.action_lock:
            actions = list(self.actions.values())
        return render_page(actions, lang or DEFAULT_LANGUAGE, token_param)

    def request_language(self, query, headers=None):
        values = urllib.parse.parse_qs(query)
        requested = values.get("lang", [None])[0]
        if requested:
            return normalize_language(requested)
        accept_language = headers.get("Accept-Language") if headers else None
        return language_from_accept_header(accept_language)

    def _handle_signal(self, signum, _frame):
        LOG.info("received signal %s, stopping", signum)
        self.stop_event.set()