commit f8971c4821920e0080733b212293617498eeb434 Author: Hartmut Bauer Date: Wed Dec 17 17:02:05 2025 +0100 initial commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..95044d4 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +/div/ diff --git a/.idea/.gitignore b/.idea/.gitignore new file mode 100644 index 0000000..ab1f416 --- /dev/null +++ b/.idea/.gitignore @@ -0,0 +1,10 @@ +# Default ignored files +/shelf/ +/workspace.xml +# Ignored default folder with query files +/queries/ +# Datasource local storage ignored files +/dataSources/ +/dataSources.local.xml +# Editor-based HTTP Client requests +/httpRequests/ diff --git a/.idea/bravia8A.iml b/.idea/bravia8A.iml new file mode 100644 index 0000000..0f36c22 --- /dev/null +++ b/.idea/bravia8A.iml @@ -0,0 +1,10 @@ + + + + + + + + + + \ No newline at end of file diff --git a/.idea/inspectionProfiles/profiles_settings.xml b/.idea/inspectionProfiles/profiles_settings.xml new file mode 100644 index 0000000..105ce2d --- /dev/null +++ b/.idea/inspectionProfiles/profiles_settings.xml @@ -0,0 +1,6 @@ + + + + \ No newline at end of file diff --git a/.idea/misc.xml b/.idea/misc.xml new file mode 100644 index 0000000..e2b27f0 --- /dev/null +++ b/.idea/misc.xml @@ -0,0 +1,7 @@ + + + + + + \ No newline at end of file diff --git a/.idea/modules.xml b/.idea/modules.xml new file mode 100644 index 0000000..0ff3492 --- /dev/null +++ b/.idea/modules.xml @@ -0,0 +1,8 @@ + + + + + + + + \ No newline at end of file diff --git a/.idea/vcs.xml b/.idea/vcs.xml new file mode 100644 index 0000000..94a25f7 --- /dev/null +++ b/.idea/vcs.xml @@ -0,0 +1,6 @@ + + + + + + \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..0079f32 --- /dev/null +++ b/README.md @@ -0,0 +1,15 @@ +# Control Sony BRAVIA 8A via HTTPS in Home Assistant + +The BRAVIA 8A uses only HTTPS for communication, which makes the Home Assistant integration SonyBravia unusable. This project provides a workaround. + +We get the directory ```pybravia``` from https://github.com/Drafteed/pybravia rename it to ```pybravia8``` and modify it to use HTTPS (changes in ```client.py```). The Module ```bravia8``` uses this. + +The main script ```bravia8_pyscript.py``` utilizes the HACS custom integration ```pyscript``` (https://github.com/custom-components/pyscript). + +## Installation + +Install ```pyscript``` following the instructions on the [pyscript GitHub repository](https://github.com/custom-components/pyscript). + +Copy ```bravia8/``` and ```pybravia8/``` to ```/pyscript/modules/```. + +Copy ```bravia8_pyscript.py``` to ```/pyscript/```. diff --git a/bravia8/__init__.py b/bravia8/__init__.py new file mode 100644 index 0000000..74d0d61 --- /dev/null +++ b/bravia8/__init__.py @@ -0,0 +1,2 @@ + +from .remote import BraviaRemote diff --git a/bravia8/remote.py b/bravia8/remote.py new file mode 100644 index 0000000..f90c357 --- /dev/null +++ b/bravia8/remote.py @@ -0,0 +1,90 @@ +import logging + +from pybravia8 import BraviaClient, BraviaError + +_LOGGER = logging.getLogger(__name__) + +class BraviaRemote: + def __init__(self, host, psk): + self.client = BraviaClient(host) + self.psk = psk + self.connected = False + + async def connect(self): + if not self.connected: + try: + await self.client.connect(psk=self.psk) + self.connected = True + except BraviaError as e: + _LOGGER.error(f"Error connecting to Bravia TV: {e}") + + async def disconnect(self): + await self.client.disconnect() + self.connected = False + + async def send_command(self, cmd): + await self.connect() + try: + _LOGGER.info(f"Sending {cmd}") + rc = await self.client.send_command(cmd) + _LOGGER.info(f"send command '{cmd}' returned: {rc}") + except BraviaError as e: + _LOGGER.error(f"Error sending command '{cmd}': {e}") + + async def turn_on(self): + await self.disconnect() + await self.connect() + try: + _LOGGER.info(f"Turning TV on") + rc = await self.client.turn_on() + _LOGGER.info(f"turn_on() returned: {rc}") + except BraviaError as e: + _LOGGER.error(f"Error turning TV on: {e}") + + async def turn_off(self): + await self.connect() + try: + _LOGGER.info(f"Turning TV off") + rc = await self.client.turn_off() + _LOGGER.info(f"turn_off() returned: {rc}") + await self.disconnect() + except BraviaError as e: + _LOGGER.error(f"Error turning TV off: {e}") + + async def volume_up(self): + await self.connect() + try: + _LOGGER.info(f"Increasing volume") + rc = await self.client.volume_up() + _LOGGER.info(f"volume_up() returned: {rc}") + except BraviaError as e: + _LOGGER.error(f"Error increasing volume: {e}") + + async def volume_down(self): + await self.connect() + try: + _LOGGER.info(f"Decreasing volume") + rc = await self.client.volume_down() + _LOGGER.info(f"volume_down() returned: {rc}") + except BraviaError as e: + _LOGGER.error(f"Error decreasing volume: {e}") + + async def mute(self, mute: bool = None): + await self.connect() + try: + _LOGGER.info(f"Toggling mute") + rc = await self.client.volume_mute(mute) + _LOGGER.info(f"mute() returned: {rc}") + except BraviaError as e: + _LOGGER.error(f"Error toggling mute: {e}") + + async def dump_commands(self): + await self.connect() + cmd_list = await self.client.get_command_list() + commands = ", ".join(cmd_list.keys()) + _LOGGER.info(f"Available commands: \n{commands}") + + async def dump_info(self): + await self.connect() + info = await self.client.get_system_info() + _LOGGER.info(f"System info: \n{info}") diff --git a/bravia8_pyscript.py b/bravia8_pyscript.py new file mode 100644 index 0000000..bb38b4e --- /dev/null +++ b/bravia8_pyscript.py @@ -0,0 +1,51 @@ +import logging + +from bravia8 import BraviaRemote + +logging.basicConfig(level=logging.DEBUG) +_LOGGER = logging.getLogger(__name__) + +HOST = "10.4.2.60" +PSK = "12345678901234567890" + +remote = BraviaRemote(host=HOST, psk=PSK) + +@service +def bravia_send_command(command=None): + remote.send_command(command) + +@service +def bravia_turn_on(): + remote.turn_on() + +@service +def bravia_turn_off(): + remote.turn_off() + +@service +def bravia_volume_up(): + remote.volume_up() + +@service +def bravia_volume_down(): + remote.volume_down() + +@service +def bravia_mute(): + remote.mute() + +@service +def bravia_mute_on(): + remote.mute(True) + +@service +def bravia_mute_off(): + remote.mute(False) + +@service +def bravia_dump_commands(): + remote.dump_commands() + +@service +def bravia_dump_info(): + remote.dump_info() diff --git a/commands.txt b/commands.txt new file mode 100644 index 0000000..a9d5e13 --- /dev/null +++ b/commands.txt @@ -0,0 +1,147 @@ +Num1: AAAAAQAAAAEAAAAAAw== +Num2: AAAAAQAAAAEAAAABAw== +Num3: AAAAAQAAAAEAAAACAw== +Num4: AAAAAQAAAAEAAAADAw== +Num5: AAAAAQAAAAEAAAAEAw== +Num6: AAAAAQAAAAEAAAAFAw== +Num7: AAAAAQAAAAEAAAAGAw== +Num8: AAAAAQAAAAEAAAAHAw== +Num9: AAAAAQAAAAEAAAAIAw== +Num0: AAAAAQAAAAEAAAAJAw== +Num11: AAAAAQAAAAEAAAAKAw== +Num12: AAAAAQAAAAEAAAALAw== +Enter: AAAAAQAAAAEAAAALAw== +GGuide: AAAAAQAAAAEAAAAOAw== +ChannelUp: AAAAAQAAAAEAAAAQAw== +ChannelDown: AAAAAQAAAAEAAAARAw== +VolumeUp: AAAAAQAAAAEAAAASAw== +VolumeDown: AAAAAQAAAAEAAAATAw== +Mute: AAAAAQAAAAEAAAAUAw== +TvPower: AAAAAQAAAAEAAAAVAw== +MediaAudioTrack: AAAAAQAAAAEAAAAXAw== +Tv: AAAAAQAAAAEAAAAkAw== +TvInput: AAAAAQAAAAEAAAAlAw== +TvAntennaCable: AAAAAQAAAAEAAAAqAw== +WakeUp: AAAAAQAAAAEAAAAuAw== +Sleep: AAAAAQAAAAEAAAAvAw== +Right: AAAAAQAAAAEAAAAzAw== +Left: AAAAAQAAAAEAAAA0Aw== +SleepTimer: AAAAAQAAAAEAAAA2Aw== +TvAnalog: AAAAAQAAAAEAAAA4Aw== +Display: AAAAAQAAAAEAAAA6Aw== +Jump: AAAAAQAAAAEAAAA7Aw== +PictureOff: AAAAAQAAAAEAAAA+Aw== +Teletext: AAAAAQAAAAEAAAA/Aw== +Video1: AAAAAQAAAAEAAABAAw== +Video2: AAAAAQAAAAEAAABBAw== +AnalogRgb1: AAAAAQAAAAEAAABDAw== +Home: AAAAAQAAAAEAAABgAw== +Exit: AAAAAQAAAAEAAABjAw== +PictureMode: AAAAAQAAAAEAAABkAw== +Confirm: AAAAAQAAAAEAAABlAw== +Up: AAAAAQAAAAEAAAB0Aw== +Down: AAAAAQAAAAEAAAB1Aw== +*AD: AAAAAgAAABoAAAA7Aw== +AudioMixUp: AAAAAgAAABoAAAA8Aw== +AudioMixDown: AAAAAgAAABoAAAA9Aw== +PhotoFrame: AAAAAgAAABoAAABVAw== +Tv_Radio: AAAAAgAAABoAAABXAw== +SyncMenu: AAAAAgAAABoAAABYAw== +Hdmi1: AAAAAgAAABoAAABaAw== +Hdmi2: AAAAAgAAABoAAABbAw== +Hdmi3: AAAAAgAAABoAAABcAw== +Hdmi4: AAAAAgAAABoAAABdAw== +TopMenu: AAAAAgAAABoAAABgAw== +PopUpMenu: AAAAAgAAABoAAABhAw== +OneTouchTimeRec: AAAAAgAAABoAAABkAw== +OneTouchView: AAAAAgAAABoAAABlAw== +ControlMenu: AAAAAgAAABoAAAB1Aw== +FootballMode: AAAAAgAAABoAAAB2Aw== +iManual: AAAAAgAAABoAAAB7Aw== +Netflix: AAAAAgAAABoAAAB8Aw== +Analog: AAAAAgAAAHcAAAANAw== +Mode3D: AAAAAgAAAHcAAABNAw== +DigitalToggle: AAAAAgAAAHcAAABSAw== +DemoSurround: AAAAAgAAAHcAAAB7Aw== +TenKey: AAAAAgAAAJcAAAAMAw== +PlayPause: AAAAAgAAAJcAAAAPAw== +BSCS: AAAAAgAAAJcAAAAQAw== +Ddata: AAAAAgAAAJcAAAAVAw== +Stop: AAAAAgAAAJcAAAAYAw== +Pause: AAAAAgAAAJcAAAAZAw== +Play: AAAAAgAAAJcAAAAaAw== +Rewind: AAAAAgAAAJcAAAAbAw== +Forward: AAAAAgAAAJcAAAAcAw== +DOT: AAAAAgAAAJcAAAAdAw== +Rec: AAAAAgAAAJcAAAAgAw== +Return: AAAAAgAAAJcAAAAjAw== +Blue: AAAAAgAAAJcAAAAkAw== +Red: AAAAAgAAAJcAAAAlAw== +Green: AAAAAgAAAJcAAAAmAw== +Yellow: AAAAAgAAAJcAAAAnAw== +SubTitle: AAAAAgAAAJcAAAAoAw== +CS: AAAAAgAAAJcAAAArAw== +BS: AAAAAgAAAJcAAAAsAw== +Digital: AAAAAgAAAJcAAAAyAw== +Options: AAAAAgAAAJcAAAA2Aw== +Media: AAAAAgAAAJcAAAA4Aw== +Prev: AAAAAgAAAJcAAAA8Aw== +Next: AAAAAgAAAJcAAAA9Aw== +DpadCenter: AAAAAgAAAJcAAABKAw== +CursorLeft: AAAAAgAAAJcAAABNAw== +CursorRight: AAAAAgAAAJcAAABOAw== +CursorUp: AAAAAgAAAJcAAABPAw== +CursorDown: AAAAAgAAAJcAAABQAw== +ShopRemoteControlForcedDynamic: AAAAAgAAAJcAAABqAw== +FlashPlus: AAAAAgAAAJcAAAB4Aw== +FlashMinus: AAAAAgAAAJcAAAB5Aw== +DemoMode: AAAAAgAAAJcAAAB8Aw== +ClosedCaption: AAAAAgAAAKQAAAAQAw== +Component1: AAAAAgAAAKQAAAA2Aw== +Component2: AAAAAgAAAKQAAAA3Aw== +Wide: AAAAAgAAAKQAAAA9Aw== +EPG: AAAAAgAAAKQAAABbAw== +PAP: AAAAAgAAAKQAAAB3Aw== +AdvancedBSCS: AAAAAgAAAMQAAAAMAw== +AudioOutput_AudioSystem: AAAAAgAAAMQAAAAiAw== +AudioOutput_Toggle: AAAAAgAAAMQAAAAkAw== +DarwinLoop: AAAAAgAAAMQAAAAlAw== +ApplicationLauncher: AAAAAgAAAMQAAAAqAw== +AcousticCenterSync: AAAAAgAAAMQAAAArAw== +Assists: AAAAAgAAAMQAAAA7Aw== +FeaturedApp: AAAAAgAAAMQAAABEAw== +FeaturedAppVOD: AAAAAgAAAMQAAABFAw== +GooglePlay: AAAAAgAAAMQAAABGAw== +YouTube: AAAAAgAAAMQAAABHAw== +RecorderMenu: AAAAAgAAAMQAAABIAw== +STBMenu: AAAAAgAAAMQAAABJAw== +ActionMenu: AAAAAgAAAMQAAABLAw== +Help: AAAAAgAAAMQAAABNAw== +TvSatellite: AAAAAgAAAMQAAABOAw== +AndroidMenu: AAAAAgAAAMQAAABPAw== +MicPressedFromPairedRC: AAAAAgAAAMQAAABuAw== +MicPressedFromUnpairedRC: AAAAAgAAAMQAAABvAw== +GoogleDashboard: AAAAAgAAAMQAAABwAw== +AutoParingRequest: AAAAAgAAAMQAAAB7Aw== +B2bCustomizeKey: AAAAAgAAAMQAAAB8Aw== +WirelessSubwoofer: AAAAAgAAAMQAAAB+Aw== +PartnerApp1: AAAAAwAACB8AAAAAAw== +PartnerApp2: AAAAAwAACB8AAAABAw== +PartnerApp3: AAAAAwAACB8AAAACAw== +PartnerApp4: AAAAAwAACB8AAAADAw== +PartnerApp5: AAAAAwAACB8AAAAEAw== +PartnerApp6: AAAAAwAACB8AAAAFAw== +PartnerApp7: AAAAAwAACB8AAAAGAw== +PartnerApp8: AAAAAwAACB8AAAAHAw== +PartnerApp9: AAAAAwAACB8AAAAIAw== +PartnerApp10: AAAAAwAACB8AAAAJAw== +PartnerApp11: AAAAAwAACB8AAAAKAw== +PartnerApp12: AAAAAwAACB8AAAALAw== +PartnerApp13: AAAAAwAACB8AAAAMAw== +PartnerApp14: AAAAAwAACB8AAAANAw== +PartnerApp15: AAAAAwAACB8AAAAOAw== +PartnerApp16: AAAAAwAACB8AAAAPAw== +PartnerApp17: AAAAAwAACB8AAAAQAw== +PartnerApp18: AAAAAwAACB8AAAARAw== +PartnerApp19: AAAAAwAACB8AAAASAw== +PartnerApp20: AAAAAwAACB8AAAATAw== diff --git a/pybravia8/__init__.py b/pybravia8/__init__.py new file mode 100644 index 0000000..549bc9e --- /dev/null +++ b/pybravia8/__init__.py @@ -0,0 +1,20 @@ +"""Python library for remote control of Sony Bravia TV.""" + +""" *hb* + + Modified for for Sony Bravia 8A: + Use https with self signed certificate -> client.py +""" +# flake8: noqa +from .client import BraviaClient +from .exceptions import ( + BraviaAuthError, + BraviaConnectionError, + BraviaConnectionTimeout, + BraviaError, + BraviaNotFound, + BraviaNotSupported, + BraviaTurnedOff, +) + +__version__ = "0.3.4.hb" diff --git a/pybravia8/client.py b/pybravia8/client.py new file mode 100644 index 0000000..af4a361 --- /dev/null +++ b/pybravia8/client.py @@ -0,0 +1,627 @@ +"""Python library for remote control of Sony Bravia TV.""" +from __future__ import annotations + +import asyncio +import logging +import socket +from contextlib import suppress +from datetime import datetime, timedelta +from types import TracebackType +from typing import Any + +from aiohttp import BasicAuth, ClientError, ClientSession, CookieJar + +from .const import ( + CODE_POWER_ON, + DEFAULT_AUDIO_TARGET, + DEFAULT_TIMEOUT, + DEFAULT_VERSION, + PAIR_PIN, + SERVICE_ACCESS_CONTROL, + SERVICE_APP_CONTROL, + SERVICE_AUDIO, + SERVICE_AV_CONTENT, + SERVICE_GUIDE, + SERVICE_IRCC, + SERVICE_SYSTEM, + SERVICE_VIDEO, +) +from .exceptions import ( + BraviaAuthError, + BraviaConnectionError, + BraviaConnectionTimeout, + BraviaError, + BraviaNotFound, + BraviaNotSupported, + BraviaTurnedOff, +) +from .util import normalize_cookies + +_LOGGER = logging.getLogger(__name__) + + +class BraviaClient: + """Represent a Bravia Client.""" + + def __init__( + self, host: str, mac: str | None = None, session: ClientSession | None = None + ) -> None: + """Initialize the device.""" + self.host = host + self.mac = mac + self._session = session + self._auth: BasicAuth | None = None + self._psk: str | None = None + self._ircc_time: datetime | None = None + self._ircc_endpoint = SERVICE_IRCC + self._commands: dict[str, str] = {} + + async def connect( + self, + pin: str | None = None, + psk: str | None = None, + clientid: str | None = None, + nickname: str | None = None, + ) -> None: + """Connect to device with PIN or PSK.""" + self._psk = pin[4:] if pin and pin[:4] == "psk:" else psk + + _LOGGER.debug( + "Connect with pin: %s, psk: %s, clientid: %s, nickname: %s", + pin, + self._psk, + clientid, + nickname, + ) + + if self._psk is None: + assert pin is not None + assert clientid is not None + assert nickname is not None + await self.register(pin, clientid, nickname) + else: + if self._session is not None: + self._auth = None + self._session.cookie_jar.clear() + + system_info = await self.get_system_info() + if not system_info: + raise BraviaNotSupported + + if self._psk: + _LOGGER.debug("Connected with PSK") + else: + _LOGGER.debug( + "Connected with PIN, cookie len: %s", len(self._session._cookie_jar) + ) + + async def register(self, pin: str, clientid: str, nickname: str) -> None: + """Register the device with PIN.""" + self._auth = BasicAuth("", pin) + params = [ + {"clientid": clientid, "nickname": nickname, "level": "private"}, + [{"value": "yes", "function": "WOL"}], + ] + await self.send_rest_req( + SERVICE_ACCESS_CONTROL, + "actRegister", + params, + ) + + async def pair(self, clientid: str, nickname: str) -> None: + """Register with PIN "0000" to start the pairing process on the TV.""" + with suppress(BraviaAuthError): + await self.register(PAIR_PIN, clientid, nickname) + + async def disconnect(self) -> None: + """Close connection.""" + if self._session: + await self._session.close() + self._auth = None + self._psk = None + self._session = None + + async def send_wol_req(self) -> bool: + """Send WOL packet to device.""" + if self.mac is None: + return False + mac = self.mac.replace(":", "") + packet = bytes.fromhex("F" * 12 + mac * 16) + + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: + sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) + sock.sendto(packet, ("", 9)) + return True + + async def send_req( + self, + url: str, + data: Any = None, + headers: dict[str, Any] | None = None, + json: bool = True, + timeout: int = DEFAULT_TIMEOUT, + ) -> Any: + """Send HTTP request.""" + result = {} if json else False + + if self._session is None: + self._session = ClientSession( + cookie_jar=CookieJar(unsafe=True, quote_cookie=False) + ) + + if headers is None: + headers = {} + + if self._psk: + headers["X-Auth-PSK"] = self._psk + + headers["Cache-Control"] = "no-cache" + headers["Connection"] = "keep-alive" + + _LOGGER.debug("Request %s, data: %s, headers: %s", url, data, headers) + + try: + if json: + # *hb* + ssl=False + response = await self._session.post( + url, json=data, headers=headers, timeout=timeout, auth=self._auth, ssl=False + ) + else: + # *hb* + ssl=False + response = await self._session.post( + url, data=data, headers=headers, timeout=timeout, auth=self._auth, ssl=False + ) + + _LOGGER.debug("Response status: %s", response.status) + + # Normalize non RFC-compliant cookie + # https://github.com/Drafteed/pybravia/issues/1#issuecomment-1237452709 + cookies = response.headers.getall("set-cookie", None) + if cookies: + normalized_cookies = normalize_cookies(cookies) + self._session.cookie_jar.update_cookies( + cookies=normalized_cookies, response_url=response.url + ) + + if response.status == 200: + result = await response.json() if json else True + _LOGGER.debug("Response result: %s", result) + if response.status == 404: + raise BraviaNotFound + if response.status in [401, 403]: + raise BraviaAuthError + except ClientError as err: + _LOGGER.debug("Request error %s", err) + raise BraviaConnectionError from err + except ConnectionError as err: + _LOGGER.debug("Connection error %s", err) + raise BraviaConnectionError from err + except asyncio.exceptions.TimeoutError as err: + _LOGGER.debug("Request timeout %s", err) + raise BraviaConnectionTimeout from err + + return result + + async def send_ircc_req(self, code: str, timeout: int = DEFAULT_TIMEOUT) -> bool: + """Send IRCC request to device.""" + + # After about 13 minutes of inactivity, some TVs + # ignores the first command without giving any error. + # This make an empty request to 'wake up' the api. + # https://github.com/dcnielsen90/python-bravia-tv/issues/15 + # The endpoint of some TVs is case sensitive, this also detects this. + # https://github.com/home-assistant/core/issues/86132 + if code != "": + time = datetime.now() + if not self._ircc_time or (time - self._ircc_time) > timedelta(minutes=10): + try: + await self.send_ircc_req("") + except BraviaNotFound: + self._ircc_endpoint = self._ircc_endpoint.swapcase() + except BraviaError: + pass + self._ircc_time = time + + # *hb* http -> https + url = f"https://{self.host}/sony/{self._ircc_endpoint}" + headers = { + "SOAPACTION": '"urn:schemas-sony-com:service:IRCC:1#X_SendIRCC"', + "Content-Type": "text/xml; charset=UTF-8", + } + data = ( + "' + "" + '' + f"{code}" + "" + "" + "" + ) + + return await self.send_req( + url=url, data=data, headers=headers, json=False, timeout=timeout + ) + + async def send_rest_req( + self, + service: str, + method: str, + params: Any = None, + headers: dict[str, Any] | None = None, + version: str = DEFAULT_VERSION, + timeout: int = DEFAULT_TIMEOUT, + ) -> Any: + """Send REST request to device.""" + # *hb* http -> https + url = f"https://{self.host}/sony/{service}" + params = params if isinstance(params, list) else [params] if params else [] + data = { + "method": method, + "params": params, + "id": 1, + "version": version, + } + + resp = await self.send_req( + url=url, data=data, headers=headers, json=True, timeout=timeout + ) + + if error := resp.get("error"): + if "not power-on" in error: + raise BraviaTurnedOff + + return resp + + async def send_rest_quick(self, *args: Any, **kwargs: Any) -> bool: + """Send and quick check REST request to device.""" + resp = await self.send_rest_req(*args, **kwargs) + return "result" in resp + + async def send_command(self, command: str) -> bool: + """Send IRCC command to device.""" + code = await self.get_command_code(command) + if code is None: + return False + return await self.send_ircc_req(code) + + async def get_sound_settings(self) -> str: + """Get current sound output.""" + resp = await self.send_rest_req( + SERVICE_AUDIO, + "getSoundSettings", + {"target": "outputTerminal"}, + version="1.1", + ) + result = resp.get("result", [[{}]])[0][0] + return result.get("currentValue", "") + + async def get_power_status(self) -> str: + """Get current power status.""" + resp = await self.send_rest_req(SERVICE_SYSTEM, "getPowerStatus", timeout=5) + result = resp.get("result", [{}])[0] + return result.get("status", "off") + + async def get_power_saving_mode(self) -> str: + """Get current power saving mode.""" + resp = await self.send_rest_req(SERVICE_SYSTEM, "getPowerSavingMode") + result = resp.get("result", [{}])[0] + return result.get("mode", "") + + async def get_system_info(self) -> dict[str, str]: + """Get general information of the device.""" + resp = await self.send_rest_req(SERVICE_SYSTEM, "getSystemInformation") + result = resp.get("result", [{}])[0] + self.mac = result.get("macAddr", self.mac) + return result + + async def get_api_info(self, services: list[str]) -> list[dict[str, Any]]: + """Get supported services and their information.""" + resp = await self.send_rest_req( + SERVICE_GUIDE, + "getSupportedApiInfo", + {"services": services}, + ) + result = resp.get("result", [[]])[0] + return result + + async def get_wol_mode(self) -> bool: + """Get information on the device's WOL mode settings.""" + resp = await self.send_rest_req(SERVICE_SYSTEM, "getWolMode") + result = resp.get("result", [{}])[0] + return result.get("enabled", False) + + async def get_led_status(self) -> dict[str, str]: + """Get the LED Indicator mode and status.""" + resp = await self.send_rest_req(SERVICE_SYSTEM, "getLEDIndicatorStatus") + result = resp.get("result", [{}])[0] + return result + + async def get_remote_info(self) -> list[Any]: + """Get information of the device's remote controller.""" + resp = await self.send_rest_req(SERVICE_SYSTEM, "getRemoteControllerInfo") + result = resp.get("result", [{}, []]) + return result + + async def get_command_list(self) -> dict[str, str]: + """Get list of all IRCC commands.""" + result = await self.get_remote_info() + return {x["name"]: x["value"] for x in result[1]} + + async def get_command_code(self, command: str) -> str | None: + """Get IRCC command code by name.""" + if not self._commands: + self._commands = await self.get_command_list() + return self._commands.get(command) + + async def get_volume_info( + self, target: str = DEFAULT_AUDIO_TARGET + ) -> dict[str, Any]: + """Get the sound volume information with preferred target.""" + result = await self.get_volume_info_full() + value = {} + for output in result: + value = output + if target == output.get("target"): + break + return value + + async def get_volume_info_full(self) -> list[dict[str, Any]]: + """Get information about the sound volume and mute status.""" + resp = await self.send_rest_req(SERVICE_AUDIO, "getVolumeInformation") + result = resp.get("result", [[]])[0] + return result + + async def get_picture_setting(self, target: str = "") -> list[dict[str, Any]]: + """Get information about the picture settings.""" + resp = await self.send_rest_req( + SERVICE_VIDEO, "getPictureQualitySettings", {"target": target} + ) + result = resp.get("result", [[]])[0] + return result + + async def get_app_list(self) -> list[dict[str, str]]: + """Get list of applications.""" + resp = await self.send_rest_req(SERVICE_APP_CONTROL, "getApplicationList") + result = resp.get("result", [[]])[0] + return result + + async def get_scheme_list(self) -> list[str]: + """Get list of schemes that the device can handle.""" + resp = await self.send_rest_req(SERVICE_AV_CONTENT, "getSchemeList") + result = resp.get("result", [[]])[0] + return [d["scheme"] for d in result if "scheme" in d] + + async def get_source_list(self, scheme: str) -> list[str]: + """Get list of sources in the scheme.""" + resp = await self.send_rest_req( + SERVICE_AV_CONTENT, + "getSourceList", + {"scheme": scheme}, + ) + result = resp.get("result", [[]])[0] + return [d["source"] for d in result if "source" in d] + + async def get_content_count(self, source: str) -> int: + """Get count of contents in the source.""" + + # Extended timeout due to + # https://github.com/home-assistant/core/issues/91781#issuecomment-1532151925 + resp = await self.send_rest_req( + SERVICE_AV_CONTENT, + "getContentCount", + {"source": source}, + timeout=20, + ) + result = resp.get("result", [{}])[0] + return result.get("count", 0) + + async def get_content_list( + self, source: str, index: int = 0, count: int = 50 + ) -> list[dict[str, Any]]: + """Get list of contents in the source.""" + + # Extended timeout due to + # https://github.com/home-assistant/core/issues/91781#issuecomment-1532168838 + resp = await self.send_rest_req( + SERVICE_AV_CONTENT, + "getContentList", + {"source": source, "stIdx": index, "cnt": count}, + timeout=20, + ) + result = resp.get("result", [[]])[0] + return result + + async def get_content_list_full(self, source: str) -> list[dict[str, Any]]: + """Get full list of contents in the source.""" + result = [] + total = await self.get_content_count(source) + for index in range(0, total, 50): + count = min(50, total - index) + resp = await self.get_content_list(source, index, count) + result.extend(resp) + return result + + async def get_content_list_all(self, scheme: str) -> list[dict[str, Any]]: + """Get full list of contents in the scheme.""" + result = [] + sources = await self.get_source_list(scheme) + for source in sources: + resp = await self.get_content_list_full(source) + result.extend(resp) + return result + + async def get_external_status( + self, version: str = DEFAULT_VERSION + ) -> list[dict[str, Any]]: + """Get information of all external input sources.""" + resp = await self.send_rest_req( + SERVICE_AV_CONTENT, + "getCurrentExternalInputsStatus", + version=version, + ) + result = resp.get("result", [[]])[0] + return result + + async def get_playing_info(self) -> dict[str, Any]: + """Get information of the currently playing content.""" + resp = await self.send_rest_req(SERVICE_AV_CONTENT, "getPlayingContentInfo") + result = resp.get("result", [{}])[0] + return result + + async def set_picture_setting(self, target: str, value: str) -> bool: + """Set a picture setting to a new value.""" + return await self.send_rest_req( + SERVICE_VIDEO, + "setPictureQualitySettings", + {"settings": [{"target": target, "value": value}]}, + version="1.0", + ) + + async def set_wol_mode(self, mode: bool) -> bool: + """Set WOL mode settings of the device.""" + return await self.send_rest_quick( + SERVICE_SYSTEM, "setWolMode", {"enabled": mode} + ) + + async def set_led_status(self, mode: str, status: str | None = None) -> bool: + """Set the LED Indicator mode and status.""" + return await self.send_rest_quick( + SERVICE_SYSTEM, + "setLEDIndicatorStatus", + {"mode": mode, "status": status}, + version="1.1", + ) + + async def set_active_app(self, uri: str) -> bool: + """Launch an application by URI.""" + return await self.send_rest_quick( + SERVICE_APP_CONTROL, "setActiveApp", {"uri": uri} + ) + + async def set_play_content(self, uri: str) -> bool: + """Play content by URI.""" + return await self.send_rest_quick( + SERVICE_AV_CONTENT, "setPlayContent", {"uri": uri} + ) + + async def set_sound_settings(self, output: str) -> bool: + """Change current sound output.""" + return await self.send_rest_quick( + SERVICE_AUDIO, + "setSoundSettings", + [{"settings": [{"value": output, "target": "outputTerminal"}]}], + version="1.1", + ) + + async def set_power_status(self, status: bool) -> bool: + """Change the current power status of the device.""" + return await self.send_rest_quick( + SERVICE_SYSTEM, "setPowerStatus", {"status": status} + ) + + async def set_power_saving_mode(self, mode: str) -> bool: + """Change the setting of the power saving mode.""" + return await self.send_rest_quick( + SERVICE_SYSTEM, "setPowerSavingMode", {"mode": mode} + ) + + async def set_text_form(self, text: str) -> bool: + """Input text on the field of the software keyboard.""" + return await self.send_rest_quick(SERVICE_APP_CONTROL, "setTextForm", text) + + async def turn_on(self) -> bool: + """Turn on the device.""" + await self.send_wol_req() + with suppress(BraviaError): + if await self.get_power_status() != "active": + with suppress(BraviaError): + await self.set_power_status(True) + with suppress(BraviaError): + await self.send_ircc_req(CODE_POWER_ON) + return True + + async def turn_off(self) -> bool: + """Turn off the device.""" + return await self.set_power_status(False) + + async def volume_up(self, step: int = 1, **kwargs: str) -> bool: + """Increase the volume.""" + level = f"+{step}" + return await self.volume_level(level, **kwargs) + + async def volume_down(self, step: int = 1, **kwargs: str) -> bool: + """Decrease the volume.""" + level = f"-{step}" + return await self.volume_level(level, **kwargs) + + async def volume_level( + self, + level: int | str, + target: str = DEFAULT_AUDIO_TARGET, + ui_mode: str | None = None, + version: str = DEFAULT_VERSION, + ) -> bool: + """Change the audio volume level.""" + params = {"target": target, "volume": str(level)} + if ui_mode is not None: + params["ui"] = ui_mode + return await self.send_rest_quick( + SERVICE_AUDIO, "setAudioVolume", params, version=version + ) + + async def volume_mute(self, mute: bool | None = None) -> bool: + """Change the audio mute status.""" + if mute is None: + volume_info = await self.get_volume_info() + mute = not volume_info.get("mute") + return await self.send_rest_quick( + SERVICE_AUDIO, "setAudioMute", {"status": mute} + ) + + async def play(self) -> bool: + """Send play command.""" + return await self.send_command("Play") + + async def pause(self) -> bool: + """Send pause command.""" + return await self.send_command("Pause") + + async def stop(self) -> bool: + """Send stop command.""" + return await self.send_command("Stop") + + async def next_track(self) -> bool: + """Send next track command.""" + return await self.send_command("Next") + + async def previous_track(self) -> bool: + """Send previous track command.""" + return await self.send_command("Prev") + + async def channel_up(self) -> bool: + """Send next channel command.""" + return await self.send_command("ChannelUp") + + async def channel_down(self) -> bool: + """Send previous channel command.""" + return await self.send_command("ChannelDown") + + async def terminate_apps(self) -> bool: + """Terminate all opened applications.""" + return await self.send_rest_quick(SERVICE_APP_CONTROL, "terminateApps") + + async def reboot(self) -> bool: + """Send command to reboot the device.""" + return await self.send_rest_quick(SERVICE_SYSTEM, "requestReboot") + + async def __aenter__(self) -> BraviaClient: + """Connect the client with context manager.""" + return self + + async def __aexit__( + self, exc_type: Exception, exc_value: str, traceback: TracebackType + ) -> None: + """Disconnect from context manager.""" + await self.disconnect() diff --git a/pybravia8/const.py b/pybravia8/const.py new file mode 100644 index 0000000..335ca51 --- /dev/null +++ b/pybravia8/const.py @@ -0,0 +1,21 @@ +"""Bravia constants.""" +from __future__ import annotations + +from typing import Final + +CODE_POWER_ON: Final = "AAAAAQAAAAEAAAAuAw==" + +DEFAULT_AUDIO_TARGET: Final = "speaker" +DEFAULT_TIMEOUT: Final = 10 +DEFAULT_VERSION: Final = "1.0" + +PAIR_PIN: Final = "0000" + +SERVICE_ACCESS_CONTROL: Final = "accessControl" +SERVICE_APP_CONTROL: Final = "appControl" +SERVICE_AUDIO: Final = "audio" +SERVICE_VIDEO: Final = "video" +SERVICE_AV_CONTENT: Final = "avContent" +SERVICE_GUIDE: Final = "guide" +SERVICE_IRCC: Final = "ircc" +SERVICE_SYSTEM: Final = "system" diff --git a/pybravia8/exceptions.py b/pybravia8/exceptions.py new file mode 100644 index 0000000..492702c --- /dev/null +++ b/pybravia8/exceptions.py @@ -0,0 +1,29 @@ +"""Bravia exceptions.""" + + +class BraviaError(Exception): + """Base Bravia exception.""" + + +class BraviaAuthError(BraviaError): + """Raised to indicate auth error.""" + + +class BraviaNotFound(BraviaError): + """Raised to indicate not found error.""" + + +class BraviaNotSupported(BraviaError): + """Raised to indicate not supported error.""" + + +class BraviaConnectionError(BraviaError): + """Raised to indicate connection error.""" + + +class BraviaConnectionTimeout(BraviaError): + """Raised to indicate connection timeout.""" + + +class BraviaTurnedOff(BraviaError): + """Raised to indicate that TV is turned off and do not respond.""" diff --git a/pybravia8/py.typed b/pybravia8/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/pybravia8/util.py b/pybravia8/util.py new file mode 100644 index 0000000..579ead5 --- /dev/null +++ b/pybravia8/util.py @@ -0,0 +1,17 @@ +"""Utils and helpers.""" +from __future__ import annotations + +import re +from http.cookies import SimpleCookie + +REGEXP_COOKIE_EXPIRES = re.compile("(;\\s?expires=(.*)(;|$))", re.IGNORECASE) + + +def normalize_cookies(cookies: list[str]) -> SimpleCookie: + """Normalize non RFC-compliant cookies.""" + result: SimpleCookie = SimpleCookie() + + for cookie in cookies: + result.load(REGEXP_COOKIE_EXPIRES.sub("", cookie)) + + return result