commit f12556bb439ae7b54227aa189bf191deeceaee31 Author: Andres Eduardo Garcia Marquez Date: Mon Mar 2 00:54:08 2026 -0500 feat: initial release — Samsung TV MCP server 15 tools for controlling Samsung Tizen TVs via local network. Auto-discovery (SSDP), WebSocket remote, UPnP volume/DLNA, app management, browser control, and Wake-on-LAN. Tested on Samsung UN65TU7000KXZL (2020, Crystal UHD). Co-Authored-By: Claude Opus 4.6 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..2d26841 --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +__pycache__/ +*.pyc +token.json +.venv/ +*.egg-info/ diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..794351f --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Andres Garcia + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..8141b56 --- /dev/null +++ b/README.md @@ -0,0 +1,108 @@ +# Samsung TV MCP Server + +A [Model Context Protocol](https://modelcontextprotocol.io/) server that lets AI assistants control Samsung Smart TVs (Tizen OS, 2016+) over the local network. No cloud, no SmartThings account required. + +> **Early release** — Core functionality works and has been tested on a Samsung UN65TU7000 (2020, Tizen). Not all tools have been exhaustively tested across different TV models and firmware versions. Bug reports and PRs are welcome. + +## Features + +- **15 tools** for complete TV control via natural language +- **Auto-discovery** — finds Samsung TVs on your network via SSDP +- **Zero config** — no API keys, no cloud, everything runs locally +- **Multi-protocol** — combines WebSocket, REST, UPnP/DLNA, and Wake-on-LAN +- **Persistent auth** — first connection requires TV approval, then uses saved token +- **Auto-reconnect** — recovers from dropped WebSocket connections + +## Tools + +| Tool | Description | +|------|-------------| +| `tv_discover` | Scan LAN for Samsung Smart TVs | +| `tv_info` | Get TV status (model, IP, power, volume) | +| `tv_power` | Turn TV on (WoL) or off | +| `tv_key` | Send any remote control key press | +| `tv_keys` | Send a sequence of key presses | +| `tv_navigate` | Semantic navigation (home, back, source, menu...) | +| `tv_volume` | Get/set volume (0-100), mute/unmute | +| `tv_channel` | Change channel by number or up/down | +| `tv_apps` | List all installed apps | +| `tv_launch` | Launch app by name ("netflix") or ID | +| `tv_close_app` | Close a running app | +| `tv_browser` | Open a URL in the TV's browser | +| `tv_text` | Type text into active input fields | +| `tv_cursor` | Move virtual cursor on screen | +| `tv_media` | Play media via DLNA (video/audio/images) with transport controls | + +## Requirements + +- Python 3.10+ +- Samsung Smart TV with Tizen OS (2016 or newer) +- TV and MCP server on the same local network + +## Installation + +```bash +pip install "samsungtvws[encrypted]" "mcp[cli]" +``` + +### Claude Code + +```bash +claude mcp add samsung-tv -- python /path/to/samsung-tv-mcp/main.py +``` + +### Claude Desktop + +Add to `claude_desktop_config.json`: + +```json +{ + "mcpServers": { + "samsung-tv": { + "command": "python", + "args": ["/path/to/samsung-tv-mcp/main.py"] + } + } +} +``` + +## First Connection + +On the first WebSocket connection, the TV will show a popup asking to allow "ClaudeCode". Accept it once — a token is saved to `token.json` for future sessions. + +To minimize prompts, go to **Settings > General > External Device Manager > Device Connection Manager** and set **Access Notification** to "First Time Only". + +## Supported App Aliases + +You can use friendly names instead of numeric IDs: + +`netflix`, `youtube`, `prime`, `disney`, `spotify`, `apple tv`, `hbo`, `max`, `plex`, `browser`, `steam link`, `twitch`, `tiktok`, `tubi`, `pluto`, `paramount`, `gallery`, `smartthings` + +## Architecture + +``` +main.py → FastMCP tools (thin layer, ~120 lines) +tv.py → SamsungTV client (WebSocket + UPnP + SSDP + WoL, ~250 lines) +``` + +- **WebSocket** (port 8002/8001): remote keys, apps, browser, text input, cursor +- **REST** (port 8001): device info, app launch/close +- **UPnP SOAP** (port 9197): precise volume control, DLNA media playback +- **Wake-on-LAN**: power on from standby + +## Known Limitations + +- Wake-on-LAN may not work reliably on all models (requires TV config) +- No direct HDMI source switching via API (must navigate the source menu with keys) +- No brightness/contrast/picture settings control +- No screenshot capability +- Text input only works when the TV's virtual keyboard is active +- DRM-protected content cannot be streamed via DLNA + +## Tested On + +- Samsung UN65TU7000KXZL (2020, Crystal UHD, Tizen, Colombia) + +## License + +MIT diff --git a/main.py b/main.py new file mode 100644 index 0000000..cf39bd7 --- /dev/null +++ b/main.py @@ -0,0 +1,298 @@ +"""Samsung Smart TV MCP Server — Control your TV with natural language.""" + +from __future__ import annotations + +import logging +from typing import Any, Optional + +from mcp.server.fastmcp import FastMCP +from tv import SamsungTV, discover + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(name)s] %(levelname)s: %(message)s", +) + +mcp = FastMCP("samsung-tv") +_tv = SamsungTV() + + +def _ok(message: str = "Done", **data: Any) -> dict[str, Any]: + return {"success": True, "message": message, **data} + + +def _err(message: str) -> dict[str, Any]: + return {"success": False, "message": message} + + +def _safe(fn, *args, **kwargs) -> dict[str, Any]: + try: + return fn(*args, **kwargs) + except Exception as e: + logging.getLogger("samsung-tv").error("%s: %s", fn.__name__, e) + return _err(str(e)) + + +# ── Discovery & Info ───────────────────────────────────────────── + + +@mcp.tool() +def tv_discover() -> dict[str, Any]: + """Scan the local network for Samsung Smart TVs via SSDP. + + Returns a list of found TVs with their IP, name, and model. + """ + return _safe(lambda: _ok("Scan complete", tvs=discover())) + + +@mcp.tool() +def tv_info() -> dict[str, Any]: + """Get TV status: model, IP, power state, current volume, resolution. + + Use this first to verify the TV is reachable and powered on. + """ + return _safe(lambda: _ok("TV info retrieved", **_tv.info())) + + +# ── Power ──────────────────────────────────────────────────────── + + +@mcp.tool() +def tv_power(action: str = "off") -> dict[str, Any]: + """Turn the TV on or off. + + Args: + action: "on" (Wake-on-LAN) or "off" (power key). Default "off". + """ + def _do(): + if action.lower() == "on": + _tv.power_on() + return _ok("Wake-on-LAN packet sent. TV should turn on in a few seconds.") + _tv.power_off() + return _ok("TV power off command sent") + return _safe(_do) + + +# ── Remote Keys ────────────────────────────────────────────────── + + +@mcp.tool() +def tv_key(key: str, times: int = 1) -> dict[str, Any]: + """Send a remote control key press to the TV. + + Args: + key: Key name like "VOLUP", "MUTE", "POWER", "HDMI", "PLAY", etc. + The KEY_ prefix is added automatically if missing. + times: Number of times to press the key (default 1). + + Common keys: POWER, VOLUP, VOLDOWN, MUTE, CHUP, CHDOWN, SOURCE, HDMI, + UP, DOWN, LEFT, RIGHT, ENTER, RETURN, EXIT, HOME, MENU, GUIDE, INFO, + PLAY, PAUSE, STOP, FF, REWIND, RED, GREEN, YELLOW, BLUE, 0-9, + PMODE, DYNAMIC, STANDARD, MOVIE1, GAME, SLEEP, CAPTION, APP_LIST. + """ + return _safe(lambda: (_tv.send_key(key, times), _ok(f"Sent {key} x{times}"))[1]) + + +@mcp.tool() +def tv_keys(keys: str, delay: float = 0.3) -> dict[str, Any]: + """Send a sequence of key presses with configurable delay between them. + + Args: + keys: Comma-separated key names, e.g. "HOME, DOWN, DOWN, ENTER". + delay: Seconds between each key press (default 0.3). + + Use this for menu navigation or complex sequences. + """ + key_list = [k.strip() for k in keys.split(",") if k.strip()] + return _safe(lambda: (_tv.send_keys(key_list, delay), _ok(f"Sent {len(key_list)} keys: {keys}"))[1]) + + +@mcp.tool() +def tv_navigate(action: str) -> dict[str, Any]: + """Quick navigation with semantic names instead of raw key codes. + + Args: + action: One of: home, back, exit, menu, source, guide, info, tools, + up, down, left, right, enter/ok, play, pause, stop, ff, rewind. + """ + return _safe(lambda: (_tv.navigate(action), _ok(f"Navigated: {action}"))[1]) + + +# ── Volume & Channel ───────────────────────────────────────────── + + +@mcp.tool() +def tv_volume( + level: Optional[int] = None, + action: Optional[str] = None, +) -> dict[str, Any]: + """Control TV volume. Get current volume, set to exact level, or mute/unmute. + + Args: + level: Set volume to this exact value (0-100). Omit to just read. + action: "up", "down", "mute", "unmute". Omit to just read or use level. + + Examples: tv_volume() → get current, tv_volume(level=25) → set to 25, + tv_volume(action="mute") → toggle mute. + """ + def _do(): + if level is not None: + _tv.set_volume(level) + return _ok(f"Volume set to {level}", volume=level) + if action: + a = action.lower() + if a == "up": + _tv.send_key("KEY_VOLUP") + return _ok("Volume up") + if a == "down": + _tv.send_key("KEY_VOLDOWN") + return _ok("Volume down") + if a == "mute": + _tv.set_mute(True) + return _ok("Muted") + if a == "unmute": + _tv.set_mute(False) + return _ok("Unmuted") + return _err(f"Unknown action '{a}'. Use: up, down, mute, unmute") + vol = _tv.get_volume() + muted = _tv.get_mute() + return _ok(f"Volume: {vol}, Muted: {muted}", volume=vol, muted=muted) + return _safe(_do) + + +@mcp.tool() +def tv_channel( + number: Optional[int] = None, + direction: Optional[str] = None, +) -> dict[str, Any]: + """Change TV channel by number or direction. + + Args: + number: Channel number to switch to (e.g. 42). + direction: "up" or "down" to go to next/previous channel. + + Provide either number or direction, not both. + """ + return _safe(lambda: (_tv.channel(number, direction), _ok(f"Channel changed"))[1]) + + +# ── Apps ───────────────────────────────────────────────────────── + + +@mcp.tool() +def tv_apps() -> dict[str, Any]: + """List all installed apps on the TV with their IDs. + + Returns app names and IDs that can be used with tv_launch. + """ + return _safe(lambda: _ok("Apps retrieved", apps=_tv.list_apps())) + + +@mcp.tool() +def tv_launch( + app: str, + deep_link: Optional[str] = None, +) -> dict[str, Any]: + """Launch an app on the TV by name or ID. + + Args: + app: App name ("netflix", "youtube", "spotify", "disney", "prime", + "browser", "plex", "hbo", "max") or app ID. + deep_link: Optional deep link parameter to open specific content. + + The app name is case-insensitive and supports fuzzy matching. + """ + return _safe(lambda: (_tv.launch_app(app, deep_link), _ok(f"Launched {app}"))[1]) + + +@mcp.tool() +def tv_close_app(app: str) -> dict[str, Any]: + """Close a running app on the TV. + + Args: + app: App name or ID (same as tv_launch). + """ + return _safe(lambda: (_tv.close_app(app), _ok(f"Closed {app}"))[1]) + + +# ── Browser & Text ─────────────────────────────────────────────── + + +@mcp.tool() +def tv_browser(url: str) -> dict[str, Any]: + """Open a URL in the TV's built-in web browser. + + Args: + url: The full URL to open (e.g. "https://google.com"). + """ + return _safe(lambda: (_tv.open_browser(url), _ok(f"Opened {url}"))[1]) + + +@mcp.tool() +def tv_text(text: str) -> dict[str, Any]: + """Type text into the currently active input field on the TV. + + Args: + text: The text to type. Only works when a text input is active + (virtual keyboard is visible on the TV screen). + """ + return _safe(lambda: (_tv.send_text(text), _ok(f"Typed text ({len(text)} chars)"))[1]) + + +# ── Cursor ─────────────────────────────────────────────────────── + + +@mcp.tool() +def tv_cursor(x: int, y: int, duration: int = 500) -> dict[str, Any]: + """Move the virtual cursor/pointer on the TV screen. + + Args: + x: Horizontal position (pixels from left). + y: Vertical position (pixels from top). + duration: Movement duration in milliseconds (default 500). + """ + return _safe(lambda: (_tv.move_cursor(x, y, duration), _ok(f"Cursor moved to ({x}, {y})"))[1]) + + +# ── DLNA Media ─────────────────────────────────────────────────── + + +@mcp.tool() +def tv_media( + action: str = "status", + url: Optional[str] = None, + title: Optional[str] = None, + seek_to: Optional[str] = None, +) -> dict[str, Any]: + """Play media on the TV via DLNA or control current playback. + + Args: + action: "play_url" to start playing from URL, or "play", "pause", + "stop", "seek", "status" to control current media. + url: Media URL (required for "play_url"). Supports video, audio, images. + title: Display title for the media (optional, default "Media"). + seek_to: Time position for seek, format "HH:MM:SS" (e.g. "00:05:30"). + + Examples: + tv_media(action="play_url", url="http://server/video.mp4") + tv_media(action="pause") + tv_media(action="seek", seek_to="00:10:00") + tv_media(action="status") → returns current position and state. + """ + def _do(): + if action == "play_url": + if not url: + return _err("URL required for play_url action") + _tv.play_media(url, title or "Media") + return _ok(f"Playing: {url}") + if action == "seek" and not seek_to: + return _err("seek_to required for seek action (format HH:MM:SS)") + result = _tv.media_control(action, target=seek_to) + if action == "status": + return _ok("Playback status", **result) + return _ok(f"Media {action} executed", **result) + return _safe(_do) + + +if __name__ == "__main__": + mcp.run(transport="stdio") diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..97f20dd --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,13 @@ +[project] +name = "samsung-tv-mcp" +version = "1.0.0" +description = "MCP server for Samsung Smart TV control via local network" +requires-python = ">=3.10" +dependencies = [ + "samsungtvws[encrypted]>=2.7.0", + "mcp[cli]>=1.0.0", +] + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" diff --git a/tv.py b/tv.py new file mode 100644 index 0000000..7678a4c --- /dev/null +++ b/tv.py @@ -0,0 +1,500 @@ +"""Samsung Smart TV client — WebSocket + UPnP + SSDP + WoL.""" + +from __future__ import annotations + +import json +import logging +import socket +import time +import xml.etree.ElementTree as ET +from pathlib import Path +from typing import Any +from urllib.parse import urlparse +from urllib.request import Request, urlopen +from urllib.error import URLError + +from samsungtvws import SamsungTVWS + +log = logging.getLogger("samsung-tv") + +TOKEN_FILE = str(Path(__file__).parent / "token.json") +_UPNP_NS = "urn:schemas-upnp-org:service" +_SOAP_ENV = ( + '' + '' + "{body}" +) + +# App aliases → official IDs (region-independent first, then fallbacks) +APP_ALIASES: dict[str, list[str]] = { + "netflix": ["11101200001", "3201907018807"], + "youtube": ["111299001912"], + "prime": ["3201512006785", "3201910019365"], + "disney": ["3201901017640"], + "spotify": ["3201606009684"], + "apple tv": ["3201807016597"], + "hbo": ["3201601007230", "3202301029760"], + "max": ["3202301029760", "3201601007230"], + "plex": ["3201512006963"], + "browser": ["org.tizen.browser", "3201907018784"], + "steam link": ["3201702011851"], + "twitch": ["3202203026841"], + "tiktok": ["3202008021577"], + "tubi": ["3201504001965"], + "pluto": ["3201808016802"], + "paramount": ["3201710014981"], + "gallery": ["3201710015037"], + "smartthings": ["3201710015016"], +} + +NAVIGATE_KEYS = { + "home": "KEY_HOME", + "back": "KEY_RETURN", + "exit": "KEY_EXIT", + "menu": "KEY_MENU", + "source": "KEY_SOURCE", + "guide": "KEY_GUIDE", + "info": "KEY_INFO", + "tools": "KEY_TOOLS", + "up": "KEY_UP", + "down": "KEY_DOWN", + "left": "KEY_LEFT", + "right": "KEY_RIGHT", + "enter": "KEY_ENTER", + "ok": "KEY_ENTER", + "play": "KEY_PLAY", + "pause": "KEY_PAUSE", + "stop": "KEY_STOP", + "ff": "KEY_FF", + "rewind": "KEY_REWIND", +} + + +# ── SSDP Discovery ────────────────────────────────────────────── + + +def discover(timeout: float = 5.0) -> list[dict[str, Any]]: + """Discover Samsung TVs on the local network via SSDP.""" + msg = ( + "M-SEARCH * HTTP/1.1\r\n" + "HOST: 239.255.255.250:1900\r\n" + 'MAN: "ssdp:discover"\r\n' + f"MX: {int(timeout)}\r\n" + "ST: urn:schemas-upnp-org:device:MediaRenderer:1\r\n\r\n" + ) + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) + sock.settimeout(timeout) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.sendto(msg.encode(), ("239.255.255.250", 1900)) + + locations: set[str] = set() + tvs: list[dict[str, Any]] = [] + try: + while True: + data, _ = sock.recvfrom(4096) + text = data.decode(errors="ignore") + if "samsung" not in text.lower(): + continue + for line in text.splitlines(): + if line.upper().startswith("LOCATION:"): + locations.add(line.split(":", 1)[1].strip()) + except socket.timeout: + pass + finally: + sock.close() + + for loc in locations: + try: + resp = urlopen(loc, timeout=3).read().decode() + root = ET.fromstring(resp) + ns = {"d": "urn:schemas-upnp-org:device-1-0"} + dev = root.find(".//d:device", ns) + if dev is None: + continue + fn = dev.findtext("d:friendlyName", "", ns) + mn = dev.findtext("d:modelName", "", ns) + mfr = dev.findtext("d:manufacturer", "", ns) + if "samsung" not in (mfr or "").lower(): + continue + ip = urlparse(loc).hostname + tvs.append({ + "ip": ip, + "name": fn, + "model": mn, + "manufacturer": mfr, + "location": loc, + }) + except Exception: + continue + return tvs + + +# ── UPnP SOAP helpers ─────────────────────────────────────────── + + +def _soap_call( + ip: str, control_url: str, service: str, action: str, args: str = "" +) -> str: + """Execute a UPnP SOAP action and return raw XML response.""" + body = f'{args}' + envelope = _SOAP_ENV.format(body=body) + url = f"http://{ip}:9197{control_url}" + req = Request( + url, + data=envelope.encode(), + headers={ + "Content-Type": 'text/xml; charset="utf-8"', + "SOAPAction": f'"{_UPNP_NS}:{service}:1#{action}"', + }, + ) + return urlopen(req, timeout=5).read().decode() + + +def _soap_value(xml_text: str, tag: str) -> str | None: + """Extract a single value from SOAP response XML.""" + root = ET.fromstring(xml_text) + for elem in root.iter(): + if elem.tag.endswith(tag): + return elem.text + return None + + +# ── Wake-on-LAN ───────────────────────────────────────────────── + + +def wake_on_lan(mac: str) -> None: + """Send WoL magic packet to power on the TV.""" + mac_bytes = bytes.fromhex(mac.replace(":", "").replace("-", "")) + packet = b"\xff" * 6 + mac_bytes * 16 + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s: + s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) + s.sendto(packet, ("255.255.255.255", 9)) + + +# ── Main TV Client ────────────────────────────────────────────── + + +class SamsungTV: + """Unified Samsung TV controller combining WebSocket, REST, and UPnP.""" + + def __init__(self, ip: str | None = None): + self._ip = ip + self._ws: SamsungTVWS | None = None + self._info: dict[str, Any] | None = None + self._mac: str | None = None + + # ── Connection ─────────────────────────────────────────── + + def _ensure_ip(self) -> str: + if self._ip: + return self._ip + tvs = discover(timeout=4.0) + if not tvs: + raise ConnectionError("No Samsung TV found on the network") + self._ip = tvs[0]["ip"] + log.info("Auto-discovered TV at %s (%s)", self._ip, tvs[0].get("name")) + return self._ip + + def _ensure_ws(self) -> SamsungTVWS: + if self._ws is not None: + return self._ws + ip = self._ensure_ip() + try: + self._ws = SamsungTVWS( + host=ip, port=8002, token_file=TOKEN_FILE, name="ClaudeCode" + ) + self._ws.open() + log.info("Connected via WSS:8002") + except Exception: + log.warning("WSS:8002 failed, trying WS:8001") + self._ws = SamsungTVWS( + host=ip, port=8001, token_file=TOKEN_FILE, name="ClaudeCode" + ) + self._ws.open() + return self._ws + + def _reconnect(self) -> SamsungTVWS: + self._ws = None + return self._ensure_ws() + + def _send_ws(self, method: str, **kwargs: Any) -> Any: + """Send WebSocket command with auto-reconnect.""" + ws = self._ensure_ws() + for attempt in range(2): + try: + return getattr(ws, method)(**kwargs) + except Exception as e: + if attempt == 0: + log.warning("WS error (%s), reconnecting...", e) + ws = self._reconnect() + else: + raise + + # ── Device Info ────────────────────────────────────────── + + def info(self) -> dict[str, Any]: + ip = self._ensure_ip() + try: + raw = urlopen(f"http://{ip}:8001/api/v2/", timeout=5).read() + data = json.loads(raw) + device = data.get("device", {}) + self._mac = device.get("wifiMac") + # Enrich with UPnP volume + try: + vol = self.get_volume() + device["currentVolume"] = vol + except Exception: + pass + self._info = device + return { + "name": device.get("name", "Unknown"), + "model": device.get("modelName", "Unknown"), + "ip": device.get("ip", ip), + "mac": self._mac, + "power": device.get("PowerState", "unknown"), + "os": device.get("OS", "Tizen"), + "resolution": device.get("resolution", "unknown"), + "network": device.get("networkType", "unknown"), + "volume": device.get("currentVolume"), + } + except (URLError, OSError): + return {"power": "off", "ip": ip, "message": "TV appears to be off"} + + # ── Power ──────────────────────────────────────────────── + + def power_off(self) -> None: + self._send_ws("send_key", key="KEY_POWER") + + def power_on(self, mac: str | None = None) -> None: + target_mac = mac or self._mac + if not target_mac: + # Try to get MAC from previous info + try: + self.info() + target_mac = self._mac + except Exception: + pass + if not target_mac: + raise ValueError("MAC address required for Wake-on-LAN. Get it with tv_info first.") + wake_on_lan(target_mac) + + # ── Keys ───────────────────────────────────────────────── + + def send_key(self, key: str, times: int = 1) -> None: + normalized = key.upper() + if not normalized.startswith("KEY_"): + normalized = f"KEY_{normalized}" + for i in range(times): + self._send_ws("send_key", key=normalized) + if i < times - 1: + time.sleep(0.15) + + def send_keys(self, keys: list[str], delay: float = 0.3) -> None: + for i, key in enumerate(keys): + self.send_key(key) + if i < len(keys) - 1: + time.sleep(delay) + + def navigate(self, action: str) -> None: + key = NAVIGATE_KEYS.get(action.lower()) + if not key: + raise ValueError( + f"Unknown action '{action}'. Valid: {', '.join(NAVIGATE_KEYS)}" + ) + self.send_key(key) + + # ── Volume ─────────────────────────────────────────────── + + def get_volume(self) -> int: + ip = self._ensure_ip() + xml = _soap_call( + ip, "/upnp/control/RenderingControl1", "RenderingControl", + "GetVolume", "0Master", + ) + val = _soap_value(xml, "CurrentVolume") + return int(val) if val else -1 + + def set_volume(self, level: int) -> None: + ip = self._ensure_ip() + level = max(0, min(100, level)) + _soap_call( + ip, "/upnp/control/RenderingControl1", "RenderingControl", + "SetVolume", + f"0Master" + f"{level}", + ) + + def get_mute(self) -> bool: + ip = self._ensure_ip() + xml = _soap_call( + ip, "/upnp/control/RenderingControl1", "RenderingControl", + "GetMute", "0Master", + ) + val = _soap_value(xml, "CurrentMute") + return val == "1" if val else False + + def set_mute(self, mute: bool) -> None: + ip = self._ensure_ip() + _soap_call( + ip, "/upnp/control/RenderingControl1", "RenderingControl", + "SetMute", + f"0Master" + f"{'1' if mute else '0'}", + ) + + # ── Channel ────────────────────────────────────────────── + + def channel(self, number: int | None = None, direction: str | None = None) -> None: + if number is not None: + for digit in str(number): + self.send_key(f"KEY_{digit}") + time.sleep(0.15) + time.sleep(0.3) + self.send_key("KEY_ENTER") + elif direction: + key = "KEY_CHUP" if direction.lower() == "up" else "KEY_CHDOWN" + self.send_key(key) + else: + raise ValueError("Provide either number or direction ('up'/'down')") + + # ── Apps ───────────────────────────────────────────────── + + def _resolve_app_id(self, name_or_id: str) -> str: + alias = name_or_id.lower().strip() + if alias in APP_ALIASES: + return APP_ALIASES[alias][0] + # Check if it looks like an app ID already + if name_or_id.replace(".", "").replace("_", "").isalnum() and ( + len(name_or_id) > 8 or "." in name_or_id + ): + return name_or_id + # Fuzzy match against aliases + for key, ids in APP_ALIASES.items(): + if alias in key or key in alias: + return ids[0] + return name_or_id + + def list_apps(self) -> list[dict[str, Any]]: + ws = self._ensure_ws() + raw = ws.app_list() + if not raw: + return [] + return [{"id": a.get("appId"), "name": a.get("name", "Unknown")} for a in raw] + + def launch_app( + self, name_or_id: str, meta_tag: str | None = None + ) -> None: + app_id = self._resolve_app_id(name_or_id) + ip = self._ensure_ip() + # Try REST first (more reliable for launching) + try: + req = Request(f"http://{ip}:8001/api/v2/applications/{app_id}", method="POST") + urlopen(req, timeout=5) + return + except Exception: + pass + # Fallback to WebSocket + self._send_ws( + "run_app", app_id=app_id, app_type=2, meta_tag=meta_tag or "" + ) + + def close_app(self, name_or_id: str) -> None: + app_id = self._resolve_app_id(name_or_id) + ip = self._ensure_ip() + req = Request( + f"http://{ip}:8001/api/v2/applications/{app_id}", method="DELETE" + ) + urlopen(req, timeout=5) + + # ── Browser ────────────────────────────────────────────── + + def open_browser(self, url: str) -> None: + self._send_ws("open_browser", url=url) + + # ── Text Input ─────────────────────────────────────────── + + def send_text(self, text: str) -> None: + self._send_ws("send_text", text=text) + + # ── Cursor ─────────────────────────────────────────────── + + def move_cursor(self, x: int, y: int, duration: int = 500) -> None: + self._send_ws("move_cursor", x=x, y=y, duration=duration) + + # ── DLNA Media ─────────────────────────────────────────── + + def play_media(self, url: str, title: str = "Media") -> None: + ip = self._ensure_ip() + meta = ( + f'' + f"{title}" + ) + _soap_call( + ip, "/upnp/control/AVTransport1", "AVTransport", + "SetAVTransportURI", + f"0" + f"{url}" + f"{meta}", + ) + time.sleep(0.5) + _soap_call( + ip, "/upnp/control/AVTransport1", "AVTransport", + "Play", "01", + ) + + def media_control(self, action: str, target: str | None = None) -> dict[str, Any]: + ip = self._ensure_ip() + action_lower = action.lower() + + if action_lower == "play": + _soap_call( + ip, "/upnp/control/AVTransport1", "AVTransport", + "Play", "01", + ) + elif action_lower == "pause": + _soap_call( + ip, "/upnp/control/AVTransport1", "AVTransport", + "Pause", "0", + ) + elif action_lower == "stop": + _soap_call( + ip, "/upnp/control/AVTransport1", "AVTransport", + "Stop", "0", + ) + elif action_lower == "seek" and target: + _soap_call( + ip, "/upnp/control/AVTransport1", "AVTransport", + "Seek", + f"0" + f"REL_TIME{target}", + ) + elif action_lower == "status": + xml_t = _soap_call( + ip, "/upnp/control/AVTransport1", "AVTransport", + "GetTransportInfo", "0", + ) + xml_p = _soap_call( + ip, "/upnp/control/AVTransport1", "AVTransport", + "GetPositionInfo", "0", + ) + return { + "state": _soap_value(xml_t, "CurrentTransportState"), + "position": _soap_value(xml_p, "RelTime"), + "duration": _soap_value(xml_p, "TrackDuration"), + "uri": _soap_value(xml_p, "TrackURI"), + } + else: + raise ValueError( + f"Unknown action '{action}'. Valid: play, pause, stop, seek, status" + ) + return {"action": action_lower, "done": True} + + def close(self) -> None: + if self._ws: + try: + self._ws.close() + except Exception: + pass + self._ws = None