"""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()