586 lines
26 KiB
Python
586 lines
26 KiB
Python
"""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
|
|
# A fresh target must not inherit a long idle reconnection delay.
|
|
self._retry_at = min(self._retry_at, self._loop.time() + self._transition_retry_delay())
|
|
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}')
|
|
retry_delay = self._backoff
|
|
if self.requested_state is not None:
|
|
retry_delay = min(retry_delay, self._transition_retry_delay())
|
|
self._retry_at = self._loop.time() + retry_delay
|
|
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)
|
|
|
|
def _transition_retry_delay(self):
|
|
return max(0.2, float(self.args.get('transition_retry_delay', 2)))
|
|
|
|
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 = self._transition_retry_delay()
|
|
# 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 self._transition_retry_delay()
|
|
if not plugged:
|
|
if target == 'power_off':
|
|
return None
|
|
await self.turn_on(self.power_plug)
|
|
return 5
|
|
if not powered:
|
|
if target == 'tv_off':
|
|
return None
|
|
await self._wake_up()
|
|
return 10
|
|
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()
|