lgtv/lgtv_mqtt.py

#!/usr/bin/env python3
"""
LG TV MQTT bridge for Home Assistant.

Runs locally on steel141 as user `seth` (where bscpylgtv lives) via systemd.
Subscribes to command topic, runs lgtv.sh, publishes state back.
Registers the TV as a switch via MQTT discovery so HA picks it up automatically.

Topics:
  homeassistant/switch/lgtv/config    -- discovery payload (published on connect)
  lgtv/switch/state                   -- current state: ON / OFF
  lgtv/switch/command                 -- receives: ON / OFF
"""

import os
import subprocess
import time
import json
import logging
import paho.mqtt.client as mqtt
from concurrent.futures import ThreadPoolExecutor

worker = ThreadPoolExecutor(max_workers=1)

# --- Config ---
MQTT_HOST     = "192.168.0.154"   # CT 301 Mosquitto
MQTT_PORT     = 1883
MQTT_USER     = "lgtv"
MQTT_PASS     = os.environ.get("MQTT_PASSWORD", "")

LGTV_SCRIPT   = "/usr/local/bin/lgtv.sh"

DISCOVERY_TOPIC = "homeassistant/switch/lgtv/config"
STATE_TOPIC     = "lgtv/switch/state"
COMMAND_TOPIC   = "lgtv/switch/command"

POLL_INTERVAL   = 30   # seconds between state polls

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [lgtv-mqtt] %(levelname)s %(message)s",
)
log = logging.getLogger(__name__)


def run(cmd, timeout=60):
    # Every caller (poll loop, on_connect, commands) routes through here, so a
    # stalled helper becomes a plain failure instead of an exception.
    try:
        result = subprocess.run(
            cmd, shell=True, capture_output=True, text=True, timeout=timeout
        )
    except subprocess.TimeoutExpired:
        log.warning(f"timed out after {timeout}s: {cmd}")
        return "timeout", 124
    return result.stdout.strip(), result.returncode


def get_tv_state():
    """Returns 'ON' or 'OFF' by using the same probe logic as lgtv.sh status."""
    out, rc = run(f"{LGTV_SCRIPT} status")
    if rc != 0:
        log.warning(f"status probe failed rc={rc} out={out}")
        return "OFF"
    return "ON" if "TV: ON" in out else "OFF"


def tv_on():
    out, rc = run(f"{LGTV_SCRIPT} on")
    log.info(f"tv_on: rc={rc} out={out}")
    return rc == 0


def tv_off():
    out, rc = run(f"{LGTV_SCRIPT} off")
    log.info(f"tv_off: rc={rc} out={out}")
    return rc == 0


def publish_discovery(client):
    payload = {
        "name": "LG TV",
        "unique_id": "lgtv_f801b42a26ed",
        "state_topic": STATE_TOPIC,
        "command_topic": COMMAND_TOPIC,
        "payload_on": "ON",
        "payload_off": "OFF",
        "state_on": "ON",
        "state_off": "OFF",
        "optimistic": False,
        "device": {
            "identifiers": ["lgtv_f801b42a26ed"],
            "name": "LG TV",
            "manufacturer": "LG",
            "model": "WebOS TV",
        },
        "icon": "mdi:television",
        "availability_topic": "lgtv/switch/availability",
        "payload_available": "online",
        "payload_not_available": "offline",
    }
    client.publish(DISCOVERY_TOPIC, json.dumps(payload), retain=True)
    client.publish("lgtv/switch/availability", "online", retain=True)
    log.info("Published MQTT discovery payload")


def on_connect(client, userdata, flags, rc):
    if rc == 0:
        log.info("Connected to MQTT broker")
        publish_discovery(client)
        client.subscribe(COMMAND_TOPIC)
        log.info(f"Subscribed to {COMMAND_TOPIC}")
        # Publish current state on connect
        state = get_tv_state()
        client.publish(STATE_TOPIC, state, retain=True)
        log.info(f"Initial state: {state}")
    else:
        log.error(f"MQTT connect failed: rc={rc}")


def on_message(client, userdata, msg):
    # Hand off to one worker thread: an `off` can block up to 60s, which starves
    # paho's keepalive (broker drops us, queued commands are lost). The single
    # worker keeps commands in order.
    worker.submit(safe_handle, client, msg)


def safe_handle(client, msg):
    # An exception escaping a paho callback kills the network thread while the
    # poll loop keeps the process alive — a silent zombie systemd never restarts.
    try:
        handle_command(client, msg)
    except Exception:
        log.exception("command handling failed")


def handle_command(client, msg):
    command = msg.payload.decode(errors="replace").strip().upper()
    log.info(f"Received command: {command}")

    if command == "ON":
        ok = tv_on()
    elif command == "OFF":
        ok = tv_off()
    else:
        log.warning(f"Unknown command: {command}")
        return

    if not ok:
        # Failed command: report what the TV actually is, not what was asked.
        state = get_tv_state()
        client.publish(STATE_TOPIC, state, retain=True)
        log.warning(f"{command} failed; published probed state: {state}")
    elif command == "ON":
        # Publish ON optimistically immediately — WoL takes ~20s to appear on network.
        # The 30s poll loop will correct it if it failed.
        client.publish(STATE_TOPIC, "ON", retain=True)
        log.info("State published: ON (optimistic)")
    else:
        time.sleep(3)
        client.publish(STATE_TOPIC, "OFF", retain=True)
        log.info("State published: OFF")


def main():
    client = mqtt.Client(client_id="lgtv-bridge")
    client.username_pw_set(MQTT_USER, MQTT_PASS)
    client.on_connect = on_connect
    client.on_message = on_message

    client.will_set("lgtv/switch/availability", "offline", retain=True)

    log.info(f"Connecting to {MQTT_HOST}:{MQTT_PORT}...")
    client.connect(MQTT_HOST, MQTT_PORT, keepalive=60)
    client.loop_start()

    last_poll = 0
    try:
        while True:
            now = time.time()
            if now - last_poll >= POLL_INTERVAL:
                state = get_tv_state()
                client.publish(STATE_TOPIC, state, retain=True)
                log.info(f"Polled state: {state}")
                last_poll = now
            time.sleep(5)
    except KeyboardInterrupt:
        log.info("Shutting down")
    finally:
        client.publish("lgtv/switch/availability", "offline", retain=True)
        client.loop_stop()
        client.disconnect()


if __name__ == "__main__":
    main()