"""HTTP based control of Bravia TVs. Tested with the 2012 KDL-40HX850. """ import dataclasses import hashlib import logging import socket import threading import typing import urllib.parse import xml.etree.ElementTree from typing import Any, Dict, Iterable, Optional, Sequence from xml.etree.ElementTree import Element import requests from . import Channel, Device, ssdp logger = logging.getLogger('janet.bravia') _IRCC = """ {code} """ @dataclasses.dataclass class SystemInformation: """System information.""" name: str generation: str area: str language: str country: str model_name: str @dataclasses.dataclass class StatusItem: """A single item in the status.""" name: str source: str title: Optional[str] = None provider: Optional[str] = None @dataclasses.dataclass class Command: """A remote control command.""" name: str type: str value: str def _attrib_factory(node: Element, schema: Any) -> Any: """Returns a new instance of `schema` using attributes from `node`.""" values = {} for field in dataclasses.fields(schema): values[field.name] = node.attrib[_camel_case(field.name)] return schema(**values) def _remote_commands_factory(nodes: Iterable[Element]) -> Dict[str, Command]: commands = [_attrib_factory(n, Command) for n in nodes] return {c.name: c for c in commands} def _status_map_factory(nodes: Iterable[Element]) -> Sequence[StatusItem]: items = [] for node in nodes: values = {'name': node.attrib['name']} for item in node: values[item.attrib['field']] = item.attrib['value'] keep = {} for f in dataclasses.fields(StatusItem): keep[f.name] = values.get(f.name, None) items.append(StatusItem(**keep)) return items @dataclasses.dataclass class Status: """Status of the TV.""" status: Dict[str, StatusItem] = dataclasses.field( default_factory=_status_map_factory) @dataclasses.dataclass class RemoteCommands: """All remote commands accepted by the TV.""" command: Dict[str, Command] = dataclasses.field( default_factory=_remote_commands_factory) def _camel_case(snake: str) -> str: """Convert a snake_case name to camelCase.""" words = snake.split('_') words = [words[0]] + [x.capitalize() for x in words[1:]] return ''.join(words) def _extract_xml(text: str, schema: Any) -> Any: """Extract an XML tree as a dataclass instance.""" root = xml.etree.ElementTree.fromstring(text) values = {} for field in dataclasses.fields(schema): path = _camel_case(field.name) nodes = root.findall(path) if field.type == str: values[field.name] = nodes[0].text elif field.default_factory is not dataclasses.MISSING: values[field.name] = field.default_factory(nodes) else: logger.warning('Skipping unhandled field type %s', field) return schema(**values) class Client: def __init__(self, host: str): self.host = host self.path = host + '/cers/api/' id = hashlib.sha1(socket.getfqdn().encode()).hexdigest() self.device_id = f'braviamqtt:{id}' self.device_info = 'bravia.py' def _get(self, path: str, params: Optional[Dict[str, str]] = None) -> str: """Perform a GET with authentication.""" headers = { 'X-CERS-DEVICE-ID': self.device_id, 'X-CERS-DEVICE-INFO': self.device_info, } r = requests.get(self.path + path, params=params, headers=headers, timeout=3) r.raise_for_status() return r.text def get_system_information(self) -> SystemInformation: return _extract_xml(self._get('getSystemInformation'), SystemInformation) def register(self) -> None: """Register with the TV. This is a no-op if the client has already registered. Otherwise the TV will show a prompt asking to approve the new connection. """ self._get('register', params={ 'name': 'braviamqtt', 'registrationType': 'initial', 'deviceId': self.device_id, }) def get_remote_command_list(self) -> RemoteCommands: """Fetch all of the supported remote commands.""" return _extract_xml(self._get('getRemoteCommandList'), RemoteCommands) def get_status(self) -> Status: """Fetch the current high level status.""" return _extract_xml(self._get('getStatus'), Status) def ircc(self, code: str) -> None: """Send a remote control command.""" path = '/IRCC' headers = { 'soapaction': 'urn:schemas-sony-com:service:IRCC:1#X_SendIRCC', 'X-CERS-DEVICE-ID': self.device_id, 'X-CERS-DEVICE-INFO': self.device_info, } body = _IRCC.strip().format(code=code) r = requests.post(self.host + path, data=body, headers=headers) r.raise_for_status() class Bridge: def __init__(self, location: str, listeners: Channel): self._listeners = listeners self._location = location url = urllib.parse.urlparse(location) host = url.netloc.split(':')[0] self._bravia = Client(f'{url.scheme}://{host}') self._device = None def start(self): info = self._bravia.get_system_information() self._device = Device( identifiers=[f'bravia_{self._location}'], name=info.name, model=info.model_name, sw_version=info.generation, manufacturer='SONY', kind='bravia', available=True, ) self._listeners.put(self._device, self._device) self._listeners.put(self._device, info) self._bravia.register() commands = self._bravia.get_remote_command_list() self._listeners.put(self._device, commands) def available(self): status = self._bravia.get_status() for item in status.status: self._listeners.put(self._device, item) def stop(self): if self._device: self._device.available = False self._listeners.put(self._device, self._device) class Discoverer: def __init__(self): self._listeners = Channel() self._devices = {} self._lock = threading.Lock() def ssdp(self, device: Device, entity: Any) -> None: if not isinstance(entity, ssdp.Service): return entity = typing.cast(ssdp.Service, entity) if entity.st != 'urn:schemas-sony-com:service:IRCC:1': return if entity.available: with self._lock: if entity.usn not in self._devices: dev = Bridge(entity.location, self._listeners) self._devices[entity.usn] = dev dev.start() self._devices[entity.usn].available() else: with self._lock: dev = self._devices.get(entity.usn, None) if dev is not None: dev.stop() del self._devices[entity.usn] def listen(self, listener) -> None: self._listeners.listen(listener)