diff --git a/apps/global/homecinema.yaml b/apps/global/homecinema.yaml index e219e97..1593e74 100644 --- a/apps/global/homecinema.yaml +++ b/apps/global/homecinema.yaml @@ -1,5 +1,5 @@ samsung_the_frame: - module: samsungtheframe + module: samsungtheframeasync class: SamsungTheFrame volume_up_events: diff --git a/apps/samsungtheframeasync.py b/apps/samsungtheframeasync.py new file mode 100644 index 0000000..1bb02d0 --- /dev/null +++ b/apps/samsungtheframeasync.py @@ -0,0 +1,577 @@ +"""Persistent, non-blocking alternative to samsungtheframe. + +Keep ``class: SamsungTheFrame`` and change only ``module: SamsungTheFrameAsync``. +No YAML changes are needed. Network I/O runs on AppDaemon's asyncio loop. +""" + +import asyncio +import base64 +from contextlib import suppress +import json +import math +from pathlib import Path +import random +import socket +import uuid + +import aiohttp +from yarl import URL + +from ad_toolbox.smartobject import SmartObject +from ad_toolbox.eventhandler import EventHandler +import ad_toolbox.smartcondition as SmartCondition + + +class TVError(Exception): + """A failed connection or a rejected/malformed TV response.""" + + +NETWORK_ERRORS = (TVError, OSError, TimeoutError, aiohttp.ClientError, ValueError) + + +class _TVConnection: + """One reader per channel; Art responses and notifications share that reader. + + This intentionally uses aiohttp directly: the installed samsungtvws Art + client is synchronous, and cannot share recv() with a notification listener. + """ + + def __init__(self, session, host, token_file, on_event, timeout=2): + self.session = session + self.host = host + self.token_file = Path(token_file) + self.on_event = on_event + self.timeout = timeout + self.token = None + self.sockets = {} + self.readers = {} + self.pending = {} + self.locks = {name: asyncio.Lock() for name in ('art', 'remote')} + self.remote_lock = asyncio.Lock() + self.art_lock = asyncio.Lock() + self.closing = False + + async def load_token(self): + try: + self.token = (await asyncio.to_thread(self.token_file.read_text)).strip() or None + except FileNotFoundError: + pass + + def alive(self, channel): + ws = self.sockets.get(channel) + reader = self.readers.get(channel) + return ws is not None and not ws.closed and reader is not None and not reader.done() + + async def connect(self, channel): + async with self.locks[channel]: + if self.alive(channel): + return self.sockets[channel] + await self.close_channel(channel) + endpoint = 'com.samsung.art-app' if channel == 'art' else 'samsung.remote.control' + query = {'name': base64.b64encode(b'SamsungTvRemote').decode()} + if self.token: + query['token'] = self.token + url = URL.build(scheme='wss', host=self.host, port=8002, + path=f'/api/v2/channels/{endpoint}', query=query) + ws = None + try: + # Bound the whole handshake, including TV authorization and ready. + async with asyncio.timeout(self.timeout): + ws = await self.session.ws_connect(url, ssl=False, heartbeat=20, + timeout=self.timeout) + connected = False + while True: + message = await ws.receive() + frame = self._frame(message) + event = frame.get('event') + if event in ('ms.channel.unauthorized', 'ms.error'): + raise TVError('TV rejected connection; check TV authorization') + if event == 'ms.channel.connect': + connected = True + data = frame.get('data', {}) + if not isinstance(data, dict): + raise TVError('Invalid connection response') + token = data.get('token') + if token and str(token) != self.token: + self.token = str(token) + await asyncio.to_thread(self.token_file.write_text, self.token) + if connected and (channel == 'remote' or event == 'ms.channel.ready'): + break + except BaseException: + if ws is not None: + await ws.close() + raise + self.sockets[channel] = ws + self.readers[channel] = asyncio.create_task(self._read(channel, ws)) + return ws + + @staticmethod + def _frame(message): + if message.type != aiohttp.WSMsgType.TEXT: + raise TVError('TV WebSocket closed or returned a non-text message') + frame = json.loads(message.data) + if not isinstance(frame, dict): + raise TVError('Invalid TV WebSocket response') + return frame + + def _dispatch(self, frame): + if frame.get('event') != 'd2d_service_message': + if frame.get('event') == 'ms.error': + raise TVError('TV rejected WebSocket command') + return + data = frame.get('data', {}) + data = json.loads(data) if isinstance(data, str) else data + if not isinstance(data, dict): + raise TVError('Invalid Art response') + request_id = data.get('request_id') or data.get('id') + if request_id is not None and not isinstance(request_id, str): + raise TVError('Invalid Art request ID') + pending = self.pending.get(request_id) + if pending is not None and not pending.done(): + if data.get('event') == 'error': + pending.set_exception(TVError(f"Art request rejected: {data.get('error_code')}")) + else: + pending.set_result(data) + # Correlated status responses are applied by _refresh(), which can + # reject a response older than a notification received while waiting. + if pending is None or data.get('event') != 'artmode_status': + self.on_event(data) + + async def _read(self, channel, ws): + try: + async for message in ws: + self._dispatch(self._frame(message)) + except NETWORK_ERRORS: + pass + finally: + if channel == 'art': + for future in tuple(self.pending.values()): + if not future.done(): + future.set_exception(TVError('Art connection lost')) + await ws.close() + if not self.closing: + self.on_event({'event': 'connection_lost', 'channel': channel}) + + async def art_request(self, request, **params): + async with self.art_lock: + ws = await self.connect('art') + request_id = str(uuid.uuid4()) + future = asyncio.get_running_loop().create_future() + self.pending[request_id] = future + try: + async with asyncio.timeout(self.timeout): + await ws.send_json({'method': 'ms.channel.emit', 'params': { + 'event': 'art_app_request', 'to': 'host', + 'data': json.dumps({'request': request, 'id': request_id, + 'request_id': request_id, **params})}}) + return await future + except (TimeoutError, aiohttp.ClientError, OSError): + # Drop a timed-out channel: a delayed response must not satisfy + # the next request, and the next attempt needs a clean handshake. + await self.close_channel('art') + raise + finally: + self.pending.pop(request_id, None) + if not future.done(): + future.cancel() + + async def send_key(self, key, hold=0): + async with self.remote_lock: + ws = await self.connect('remote') + + async def send(cmd): + async with asyncio.timeout(self.timeout): + await ws.send_json({'method': 'ms.remote.control', 'params': { + 'Cmd': cmd, 'DataOfCmd': key, 'Option': 'false', + 'TypeOfRemote': 'SendRemoteKey'}}) + try: + await send('Press' if hold else 'Click') + if hold: + try: + await asyncio.sleep(hold) + finally: + # Even a cancelled transition must release a held key. + await send('Release') + except NETWORK_ERRORS: + await self.close_channel('remote') + raise + + async def app_status(self, app_id): + url = URL.build(scheme='https', host=self.host, port=8002, + path=f'/api/v2/applications/{app_id}') + async with self.session.get(url, ssl=False) as response: + response.raise_for_status() + result = await response.json(content_type=None) + if not isinstance(result, dict) or not isinstance(result.get('visible'), bool): + raise TVError('Invalid application status response') + return result + + async def close_channel(self, channel): + reader = self.readers.pop(channel, None) + ws = self.sockets.pop(channel, None) + if reader is not None: + reader.cancel() + with suppress(asyncio.CancelledError): + await reader + if ws is not None: + await ws.close() + + async def close(self): + self.closing = True + for channel in ('art', 'remote'): + await self.close_channel(channel) + + +class SamsungTheFrame(SmartObject): + """Same YAML contract as samsungtheframe.SamsungTheFrame. + + Optional settings: network_timeout (2 s), reconnect_delay (5 s, exponential + backoff capped at 60 s), fallback_refresh (60 s), power_threshold (35 W). + State/power callbacks only signal the async workers; they never contact TV. + """ + + VALID_STATES = ('art_mode', 'tv_off', 'hdmi', 'power_off', 'manual') + + def on_initialize_smart_object(self): + super().on_initialize_smart_object() + self._loop = self.AD.loop + self._stopped = False + self._wake = asyncio.Event() + self._ready = asyncio.Event() + self._volume = asyncio.Queue(maxsize=8) + self._target_task = None + self._tasks = set() + self._publish_lock = asyncio.Lock() + self._generation = 0 + self.requested_state = None + self._observed = None + self._observation_revision = 0 + self._refresh_at = 0 + self._retry_at = 0 + self._backoff = float(self.args.get('reconnect_delay', 5)) + self._client = None + self.power_plug = self.args['input_sensors']['power_plug'] + self.power_plug_power = self.args['input_sensors']['power_plug_power'] + outputs = self.args['output_sensors'] + self.is_on_entity = self.create_entity(outputs['is_on']) + self.is_art_mode_entity = self.create_entity(outputs['is_art_mode']) + self.tv_auto_shutdown_entity = self.create_entity( + outputs['tv_auto_shutdown'], + state=self.get_state(outputs['tv_auto_shutdown']) or 'off') + self.app_list = self.args.get('app_list', []) + self.tv_states_evaluators = [] + self.listen_state(self.update_sensors_cb, self.power_plug) + self.listen_state(self.update_sensors_cb, self.power_plug_power) + if 'tv_state' in self.args: + self.listen_state(self.on_update_tv_state_sensor, self.args['tv_state']) + self.request_tv_state(self.get_state(self.args['tv_state'])) + if 'tv_states' in self.args: + for state, condition in self.args['tv_states'].items(): + self.tv_states_evaluators.append((SmartCondition.Evaluator( + self, condition, condition_name=state, + on_update_cb=self.on_update_tv_states), state)) + self.on_update_tv_states() + self.volume_event_handlers = [] + for direction in ('up', 'down'): + if self.args.get(f'volume_{direction}_events'): + self.volume_event_handlers.append(EventHandler( + self, self.args[f'volume_{direction}_events'], + self.on_volume_event, direction)) + self._main_task = self.create_task(self._run()) + self.log_info('SamsungTheFrameAsync initialized (persistent async WebSockets)') + + def _post(self, callback, *args): + if not self._stopped: + self._loop.call_soon_threadsafe(callback, *args) + + def update_sensors_cb(self, *args): + self._post(self._wake.set) + + def on_update_tv_state_sensor(self, entity, attribute, old, new, kwargs): + if old != new: + self.request_tv_state(new) + + def on_update_tv_states(self): + for evaluator, state in self.tv_states_evaluators: + if evaluator.evaluate(False) == SmartCondition.Result.Succeeded: + evaluator.log_evaluation_result() + self.request_tv_state(state) + break + + def request_tv_state(self, state): + if state not in self.VALID_STATES: + self.log_info(f'Ignoring unavailable or invalid TV target: {state!r}') + return + self._post(self._set_target, state) + + def _set_target(self, state): + if self._stopped: + return + self._generation += 1 + self.requested_state = None if state == 'manual' else state + previous = self._target_task + if previous is not None and not previous.cancelling(): + previous.cancel() + self._target_task = self._spawn( + self._target_runner(previous, state, self._generation)) + self._wake.set() + + def on_volume_event(self, event_name, event_data, direction): + self._post(self._queue_volume, direction) + + def _queue_volume(self, direction): + # Drop repetitions instead of replaying stale volume presses after an outage. + if not self._stopped and not self._volume.full(): + self._volume.put_nowait((direction, self._loop.time())) + + def _on_art_event(self, data): + event = data.get('event') + mode = data.get('status') if event == 'art_mode_changed' else data.get('value') + if event in ('art_mode_changed', 'artmode_status') and mode in ('on', 'off'): + self._observed = (mode == 'off', mode == 'on') + self._observation_revision += 1 + elif event == 'go_to_standby': + self._observed = (False, False) + self._observation_revision += 1 + elif event == 'wakeup': + self._refresh_at = 0 + # Connection loss never means TV off. Keep the last confirmed state. + self._wake.set() + + async def _power(self): + plug = await self.get_state(self.power_plug) + if plug == 'off': + return False, False + if plug != 'on': + return None, None + try: + power = float(await self.get_state(self.power_plug_power)) + if not math.isfinite(power): + return True, None + return True, power > float(self.args.get('power_threshold', 35)) + except (TypeError, ValueError): + return True, None + + async def _publish(self, is_on, is_art): + async with self._publish_lock: + # Another reader may have delivered a newer state while HA was busy. + if self._observed is not None: + is_on, is_art = self._observed + if not is_on and await self.is_on_entity.get_state() == 'on': + await self._write(self.tv_auto_shutdown_entity, 'off') + await self._write(self.is_art_mode_entity, 'on' if is_art else 'off') + await self._write(self.is_on_entity, 'on' if is_on else 'off') + + async def _write(self, entity, value): + if await entity.get_state() != value: + await entity.set_state(state=value) + + async def _refresh(self): + revision = self._observation_revision + data = await self._client.art_request('get_artmode_status') + mode = data.get('value') + if mode not in ('on', 'off'): + raise TVError('Invalid art mode response') + # A notification arriving during the request takes precedence. + if self._observation_revision == revision: + self._observed = (mode == 'off', mode == 'on') + self._refresh_at = self._loop.time() + float(self.args.get('fallback_refresh', 60)) + + def _spawn(self, coro): + task = asyncio.create_task(coro) + self._tasks.add(task) + + def done(future): + self._tasks.discard(future) + if not future.cancelled() and future.exception() is not None: + self.log_warning(f'TV background task failed: {future.exception()}') + + task.add_done_callback(done) + return task + + async def _run(self): + volume_task = None + try: + timeout = float(self.args.get('network_timeout', 2)) + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=timeout)) as session: + self._client = _TVConnection( + session, self.args['ip_address'], + Path(str(self.AD.app_dir)) / 'theframe_token.txt', + self._on_art_event, timeout) + try: + await self._client.load_token() + self._ready.set() + volume_task = self._spawn(self._volume_worker()) + while not self._stopped: + self._wake.clear() + try: + plugged, powered = await self._power() + now = self._loop.time() + if plugged is False or powered is False: + self._observed = (False, False) + self._observation_revision += 1 + self._refresh_at = 0 + for channel in ('art', 'remote'): + await self._client.close_channel(channel) + elif plugged is True and now >= self._retry_at: + if not self._client.alive('art') or now >= self._refresh_at: + await self._refresh() + self._retry_at = 0 + self._backoff = float(self.args.get('reconnect_delay', 5)) + if self._observed is not None: + await self._publish(*self._observed) + except NETWORK_ERRORS as error: + self.log_info(f'TV refresh failed: {type(error).__name__}: {error}') + self._retry_at = self._loop.time() + self._backoff + self._backoff = min(60, self._backoff * 2) + # One refresh worker coalesces power events. WebSocket + # notifications wake it immediately; no network polling + # occurs for every consumption fluctuation. + deadline = (self._retry_at if self._retry_at > self._loop.time() + else self._refresh_at) + if not self._client.alive('art') and self._retry_at <= self._loop.time(): + deadline = self._loop.time() + delay = max(0.2, min(60, deadline - self._loop.time())) + if powered is False or plugged is not True: + delay = 60 + with suppress(TimeoutError): + await asyncio.wait_for(self._wake.wait(), delay) + finally: + tasks = list(self._tasks) + for task in tasks: + if not task.cancelling(): + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + await self._client.close() + except asyncio.CancelledError: + raise + except Exception as error: + self.log_warning(f'SamsungTheFrameAsync stopped unexpectedly: {error}') + finally: + self._stopped = True + + async def _volume_worker(self): + while True: + direction, received = await self._volume.get() + if (self._loop.time() - received > 1 or self._observed != (True, False) + or self._client.remote_lock.locked()): + continue + try: + await self._client.send_key('KEY_VOLUP' if direction == 'up' else 'KEY_VOLDOWN') + except NETWORK_ERRORS as error: + self.log_info(f'Volume {direction} failed (not replayed): {error}') + + async def _wake_up(self): + if 'wake_up_service' in self.args: + await self.call_service(self.args['wake_up_service']) + elif 'wol_switch' in self.args: + await self.turn_on(self.args['wol_switch']) + if 'mac_address' in self.args: + mac = bytes.fromhex(self.args['mac_address'].replace(':', '').replace('-', '')) + if len(mac) != 6: + raise ValueError('Invalid Wake-on-LAN MAC address') + packet = b'\xff' * 6 + mac * 16 + destinations = {'255.255.255.255', self.args.get('ip_address'), + self.args.get('broadcast_address')} - {None} + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: + sock.setblocking(False) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) + for destination in destinations: + await self._loop.sock_sendto(sock, packet, (destination, 9)) + + async def _select_random_art(self): + data = await self._client.art_request('get_content_list', category=None) + available = data.get('content_list', []) + if isinstance(available, str): + available = json.loads(available) + if not isinstance(available, list): + raise TVError('Invalid artwork list') + if available: + selected = random.choice(available) + await self._client.art_request('select_image', content_id=selected['content_id'], show=True) + + async def _target_runner(self, previous, target, generation): + if previous is not None: + with suppress(asyncio.CancelledError): + await asyncio.shield(previous) + if target == 'manual' or self._stopped or generation != self._generation: + return + await self._ready.wait() + self.log_info(f'Requesting TV to switch to {target}') + if target == 'hdmi' and await self.is_on_entity.get_state() == 'off': + await self._write(self.tv_auto_shutdown_entity, 'on') + while generation == self._generation and not self._stopped: + try: + delay = await self._transition(target) + if delay is None: + self.log_info(f'The TV is confirmed in {target}') + if generation == self._generation: + self.requested_state = None + return + except NETWORK_ERRORS + (KeyError, TypeError) as error: + self.log_info(f'TV transition failed: {type(error).__name__}: {error}') + delay = 30 + # Cancellation interrupts both network waits and confirmation delays. + await asyncio.sleep(delay) + self._refresh_at = 0 + self._wake.set() + + async def _transition(self, target): + plugged, powered = await self._power() + if plugged is None or powered is None: + return 30 + if not plugged: + if target == 'power_off': + return None + await self.turn_on(self.power_plug) + return 60 + if not powered: + if target == 'tv_off': + return None + await self._wake_up() + return 60 + if self._loop.time() < self._retry_at: + return max(1, self._retry_at - self._loop.time()) + # Always confirm the current state before a toggle. Serialize Art reads + # with the monitor; remote key sequences have a separate lock. + await self._refresh() + is_on, is_art = self._observed + await self._publish(is_on, is_art) + if target == 'tv_off': + if is_on: + await self._client.send_key('KEY_POWER') + return 10 + if is_art: + await self._client.send_key('KEY_POWER', hold=4) + return 10 + elif target == 'art_mode': + if not is_art: + await self._client.send_key('KEY_POWER') + return 10 + elif target == 'hdmi': + if not is_on: + await self._client.send_key('KEY_POWER') + return 5 + for app in self.app_list: + if (await self._client.app_status(str(app)))['visible']: + await self._client.send_key('KEY_HDMI') + return 5 + elif target == 'power_off': + if is_on: + await self._client.send_key('KEY_POWER') + return 10 + if not is_art: + await self._wake_up() + return 60 + await self._select_random_art() + await asyncio.sleep(5) + await self.turn_off(self.power_plug) + return 5 + return None + + def terminate(self): + self._stopped = True + task = getattr(self, '_main_task', None) + if task is not None: + self._loop.call_soon_threadsafe(task.cancel) + super().terminate() diff --git a/docs/samsungtheframeasync.md b/docs/samsungtheframeasync.md new file mode 100644 index 0000000..4a1f1d7 --- /dev/null +++ b/docs/samsungtheframeasync.md @@ -0,0 +1,84 @@ +# SamsungTheFrameAsync + +Alternative asynchrone à `apps/samsungtheframe.py`. La classe reste +`SamsungTheFrame` et utilise les mêmes entrées, sorties, consignes, événements +de volume, services de réveil et fichier `apps/theframe_token.txt`. + +## Basculement manuel + +Dans `apps/global/homecinema.yaml`, remplacer uniquement : + +```yaml +module: samsungtheframe +``` + +par : + +```yaml +module: SamsungTheFrameAsync +``` + +Conserver `class: SamsungTheFrame`. Revenir au nom de module précédent pour +restaurer l'ancienne implémentation. Ne pas lancer les deux apps simultanément : +elles piloteraient la même TV et les mêmes capteurs. Le YAML n'a pas été modifié +lors de l'ajout de ce module. + +## Fonctionnement + +- Les callbacks AppDaemon transmettent les demandes à la boucle asyncio et + retournent immédiatement, sans communication réseau avec la TV. +- Une connexion WebSocket Art reste ouverte tant que la TV est alimentée et que + sa consommation dépasse le seuil. Un lecteur unique reçoit notifications et + réponses, corrélées par identifiant de requête. +- `art_mode_changed`, `artmode_status`, `go_to_standby` et `wakeup` sont traités. + Les notifications mettent à jour les capteurs sans attendre le prochain relevé + de consommation. Leur disponibilité dépend du modèle et du firmware. +- Une connexion télécommande distincte est ouverte à la première commande puis + réutilisée. Les commandes sont sérialisées ; une touche maintenue est relâchée + même si la transition est annulée. +- Une lecture Art de secours a lieu toutes les 60 secondes. Les fluctuations de + consommation réveillent le moniteur sans multiplier ces lectures. +- Une perte réseau conserve les derniers états connus. Une consommation sous + le seuil ou une prise explicitement éteinte confirme l'arrêt et ferme les + connexions. Une consommation indisponible n'est pas assimilée à un arrêt. +- Après un échec de lecture, reconnexion avec attente progressive : 5, 10, 20, + 40, puis 60 secondes. Les événements de consommation ne contournent pas cette + attente. Les transitions échouées sont revérifiées après 30 secondes. +- Les cinq consignes sont conservées : `manual`, `art_mode`, `tv_off`, `hdmi`, + `power_off`. Une nouvelle consigne annule la transition précédente, y compris + l'attente de 5 secondes avant coupure électrique. Une commande Power échouée + n'est pas immédiatement rejouée : l'état est relu avant une nouvelle action. +- Les répétitions de volume en attente expirent après une seconde ; la file est + limitée à huit événements. Les échecs ne sont pas rejoués. +- À l'arrêt/rechargement, les tâches sont annulées et les connexions fermées. + +Les sockets utilisent `aiohttp`, déjà fourni par AppDaemon. Aucun thread de +réseau supplémentaire n'est créé. Les seules opérations envoyées à un executor +sont la lecture et l'écriture du petit fichier de token. + +## Réglages facultatifs + +Ces clés peuvent être ajoutées à l'app ; elles ne sont pas nécessaires au +basculement : + +```yaml +network_timeout: 2 # secondes, connexion/autorisation et requêtes TV +reconnect_delay: 5 # attente initiale, doublée jusqu'à 60 secondes +fallback_refresh: 60 # intervalle de lecture de secours, en secondes +power_threshold: 35 # watts, même seuil que l'ancienne app +``` + +Le timeout s'applique à chaque opération, pas à une transition complète. Les +attentes de réveil/confirmation et l'appui long Power utilisent `asyncio` et +n'occupent pas les threads de callbacks AppDaemon. + +## Vérification + +```sh +python -m unittest discover -s tests -p 'test_samsungtheframe_async.py' -v +``` + +Les tests simulent les messages et sockets TV : persistance, notifications, +concurrence, délais, reconnexion, transitions et annulation. Ils ne contactent pas +la TV réelle. La réception des événements et les commandes doivent encore être +validées sur le téléviseur lors du basculement manuel. diff --git a/tests/test_samsungtheframe.py b/tests/test_samsungtheframe.py new file mode 100644 index 0000000..cf4b9fd --- /dev/null +++ b/tests/test_samsungtheframe.py @@ -0,0 +1,149 @@ +import sys +from pathlib import Path +from types import SimpleNamespace +import unittest +from unittest.mock import Mock, patch + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'apps')) +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'apps' / 'ad_toolbox')) +from samsungtheframe import SamsungTheFrame +import requests +import websocket + + +class SamsungTheFrameTests(unittest.TestCase): + def app(self): + app = object.__new__(SamsungTheFrame) + app.args = {'ip_address': '192.0.2.1'} + app.AD = SimpleNamespace(app_dir='/tmp') + app.update_sensors_handle = None + app.update_sensors_is_retry = False + app.requested_state_confirm_cb_handle = None + app.volume_connection = None + app.power_plug = 'switch.tv' + app.power_plug_power = 'sensor.tv_power' + app.get_state = Mock(side_effect=lambda e: 'on' if e == app.power_plug else '60') + for name in ('is_on_entity', 'is_art_mode_entity', 'tv_auto_shutdown_entity'): + entity = Mock() + entity.get_state.return_value = 'on' + setattr(app, name, entity) + app.log_info = Mock() + app.log = Mock() + app.run_in = Mock(side_effect=lambda *a, **kw: object()) + app.cancel_timer = Mock() + app.turn_off = Mock() + return app + + def test_timeout_propagates_to_art_and_rest(self): + a = self.app() + tv = a.connect() + self.assertEqual(tv.timeout, 2) + self.assertEqual(tv.art().timeout, 2) + self.assertEqual(tv._get_rest_api().timeout, 2) + + def test_sensor_burst_only_schedules_one_read(self): + a = self.app() + a.connect = Mock() + for _ in range(10): + a.update_sensors_cb() + a.run_in.assert_called_once_with(a.refresh_sensors_cb, 1) + a.connect.assert_not_called() + + def test_failed_refresh_has_one_retry_replaced_by_new_reading(self): + a = self.app() + a.update_sensors = Mock(return_value=(False, 30)) + a.refresh_sensors_cb({}) + retry = a.update_sensors_handle + a.schedule_sensor_update(30, retry=True) + self.assertEqual(a.run_in.call_count, 1) + a.update_sensors_cb() + a.cancel_timer.assert_called_once_with(retry) + self.assertEqual(a.run_in.call_count, 2) + self.assertFalse(a.update_sensors_is_retry) + + def test_read_errors_preserve_states_and_close_both_connections(self): + errors = [websocket.WebSocketTimeoutException('timeout'), + ConnectionRefusedError(), requests.Timeout(), KeyError('value')] + for error in errors: + with self.subTest(error=type(error).__name__): + a = self.app() + tv, art = Mock(), Mock() + tv.art.return_value = art + art.get_artmode.side_effect = error + a.connect = Mock(return_value=tv) + self.assertEqual(a.update_sensors(), (False, 30)) + art.close.assert_called_once() + tv.close.assert_called_once() + a.is_on_entity.set_state.assert_not_called() + a.is_art_mode_entity.set_state.assert_not_called() + + def test_invalid_readings_do_not_mark_tv_off(self): + for power in ('unavailable', None, 'nan', 'inf'): + with self.subTest(power=power): + a = self.app() + a.get_state.side_effect = lambda e: 'on' if e == a.power_plug else power + a.connect = Mock() + self.assertEqual(a.update_sensors(), (False, 30)) + a.connect.assert_not_called() + a.is_on_entity.set_state.assert_not_called() + + def test_successful_read_consumes_queued_refresh(self): + a = self.app() + a.schedule_sensor_update() + pending = a.update_sensors_handle + tv = Mock() + tv.art.return_value.get_artmode.return_value = 'off' + a.connect = Mock(return_value=tv) + self.assertEqual(a.update_sensors(), (True, 0)) + a.cancel_timer.assert_called_once_with(pending) + self.assertIsNone(a.update_sensors_handle) + a.is_art_mode_entity.set_state.assert_called_once_with(state='off') + tv.art.return_value.close.assert_called_once() + + def test_art_selection_reuses_one_connection(self): + a = self.app() + tv, art = Mock(), Mock() + tv.art.return_value = art + art.available.return_value = [{'content_id': 'image'}] + a.connect = Mock(return_value=tv) + self.assertTrue(a.select_random_art()) + tv.art.assert_called_once() + art.select_image.assert_called_once_with('image', show=True) + art.close.assert_called_once() + + def test_command_timeout_closes_connection_and_schedules_confirmation(self): + a = self.app() + a.requested_state = 'tv_off' + a.update_sensors = Mock(return_value=(True, 0)) + tv = Mock() + tv.shortcuts.return_value.power.side_effect = websocket.WebSocketTimeoutException() + a.connect = Mock(return_value=tv) + self.assertFalse(a._process_state_request()) + tv.close.assert_called_once() + a.run_in.assert_called_once_with(a.confirm_change_state_cb, 30) + + def test_power_off_wait_is_scheduled_and_cancelled_by_new_request(self): + a = self.app() + a.requested_state = 'power_off' + a.is_on_entity.get_state.return_value = 'off' + a.update_sensors = Mock(return_value=(True, 0)) + a.select_random_art = Mock(return_value=True) + self.assertFalse(a._process_state_request()) + a.turn_off.assert_not_called() + a.run_in.assert_called_once_with(a.power_off_cb, 5) + pending = a.requested_state_confirm_cb_handle + a.request_tv_state('manual') + a.cancel_timer.assert_called_once_with(pending) + a.power_off_cb({}) + a.turn_off.assert_not_called() + + def test_power_off_timer_cuts_power_and_schedules_confirmation(self): + a = self.app() + a.requested_state = 'power_off' + a.power_off_cb({}) + a.turn_off.assert_called_once_with(a.power_plug) + a.run_in.assert_called_once_with(a.confirm_change_state_cb, 5) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_samsungtheframe_async.py b/tests/test_samsungtheframe_async.py new file mode 100644 index 0000000..b1840b6 --- /dev/null +++ b/tests/test_samsungtheframe_async.py @@ -0,0 +1,420 @@ +import asyncio +import json +from pathlib import Path +import sys +import tempfile +from types import SimpleNamespace +import unittest +from unittest.mock import AsyncMock, Mock, patch + +import aiohttp + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'apps')) +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'apps' / 'ad_toolbox')) +from SamsungTheFrameAsync import SamsungTheFrame, TVError, _TVConnection + + +def message(data): + return SimpleNamespace(type=aiohttp.WSMsgType.TEXT, data=json.dumps(data)) + + +class FakeSocket: + def __init__(self, ready=True): + self.messages = asyncio.Queue() + self.closed = False + self.sent = [] + self.respond = True + self.messages.put_nowait(message({'event': 'ms.channel.connect', 'data': {}})) + if ready: + self.messages.put_nowait(message({'event': 'ms.channel.ready'})) + + async def receive(self): + return await self.messages.get() + + def __aiter__(self): + return self + + async def __anext__(self): + msg = await self.receive() + if msg.type == aiohttp.WSMsgType.CLOSED: + raise StopAsyncIteration + return msg + + async def send_json(self, payload): + self.sent.append(payload) + if payload['method'] == 'ms.channel.emit' and self.respond: + request = json.loads(payload['params']['data']) + self.push({'event': 'artmode_status', 'value': 'on', + 'request_id': request['request_id']}) + + def push(self, data): + self.messages.put_nowait(message({'event': 'd2d_service_message', + 'data': json.dumps(data)})) + + async def close(self): + self.closed = True + self.messages.put_nowait(SimpleNamespace(type=aiohttp.WSMsgType.CLOSED)) + + +class ConnectionTests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.temp = tempfile.TemporaryDirectory() + self.ws = FakeSocket() + self.session = Mock() + self.session.ws_connect = AsyncMock(return_value=self.ws) + self.events = Mock() + self.client = _TVConnection(self.session, '192.0.2.1', + Path(self.temp.name) / 'token', self.events, .05) + + async def asyncTearDown(self): + await self.client.close() + self.temp.cleanup() + + async def test_art_requests_reuse_socket_and_reader(self): + first = await self.client.art_request('get_artmode_status') + reader = self.client.readers['art'] + second = await self.client.art_request('get_artmode_status') + self.assertEqual(first['value'], 'on') + self.assertEqual(second['value'], 'on') + self.session.ws_connect.assert_awaited_once() + self.assertIs(self.client.readers['art'], reader) + self.assertFalse(reader.done()) + self.assertEqual(self.client.pending, {}) + self.events.assert_not_called() # Query responses applied by refresh. + + async def test_unsolicited_notifications_delivered_while_idle(self): + await self.client.connect('art') + self.ws.push({'event': 'art_mode_changed', 'status': 'off'}) + await asyncio.sleep(.01) + self.events.assert_called_once_with({'event': 'art_mode_changed', 'status': 'off'}) + + async def test_wrong_request_id_cannot_satisfy_request(self): + self.ws.respond = False + task = asyncio.create_task(self.client.art_request('get_artmode_status')) + await asyncio.sleep(.01) + self.ws.push({'event': 'artmode_status', 'value': 'off', 'id': 'wrong'}) + with self.assertRaises(TimeoutError): + await task + self.assertEqual(self.client.pending, {}) + self.assertTrue(self.ws.closed) + + async def test_disconnect_fails_pending_request_without_waiting_for_timeout(self): + self.ws.respond = False + task = asyncio.create_task(self.client.art_request('get_artmode_status')) + await asyncio.sleep(.01) + await self.ws.close() + with self.assertRaises(TVError): + await task + self.assertEqual(self.client.pending, {}) + + async def test_request_rejection_is_not_mistaken_for_state(self): + self.ws.respond = False + task = asyncio.create_task(self.client.art_request('get_artmode_status')) + await asyncio.sleep(.01) + request = json.loads(self.ws.sent[0]['params']['data']) + self.ws.push({'event': 'error', 'request_id': request['request_id'], 'error_code': -1}) + with self.assertRaises(TVError): + await task + + async def test_cancel_request_removes_pending_future(self): + self.ws.respond = False + task = asyncio.create_task(self.client.art_request('get_artmode_status')) + await asyncio.sleep(.01) + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + self.assertEqual(self.client.pending, {}) + + async def test_timeout_covers_handshake_and_does_not_block_loop(self): + self.ws = FakeSocket(ready=False) + self.session.ws_connect.return_value = self.ws + task = asyncio.create_task(self.client.connect('art')) + await asyncio.sleep(.01) + self.assertFalse(task.done()) + with self.assertRaises(TimeoutError): + await task + self.assertTrue(self.ws.closed) + + async def test_unauthorized_handshake_closes_socket(self): + self.ws.messages = asyncio.Queue() + self.ws.messages.put_nowait(message({'event': 'ms.channel.unauthorized'})) + with self.assertRaises(TVError): + await self.client.connect('art') + self.assertTrue(self.ws.closed) + + async def test_remote_reuse_and_cancellation_releases_held_power_key(self): + task = asyncio.create_task(self.client.send_key('KEY_POWER', hold=4)) + await asyncio.sleep(.01) + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + await self.client.send_key('KEY_VOLUP') + self.assertEqual([v['params']['Cmd'] for v in self.ws.sent], + ['Press', 'Release', 'Click']) + self.session.ws_connect.assert_awaited_once() + + async def test_close_cancels_reader(self): + await self.client.connect('art') + reader = self.client.readers['art'] + await self.client.close() + self.assertTrue(reader.done()) + self.assertEqual(self.client.readers, {}) + self.assertEqual(self.client.sockets, {}) + + async def test_corrupt_frame_is_contained_and_requests_fail(self): + self.ws.respond = False + task = asyncio.create_task(self.client.art_request('get_artmode_status')) + await asyncio.sleep(.01) + self.ws.messages.put_nowait(SimpleNamespace(type=aiohttp.WSMsgType.TEXT, data='{bad')) + with self.assertRaises(TVError): + await task + self.assertTrue(self.ws.closed) + + async def test_reconnect_creates_one_fresh_socket(self): + await self.client.art_request('get_artmode_status') + await self.ws.close() + await asyncio.sleep(.01) + fresh = FakeSocket() + self.session.ws_connect.return_value = fresh + await self.client.art_request('get_artmode_status') + self.assertEqual(self.session.ws_connect.await_count, 2) + self.assertTrue(self.client.alive('art')) + + +def fake_entity(value): + entity = Mock() + state = {'value': value} + entity.get_state = AsyncMock(side_effect=lambda: state['value']) + async def set_state(*, state): + state_holder['value'] = state + state_holder = state + entity.set_state = AsyncMock(side_effect=set_state) + return entity + + +class ApplicationTests(unittest.IsolatedAsyncioTestCase): + def app(self): + app = object.__new__(SamsungTheFrame) + app.AD = SimpleNamespace(app_dir='/tmp') + app.args = {'ip_address': '192.0.2.1', 'reconnect_delay': .05} + app._loop = asyncio.get_running_loop() + app._stopped = False + app._wake = asyncio.Event() + app._ready = asyncio.Event() + app._ready.set() + app._publish_lock = asyncio.Lock() + app._volume = asyncio.Queue(maxsize=8) + app._tasks = set() + app._target_task = None + app._generation = 0 + app._observed = (False, True) + app._observation_revision = 0 + app._refresh_at = 0 + app._retry_at = 0 + app._backoff = .05 + app.requested_state = None + app.power_plug, app.power_plug_power = 'switch.tv', 'sensor.power' + app.get_state = AsyncMock(side_effect=lambda e: 'on' if e == app.power_plug else '60') + app.is_on_entity = fake_entity('off') + app.is_art_mode_entity = fake_entity('on') + app.tv_auto_shutdown_entity = fake_entity('off') + app.turn_on = AsyncMock() + app.turn_off = AsyncMock() + app.call_service = AsyncMock() + app.log_info = Mock() + app.log_warning = Mock() + app._client = Mock() + for method in ('art_request', 'send_key', 'app_status', 'close', 'close_channel', 'load_token'): + setattr(app._client, method, AsyncMock()) + app._client.art_request.return_value = {'event': 'artmode_status', 'value': 'on'} + app._client.remote_lock = asyncio.Lock() + app.app_list = [] + return app + + async def asyncTearDown(self): + tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()] + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + + async def test_existing_yaml_contract_initializes_without_tv_io(self): + app = self.app() + app.args.update({ + 'input_sensors': {'power_plug': 'switch.tv', 'power_plug_power': 'sensor.power'}, + 'output_sensors': {'is_on': 'binary_sensor.tv_on', + 'is_art_mode': 'binary_sensor.art', + 'tv_auto_shutdown': 'binary_sensor.shutdown'}, + 'tv_state': 'sensor.target', 'wake_up_service': 'esphome/wake_tv', + 'volume_up_events': ['volume_up'], 'volume_down_events': ['volume_down']}) + app.AD.loop = app._loop + app.get_state = Mock(return_value='off') + app.create_entity = Mock(side_effect=lambda *a, **kw: Mock()) + app.listen_state = Mock() + app._post = Mock() + app.create_task = Mock(side_effect=lambda coro: coro.close()) + with patch('SamsungTheFrameAsync.EventHandler') as handler: + app.on_initialize_smart_object() + self.assertEqual(app.listen_state.call_count, 3) + self.assertEqual(handler.call_count, 2) + self.assertEqual(app.create_entity.call_count, 3) + app.call_service.assert_not_called() + self.assertEqual(type(app).__name__, 'SamsungTheFrame') + + async def test_callbacks_only_post_work_to_loop(self): + app = self.app() + app._loop = Mock() + app.update_sensors_cb() + app.request_tv_state('hdmi') + app.on_volume_event('volume', {}, 'up') + self.assertEqual(app._loop.call_soon_threadsafe.call_count, 3) + app._client.send_key.assert_not_called() + app._client.art_request.assert_not_called() + + async def test_notification_updates_state_and_disconnect_preserves_it(self): + app = self.app() + app._on_art_event({'event': 'art_mode_changed', 'status': 'off'}) + self.assertEqual(app._observed, (True, False)) + self.assertTrue(app._wake.is_set()) + app._on_art_event({'event': 'connection_lost', 'channel': 'art'}) + self.assertEqual(app._observed, (True, False)) + app._on_art_event({'event': 'go_to_standby'}) + self.assertEqual(app._observed, (False, False)) + + async def test_late_query_cannot_overwrite_newer_standby_notification(self): + app = self.app() + async def request(*args): + app._on_art_event({'event': 'go_to_standby'}) + return {'value': 'on'} + app._client.art_request.side_effect = request + await app._refresh() + self.assertEqual(app._observed, (False, False)) + + async def test_concurrent_queries_apply_latest_response(self): + app = self.app() + calls = 0 + async def request(*args): + nonlocal calls + calls += 1 + value = 'on' if calls == 1 else 'off' + await asyncio.sleep(.01 if value == 'on' else .02) + return {'value': value} + app._client.art_request.side_effect = request + await asyncio.gather(app._refresh(), app._refresh()) + self.assertEqual(app._observed, (True, False)) + + async def test_unknown_consumption_is_not_off(self): + app = self.app() + for invalid in ('unavailable', None, 'nan', 'inf'): + app.get_state.side_effect = lambda e: 'on' if e == app.power_plug else invalid + self.assertEqual(await app._power(), (True, None)) + + async def test_hdmi_shutdown_flag_survives_initial_art_mode_read(self): + app = self.app() + await app.tv_auto_shutdown_entity.set_state(state='on') + await app._publish(False, True) + self.assertEqual(await app.tv_auto_shutdown_entity.get_state(), 'on') + app.is_on_entity = fake_entity('on') + await app._publish(False, True) + self.assertEqual(await app.tv_auto_shutdown_entity.get_state(), 'off') + + async def test_power_modes_issue_expected_commands(self): + for target, mode, hold in [('art_mode', 'off', 0), ('tv_off', 'on', 4), + ('tv_off', 'off', 0), ('hdmi', 'on', 0)]: + with self.subTest(target=target, mode=mode): + app = self.app() + app._client.art_request.return_value = {'value': mode} + self.assertIn(await app._transition(target), (5, 10)) + if hold: + app._client.send_key.assert_awaited_once_with('KEY_POWER', hold=4) + else: + app._client.send_key.assert_awaited_once_with('KEY_POWER') + + async def test_hdmi_preserves_app_visibility_check(self): + app = self.app() + app.app_list = ['netflix', 'prime'] + app._client.art_request.return_value = {'value': 'off'} + app._client.app_status.side_effect = [{'visible': False}, {'visible': True}] + self.assertEqual(await app._transition('hdmi'), 5) + app._client.send_key.assert_awaited_once_with('KEY_HDMI') + + async def test_off_plug_wakes_for_art_but_satisfies_power_off(self): + app = self.app() + app.get_state.return_value = 'off' + app.get_state.side_effect = None + self.assertIsNone(await app._transition('power_off')) + app.turn_on.assert_not_called() + self.assertEqual(await app._transition('art_mode'), 60) + app.turn_on.assert_awaited_once_with(app.power_plug) + app._client.art_request.assert_not_called() + + async def test_manual_cancels_pending_power_cut(self): + app = self.app() + app._select_random_art = AsyncMock() + app._set_target('power_off') + await asyncio.sleep(.02) + app._select_random_art.assert_awaited_once() + app._set_target('manual') + await asyncio.sleep(.02) + app.turn_off.assert_not_called() + self.assertIsNone(app.requested_state) + self.assertFalse(app._tasks) + + async def test_volume_queue_bounded_and_old_events_dropped(self): + app = self.app() + app._observed = (True, False) + for _ in range(30): + app._queue_volume('up') + self.assertEqual(app._volume.qsize(), 8) + for _ in range(8): + app._volume.get_nowait() + app._volume.put_nowait(('up', app._loop.time() - 5)) + app._volume.put_nowait(('down', app._loop.time())) + task = asyncio.create_task(app._volume_worker()) + await asyncio.sleep(.01) + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + app._client.send_key.assert_awaited_once_with('KEY_VOLDOWN') + + async def test_monitor_coalesces_sensor_bursts_and_shutdown_closes_client(self): + app = self.app() + client = app._client + with patch('SamsungTheFrameAsync._TVConnection', return_value=client): + task = asyncio.create_task(app._run()) + await asyncio.sleep(.02) + for _ in range(30): + app.update_sensors_cb() + await asyncio.sleep(.02) + client.art_request.assert_awaited_once() + app._on_art_event({'event': 'art_mode_changed', 'status': 'off'}) + await asyncio.sleep(.02) + self.assertEqual(await app.is_on_entity.get_state(), 'on') + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + client.close.assert_awaited_once() + self.assertFalse(app._tasks) + + async def test_power_events_cannot_bypass_reconnect_backoff(self): + app = self.app() + client = app._client + app._backoff = .5 + app.args['reconnect_delay'] = .5 + client.alive.return_value = False + client.art_request.side_effect = TVError('offline') + with patch('SamsungTheFrameAsync._TVConnection', return_value=client): + task = asyncio.create_task(app._run()) + await asyncio.sleep(.01) + for _ in range(10): + app._wake.set() + await asyncio.sleep(.001) + self.assertEqual(client.art_request.await_count, 1) + await asyncio.sleep(.6) + self.assertGreaterEqual(client.art_request.await_count, 2) + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + + +if __name__ == '__main__': + unittest.main()