lgtv/lgtv.sh

#!/bin/bash
# LG TV control — runs locally on steel141 (192.168.0.141), on the TV's LAN.
#
# Design notes (see DECISIONS.md):
#   * No SSH hop. This box is already on the TV's subnet with bscpylgtv +
#     network broadcast, and is effectively always-on. The old version SSHed
#     to seth@192.168.0.141 from itself — pointless overhead + a failure mode.
#   * IP discovery is cached. We persist the last-known TV IP and probe it
#     directly (fast TCP). A full /24 ping-sweep only runs on a cache miss,
#     so the 30s HA state poll no longer hammers the network and no longer
#     flaps when ARP entries go stale.
#   * WoL is sent unicast (to last-known IP) AND broadcast, multiple ports,
#     several bursts — the robust-as-software-can-be wake path. If the TV
#     still won't wake, it's the TV's own settings (Quick Start+ / Mobile TV
#     On / wired Ethernet), which no script can fix.
#
# Usage: lgtv.sh [on|off|status]

set -euo pipefail

TV_MAC="f8:01:b4:2a:26:ed"
BROADCAST="192.168.0.255"
STATE_FILE="${LGTV_STATE:-${HOME}/.cache/lgtv/last_ip}"
WEBOS_PORTS="3000 3001"

mkdir -p "$(dirname "$STATE_FILE")"

# Probe a single IP for an open WebOS port. Echo the IP and succeed if reachable.
probe_ip() {
    local ip="$1"
    IP="$ip" PORTS="$WEBOS_PORTS" python3 - <<'PY'
import os, socket, sys
ip = os.environ['IP']
for port in os.environ['PORTS'].split():
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(1.2)
    try:
        s.connect((ip, int(port)))
        print(ip)
        sys.exit(0)
    except Exception:
        pass
    finally:
        s.close()
sys.exit(1)
PY
}

# Find the TV's IP. Try the cache first (fast), fall back to an ARP sweep.
find_tv_ip() {
    # 1. Cached IP — the common path, one quick TCP probe.
    if [[ -s "$STATE_FILE" ]]; then
        local cached
        cached="$(<"$STATE_FILE")"
        if [[ -n "$cached" ]] && probe_ip "$cached" >/dev/null 2>&1; then
            echo "$cached"
            return 0
        fi
    fi

    # 2. Cache miss — sweep the /24 to repopulate ARP, match the TV's MAC,
    #    confirm a WebOS port is open, then re-cache.
    local found
    found="$(TV_MAC="$TV_MAC" PORTS="$WEBOS_PORTS" python3 - <<'PY'
import os, socket, subprocess
mac = os.environ['TV_MAC'].lower()
ports = os.environ['PORTS'].split()

subprocess.run(
    'for i in $(seq 1 254); do ping -c1 -W1 192.168.0.$i >/dev/null 2>&1 & done; wait',
    shell=True,
)

neigh = subprocess.check_output('ip neigh show', shell=True, text=True)
for line in neigh.splitlines():
    low = line.lower()
    if mac not in low or 'failed' in low or 'incomplete' in low:
        continue
    ip = line.split()[0]
    for port in ports:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(1.2)
        try:
            s.connect((ip, int(port)))
            print(ip)
            raise SystemExit(0)
        except Exception:
            pass
        finally:
            s.close()
print('')
PY
)"

    if [[ -n "$found" ]]; then
        echo "$found" > "$STATE_FILE"
        echo "$found"
        return 0
    fi
    return 1
}

tv_on() {
    # Unicast to last-known IP (if we have one) + broadcast, ports 9 & 7,
    # three bursts. WoL is fire-and-forget; the TV has no IP to talk to yet.
    local target=""
    [[ -s "$STATE_FILE" ]] && target="$(<"$STATE_FILE")"

    echo "Sending Wake-on-LAN to $TV_MAC (broadcast${target:+ + unicast $target})..."
    TV_MAC="$TV_MAC" BROADCAST="$BROADCAST" TARGET="$target" python3 - <<'PY'
import os, socket, time
mac = os.environ['TV_MAC']
magic = b'\xff' * 6 + bytes.fromhex(mac.replace(':', '')) * 16
dests = [os.environ['BROADCAST']]
if os.environ.get('TARGET'):
    dests.append(os.environ['TARGET'])

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
for _ in range(3):
    for dest in dests:
        for port in (9, 7):
            try:
                s.sendto(magic, (dest, port))
            except OSError:
                pass
    time.sleep(0.4)
s.close()
print('WoL bursts sent')
PY
}

tv_off() {
    echo "Finding TV..."
    local ip
    if ! ip="$(find_tv_ip)"; then
        echo "ERROR: TV not found on network (already off?)"
        exit 1
    fi
    echo "TV at $ip — sending WebOS power_off..."
    # Retry a few times: a TV woken seconds ago answers TCP :3000 but isn't
    # ready for a WebSocket control session for ~20-30s. The common case
    # (TV already up) succeeds on the first try in ~3s.
    local attempt
    for attempt in 1 2; do
        if TV_IP="$ip" python3 - <<'PY'
import asyncio, os
from bscpylgtv import WebOsClient

# Pass the key store path explicitly. bscpylgtv otherwise defaults to a
# CWD-relative ./.aiopylgtv.sqlite (and its $USER_HOME fallback isn't $HOME),
# so under systemd (cwd=/) it can't find the TV's pairing key. Absolute path
# = works regardless of cwd/env. Override with LGTV_KEY_FILE if needed.
KEY = os.environ.get('LGTV_KEY_FILE') or os.path.expanduser('~/.aiopylgtv.sqlite')

async def main():
    client = await WebOsClient.create(os.environ['TV_IP'], key_file_path=KEY)
    await client.connect()
    await client.power_off()
    await asyncio.sleep(1)
    await client.disconnect()

asyncio.run(main())
print('power_off sent')
PY
        then
            echo "Done."
            return 0
        fi
        echo "  power_off attempt $attempt failed (TV may still be waking); retrying..."
        sleep 10
    done
    echo "ERROR: power_off failed after 2 attempts"
    exit 1
}

tv_status() {
    if find_tv_ip >/dev/null 2>&1; then
        echo "TV: ON at $(<"$STATE_FILE")"
    else
        echo "TV: OFF (not found on network)"
    fi
}

case "${1:-}" in
    on)     tv_on ;;
    off)    tv_off ;;
    status) tv_status ;;
    *)      echo "Usage: $0 [on|off|status]"; exit 1 ;;
esac