mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-04 06:15:09 +00:00
ChuckandClaude Opus 5.5
e3c85cece6
Optional web login, off by default: a device that sets no password behaves exactly as before. Set under General > Security; then every page and API route needs a session login or an API token (Authorization: Bearer). Loopback, the Wi-Fi setup flow in AP mode, static files, captive-portal probes and a reduced /api/v3/health stay open. Secrets live in the web_auth section of config_secrets.json and no API returns them. scripts/reset_web_password.py turns login off. Stacked on #674. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
557 lines
23 KiB
Python
557 lines
23 KiB
Python
#!/usr/bin/env python3
|
|
"""Control a LEDMatrix display from Home Assistant over MQTT.
|
|
|
|
The bridge owns no display logic. It subscribes to one command topic and
|
|
turns each message into a call against the same api_v3 routes the web UI
|
|
uses, so behaviour stays in one place and this stays a translation layer.
|
|
|
|
On connect it publishes Home Assistant MQTT Discovery config, so a matrix
|
|
appears in HA as real entities rather than something you drive with
|
|
`mqtt.publish` by hand:
|
|
|
|
select.ledmatrix_display_mode every mode across enabled plugins;
|
|
choosing one force-displays it
|
|
button.ledmatrix_stop_display back to normal rotation
|
|
switch.ledmatrix_power the display service, on or off
|
|
number.ledmatrix_brightness 0-100
|
|
|
|
Anything the entities do not cover is still reachable by publishing JSON
|
|
to the command topic:
|
|
|
|
{"action": "display", "mode": "nfl_live"}
|
|
{"action": "display", "plugin_id": "starlark-apps", "mode": "aquarium",
|
|
"duration": 300, "pinned": true}
|
|
{"action": "stop_display"}
|
|
{"action": "power", "state": "on" | "off"}
|
|
{"action": "brightness", "value": 75}
|
|
{"action": "refresh"} re-publish discovery after installing a plugin
|
|
|
|
Every command publishes its result to <command_topic>/status.
|
|
|
|
Run it with `python3 ledmatrix_mqtt_bridge.py [--config PATH]`, or install
|
|
ledmatrix-mqtt-bridge.service.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import logging
|
|
import os
|
|
import signal
|
|
import sys
|
|
import threading
|
|
from typing import Any, Callable, Dict, List, Optional
|
|
|
|
import requests
|
|
|
|
logger = logging.getLogger("ledmatrix-mqtt-bridge")
|
|
|
|
DISCOVERY_PREFIX = "homeassistant"
|
|
DEVICE_ID = "ledmatrix"
|
|
DEVICE_INFO = {
|
|
"identifiers": [DEVICE_ID],
|
|
"name": "LEDMatrix",
|
|
"manufacturer": "ChuckBuilds",
|
|
"model": "LEDMatrix Display",
|
|
}
|
|
|
|
DEFAULTS = {
|
|
"mqtt_host": "localhost",
|
|
"mqtt_port": 1883,
|
|
"mqtt_username": None,
|
|
"mqtt_password": None, # nosec B105 - "no password configured", not a credential
|
|
"mqtt_client_id": "ledmatrix-mqtt-bridge",
|
|
"mqtt_topic": "ledmatrix/command",
|
|
"mqtt_tls": False,
|
|
"mqtt_tls_insecure": False,
|
|
"ledmatrix_api_base": "http://localhost:5000",
|
|
# Only needed when the web interface's optional login is on AND the bridge
|
|
# reaches it from another machine: requests from the Pi itself never need
|
|
# one. Create it under General > Security; sent as a Bearer token.
|
|
"ledmatrix_api_token": None,
|
|
"request_timeout": 15,
|
|
"on_demand_duration": None,
|
|
"log_level": "INFO",
|
|
}
|
|
|
|
|
|
class ConfigError(Exception):
|
|
"""The bridge cannot start with the configuration it was given."""
|
|
|
|
|
|
def load_config(path: str) -> Dict[str, Any]:
|
|
"""Read bridge_config.json, overlaid on DEFAULTS.
|
|
|
|
Every value may also come from the environment as LEDMATRIX_MQTT_<KEY>,
|
|
which is how a password stays out of a file that has to be world-readable
|
|
for the service user.
|
|
"""
|
|
config = dict(DEFAULTS)
|
|
if os.path.isfile(path):
|
|
with open(path, encoding="utf-8") as handle:
|
|
try:
|
|
loaded = json.load(handle)
|
|
except json.JSONDecodeError as err:
|
|
raise ConfigError(f"{path} is not valid JSON: {err}") from err
|
|
if not isinstance(loaded, dict):
|
|
raise ConfigError(f"{path} must contain a JSON object")
|
|
config.update(loaded)
|
|
else:
|
|
logger.warning("No config file at %s - using defaults and environment", path)
|
|
|
|
for key in DEFAULTS:
|
|
env_value = os.environ.get(f"LEDMATRIX_MQTT_{key.upper()}")
|
|
if env_value is not None:
|
|
config[key] = env_value
|
|
|
|
for key in ("mqtt_port", "request_timeout"):
|
|
try:
|
|
config[key] = int(config[key])
|
|
except (TypeError, ValueError) as err:
|
|
raise ConfigError(f"{key} must be a whole number, got {config[key]!r}") from err
|
|
for key in ("mqtt_tls", "mqtt_tls_insecure"):
|
|
config[key] = str(config[key]).lower() in ("1", "true", "yes", "on")
|
|
|
|
if config.get("mqtt_password") == "REPLACE_WITH_YOUR_ACTUAL_MQTT_PASSWORD":
|
|
raise ConfigError(
|
|
"mqtt_password is still the example placeholder - set a real password, "
|
|
"or remove the key if your broker allows anonymous connections")
|
|
return config
|
|
|
|
|
|
class LEDMatrixClient:
|
|
"""The api_v3 calls the bridge needs, and nothing else.
|
|
|
|
Everything goes through the HTTP API rather than the filesystem, so the
|
|
bridge does not have to live on the Pi, does not need read access to
|
|
config.json, and cannot drift from the web UI's own behaviour.
|
|
"""
|
|
|
|
def __init__(self, api_base: str, timeout: int = 15,
|
|
session: Optional[requests.Session] = None,
|
|
api_token: Optional[str] = None):
|
|
self.api_base = api_base.rstrip("/")
|
|
self.timeout = timeout
|
|
self.session = session or requests.Session()
|
|
if api_token:
|
|
self.session.headers["Authorization"] = f"Bearer {api_token}"
|
|
|
|
def _call(self, method: str, path: str, **kwargs) -> Dict[str, Any]:
|
|
url = f"{self.api_base}/api/v3{path}"
|
|
response = self.session.request(method, url, timeout=self.timeout, **kwargs)
|
|
try:
|
|
body = response.json()
|
|
except ValueError:
|
|
body = {}
|
|
if response.status_code >= 400 or body.get("status") == "error":
|
|
message = body.get("message") or f"HTTP {response.status_code}"
|
|
raise RuntimeError(f"{method} {path} failed: {message}")
|
|
return body.get("data", body)
|
|
|
|
def list_modes(self) -> List[Dict[str, Any]]:
|
|
"""Every display mode that can be force-displayed, newest discovery.
|
|
|
|
/display/modes triggers plugin discovery itself, which matters because
|
|
discovery is lazy: a bridge that never opens the dashboard would
|
|
otherwise see nothing at all.
|
|
"""
|
|
return self._call("GET", "/display/modes").get("modes", [])
|
|
|
|
def display_status(self) -> Dict[str, Any]:
|
|
return self._call("GET", "/display/on-demand/status")
|
|
|
|
def start_on_demand(self, mode: str, plugin_id: Optional[str] = None,
|
|
duration: Optional[int] = None, pinned: bool = False) -> Dict[str, Any]:
|
|
payload: Dict[str, Any] = {"mode": mode, "pinned": pinned}
|
|
if plugin_id:
|
|
# find_plugin_for_mode only sees modes declared in a static
|
|
# manifest, so a plugin whose modes are generated -- each installed
|
|
# Starlark app is one -- 404s when plugin_id is omitted. Sending it
|
|
# skips that lookup. /display/modes reports it for every mode.
|
|
payload["plugin_id"] = plugin_id
|
|
if duration:
|
|
payload["duration"] = int(duration)
|
|
return self._call("POST", "/display/on-demand/start", json=payload)
|
|
|
|
def stop_on_demand(self) -> Dict[str, Any]:
|
|
return self._call("POST", "/display/on-demand/stop", json={})
|
|
|
|
def set_power(self, on: bool) -> Dict[str, Any]:
|
|
action = "start_display" if on else "stop_display"
|
|
return self._call("POST", "/system/action", json={"action": action})
|
|
|
|
def get_brightness(self) -> Optional[int]:
|
|
config = self._call("GET", "/config/main")
|
|
value = config.get("display", {}).get("hardware", {}).get("brightness")
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
def set_brightness(self, value: int) -> Dict[str, Any]:
|
|
return self._call("POST", "/config/main", json={"brightness": int(value)})
|
|
|
|
|
|
class CommandHandler:
|
|
"""Turns one decoded MQTT payload into one API call.
|
|
|
|
Kept free of MQTT so it can be tested against a fake client: the failure
|
|
modes worth pinning are all in here (an unknown mode, an out-of-range
|
|
brightness, a mode name that needs its plugin_id attached).
|
|
"""
|
|
|
|
def __init__(self, client: LEDMatrixClient, default_duration: Optional[int] = None):
|
|
self.client = client
|
|
self.default_duration = default_duration
|
|
self._modes_by_name: Dict[str, Dict[str, Any]] = {}
|
|
|
|
def refresh_modes(self) -> List[Dict[str, Any]]:
|
|
modes = self.client.list_modes()
|
|
self._modes_by_name = {m["mode"]: m for m in modes}
|
|
# Home Assistant's select shows labels, so accept them back as well --
|
|
# otherwise picking "Simple Clock" in a dashboard is not a mode name.
|
|
for entry in modes:
|
|
self._modes_by_name.setdefault(entry.get("name") or entry["mode"], entry)
|
|
return modes
|
|
|
|
@property
|
|
def known_modes(self) -> List[Dict[str, Any]]:
|
|
return list({id(v): v for v in self._modes_by_name.values()}.values())
|
|
|
|
def handle(self, payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
action = payload.get("action")
|
|
handlers: Dict[str, Callable[[Dict[str, Any]], Dict[str, Any]]] = {
|
|
"display": self._display,
|
|
"stop_display": lambda _p: self._ok(self.client.stop_on_demand()),
|
|
"power": self._power,
|
|
"brightness": self._brightness,
|
|
"refresh": lambda _p: self._ok({"modes": len(self.refresh_modes())}),
|
|
}
|
|
handler = handlers.get(action)
|
|
if handler is None:
|
|
return self._error(f"Unknown action {action!r}; expected one of "
|
|
f"{', '.join(sorted(handlers))}")
|
|
try:
|
|
return handler(payload)
|
|
except (requests.RequestException, RuntimeError) as err:
|
|
logger.error("Command %s failed: %s", action, err)
|
|
return self._error(str(err))
|
|
|
|
def _display(self, payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
mode = payload.get("mode")
|
|
plugin_id = payload.get("plugin_id")
|
|
if not mode and not plugin_id:
|
|
return self._error("display requires 'mode' or 'plugin_id'")
|
|
|
|
known = self._modes_by_name.get(mode) if mode else None
|
|
if known is None and mode and not plugin_id:
|
|
# One retry against a fresh listing: a plugin installed since the
|
|
# last refresh is the common reason a valid mode looks unknown.
|
|
self.refresh_modes()
|
|
known = self._modes_by_name.get(mode)
|
|
if known is not None:
|
|
mode = known["mode"]
|
|
plugin_id = plugin_id or known.get("plugin_id")
|
|
|
|
duration = payload.get("duration", self.default_duration)
|
|
pinned = bool(payload.get("pinned", False))
|
|
result = self.client.start_on_demand(
|
|
mode=mode, plugin_id=plugin_id, duration=duration, pinned=pinned)
|
|
return self._ok(result, mode=mode, plugin_id=plugin_id)
|
|
|
|
def _power(self, payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
state = str(payload.get("state", "")).strip().lower()
|
|
if state not in ("on", "off"):
|
|
return self._error("power requires 'state' of 'on' or 'off'")
|
|
return self._ok(self.client.set_power(state == "on"), state=state)
|
|
|
|
def _brightness(self, payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
raw = payload.get("value")
|
|
try:
|
|
value = int(float(raw))
|
|
except (TypeError, ValueError):
|
|
return self._error(f"brightness requires a number, got {raw!r}")
|
|
if not 0 <= value <= 100:
|
|
return self._error(f"brightness must be between 0 and 100, got {value}")
|
|
return self._ok(self.client.set_brightness(value), value=value)
|
|
|
|
@staticmethod
|
|
def _ok(result: Any, **extra) -> Dict[str, Any]:
|
|
return {"status": "success", "result": result, **extra}
|
|
|
|
@staticmethod
|
|
def _error(message: str) -> Dict[str, Any]:
|
|
return {"status": "error", "message": message}
|
|
|
|
|
|
def discovery_messages(command_topic: str, state_topic: str, availability_topic: str,
|
|
mode_labels: List[str]) -> List[Dict[str, Any]]:
|
|
"""The retained MQTT Discovery configs, as {topic, payload} pairs.
|
|
|
|
Pure, so the entity shapes can be asserted without a broker. Every entity
|
|
shares one availability topic, which is also the bridge's last will -- HA
|
|
then shows the matrix as unavailable when the bridge dies, instead of
|
|
leaving stale controls that silently do nothing.
|
|
"""
|
|
common = {
|
|
"device": DEVICE_INFO,
|
|
"availability_topic": availability_topic,
|
|
"payload_available": "online",
|
|
"payload_not_available": "offline",
|
|
}
|
|
return [
|
|
{
|
|
"topic": f"{DISCOVERY_PREFIX}/select/{DEVICE_ID}/display_mode/config",
|
|
"payload": {
|
|
**common,
|
|
"name": "Display Mode",
|
|
"unique_id": f"{DEVICE_ID}_display_mode",
|
|
"command_topic": command_topic,
|
|
"command_template": '{"action": "display", "mode": "{{ value }}"}',
|
|
"state_topic": state_topic,
|
|
"value_template": "{{ value_json.mode }}",
|
|
"options": mode_labels,
|
|
"icon": "mdi:view-dashboard",
|
|
},
|
|
},
|
|
{
|
|
"topic": f"{DISCOVERY_PREFIX}/button/{DEVICE_ID}/stop_display/config",
|
|
"payload": {
|
|
**common,
|
|
"name": "Stop Display",
|
|
"unique_id": f"{DEVICE_ID}_stop_display",
|
|
"command_topic": command_topic,
|
|
"payload_press": '{"action": "stop_display"}',
|
|
"icon": "mdi:stop",
|
|
},
|
|
},
|
|
{
|
|
"topic": f"{DISCOVERY_PREFIX}/switch/{DEVICE_ID}/power/config",
|
|
"payload": {
|
|
**common,
|
|
"name": "Power",
|
|
"unique_id": f"{DEVICE_ID}_power",
|
|
"command_topic": command_topic,
|
|
"payload_on": '{"action": "power", "state": "on"}',
|
|
"payload_off": '{"action": "power", "state": "off"}',
|
|
"state_topic": state_topic,
|
|
"value_template": "{{ 'ON' if value_json.power else 'OFF' }}",
|
|
"state_on": "ON",
|
|
"state_off": "OFF",
|
|
"icon": "mdi:power",
|
|
},
|
|
},
|
|
{
|
|
"topic": f"{DISCOVERY_PREFIX}/number/{DEVICE_ID}/brightness/config",
|
|
"payload": {
|
|
**common,
|
|
"name": "Brightness",
|
|
"unique_id": f"{DEVICE_ID}_brightness",
|
|
"command_topic": command_topic,
|
|
"command_template": '{"action": "brightness", "value": {{ value }}}',
|
|
"state_topic": state_topic,
|
|
"value_template": "{{ value_json.brightness }}",
|
|
"min": 0,
|
|
"max": 100,
|
|
"step": 1,
|
|
"icon": "mdi:brightness-6",
|
|
},
|
|
},
|
|
]
|
|
|
|
|
|
def warn_if_cleartext(config: Dict[str, Any]) -> bool:
|
|
"""Say so, once, when a broker password is going over an unencrypted link.
|
|
|
|
The shipped example has TLS on, so reaching here means somebody turned it
|
|
off deliberately -- which is legitimate (the Mosquitto add-on is plaintext
|
|
on 1883) but should not be silent when there is a password to lose. Returns
|
|
whether it warned, so the decision is testable without a broker.
|
|
"""
|
|
if config.get("mqtt_tls") or not config.get("mqtt_password"):
|
|
return False
|
|
logger.warning(
|
|
'mqtt_tls is off and a password is set: the broker password and every '
|
|
'command are sent unencrypted. Set "mqtt_tls": true (port 8883 on most '
|
|
'brokers) unless this is a trusted, isolated network.')
|
|
return True
|
|
|
|
|
|
def read_state(client: LEDMatrixClient) -> Dict[str, Any]:
|
|
"""The state every entity reads, so HA opens on real values.
|
|
|
|
Each field is fetched independently: a matrix with its display service
|
|
stopped still has a brightness worth showing, and one unreachable field
|
|
should not blank the rest.
|
|
"""
|
|
state: Dict[str, Any] = {"power": False, "mode": None, "brightness": None}
|
|
try:
|
|
status = client.display_status()
|
|
state["power"] = bool(status.get("service", {}).get("active"))
|
|
on_demand = status.get("state", {})
|
|
if on_demand.get("active"):
|
|
state["mode"] = on_demand.get("mode")
|
|
except (requests.RequestException, RuntimeError) as err:
|
|
logger.debug("Could not read display status: %s", err)
|
|
try:
|
|
state["brightness"] = client.get_brightness()
|
|
except (requests.RequestException, RuntimeError) as err:
|
|
logger.debug("Could not read brightness: %s", err)
|
|
return state
|
|
|
|
|
|
class Bridge:
|
|
"""MQTT wiring around CommandHandler."""
|
|
|
|
def __init__(self, config: Dict[str, Any]):
|
|
self.config = config
|
|
self.command_topic = config["mqtt_topic"]
|
|
self.status_topic = f"{self.command_topic}/status"
|
|
self.state_topic = f"{self.command_topic}/state"
|
|
self.availability_topic = f"{self.command_topic}/availability"
|
|
self.client = LEDMatrixClient(config["ledmatrix_api_base"], config["request_timeout"],
|
|
api_token=config.get("ledmatrix_api_token") or None)
|
|
self.handler = CommandHandler(self.client, config.get("on_demand_duration"))
|
|
self._stop = threading.Event()
|
|
self._mqtt = None
|
|
|
|
# -- MQTT callbacks (paho-mqtt 2.x VERSION2 signatures) ------------------
|
|
|
|
def _on_connect(self, client, _userdata, _flags, reason_code, _properties=None):
|
|
if getattr(reason_code, "is_failure", reason_code != 0):
|
|
logger.error("MQTT connection refused: %s", reason_code)
|
|
return
|
|
logger.info("Connected to MQTT broker; subscribing to %s", self.command_topic)
|
|
client.subscribe(self.command_topic, qos=1)
|
|
client.publish(self.availability_topic, "online", qos=1, retain=True)
|
|
# Re-publish on every reconnect, not just the first connect: a broker
|
|
# restart drops retained discovery configs, and HA would otherwise be
|
|
# left with entities it can no longer describe.
|
|
self.publish_discovery()
|
|
self.publish_state()
|
|
|
|
def _on_message(self, _client, _userdata, message):
|
|
try:
|
|
payload = json.loads(message.payload.decode("utf-8"))
|
|
except (UnicodeDecodeError, json.JSONDecodeError) as err:
|
|
logger.warning("Ignoring unparseable message on %s: %s", message.topic, err)
|
|
self._publish(self.status_topic, {"status": "error", "message": f"bad payload: {err}"})
|
|
return
|
|
if not isinstance(payload, dict):
|
|
self._publish(self.status_topic,
|
|
{"status": "error", "message": "payload must be a JSON object"})
|
|
return
|
|
|
|
logger.info("Command: %s", payload)
|
|
result = self.handler.handle(payload)
|
|
self._publish(self.status_topic, result)
|
|
# The API applies changes asynchronously (the controller polls its
|
|
# mailbox), so read state back rather than assuming the command took.
|
|
self.publish_state()
|
|
|
|
# -- publishing ---------------------------------------------------------
|
|
|
|
def _publish(self, topic: str, payload: Any, retain: bool = False) -> None:
|
|
if self._mqtt is None:
|
|
return
|
|
body = payload if isinstance(payload, str) else json.dumps(payload)
|
|
self._mqtt.publish(topic, body, qos=1, retain=retain)
|
|
|
|
def publish_discovery(self) -> None:
|
|
try:
|
|
modes = self.handler.refresh_modes()
|
|
except (requests.RequestException, RuntimeError) as err:
|
|
logger.error("Could not list display modes: %s", err)
|
|
modes = self.handler.known_modes
|
|
labels = sorted({m.get("name") or m["mode"] for m in modes})
|
|
for message in discovery_messages(self.command_topic, self.state_topic,
|
|
self.availability_topic, labels):
|
|
self._publish(message["topic"], message["payload"], retain=True)
|
|
logger.info("Published discovery for %d display mode(s)", len(labels))
|
|
|
|
def publish_state(self) -> None:
|
|
self._publish(self.state_topic, read_state(self.client), retain=True)
|
|
|
|
# -- lifecycle ----------------------------------------------------------
|
|
|
|
def run(self) -> int:
|
|
try:
|
|
import paho.mqtt.client as mqtt
|
|
except ImportError:
|
|
logger.error("paho-mqtt is not installed: pip install -r requirements.txt")
|
|
return 1
|
|
|
|
# VERSION2 is the current callback API. The compatibility note in
|
|
# CLAUDE.md is about code written against the v1 signatures; this file
|
|
# is written against v2 and requires paho-mqtt >= 2.0.
|
|
self._mqtt = mqtt.Client(
|
|
mqtt.CallbackAPIVersion.VERSION2,
|
|
client_id=self.config["mqtt_client_id"])
|
|
if self.config.get("mqtt_username"):
|
|
self._mqtt.username_pw_set(self.config["mqtt_username"],
|
|
self.config.get("mqtt_password"))
|
|
if self.config.get("mqtt_tls"):
|
|
self._mqtt.tls_set()
|
|
if self.config.get("mqtt_tls_insecure"):
|
|
logger.warning("TLS certificate verification is disabled (mqtt_tls_insecure)")
|
|
self._mqtt.tls_insecure_set(True)
|
|
else:
|
|
warn_if_cleartext(self.config)
|
|
|
|
self._mqtt.will_set(self.availability_topic, "offline", qos=1, retain=True)
|
|
self._mqtt.on_connect = self._on_connect
|
|
self._mqtt.on_message = self._on_message
|
|
|
|
logger.info("Connecting to %s:%s", self.config["mqtt_host"], self.config["mqtt_port"])
|
|
try:
|
|
self._mqtt.connect(self.config["mqtt_host"], self.config["mqtt_port"], keepalive=60)
|
|
except OSError as err:
|
|
logger.error("Could not reach the MQTT broker: %s", err)
|
|
return 1
|
|
|
|
self._mqtt.loop_start()
|
|
try:
|
|
while not self._stop.wait(30):
|
|
# HA is told the truth about state that changed outside the
|
|
# bridge -- somebody using the web UI, or an on-demand window
|
|
# expiring on its own.
|
|
self.publish_state()
|
|
finally:
|
|
self._publish(self.availability_topic, "offline", retain=True)
|
|
self._mqtt.loop_stop()
|
|
self._mqtt.disconnect()
|
|
return 0
|
|
|
|
def stop(self, *_args) -> None:
|
|
logger.info("Shutting down")
|
|
self._stop.set()
|
|
|
|
|
|
def main(argv: Optional[List[str]] = None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
|
parser.add_argument(
|
|
"--config",
|
|
default=os.path.join(os.path.dirname(os.path.abspath(__file__)), "bridge_config.json"),
|
|
help="Path to bridge_config.json (default: alongside this script)")
|
|
args = parser.parse_args(argv)
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s - %(levelname)s - %(name)s - %(message)s")
|
|
try:
|
|
config = load_config(args.config)
|
|
except ConfigError as err:
|
|
logger.error("%s", err)
|
|
return 1
|
|
logging.getLogger().setLevel(str(config.get("log_level", "INFO")).upper())
|
|
|
|
bridge = Bridge(config)
|
|
signal.signal(signal.SIGTERM, bridge.stop)
|
|
signal.signal(signal.SIGINT, bridge.stop)
|
|
return bridge.run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|