Added support for power plug and spam sensor

.gitignore

@@ -4,3 +4,4 @@ __pycache__/
 *.pyc
 logs/
 apps/data/
+secrets.yaml

appdaemon.yaml

@@ -13,6 +13,29 @@ appdaemon:
       token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJiZThhYWI2ZTlkNmQ0NWU2YTk1ODU5OWJjOWM3MWJkYiIsImlhdCI6MTc3NjEwOTc0NCwiZXhwIjoyMDkxNDY5NzQ0fQ.BorkjFjWlWCZqnUa9-NMUxGsDiupDoRZ3cEgsmeSofM
       cert_verify: True
       ws_timeout: 00:10:00
+    MQTT:
+      type: mqtt
+      namespace: mqtt
+      client_host: 10.0.0.21
+      client_port: 1883
+      client_id: appdaemon_radiator_heating
+      client_user: !secret boiler_mqtt_user
+      client_password: !secret boiler_mqtt_password
+      client_timeout: 30
+      client_topics:
+        - appdaemon/radiator_heating/command/+
+        - appdaemon/radiator_heating/heartbeat
+        - themaplus-condens/bai/SetMode
+        - themaplus-condens/bai/Statenumber
+        - themaplus-condens/bai/FlowTemp
+        - homeassistant/status
+      birth_topic: appdaemon/radiator_heating/connection
+      birth_payload: online
+      birth_retain: true
+      will_topic: appdaemon/radiator_heating/connection
+      will_payload: offline
+      will_retain: true
+      shutdown_payload: offline
 http:
   url: http://0.0.0.0:5050
 admin:

apps/boilerclimate.py

@@ -1,14 +1,11 @@
-"""MQTT-discovered boiler climate; MQTT transport is Home Assistant's service.
+"""MQTT-discovered climate using AppDaemon's MQTT plugin and HA sensors."""
-
-Incoming MQTT is forwarded by the companion Home Assistant automation. All
-callbacks run on the app's pinned AppDaemon thread; no broker credentials.
-"""
 from datetime import datetime, timezone
 import json
 from pathlib import Path
 import time
 
 import appdaemon.plugins.hass.hassapi as hass
+import ad_toolbox.smartcondition as SmartCondition
 from boilerregulator import BoilerRegulator, finite, setmode
 from logger_interface import LoggerInterface
 
@@ -16,12 +13,23 @@ from logger_interface import LoggerInterface
 class BoilerClimate(hass.Hass, LoggerInterface):
     def initialize(self):
         self.initialize_logger_interface(self.get_ad_api(), ascii_encode=False)
+        self.mqtt_namespace = self.args.get('mqtt_namespace', 'mqtt')
+        if self.AD.plugins.get_plugin_object(self.mqtt_namespace) is None:
+            raise RuntimeError('MQTT plugin missing; configure it and restart AppDaemon')
         self.base = self.args.get('mqtt_topic', 'appdaemon/radiator_heating').rstrip('/')
         self.boiler = self.args.get('boiler_topic', 'themaplus-condens').rstrip('/')
+        self.connection_topic = self.args.get('mqtt_connection_topic', self.base + '/connection')
         self.uid = self.args.get('unique_id', 'appdaemon_radiator_heating')
+        self.entity_id = self.args.get('entity', 'climate.radiator_heating')
         self.title = self.args.get('display_name', 'Chauffage radiateurs')
         self.indoor = self.args['indoor_sensor']
-        self.outdoor = self.args['outdoor_sensor']
+        outdoors = self.args.get('outdoor_sensors')
+        if outdoors is None:
+            outdoors = [self.args['outdoor_sensor']]
+        if not isinstance(outdoors, list) or not outdoors or any(not isinstance(e, str) or not e for e in outdoors):
+            raise ValueError('outdoor_sensors must be a non-empty list of entity IDs')
+        self.outdoors = tuple(dict.fromkeys(outdoors))
+        self.selected_outdoor = None
         self.interval = finite(self.args.get('command_interval', 60))
         if not 10 <= self.interval <= 300:
             raise ValueError('command_interval must be between 10 and 300 seconds')
@@ -43,25 +51,59 @@ class BoilerClimate(hass.Hass, LoggerInterface):
         self.sensor_error = None
         self.last_status_time = 0.0
         self._state = None
-        self.listen_event(self._mqtt_event, 'APPDAEMON_BOILER_MQTT')
+        self.listen_event(self._mqtt_event, 'MQTT_MESSAGE', namespace=self.mqtt_namespace)
         self.listen_state(self._sensor_changed, self.indoor, attribute='all')
-        self.listen_state(self._sensor_changed, self.outdoor, attribute='all')
+        for entity in self.outdoors:
+            self.listen_state(self._sensor_changed, entity, attribute='all')
         self._discovery()
+        self._initialize_smart_conditions()
         self.run_every(self._tick, 'now', self.interval)
         self._control()
         self.log_info(f'Boiler climate ready: MQTT writes={self.send_commands}, room target={self.target} °C')
 
+    def _initialize_smart_conditions(self, kwargs=None):
+        if getattr(self, 'smart_conditions_evaluator', None) is not None:
+            return
+        self.smart_conditions_evaluator = None
+        if 'smart_conditions' in self.args:
+            if not self.entity_exists(self.entity_id) or self.get_state(self.entity_id, attribute='temperature') is None:
+                self.mode = 'off'
+                self.run_in(self._initialize_smart_conditions, 1)
+                return
+            constants = {**self.args.get('constants', {}), 'self': self.entity_id}
+            self.smart_conditions_evaluator = SmartCondition.Evaluator(
+                self, self.args['smart_conditions'], condition_name='smart_conditions',
+                on_change_cb=self.on_smart_conditions_change,
+                constants=constants)
+
+    def on_smart_conditions_change(self, previous, result):
+        if result == SmartCondition.Result.Disabled:
+            return
+        mode = 'heat' if result == SmartCondition.Result.Succeeded else 'off'
+        if mode != self.mode:
+            self._set_mode(mode, 'SmartCondition')
+
+    def _set_mode(self, mode, source):
+        self.log_info(f'Climate mode: {self.mode} -> {mode} ({source})')
+        self.mode = mode
+        self.controller.reset()
+        self._control(force=True)
+        self._save_safely()
+
     def _publish(self, topic, payload, retain=False):
         if not isinstance(payload, str):
             payload = json.dumps(payload, allow_nan=False, ensure_ascii=False)
-        self.call_service('mqtt/publish', topic=topic, payload=payload, qos=0, retain=retain)
+        self.call_service('mqtt/publish', topic=topic, payload=payload, qos=0,
+                          retain=retain, namespace=self.mqtt_namespace)
 
     def _discovery(self):
         device = dict(identifiers=[self.uid], name=self.title,
                       manufacturer='AppDaemon', model='Weather compensation + room PI')
-        common = dict(device=device, availability_topic=self.base + '/availability')
+        common = dict(device=device, availability=[{'topic': self.base + '/availability'},
+                                                   {'topic': self.connection_topic}],
+                      availability_mode='all')
         config = dict(common, name=None, unique_id=self.uid,
-                      default_entity_id='climate.radiator_heating',
+                      default_entity_id=self.entity_id,
                       modes=['off', 'heat'], temperature_unit='C', precision=0.1,
                       temp_step=0.5, min_temp=self.min_target, max_temp=self.max_target,
                       mode_command_topic=self.base + '/command/mode',
@@ -147,6 +189,11 @@ class BoilerClimate(hass.Hass, LoggerInterface):
 
     def _mqtt_event(self, event, data, kwargs):
         topic, payload = data.get('topic'), data.get('payload', '')
+        if topic is None:
+            if data.get('state') == 'Connected':
+                self._discovery()
+                self._control(force=True)
+            return
         if topic == 'homeassistant/status' and payload == 'online':
             self._discovery()
             self._control()
@@ -155,10 +202,7 @@ class BoilerClimate(hass.Hass, LoggerInterface):
             if payload not in ('off', 'heat'):
                 self.log_warning('Rejected invalid climate mode')
                 return
-            self.mode = payload
+            self._set_mode(payload, 'MQTT')
-            self.controller.reset()
-            self._control(force=True)
-            self._save_safely()
         elif topic == self.base + '/command/temperature':
             try:
                 target = finite(payload)
@@ -201,7 +245,8 @@ class BoilerClimate(hass.Hass, LoggerInterface):
 
     def _attributes(self):
         self._publish(self.base + '/attributes', dict(
-            indoor_sensor=self.indoor, outdoor_sensor=self.outdoor,
+            indoor_sensor=self.indoor, outdoor_sensors=list(self.outdoors),
+            outdoor_sensor=self.selected_outdoor,
             commands_enabled=self.send_commands,
             sensor_error=getattr(self, 'sensor_error', None),
             last_command_at=self.last_command_at, last_c6_echo_at=self.last_echo_at,
@@ -215,16 +260,20 @@ class BoilerClimate(hass.Hass, LoggerInterface):
         indoor = outdoor = None
         result = None
         readings, errors = {}, []
-        for entity in (self.indoor, self.outdoor):
+        sensors = (self.indoor, *self.outdoors)
+        for entity in sensors:
             try:
                 readings[entity] = self._sensor_value(entity)
             except (ValueError, TypeError, KeyError) as exc:
                 errors.append(str(exc))
-        indoor, outdoor = readings.get(self.indoor), readings.get(self.outdoor)
+        indoor = readings.get(self.indoor)
+        self.selected_outdoor = (min(self.outdoors, key=readings.__getitem__)
+                                 if all(entity in readings for entity in self.outdoors) else None)
+        outdoor = readings.get(self.selected_outdoor)
         valid = not errors
         sensor_error = '; '.join(errors) if errors else None
         # Log once per set of failing sensors, not every minute as age increases.
-        signature = tuple((entity, entity in readings) for entity in (self.indoor, self.outdoor))
+        signature = tuple((entity, entity in readings) for entity in sensors)
         previous = getattr(self, '_sensor_fault_signature', None)
         if not valid and signature != previous:
             self.log_error(f'Heating inhibited: sensor fault: {sensor_error}')
@@ -266,11 +315,17 @@ class BoilerClimate(hass.Hass, LoggerInterface):
         # on boiler command expiration. No retained actuator command, ever.
         if not hasattr(self, 'base'):
             return
+        evaluator = getattr(self, 'smart_conditions_evaluator', None)
         try:
             if self.send_commands:
                 self._publish(self.boiler + '/bai/SetMode/set', setmode(0, False))
         except Exception as exc:
             self.log_error(f'Could not publish shutdown; boiler expiration remains the fallback: {exc}')
+        if evaluator is not None:
+            try:
+                evaluator.deactivate()
+            except Exception as exc:
+                self.log_warning(f'Could not deactivate SmartCondition: {exc}')
         try:
             self._publish(self.base + '/availability', 'offline', True)
         except Exception as exc:

apps/boilerwatchdog.py

@@ -0,0 +1,50 @@
+"""Independent AppDaemon worker monitoring the boiler climate heartbeat."""
+import time
+
+import appdaemon.plugins.hass.hassapi as hass
+from logger_interface import LoggerInterface
+from boilerregulator import finite
+
+
+class BoilerWatchdog(hass.Hass, LoggerInterface):
+    def initialize(self):
+        self.initialize_logger_interface(self.get_ad_api(), ascii_encode=False)
+        self.base = self.args.get('mqtt_topic', 'appdaemon/radiator_heating').rstrip('/')
+        self.mqtt_namespace = self.args.get('mqtt_namespace', 'mqtt')
+        self.timeout = finite(self.args.get('heartbeat_timeout', 180))
+        if self.timeout < 60:
+            raise ValueError('heartbeat_timeout must be at least 60 seconds')
+        if self.AD.plugins.get_plugin_object(self.mqtt_namespace) is None:
+            raise RuntimeError('MQTT plugin missing; configure it and restart AppDaemon')
+        self.last_heartbeat = None
+        self.waiting_since = time.monotonic()
+        self.fault = False
+        self.warned = False
+        self.listen_event(self._mqtt_event, 'MQTT_MESSAGE', namespace=self.mqtt_namespace)
+        self.run_every(self._check, 'now', 10)
+        self._check()
+
+    def _mqtt_event(self, event, data, kwargs):
+        if data.get('topic') == self.base + '/heartbeat':
+            self.last_heartbeat = time.monotonic()
+            if self.warned:
+                self.log_info('Boiler climate heartbeat recovered')
+            self.fault = False
+            self.warned = False
+        elif data.get('topic') is None and data.get('state') == 'Connected':
+            self.last_heartbeat = None
+            self.waiting_since = time.monotonic()
+            self.fault = False
+            self.warned = False
+            self._check()
+
+    def _check(self, kwargs=None):
+        age = time.monotonic() - (self.waiting_since if self.last_heartbeat is None else self.last_heartbeat)
+        if self.last_heartbeat is None or age >= self.timeout:
+            if not self.fault:
+                self.call_service('mqtt/publish', namespace=self.mqtt_namespace,
+                                  topic=self.base + '/availability', payload='offline', qos=0, retain=True)
+                self.fault = True
+            if age >= self.timeout and not self.warned:
+                self.log_warning('Boiler climate unavailable: no heartbeat within the timeout')
+                self.warned = True

apps/global/heating.yaml

@@ -221,18 +221,32 @@ guestroom_heatpump:
 radiator_heating:
   module: boilerclimate
   class: BoilerClimate
+  entity: climate.radiator_heating
   display_name: Chauffage radiateurs
   unique_id: appdaemon_radiator_heating
   indoor_sensor: sensor.corridor_temperature
-  outdoor_sensor: sensor.ext_front_temperature
+  outdoor_sensors:
+    - sensor.ext_front_temperature
+    - sensor.ext_back_temperature
   target_temperature: 18
   initial_mode: heat
 
+  smart_conditions:
+    unavaibility_result: Failed #if something is unavailable in the expression, the radiator stay off
+    local_expressions:
+      outdoor_temperature: "sensor.ext_front_temperature < sensor.ext_back_temperature ? sensor.ext_front_temperature : sensor.ext_back_temperature"
+    trigger_conditions: binary_sensor.someone_home and outdoor_temperature < self.temperature
+    blocking_conditions:
+      - binary_sensor.group_windows
+      - binary_sensor.kitchen_french_door or binary_sensor.living_room_french_door or binary_sensor.bedroom_french_door      
+
   # Radiator heating enabled with the room-thermostat bridge CLOSED.
   # false keeps the climate/diagnostics but suppresses SetMode writes.
   send_commands: true
   command_interval: 60
   mqtt_topic: appdaemon/radiator_heating
+  mqtt_namespace: mqtt
+  mqtt_connection_topic: appdaemon/radiator_heating/connection
   boiler_topic: themaplus-condens
 
   # Starting values, to tune from observations with radiator circulation enabled.
@@ -247,3 +261,10 @@ radiator_heating:
     max_flow: 50
     overheat: 0.5
     restart_margin: 0.1
+
+radiator_heating_watchdog:
+  module: boilerwatchdog
+  class: BoilerWatchdog
+  mqtt_namespace: mqtt
+  mqtt_topic: appdaemon/radiator_heating
+  heartbeat_timeout: 180

apps/global/homecinema.yaml

@@ -1,5 +1,5 @@
 samsung_the_frame:
-  module: samsungtheframe
+  module: samsungtheframeasync
   class: SamsungTheFrame
 
   volume_up_events:
@@ -25,7 +25,7 @@ samsung_the_frame:
         power_off:
           trigger_conditions:
             - binary_sensor.is_everybody_sleeping or binary_sensor.is_living_room_sleeping or not binary_sensor.someone_home
-            - binary_sensor.projector_on or binary_sensor.projector_screen_down
+            - binary_sensor.projector_screen_down
         hdmi: binary_sensor.vertex2_has_input_signal and not binary_sensor.projector_screen_down
         manual: binary_sensor.is_tv_on and not binary_sensor.tv_auto_shutdown
         tv_off: not binary_sensor.living_room_occupancy
@@ -74,7 +74,7 @@ w4000i:
         time: 30
       continuous_condition.has_allm_signal_for_a_while:
         conditions: binary_sensor.vertex2_allm
-        time: 5
+        time: 10
 
   attributes_override:
     icon: sensor.projector_icon
@@ -121,11 +121,18 @@ bravia_theatre:
     trigger_conditions: sensor.vertex2_output_selected == 'Projector' or input_boolean.use_bravia_theater_for_the_frame
     blocking_conditions: not binary_sensor.vertex2_has_input_signal
 
-projection_screen:
+projection_screen_switch:
-  module: smartshutter
+  module: smartswitch
-  class: SmartShutter
+  class: SmartSwitch
-  entity: cover.living_room_projection_screen
+  entity: input_boolean.projection_screen
-  close_position: 22
+
+  virtual_sensors:
+    sensors:
+      continuous_condition.projection_screen_on_for_a_while:
+        time: 300
+        conditions: input_boolean.projection_screen
+
+  off_conditions: binary_sensor.projection_screen_on_for_a_while and binary_sensor.projector_off_for_a_while 
 
   toggle_events:
     apple_tv_mute_long_press:
@@ -137,16 +144,13 @@ projection_screen:
         new_state:
           state: "on"
 
-  # close_conditions: # close mean screen down
+projection_screen:
-  #   # when the projector is turn back on, it can happen that this condition is evaluated before projector_off_for_a_while is updated
+  module: smartshutter
-  #   # so I need to test both binary_sensor.projector_on and not binary_sensor.projector_off_for_a_while
+  class: SmartShutter
-  #   trigger_conditions: binary_sensor.projector_on or not binary_sensor.projector_off_for_a_while
+  entity: cover.living_room_projection_screen
-  #   disable_conditions:
+  close_position: 22
-  #     - binary_sensor.day_interval_night and not binary_sensor.projector_on
-  #     - binary_sensor.is_everybody_sleeping and not binary_sensor.projector_on
 
-  open_conditions: # close mean screen up
+  close_conditions: input_boolean.projection_screen
-    trigger_conditions: binary_sensor.projector_off_for_a_while
 
 vertex_2:
   module: vertex2
@@ -156,9 +160,10 @@ vertex_2:
   virtual_sensors:
     default_values:
       binary_sensor.is_tv_on: false
+      binary_sensor.someone_home: false
       binary_sensor.vertex2_has_input_signal: false
     sensors:
-      binary_sensor.projector_screen_down: cover.living_room_projection_screen.current_position < 90
+      binary_sensor.projector_screen_down: input_boolean.projection_screen
 
   input_config:
     default:
@@ -167,7 +172,6 @@ vertex_2:
     0:
       name: AppleTV
       icon: mdi:apple
-      #reboot_when_leaving: True
     1:
       name: Xbox
       icon: mdi:microsoft-xbox    
@@ -186,23 +190,20 @@ vertex_2:
       name: TV
       icon: mdi:television
       use_audio_out_audio_flags: input_boolean.use_bravia_theater_for_the_frame
-      force_plus5: 
+      force_plus5: not binary_sensor.vertex2_no_signal_spam
-        trigger_conditions: binary_sensor.is_tv_on or binary_sensor.vertex2_has_input_signal
-        blocking_conditions: not switch.the_frame_power_plug
     1:
       name: Projector
       icon: mdi:projector
       use_audio_out_audio_flags: True
-      force_plus5: switch.living_room_projector_plug
+      #force_plus5: switch.living_room_projector_plug
+
+  vertex2_power_plug:
+    entity: switch.vertex2_power_plug
+    # power_consumption: sensor.fibaro_vertex_plug_51_power
+    conditions:
+      trigger_conditions: binary_sensor.someone_home #and (input_boolean.projection_screen or binary_sensor.the_frame_is_plugged_for_a_while)
+      blocking_conditions: (binary_sensor.is_everybody_sleeping or binary_sensor.is_living_room_sleeping) and not binary_sensor.vertex2_has_input_signal
         
-  # vertex2_power_plug:
-  #   entity: switch.fibaro_vertex_plug_51
-  #   # power_consumption: sensor.fibaro_vertex_plug_51_power
-  #   conditions:
-  #     trigger_conditions: binary_sensor.someone_home
-  #     blocking_conditions:
-  #       - binary_sensor.is_everybody_sleeping or binary_sensor.is_salon_sleeping
-  #       - binary_sensor.vertex2_unreachable and not binary_sensor.is_tv_on and not binary_sensor.projector_on
 
   #filmmaker_mode_conditions: sensor.vertex2_input_select_tx0 == 'AppleTV'
 
@@ -219,12 +220,16 @@ vertex_2:
 
   vertex2_address: http://10.0.0.33
   vertex2_tcp_port: 2220
+  startup_delay: 30 # seconds after the power plug turns on
 
   video_tx_signal_sensor: sensor.vertex2_video_signal
   allm_sensor: binary_sensor.vertex2_allm
+  #tx0_plus5_sensor: binary_sensor.vertex2_tx0_plus5
+  #tx1_plus5_sensor: binary_sensor.vertex2_tx1_plus5
   audio_out_signal_sensor: sensor.vertex2_audio_signal
   output_selected_sensor: sensor.vertex2_output_selected
   input_selected_sensor: sensor.vertex2_input_selected
+  no_signal_spam_sensor: binary_sensor.vertex2_no_signal_spam
   unreachable_sensor: binary_sensor.vertex2_unreachable
   command_processing_sensor: binary_sensor.vertex2_command_processing
   command_processing_timeout: 20

apps/global/presence.yaml

@@ -41,12 +41,13 @@ occupancy_sensors:
   default_values:
       binary_sensor.someone_home: off
       binary_sensor.projector_on: off
+      binary_sensor.is_tv_on: off
 
   sensors:
     binary_sensor.living_room_occupancy:
       trigger_conditions: 
         - sensor.living_room_last_motion <= 5
-        - binary_sensor.projector_on
+        - binary_sensor.projector_on or binary_sensor.is_tv_on
     binary_sensor.kitchen_occupancy: 
       trigger_conditions: sensor.kitchen_last_motion <= 5
     binary_sensor.garage_occupancy: 

apps/global/virtualsensors.yaml

@@ -110,10 +110,10 @@ virtual_sensors:
     binary_sensor.is_playstation_on: switch.ps5_136_power
     sensor.vertex2_input_selected_icon: sensor.vertex2_input_selected.icon
     retain_condition.vertex2_has_input_signal: 
-      retain_time: 10
+      retain_time: 20
       conditions: 
         trigger_conditions: (sensor.vertex2_video_signal and sensor.vertex2_video_signal != 'no signal') or sensor.vertex2_audio_signal
-        disable_conditions: binary_sensor.vertex2_unreachable or binary_sensor.vertex2_command_processing
+        disable_conditions: self and (binary_sensor.vertex2_unreachable or binary_sensor.vertex2_command_processing)
       sensor_attributes:
         icon: sensor.vertex2_input_selected_icon
     binary_sensor.is_using_apple_tv: binary_sensor.is_apple_tv_on and binary_sensor.vertex2_has_input_signal and sensor.vertex2_input_selected == 'AppleTV'

apps/samsungtheframe.py

@@ -23,12 +23,13 @@ class SamsungTheFrame(SmartObject):
 
         self.power_plug = self.args['input_sensors']['power_plug']
         self.power_plug_power = self.args['input_sensors']['power_plug_power']
-        self.is_on_sensor = self.args['output_sensors']['is_on']
+        self.is_on_entity = self.create_entity(self.args['output_sensors']['is_on'])
-        self.is_art_mode_sensor = self.args['output_sensors']['is_art_mode']
+        self.is_art_mode_entity = self.create_entity(self.args['output_sensors']['is_art_mode'])
-        self.tv_auto_shutdown = self.args['output_sensors']['tv_auto_shutdown']
+        auto_shutdown_state = self.get_state(self.args['output_sensors']['tv_auto_shutdown'])
-
+        self.tv_auto_shutdown_entity = self.create_entity(
-        if self.get_state(self.tv_auto_shutdown) == None:
+            self.args['output_sensors']['tv_auto_shutdown'],
-            self.set_state(self.tv_auto_shutdown,state = 'off')
+            state=auto_shutdown_state if auto_shutdown_state is not None else 'off',
+        )
 
         if 'app_list' in self.args: self.app_list = self.args['app_list']
         else: self.app_list = []
@@ -85,7 +86,7 @@ class SamsungTheFrame(SmartObject):
         super().terminate()
 
     def on_volume_event(self, event_name, event_data, direction):
-        if self.get_state(self.is_on_sensor) != 'on':
+        if self.is_on_entity.get_state() != 'on':
             self.close_volume_connection()
             return
         try:
@@ -138,9 +139,9 @@ class SamsungTheFrame(SmartObject):
 
         self.log_info(f"Requesting TV to switch to {requested_state}")
 
-        if requested_state == 'hdmi' and self.get_state(self.is_on_sensor) == 'off':
+        if requested_state == 'hdmi' and self.is_on_entity.get_state() == 'off':
-            self.log_info(f"setting {self.tv_auto_shutdown} to 'on'")
+            self.log_info(f"setting {self.tv_auto_shutdown_entity.entity_id} to 'on'")
-            self.set_state(self.tv_auto_shutdown,state = 'on')
+            self.tv_auto_shutdown_entity.set_state(state = 'on')
 
         #if we already have a confirm callback, let's cancel it
         if self.requested_state_confirm_cb_handle:
@@ -161,8 +162,8 @@ class SamsungTheFrame(SmartObject):
             if not success:
                 self.requested_state_confirm_cb_handle = self.run_in(self.confirm_change_state_cb, wait_time)
                 return False
-            is_art_mode = self.get_state(self.is_art_mode_sensor) == 'on'
+            is_art_mode = self.is_art_mode_entity.get_state() == 'on'
-            is_on = self.get_state(self.is_on_sensor) == 'on'
+            is_on = self.is_on_entity.get_state() == 'on'
             is_plugged = self.get_state(self.power_plug) == 'on'
 
             if self.requested_state == 'tv_off':
@@ -267,7 +268,7 @@ class SamsungTheFrame(SmartObject):
     #     if result == SmartCondition.Result.Succeeded:
     #         self.request_tv_state('art_mode')
     #     else:
-    #         if self.get_state(self.is_art_mode_sensor) == 'on':
+    #         if self.is_art_mode_entity.get_state() == 'on':
     #             self.select_random_art()
     #         #I need to request state if I'm not in art_mode as it will cancel a potential request pending
     #         self.request_tv_state('off')
@@ -315,19 +316,19 @@ class SamsungTheFrame(SmartObject):
             #     self.log(f"update sensors failed with exception {exception_string}")
             #     return False , 5
 
-        if self.get_state(self.is_art_mode_sensor) != is_art_mode:
+        if self.is_art_mode_entity.get_state() != is_art_mode:
-            self.log_info(f"Updating sensor: {self.is_art_mode_sensor} = '{is_art_mode}'")
+            self.log_info(f"Updating sensor: {self.is_art_mode_entity.entity_id} = '{is_art_mode}'")
-            self.set_state(self.is_art_mode_sensor,state = is_art_mode)
+            self.is_art_mode_entity.set_state(state = is_art_mode)
         
         if is_on == 'off':
             self.close_volume_connection()
 
-        if self.get_state(self.is_on_sensor) != is_on:
+        if self.is_on_entity.get_state() != is_on:
-            self.log_info(f"Updating sensor: {self.is_on_sensor} = '{is_on}'")
+            self.log_info(f"Updating sensor: {self.is_on_entity.entity_id} = '{is_on}'")
-            self.set_state(self.is_on_sensor,state = is_on)
+            self.is_on_entity.set_state(state = is_on)
-            if is_on == 'off' and self.get_state(self.tv_auto_shutdown) != 'off':
+            if is_on == 'off' and self.tv_auto_shutdown_entity.get_state() != 'off':
-                self.log_info(f"setting {self.tv_auto_shutdown} to 'off'")
+                self.log_info(f"setting {self.tv_auto_shutdown_entity.entity_id} to 'off'")
-                self.set_state(self.tv_auto_shutdown,state = 'off')
+                self.tv_auto_shutdown_entity.set_state(state = 'off')
 
 
         return True, 0

apps/samsungtheframeasync.py

@@ -0,0 +1,585 @@
+"""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()

apps/sonoscontroller.py

@@ -159,10 +159,9 @@ class SonosPlayer:
             self.log(f"Creating a ValueSelector for the audio_delay")
             self.audio_delay_selector = ValueSelector(self.controller,self.controller,sensor_name = yaml_conf["audio_delay"]["entity"],super_entity_id = self.entity_id,yaml_block = yaml_conf["audio_delay"]['value'],self_initialize = True)
 
-        self.is_playing_podcast_sensor = f'binary_sensor.{player_name}_is_playing_podcast'
+        self.is_playing_podcast_entity = self.controller.create_entity(
-        is_playing_podcast_sensor_entity = self.controller.get_entity(self.is_playing_podcast_sensor)
+            f'binary_sensor.{player_name}_is_playing_podcast',
-        if not is_playing_podcast_sensor_entity.exists():
+        )
-            is_playing_podcast_sensor_entity.add()
         self.controller.listen_state(self.on_state_change,self.entity_id)
         self.controller.listen_state(self.on_update_media_artist,self.entity_id,attribute = "media_artist")
         self.update_is_playing_podcast_sensor()
@@ -401,16 +400,16 @@ class SonosPlayer:
         self.update_is_playing_podcast_sensor()
 
     def update_is_playing_podcast_sensor(self):
-        if self.is_playing_podcast_sensor and self.is_playing():
+        if self.is_playing_podcast_entity and self.is_playing():
-            prev_state = self.controller.get_state(self.is_playing_podcast_sensor)
+            prev_state = self.is_playing_podcast_entity.get_state()
             if self.controller.get_state(self.entity_id,attribute = "media_artist") == None:
                 if prev_state != 'on':
-                    self.log(f"setting {self.is_playing_podcast_sensor} to on")
+                    self.log(f"setting {self.is_playing_podcast_entity.entity_id} to on")
-                    self.controller.set_state(self.is_playing_podcast_sensor,state = "on")
+                    self.is_playing_podcast_entity.set_state(state = "on")
             else:
                 if prev_state != 'off':
-                    self.log(f"setting {self.is_playing_podcast_sensor} to off")
+                    self.log(f"setting {self.is_playing_podcast_entity.entity_id} to off")
-                    self.controller.set_state(self.is_playing_podcast_sensor,state = "off")
+                    self.is_playing_podcast_entity.set_state(state = "off")
 
     def on_update_off_conditions(self):
         if self.off_conditions_evaluator.evaluate(False) == SmartCondition.Result.Succeeded:

apps/vertex2.py

@@ -1,6 +1,6 @@
 import appdaemon.plugins.hass.hassapi as hass
 from requests import get
-from requests.exceptions import Timeout,ConnectionError
+from requests.exceptions import Timeout,ConnectionError,RequestException
 from ad_toolbox.smartobject import SmartObject
 import ad_toolbox.smartcondition as SmartCondition
 import time
@@ -15,6 +15,24 @@ class Vertex2(SmartObject):
         self._stop_interrupt_listener()
         self._interrupt_stop = threading.Event()
         self._status_requested = threading.Event()
+        self._last_logged_status = None
+        self._spam_lock = threading.RLock()
+        self._spam_last_interrupt = None
+        self._spam_snapshot = None
+        self._spam_count = 0
+        self._spam_timer = None
+        self._spam_deadline = None
+        self.no_signal_spam_sensor = None
+        if self.args.get('no_signal_spam_sensor'):
+            self.no_signal_spam_sensor = self.create_entity(
+                self.args['no_signal_spam_sensor'], state='off')
+        self._connection_lock = threading.Lock()
+        self._interrupt_connection = None
+        self.startup_delay = max(0, float(self.args.get('startup_delay', 30)))
+        self._startup_deadline = 0
+        self._startup_commands = []
+        self.power_plug_entity = self.args.get("vertex2_power_plug", {}).get("entity")
+        self._power_allowed = self.power_plug_entity is None
         self._command_processing_lock = threading.Lock()
         self._command_processing_handle = None
         self._command_processing_deadline = None
@@ -44,6 +62,11 @@ class Vertex2(SmartObject):
             )
             for index in range(2)
         }
+        self.plus5_sensors = {
+            index: self.create_entity(self.args[key], state='unavailable')
+            for index, key in enumerate(('tx0_plus5_sensor', 'tx1_plus5_sensor'))
+            if self.args.get(key)
+        }
         self._tx_video_signal_values = {0: 'unavailable', 1: 'unavailable'}
 
         self.audio_out_signal_sensor = self.create_entity(
@@ -99,7 +122,11 @@ class Vertex2(SmartObject):
             state='off',
         )
 
-        self.power_plug_entity = None
+        if "vertex2_power_plug" in self.args:
+            self.power_plug_entity = self.args["vertex2_power_plug"]["entity"]
+            self.power_plug_conditions_evaluator = SmartCondition.Evaluator(self,self.args["vertex2_power_plug"]["conditions"],condition_name = 'vertex2_power_plug',on_change_cb = self.on_change_power_plug_conditions)
+        if self.power_plug_entity is not None:
+            self.listen_state(self.on_power_plug_state_change, self.power_plug_entity)
 
         if "filmmaker_mode_conditions" in self.args:
             self.filmmaker_mode_conditions = SmartCondition.Evaluator(self,self.args['filmmaker_mode_conditions'], condition_name = "filmmaker_mode_conditions",on_change_cb = self.on_filmmaker_mode_conditions_change)
@@ -152,9 +179,6 @@ class Vertex2(SmartObject):
                 ))
             self.on_update_output_selection()
 
-        if "vertex2_power_plug" in self.args:
-            self.power_plug_entity = self.args["vertex2_power_plug"]["entity"]
-            self.power_plug_conditions_evaluator = SmartCondition.Evaluator(self,self.args["vertex2_power_plug"]["conditions"],condition_name = 'vertex2_power_plug',on_change_cb = self.on_change_power_plug_conditions)
         self.vertex2_tcp_port = int(self.args.get('vertex2_tcp_port', 2220))
         self._interrupt_thread = threading.Thread(
             target=self._interrupt_listener,
@@ -165,6 +189,65 @@ class Vertex2(SmartObject):
         self.update_vertex2_states()
         self._interrupt_thread.start()
 
+    def _reset_no_signal_spam(self, state='off'):
+        if getattr(self, 'no_signal_spam_sensor', None) is None:
+            return
+        with self._spam_lock:
+            if self._spam_timer is not None:
+                self.cancel_timer(self._spam_timer)
+            self._spam_timer = None
+            self._spam_deadline = None
+            self._spam_last_interrupt = None
+            self._spam_snapshot = None
+            self._spam_count = 0
+            if self.no_signal_spam_sensor.get_state() != state:
+                self.no_signal_spam_sensor.set_state(state=state)
+
+    def _observe_no_signal_spam(self, status):
+        if getattr(self, 'no_signal_spam_sensor', None) is None:
+            return
+        with self._spam_lock:
+            missing = ('', 'no signal')
+            if any(status[field].strip().lower() not in missing for field in ('rx0', 'tx0')):
+                self._reset_no_signal_spam()
+                return
+            snapshot = tuple(status[field].strip() for field in
+                             ('rx0', 'rx1', 'tx0', 'tx1', 'audout', 'insel'))
+            if self._spam_snapshot is not None and snapshot != self._spam_snapshot:
+                self._reset_no_signal_spam()
+            at = status.get('interrupt_at')
+            if at is None:
+                if self.no_signal_spam_sensor.get_state() == 'unavailable':
+                    self.no_signal_spam_sensor.set_state(state='off')
+                return  # Polls and commands do not count as device interrupts.
+            now = time.monotonic()
+            if now - at >= 4:
+                return  # Do not revive a streak with a delayed callback.
+            previous = self._spam_last_interrupt
+            self._spam_last_interrupt = at
+            snapshot = tuple(status[field].strip() for field in
+                             ('rx0', 'rx1', 'tx0', 'tx1', 'audout', 'insel'))
+            self._spam_count = (self._spam_count + 1
+                                if previous is not None and 2 <= at - previous <= 4
+                                and snapshot == self._spam_snapshot else 1)
+            self._spam_snapshot = snapshot
+            state = 'on' if self._spam_count >= 3 else 'off'
+            if self.no_signal_spam_sensor.get_state() != state:
+                self.no_signal_spam_sensor.set_state(state=state)
+            if self._spam_timer is not None:
+                self.cancel_timer(self._spam_timer)
+            self._spam_deadline = at + 4
+            self._spam_timer = self.run_in(
+                self._expire_no_signal_spam, max(0, self._spam_deadline - now),
+                deadline=self._spam_deadline)
+
+    def _expire_no_signal_spam(self, kwargs):
+        with self._spam_lock:
+            if kwargs['deadline'] != self._spam_deadline:
+                return
+            self._spam_timer = None
+            self._reset_no_signal_spam()
+
     def _interrupt_listener(self, interrupt_stop):
         address = urlparse(self.args['vertex2_address'])
         host = address.hostname
@@ -173,10 +256,15 @@ class Vertex2(SmartObject):
         expected = set(status_fields) | {'insel'}
 
         while not interrupt_stop.is_set():
+            if not self._vertex2_ready():
+                interrupt_stop.wait(0.5)
+                continue
             try:
                 with socket.create_connection((host, self.vertex2_tcp_port), timeout=10) as connection:
-                    if interrupt_stop.is_set():
+                    with self._connection_lock:
-                        break
+                        self._interrupt_connection = connection
+                    if interrupt_stop.is_set() or not self._vertex2_powered():
+                        continue
                     # Detect a silent reboot/network loss without polling HDMI status.
                     connection.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
                     connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 10)
@@ -188,12 +276,16 @@ class Vertex2(SmartObject):
                     buffer = ''
                     status = {}
                     status_started = None
+                    reconfigure = True
+                    pending_interrupt_at = None
 
-                    while not interrupt_stop.is_set():
+                    while not interrupt_stop.is_set() and self._vertex2_powered():
                         if status_started is None and self._status_requested.is_set():
                             self._status_requested.clear()
                             status = {}
                             status_started = time.monotonic()
+                            request_interrupt_at = pending_interrupt_at
+                            pending_interrupt_at = None
                             for command in commands:
                                 connection.sendall((command + '\r').encode('ascii'))
 
@@ -204,7 +296,7 @@ class Vertex2(SmartObject):
                                 raise OSError('Vertex2 TCP status response timed out')
                             continue
 
-                        if interrupt_stop.is_set():
+                        if interrupt_stop.is_set() or not self._vertex2_powered():
                             break
                         if not data:
                             raise OSError('Vertex2 closed the IP INTERRUPT connection')
@@ -217,8 +309,7 @@ class Vertex2(SmartObject):
                             line, buffer = buffer.split('\n', 1)
                             line = line.strip()
                             if line == 'int':
-                                self._refresh_command_processing()
+                                pending_interrupt_at = time.monotonic()
-                                self.log_info('Vertex2 IP INTERRUPT received')
                                 self._status_requested.set()
                             elif status_started is not None:
                                 if line.startswith('insel '):
@@ -235,7 +326,10 @@ class Vertex2(SmartObject):
                                     self.run_in(self._apply_vertex2_status, 0,
                                                 rx0=snapshot['rx0'], rx1=snapshot['rx1'],
                                                 tx0=snapshot['tx0'], tx1=snapshot['tx1'],
-                                                audout=snapshot['audout'], insel=snapshot['insel'])
+                                                audout=snapshot['audout'], insel=snapshot['insel'],
+                                                reconfigure=reconfigure,
+                                                interrupt_at=request_interrupt_at)
+                                    reconfigure = False
                                     status_started = None
 
                         if status_started is not None and time.monotonic() - status_started > 5:
@@ -244,6 +338,10 @@ class Vertex2(SmartObject):
             except (OSError, ValueError) as error:
                 if interrupt_stop.is_set():
                     break
+                if not self._vertex2_powered():
+                    continue
+                self._reset_no_signal_spam('unavailable')
+                self._invalidate_plus5_sensors()
                 self.unreachable_sensor.set_state(state='on')
                 self.video_tx_signal_sensor.set_state(state='unavailable')
                 for sensor in self.video_tx_signal_sensors.values():
@@ -254,11 +352,29 @@ class Vertex2(SmartObject):
                 self.audio_out_signal_sensor.set_state(state='unavailable')
                 self.log_warning(f'Vertex2 IP INTERRUPT unavailable: {error}')
                 interrupt_stop.wait(5)
+            finally:
+                with self._connection_lock:
+                    self._interrupt_connection = None
+
+    def _close_interrupt_connection(self):
+        connection_lock = getattr(self, "_connection_lock", None)
+        if connection_lock is None:
+            return
+        with connection_lock:
+            connection = self._interrupt_connection
+            if connection is not None:
+                try:
+                    connection.shutdown(socket.SHUT_RDWR)
+                except OSError:
+                    pass
+                connection.close()
 
     def _stop_interrupt_listener(self):
+        self._reset_no_signal_spam('unavailable')
         interrupt_stop = getattr(self, '_interrupt_stop', None)
         if interrupt_stop is not None:
             interrupt_stop.set()
+        self._close_interrupt_connection()
         interrupt_thread = getattr(self, '_interrupt_thread', None)
         if interrupt_thread is not None:
             # Allow the 10s connect timeout or 1s recv timeout to finish.
@@ -296,6 +412,7 @@ class Vertex2(SmartObject):
                 except KeyError: pass
 
     def on_update_input_selection(self):
+        input_selected = None
         for evaluator in self.input_selection_evaluators:
             if evaluator[0].evaluate(False) == SmartCondition.Result.Succeeded:
                 evaluator[0].log_evaluation_result()
@@ -304,7 +421,7 @@ class Vertex2(SmartObject):
 
         self.log_info("Input selection conditions changed")
 
-        if input_selected != "None":
+        if input_selected is not None and str(input_selected) != "None":
             self.select_input(input_selected)
             self.update_vertex2_states()
 
@@ -384,17 +501,101 @@ class Vertex2(SmartObject):
         else:
             self.toggle_filmmaker_mode(False)
 
+    def _vertex2_powered(self):
+        return self._power_allowed and (
+            self.power_plug_entity is None
+            or self.get_state(self.power_plug_entity) == 'on'
+        )
+
+    def _vertex2_ready(self):
+        return self._vertex2_powered() and time.monotonic() >= self._startup_deadline
+
+    def _begin_startup(self):
+        self._startup_deadline = time.monotonic() + self.startup_delay
+        self.log_info(f'Waiting {self.startup_delay:g}s for Vertex2 to start')
+
+    def _flush_startup_commands(self):
+        with self._connection_lock:
+            commands = self._startup_commands
+            self._startup_commands = []
+        for cmd, value in commands:
+            self.send_cmd(cmd, value)
+
+    def _reconfigure_vertex2(self):
+        # Conditions may not have changed while the device was powered off.
+        # Evaluate them again instead of relying on change callbacks or old commands.
+        if self.power_plug_entity is not None:
+            result = self.power_plug_conditions_evaluator.evaluate(False)
+            self.on_change_power_plug_conditions(None, result)
+        if not self._vertex2_ready():
+            return
+
+        self.log_info('Reapplying Vertex2 configuration after connection restored')
+        if self.input_selection_evaluators:
+            self.on_update_input_selection()
+        if self.output_selection_evaluators:
+            # Also reapplies audio/video EDID and audio mute for the selected output.
+            self.on_update_output_selection()
+        else:
+            selected_output = self.output_names.get(self.output_selected_sensor.get_state())
+            if selected_output in (0, 1):
+                self.apply_output_audio_settings(selected_output)
+
+        for output_index, conditions in self.output_config_conditions.items():
+            evaluator = conditions.get('force_plus5')
+            if evaluator is not None:
+                self.on_output_plus5_condition_change(
+                    output_index, None, evaluator.evaluate(False),
+                )
+        if 'filmmaker_mode_conditions' in self.args:
+            self.on_filmmaker_mode_conditions_change(
+                None, self.filmmaker_mode_conditions.evaluate(False),
+            )
+
+    def _suspend_vertex2(self):
+        self._reset_no_signal_spam()
+        self._invalidate_plus5_sensors()
+        self._close_interrupt_connection()
+        with self._connection_lock:
+            self._startup_commands.clear()
+        self._startup_deadline = 0
+        self._status_requested.clear()
+        self._update_output_video_sensors('', '')
+        self.audio_out_signal_sensor.set_state(state='unavailable')
+        self.unreachable_sensor.set_state(state='off')
+        with self._command_processing_lock:
+            self._command_processing_deadline = None
+            if self._command_processing_handle is not None:
+                self.cancel_timer(self._command_processing_handle)
+                self._command_processing_handle = None
+            if self.command_processing_sensor is not None:
+                self.command_processing_sensor.set_state(state='off')
+
+    def on_power_plug_state_change(self, entity, attribute, old, new, kwargs):
+        if self._vertex2_powered():
+            if old != 'on' and new == 'on':
+                self._begin_startup()
+            self.update_vertex2_states()
+        else:
+            self._suspend_vertex2()
+
     def on_change_power_plug_conditions(self,prev_result,result):
-        if result == SmartCondition.Result.Succeeded: 
+        self._power_allowed = result == SmartCondition.Result.Succeeded
+        if self._power_allowed:
+            if self.get_state(self.power_plug_entity) != 'on':
+                self._begin_startup()
             self.turn_on(self.power_plug_entity)
         else:
+            # Close TCP before asking Home Assistant to cut the power.
+            self._suspend_vertex2()
             self.turn_off(self.power_plug_entity)
 
-    def _refresh_command_processing(self, start=False):
+    def _start_command_processing(self, only_if_active=False):
+        # Only commands start a window; changed status may extend an active one.
         if self.command_processing_sensor is None:
             return
         with self._command_processing_lock:
-            if not start and self.command_processing_sensor.get_state() != 'on':
+            if only_if_active and self._command_processing_deadline is None:
                 return
             if self._command_processing_handle is not None:
                 self.cancel_timer(self._command_processing_handle)
@@ -417,8 +618,16 @@ class Vertex2(SmartObject):
             self.command_processing_sensor.set_state(state='off')
 
     def send_cmd(self,cmd,value = ''):
+        if not self._vertex2_powered():
+            return
+        if not self._vertex2_ready():
+            with self._connection_lock:
+                self._startup_commands.append((cmd, value))
+            return
         request = f'{self.args["vertex2_address"]}/cmd?{cmd}={value}'
-        self._refresh_command_processing(start=True)
+        if cmd in ('tx0plus5', 'tx1plus5'):
+            self._invalidate_plus5_sensors()
+        self._start_command_processing()
         try:
             r = get(request, timeout=5)
             if cmd == 'edid' and value.startswith(('audiotx', 'audioout')):
@@ -429,6 +638,43 @@ class Vertex2(SmartObject):
         except (Timeout,ConnectionError):
             self.log_warning(f"{request} is unreachable")
 
+        finally:
+            if cmd in ('tx0plus5', 'tx1plus5'):
+                # Read back the device setting even if no IP interrupt follows.
+                self._status_requested.set()
+
+    def _invalidate_plus5_sensors(self):
+        for sensor in self.plus5_sensors.values():
+            if sensor.get_state() != 'unavailable':
+                sensor.set_state(state='unavailable')
+
+    def _refresh_plus5_sensors(self):
+        if not self.plus5_sensors:
+            return
+        if not self._vertex2_ready():
+            self._invalidate_plus5_sensors()
+            return
+        try:
+            # Same read-only endpoint used by the Vertex2 configuration page.
+            response = get(f'{self.args["vertex2_address"].rstrip("/")}/ssi/confpage.ssi', timeout=5)
+            response.raise_for_status()
+            config = response.json()
+            if not isinstance(config, dict):
+                raise ValueError('expected a configuration object')
+        except (RequestException, ValueError) as error:
+            self._invalidate_plus5_sensors()
+            self.log_warning(f'Vertex2 +5V readback failed: {error}')
+            return
+        if not self._vertex2_ready():
+            self._invalidate_plus5_sensors()
+            return
+        for index in range(2):
+            value = config.get(f'tx{index}plus5')
+            state = {'0': 'off', '1': 'on'}.get(str(value), 'unavailable')
+            sensor = self.plus5_sensors.get(index)
+            if sensor is not None and sensor.get_state() != state:
+                sensor.set_state(state=state)
+
     def mute_tx0(self, mute):
         self.send_cmd("mutetx0audio","on" if mute else "off")
 
@@ -535,9 +781,8 @@ class Vertex2(SmartObject):
         return {"name": config, "icon": None}
 
     def update_vertex2_states(self):
-        if self.power_plug_entity is not None and self.get_state(self.power_plug_entity) != 'on':
+        if not self._vertex2_powered():
-            self.run_in(self._apply_vertex2_status, 0,
+            self._suspend_vertex2()
-                        rx0='', rx1='', tx0='', tx1='', audout='', insel='0 0')
             return
         self._status_requested.set()
 
@@ -597,11 +842,8 @@ class Vertex2(SmartObject):
             self.allm_sensor.set_state(state=selected_allm)
 
     def _apply_vertex2_status(self, kwargs):
-        self.log_info(
+        if not self._vertex2_powered():
-            f"Vertex2 TCP status: RX0={kwargs['rx0']!r}, RX1={kwargs['rx1']!r}, "
+            return
-            f"TX0={kwargs['tx0']!r}, TX1={kwargs['tx1']!r}, "
-            f"AUDOUT={kwargs['audout']!r}, insel={kwargs['insel']!r}"
-        )
         try:
             portselect = [int(value) for value in kwargs['insel'].split()]
             if len(portselect) != 2:
@@ -611,6 +853,22 @@ class Vertex2(SmartObject):
             self.unreachable_sensor.set_state(state='on')
             return
 
+        self._observe_no_signal_spam(kwargs)
+
+        # Ignore callback metadata (e.g. reconfigure) when comparing device state.
+        snapshot = tuple(kwargs[field].strip() for field in
+                         ('rx0', 'rx1', 'tx0', 'tx1', 'audout', 'insel'))
+        previous = self._last_logged_status
+        if snapshot != previous:
+            self._last_logged_status = snapshot
+            self.log_info(
+                f"Vertex2 TCP status: RX0={kwargs['rx0']!r}, RX1={kwargs['rx1']!r}, "
+                f"TX0={kwargs['tx0']!r}, TX1={kwargs['tx1']!r}, "
+                f"AUDOUT={kwargs['audout']!r}, insel={kwargs['insel']!r}"
+            )
+            if previous is not None:
+                self._start_command_processing(only_if_active=True)
+
         if portselect[0] == 4:  # 4 = copy TX1
             portselect[0] = portselect[1]
         if portselect[1] == 4:  # 4 = copy TX0
@@ -643,3 +901,7 @@ class Vertex2(SmartObject):
         if self.audio_out_signal_sensor.get_state() != audio_out_signal:
             self.audio_out_signal_sensor.set_state(state=audio_out_signal)
         self.unreachable_sensor.set_state(state='off')
+        self._flush_startup_commands()
+        if kwargs.get('reconfigure', False):
+            self._reconfigure_vertex2()
+        self._refresh_plus5_sensors()

docs/radiator-heating.md

@@ -1,23 +1,70 @@
 # Chauffage radiateurs — AppDaemon et C6
 
 L'app `radiator_heating` expose `climate.radiator_heating` par MQTT Discovery.
-Les publications utilisent le service `mqtt.publish` de Home Assistant, comme
+La version MQTT directe utilise le plugin MQTT d'AppDaemon pour les publications
-l'ancienne implémentation de SmartObject. Aucun identifiant MQTT supplémentaire
+et les abonnements. Le plugin Home Assistant sert à lire les thermomètres.
-ni plugin MQTT AppDaemon n'est nécessaire.
+Les commandes du climate et les retours C6 restent sur les mêmes topics.
+
+## Configuration MQTT
+
+Le plugin `MQTT` est configuré dans `appdaemon.yaml`, namespace `mqtt`.
+Les identifiants du broker sont dans `secrets.yaml` (exclu de Git) et référencés
+par `!secret`. Ils doivent être sauvegardés séparément pour une réinstallation.
+Un redémarrage complet du conteneur/service AppDaemon est nécessaire pour charger
+un nouveau plugin ; le rechargement des apps ne suffit pas.
+
+Le climate et son watchdog exigent ce plugin. Les deux anciennes automatisations
+Home Assistant ont été supprimées après vérification des commandes et de la
+télémétrie en MQTT direct. Aucun relais MQTT dans Home Assistant n'est nécessaire.
 
 ## Fichiers
 
 - `apps/boilerclimate.py` : MQTT Discovery, consigne, mode, mesures et commandes.
+- `apps/boilerwatchdog.py` : surveillance indépendante du heartbeat.
 - `apps/boilerregulator.py` : calcul indépendant de la température d'eau.
-- `apps/global/heating.yaml`, bloc `radiator_heating` : paramètres actifs de l'app.
+- `apps/global/heating.yaml`, blocs `radiator_heating` et
-- `homeassistant/packages/radiator_heating_mqtt.yaml` : copie des deux
+  `radiator_heating_watchdog` : paramètres actifs des deux apps.
-  automatisations installées via l'API de configuration de Home Assistant.
+- `appdaemon.yaml`, plugin `MQTT` : connexion broker, abonnements et Last Will.
-  Ne pas inclure aussi cette copie comme package : les automatisations existent
-  déjà dans Home Assistant. Leur modification dans cette copie ne les redéploie pas.
 
 ## Régulation
 
-Avec S la consigne ambiante, Ti la température du couloir et Te celle de devant :
+Le bloc facultatif `smart_conditions` contrôle le mode `heat`/`off` avec le même
+évaluateur que les autres apps : `Succeeded` passe en `heat`, `Failed` passe
+en `off`. L'évaluation initiale est appliquée au démarrage, puis chaque changement
+de résultat. Un résultat `Disabled` suspend l'automatisme et conserve le mode.
+Les commandes manuelles restent possibles ; elles restent en vigueur jusqu'au
+prochain changement de résultat de la condition ou redémarrage de l'app.
+Le changement de mode envoie immédiatement la commande chaudière et est persisté.
+La consigne ambiante n'est pas modifiée et les sécurités capteurs restent actives.
+Sans ce bloc, le mode reste entièrement manuel comme auparavant. Exemple :
+
+```yaml
+smart_conditions:
+  trigger_conditions:
+    - binary_sensor.heatpump_winter_mode
+```
+
+Les options standard (`blocking_conditions`, `disable_conditions`,
+`callback_delay`, constantes locales) sont prises en charge.
+
+La règle active demande `heat` uniquement si quelqu'un est à la maison,
+si le minimum extérieur est inférieur à la consigne ambiante (`self.temperature`),
+et si toutes les fenêtres et portes-fenêtres surveillées sont
+fermées. `binary_sensor.group_windows` couvre bureau, salon, chambre d'amis
+et dressing ; les portes-fenêtres cuisine, salon et chambre sont ajoutées
+séparément. La règle utilise les états booléens de présence et d'ouverture.
+Le mode est recalculé au démarrage et lors des changements de résultat de la
+règle, sans temporisation ajoutée. Après un changement de consigne, la règle
+est réévaluée lorsque Home Assistant actualise l'attribut `temperature` du climate.
+`self` désigne l'entité configurée par `entity: climate.radiator_heating`.
+Toutes les expressions lisent les états Home Assistant normalement. Au premier
+démarrage, le chauffage reste en `off` pendant l'attente de l'entité et de sa
+consigne, avant de créer l'évaluateur. Celui-ci est ensuite conservé pendant
+toute la durée de vie de l'app, y compris lors des changements de consigne.
+
+Avec S la consigne ambiante, Ti la température du couloir et Te la plus basse
+des températures extérieures (`sensor.ext_front_temperature` et
+`sensor.ext_back_temperature`, configurées dans `outdoor_sensors`) :
 
     courbe = S + pente × (S − Te) + décalage
     correction P = kp × (S − Ti)
@@ -42,7 +89,12 @@ Après arrêt, le redémarrage exige Ti <= S + 0,1 °C et une demande suffisante
 Les gains sont des valeurs initiales à ajuster après les essais de circulation,
 pas des coefficients identifiés pour cette maison.
 
-Les deux capteurs doivent fournir des valeurs finies en °C, entre −40 et 60 °C.
+Le capteur intérieur et les deux capteurs extérieurs doivent fournir des valeurs
+finies en °C, entre −40 et 60 °C. Le minimum extérieur est recalculé à chaque
+mise à jour de l'un des capteurs et à chaque cycle. L'attribut `outdoor_sensors`
+liste les sondes utilisées et `outdoor_sensor` identifie celle sélectionnée.
+Si l'une des sondes est invalide, le chauffage reste inhibé comme auparavant.
+La configuration historique `outdoor_sensor` reste acceptée pour une seule sonde.
 Une température inchangée, même ancienne, reste valide : ces sondes peuvent
 ne publier qu'en cas de changement. Aucun délai sur l'âge de la mesure n'est
 appliqué. Les états indisponibles/inconnus, les valeurs invalides et les unités
@@ -67,11 +119,21 @@ cette app, conserve les accents et le symbole °C.
 - État et diagnostics : `appdaemon/radiator_heating/state` et `/attributes`.
 - Chaudière : `themaplus-condens/bai/SetMode/set`.
 
-L'automatisation de retour transmet les commandes MQTT et les retours chaudière
+Le plugin transmet les commandes MQTT et les retours chaudière à l'événement
-à l'événement `APPDAEMON_BOILER_MQTT`, écouté par l'app. Le watchdog marque le
+`MQTT_MESSAGE` du namespace `mqtt`, écouté directement par l'app. Le watchdog
-climate indisponible après trois minutes sans heartbeat. Il ne commande PAS
+AppDaemon marque le climate indisponible après trois minutes sans heartbeat
-la chaudière : il signale l'arrêt de l'app. Configuration et états sont retenus ;
+(vérification toutes les dix secondes). Il ne commande PAS la chaudière.
-les commandes utilisateur et chaudière ne doivent jamais être retenues.
+Il utilise un worker distinct du climate. Il publie `offline` au démarrage et
+après reconnexion tant qu'aucun heartbeat récent n'est arrivé. Seul le climate
+publie `online`, selon la validité de ses mesures.
+
+La disponibilité MQTT Discovery exige simultanément la disponibilité de l'app
+(`/availability`) et celle de la connexion MQTT (`/connection`). Le plugin
+publie `online` sur `/connection` à la connexion, et le broker publie `offline`
+via le Last Will si cette connexion est perdue. Ce mécanisme ne garantit pas
+une détection immédiate et ne coupe pas le chauffage. Configuration et états
+sont retenus ; les commandes utilisateur et chaudière ne doivent jamais l'être.
+Une reconnexion MQTT republie la découverte et renouvelle la commande.
 
 `SetMode` est renouvelé chaque minute et lors d'un changement de demande.
 Les retours de télémétrie ne renouvellent pas la commande. Un arrêt ordonné

docs/samsungtheframeasync.md

@@ -0,0 +1,91 @@
+# 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 en surveillance seule. Pendant une transition, cette attente est plafonnée
+  à 2 secondes, y compris après une erreur de transition. Une nouvelle consigne
+  raccourcit aussi une attente de reconnexion déjà en cours.
+- 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
+transition_retry_delay: 2 # attente entre tentatives pendant une transition
+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.
+
+Après allumage de la prise, la transition est revérifiée après 5 secondes ;
+après une commande de réveil, après 10 secondes (au lieu de 60). Les délais
+de confirmation après une touche et avant coupure électrique sont conservés.

homeassistant/packages/radiator_heating_mqtt.yaml

@@ -1,47 +0,0 @@
-# Installed through HA's automation configuration API. This is the reviewable
-# source copy; do not also include it as a package if the automations exist.
-automation:
-  - id: appdaemon_radiator_heating_bridge
-    alias: AppDaemon - Chauffage radiateurs - Retour MQTT
-    description: Transmet les commandes du climate et les retours C6 à AppDaemon.
-    triggers:
-      - trigger: mqtt
-        topic: appdaemon/radiator_heating/command/+
-      - trigger: mqtt
-        topic: themaplus-condens/bai/SetMode
-      - trigger: mqtt
-        topic: themaplus-condens/bai/Statenumber
-      - trigger: mqtt
-        topic: themaplus-condens/bai/FlowTemp
-      - trigger: mqtt
-        topic: homeassistant/status
-    actions:
-      - event: APPDAEMON_BOILER_MQTT
-        event_data:
-          topic: '{{ trigger.topic }}'
-          payload: '{{ trigger.payload }}'
-    mode: queued
-    max: 50
-  - id: appdaemon_radiator_heating_watchdog
-    alias: AppDaemon - Chauffage radiateurs - Disponibilité
-    description: Marque le climate indisponible après trois minutes sans heartbeat.
-    triggers:
-      - trigger: mqtt
-        topic: appdaemon/radiator_heating/heartbeat
-        id: heartbeat
-      - trigger: homeassistant
-        event: start
-        id: startup
-    actions:
-      - if:
-          - condition: template
-            value_template: "{{ trigger.id == 'heartbeat' }}"
-        then:
-          - delay:
-              minutes: 3
-      - action: mqtt.publish
-        data:
-          topic: appdaemon/radiator_heating/availability
-          payload: offline
-          retain: true
-    mode: restart

tests/test_boilerclimate.py

@@ -5,12 +5,14 @@ import tempfile
 from pathlib import Path
 from types import SimpleNamespace
 import unittest
+import yaml
 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 boilerregulator import BoilerRegulator, setmode
-from boilerclimate import BoilerClimate
+from boilerclimate import BoilerClimate, SmartCondition
+from boilerwatchdog import BoilerWatchdog
 
 
 class RegulatorTests(unittest.TestCase):
@@ -68,8 +70,13 @@ class ClimateTests(unittest.TestCase):
         # Test the actual app callbacks without starting an AppDaemon server.
         app = object.__new__(BoilerClimate)
         app.base, app.boiler = 'appdaemon/test', 'boiler'
+        app.entity_id = 'climate.radiator_heating'
+        app.mqtt_namespace = 'mqtt'
+        app.connection_topic = app.base + '/connection'
         app.mode, app.target = 'heat', 18
         app.indoor, app.outdoor = 'sensor.indoor', 'sensor.outdoor'
+        app.outdoors = (app.outdoor,)
+        app.selected_outdoor = None
         app.min_target, app.max_target = 10, 23
         app.interval = 60
         app.controller = BoilerRegulator()
@@ -81,6 +88,8 @@ class ClimateTests(unittest.TestCase):
         app._publish = Mock()
         app._save = Mock()
         app.log = Mock()
+        app.entity_exists = Mock(return_value=True)
+        app.get_state = Mock(return_value=18)
         app.initialize_logger_interface(app, ascii_encode=False)
         readings = {app.indoor: 18, app.outdoor: 10}
         app._sensor_value = lambda entity: readings[entity]
@@ -89,6 +98,39 @@ class ClimateTests(unittest.TestCase):
     def writes(self, app):
         return [c for c in app._publish.call_args_list if c.args[0] == 'boiler/bai/SetMode/set']
 
+    def test_mqtt_transport_uses_plugin_namespace(self):
+        a = self.app()
+        a.call_service = Mock()
+        BoilerClimate._publish(a, 'boiler/bai/SetMode/set', setmode(40, True))
+        call = a.call_service.call_args
+        self.assertEqual(call.args, ('mqtt/publish',))
+        self.assertEqual(call.kwargs['namespace'], 'mqtt')
+        self.assertFalse(call.kwargs['retain'])
+        self.assertEqual(json.loads(call.kwargs['payload']), setmode(40, True))
+
+    def test_discovery_requires_app_and_connection_online(self):
+        a = self.app()
+        a.uid, a.title = 'test_boiler', 'Test boiler'
+        a._discovery()
+        config = a._publish.call_args_list[0].args[1]
+        self.assertNotIn('availability_topic', config)
+        self.assertEqual(config['availability_mode'], 'all')
+        self.assertEqual(config['availability'], [{'topic': a.base + '/availability'},
+                                                   {'topic': a.connection_topic}])
+
+    def test_mqtt_reconnect_renews_discovery_and_command(self):
+        a = self.app()
+        a._discovery = Mock()
+        a._mqtt_event('MQTT_MESSAGE', {'state': 'Connected', 'topic': None}, {})
+        a._discovery.assert_called_once()
+        self.assertEqual(len(self.writes(a)), 1)
+
+    def test_mqtt_disconnect_does_not_publish_or_advance_controller(self):
+        a = self.app()
+        a._mqtt_event('MQTT_MESSAGE', {'state': 'Disconnected', 'topic': None}, {})
+        a._publish.assert_not_called()
+        self.assertEqual(a.previous_time, 1000)
+
     def test_single_change_and_periodic_refresh_only(self):
         a = self.app()
         with patch('boilerclimate.time.monotonic', return_value=1000):
@@ -103,6 +145,34 @@ class ClimateTests(unittest.TestCase):
         for call in self.writes(a):
             self.assertFalse(call.args[2])
 
+    def test_lowest_outdoor_reading_controls_regulation(self):
+        a = self.app()
+        a.target = 23
+        a.outdoors = ('sensor.front', 'sensor.back')
+        readings = {a.indoor: 22.6, 'sensor.front': 30.2, 'sensor.back': 23.7}
+        a._sensor_value = lambda entity: readings[entity]
+        a._control()
+        self.assertEqual(a._state['outdoor'], 23.7)
+        self.assertEqual(a.selected_outdoor, 'sensor.back')
+        self.assertTrue(a._state['heating_requested'])
+        readings['sensor.front'] = 20
+        a._sensor_changed('sensor.front', 'all', None, None, {})
+        self.assertEqual(a._state['outdoor'], 20)
+        self.assertEqual(a.selected_outdoor, 'sensor.front')
+
+    def test_invalid_second_outdoor_sensor_still_inhibits_heat(self):
+        a = self.app()
+        a.outdoors = ('sensor.front', 'sensor.back')
+        def reading(entity):
+            if entity == 'sensor.back':
+                raise ValueError('sensor.back: unavailable')
+            return 18 if entity == a.indoor else 5
+        a._sensor_value = reading
+        a._control()
+        self.assertEqual(self.writes(a)[0].args[1]['disablehc'], 1)
+        self.assertEqual(a._state['reason'], 'invalid_sensor')
+        self.assertIn('sensor.back', a.sensor_error)
+
     def test_invalid_sensor_explicitly_inhibits_heat(self):
         a = self.app()
         a._sensor_value = Mock(side_effect=ValueError('unavailable'))
@@ -129,6 +199,133 @@ class ClimateTests(unittest.TestCase):
         a._mqtt_event('', dict(topic=a.base + '/command/mode', payload='off'), {})
         self.assertEqual(self.writes(a)[0].args[1]['disablehc'], 1)
 
+    def test_smart_conditions_without_configuration_keep_manual_mode(self):
+        a = self.app()
+        a.args = {}
+        a._initialize_smart_conditions()
+        self.assertIsNone(a.smart_conditions_evaluator)
+        self.assertEqual(a.mode, 'heat')
+        a._publish.assert_not_called()
+
+    def test_smart_condition_evaluates_initial_state_and_sensor_changes(self):
+        a = self.app()
+        a.args = {'smart_conditions': {'trigger_conditions': ['binary_sensor.test_heating']}}
+        a.entity_exists = Mock(return_value=True)
+        a.get_state = Mock(return_value='off')
+        a.listen_state = Mock(return_value='condition_callback')
+        a.cancel_listen_state = Mock()
+        a._initialize_smart_conditions()
+        self.assertEqual(a.mode, 'off')
+        self.assertEqual(self.writes(a)[-1].args[1]['disablehc'], 1)
+        callback = a.listen_state.call_args.args[0]
+        a.get_state.return_value = 'on'
+        callback('binary_sensor.test_heating', 'state', 'off', 'on', {})
+        self.assertEqual(a.mode, 'heat')
+        self.assertEqual(self.writes(a)[-1].args[1]['disablehc'], 0)
+        self.assertEqual(a.target, 18)
+        a.get_state.return_value = 'off'
+        callback('binary_sensor.test_heating', 'state', 'on', 'off', {})
+        self.assertEqual(a.mode, 'off')
+        a.terminate()
+        a.cancel_listen_state.assert_called_once_with('condition_callback')
+
+    def test_smart_condition_disabled_preserves_manual_mode(self):
+        a = self.app()
+        a.on_smart_conditions_change(None, SmartCondition.Result.Disabled)
+        self.assertEqual(a.mode, 'heat')
+        a._publish.assert_not_called()
+
+    def test_smart_condition_does_not_bypass_invalid_sensor(self):
+        a = self.app()
+        a.mode = 'off'
+        a._sensor_value = Mock(side_effect=ValueError('unavailable'))
+        a.on_smart_conditions_change(None, SmartCondition.Result.Succeeded)
+        self.assertEqual(a.mode, 'heat')
+        self.assertEqual(self.writes(a)[-1].args[1]['disablehc'], 1)
+
+    def test_condition_cleanup_failure_does_not_block_shutdown(self):
+        a = self.app()
+        a.smart_conditions_evaluator = Mock()
+        a.smart_conditions_evaluator.deactivate.side_effect = RuntimeError('cleanup failed')
+        a.terminate()
+        self.assertEqual(self.writes(a)[-1].args[1]['disablehc'], 1)
+        a._save.assert_called_once()
+
+    def test_configured_presence_weather_and_window_rule(self):
+        config = yaml.safe_load((Path(__file__).resolve().parents[1] /
+                                 'apps/global/heating.yaml').read_text())['radiator_heating']
+        baseline = {'binary_sensor.someone_home': 'on',
+                    'sensor.ext_front_temperature': '30', 'sensor.ext_back_temperature': '10',
+                    'binary_sensor.group_windows': 'off',
+                    'binary_sensor.kitchen_french_door': 'off',
+                    'binary_sensor.living_room_french_door': 'off',
+                    'binary_sensor.bedroom_french_door': 'off'}
+        cases = [({}, 'heat'),
+                 ({'binary_sensor.someone_home': 'off'}, 'off'),
+                 ({'sensor.ext_back_temperature': '18'}, 'off'),
+                 ({'sensor.ext_front_temperature': '10', 'sensor.ext_back_temperature': '30'}, 'heat')]
+        for entity in ['binary_sensor.group_windows', 'binary_sensor.kitchen_french_door',
+                       'binary_sensor.living_room_french_door', 'binary_sensor.bedroom_french_door']:
+            cases.append(({entity: 'on'}, 'off'))
+        for changes, expected in cases:
+            with self.subTest(changes=changes):
+                a = self.app()
+                a.mode = 'off'
+                a.args = config
+                states = {**baseline, **changes}
+                a.entity_exists = lambda entity: entity in states or entity == a.entity_id
+                a.get_state = lambda entity, **kwargs: a.target if entity == a.entity_id else states[entity]
+                a.listen_state = Mock(return_value='condition_callback')
+                a._initialize_smart_conditions()
+                self.assertEqual(a.mode, expected)
+                listened = {c.args[1] for c in a.listen_state.call_args_list}
+                self.assertTrue(set(baseline).issubset(listened))
+
+    def test_target_change_is_evaluated_when_ha_attribute_updates(self):
+        a = self.app()
+        a.args = {'smart_conditions': {
+            'trigger_conditions': ['sensor.outside < self.temperature']}}
+        a.entity_exists = Mock(return_value=True)
+        states = {a.entity_id: 18, 'sensor.outside': 20}
+        a.get_state = lambda entity, **kwargs: states[entity]
+        a.listen_state = Mock(return_value='condition_callback')
+        a.cancel_listen_state = Mock()
+        a._initialize_smart_conditions()
+        self.assertEqual(a.mode, 'off')  # 20 outside >= target 18.
+        evaluator = a.smart_conditions_evaluator
+        callback = next(c.args[0] for c in a.listen_state.call_args_list
+                        if c.args[1] == a.entity_id and c.kwargs.get('attribute') == 'temperature')
+        for target, mode in [(23, 'heat'), (20, 'off'), (21, 'heat')]:
+            previous_mode = a.mode
+            old_target = states[a.entity_id]
+            a._mqtt_event('', dict(topic=a.base + '/command/temperature', payload=str(target)), {})
+            self.assertEqual(a.mode, previous_mode)  # Wait for HA's attribute update.
+            states[a.entity_id] = target
+            callback(a.entity_id, 'temperature', old_target, target, {})
+            self.assertEqual(a.mode, mode)
+            self.assertEqual(a.target, target)
+            self.assertIs(a.smart_conditions_evaluator, evaluator)
+        a.cancel_listen_state.assert_not_called()
+
+    def test_smart_conditions_wait_for_discovered_entity_and_temperature(self):
+        a = self.app()
+        a.mode, a.target = 'off', 23
+        a.args = {'smart_conditions': {'trigger_conditions': ['sensor.outside < self.temperature']}}
+        a.entity_exists = Mock(return_value=False)
+        a.get_state = Mock(return_value=None)
+        a.run_in = Mock()
+        a.listen_state = Mock(return_value='condition_callback')
+        a._initialize_smart_conditions()
+        self.assertEqual(a.mode, 'off')
+        self.assertIsNone(a.smart_conditions_evaluator)
+        a.run_in.assert_called_once_with(a._initialize_smart_conditions, 1)
+        a.entity_exists.return_value = True
+        a._initialize_smart_conditions({})  # Entity exists but its target is not ready.
+        self.assertIsNone(a.smart_conditions_evaluator)
+        a.get_state = lambda entity, **kwargs: 20 if entity == 'sensor.outside' else 23
+        a._initialize_smart_conditions()
+        self.assertEqual(a.mode, 'heat')
+
     def test_native_ha_template_payload_is_accepted(self):
         a = self.app()
         a._mqtt_event('', dict(topic='boiler/bai/Statenumber', payload={'value': 30}), {})
@@ -265,5 +462,62 @@ class ClimateTests(unittest.TestCase):
         self.assertEqual(a.controller.integral, integral)
 
 
+class WatchdogTests(unittest.TestCase):
+    def app(self):
+        a = object.__new__(BoilerWatchdog)
+        a.base, a.mqtt_namespace, a.timeout = 'appdaemon/test', 'mqtt', 180
+        a.last_heartbeat, a.fault = None, False
+        a.waiting_since, a.warned = 1000, False
+        a.call_service, a.log = Mock(), Mock()
+        a.initialize_logger_interface(a, ascii_encode=False)
+        return a
+
+    def test_missing_heartbeat_publishes_offline_once(self):
+        a = self.app()
+        with patch('boilerwatchdog.time.monotonic', return_value=1000):
+            a._check()
+            a._check()
+        a.call_service.assert_called_once_with('mqtt/publish', namespace='mqtt',
+                                              topic=a.base + '/availability', payload='offline', qos=0, retain=True)
+        a.log.assert_not_called()
+        a._mqtt_event('', {'topic': a.base + '/heartbeat'}, {})
+        a.log.assert_not_called()
+
+    def test_first_heartbeat_missing_beyond_timeout_warns_once(self):
+        a = self.app()
+        with patch('boilerwatchdog.time.monotonic', return_value=1000):
+            a._check()
+        with patch('boilerwatchdog.time.monotonic', return_value=1180):
+            a._check()
+            a._check()
+        self.assertEqual(a.log.call_count, 1)
+        self.assertEqual(a.log.call_args.kwargs['level'], 'WARNING')
+
+    def test_timeout_and_recovery_do_not_override_sensor_availability(self):
+        a = self.app()
+        with patch('boilerwatchdog.time.monotonic', return_value=1000):
+            a._mqtt_event('', {'topic': a.base + '/heartbeat', 'payload': 'timestamp'}, {})
+        with patch('boilerwatchdog.time.monotonic', return_value=1179):
+            a._check()
+        a.call_service.assert_not_called()
+        with patch('boilerwatchdog.time.monotonic', return_value=1180):
+            a._check()
+        self.assertTrue(a.fault)
+        self.assertTrue(a.warned)
+        a.call_service.reset_mock()
+        a._mqtt_event('', {'topic': a.base + '/heartbeat'}, {})
+        self.assertFalse(a.fault)
+        a.call_service.assert_not_called()  # The climate alone decides online/offline for sensor faults.
+
+    def test_reconnection_waits_for_new_heartbeat(self):
+        a = self.app()
+        a.last_heartbeat = 1000
+        a._mqtt_event('', {'topic': None, 'state': 'Connected'}, {})
+        self.assertIsNone(a.last_heartbeat)
+        self.assertTrue(a.fault)
+        self.assertEqual(a.call_service.call_args.kwargs['payload'], 'offline')
+        a.log.assert_not_called()
+
+
 if __name__ == '__main__':
     unittest.main()

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

tests/test_samsungtheframe_async.py

@@ -0,0 +1,436 @@
+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 test_fresh_target_shortens_idle_reconnect_delay(self):
+        app = self.app()
+        app._retry_at = app._loop.time() + 60
+        app._transition = AsyncMock(return_value=None)
+        app._set_target('hdmi')
+        self.assertLessEqual(app._retry_at - app._loop.time(), 2)
+        await app._target_task
+
+    async def test_wake_rechecks_after_ten_seconds(self):
+        app = self.app()
+        app.args['wake_up_service'] = 'esphome/wake_tv'
+        app.get_state = AsyncMock(side_effect=lambda e: 'on' if e == app.power_plug else '0')
+        self.assertEqual(await app._transition('hdmi'), 10)
+        app.call_service.assert_awaited_once_with('esphome/wake_tv')
+        app._client.send_key.assert_not_awaited()
+
+    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'), 5)
+        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()

tests/test_vertex2.py

@@ -0,0 +1,506 @@
+from pathlib import Path
+import socket
+import sys
+import threading
+import unittest
+from unittest.mock import Mock, patch
+
+sys.path[:0] = [str(Path(__file__).resolve().parents[1] / 'apps'),
+                str(Path(__file__).resolve().parents[1] / 'apps/ad_toolbox')]
+from vertex2 import Vertex2
+import ad_toolbox.smartcondition as SmartCondition
+
+
+class Harness(Vertex2):
+    def __init__(self, plugged=True):
+        self.plugged = plugged
+        self._power_allowed = True
+        self.power_plug_entity = 'switch.vertex2'
+        self._interrupt_stop = threading.Event()
+        self._status_requested = threading.Event()
+        self._last_logged_status = None
+        self._connection_lock = threading.Lock()
+        self._interrupt_connection = None
+        self.startup_delay = 30
+        self._startup_deadline = 0
+        self._startup_commands = []
+        self._command_processing_lock = threading.Lock()
+        self._command_processing_deadline = 123
+        self._command_processing_handle = 'timer'
+        self.command_processing_timeout = 30
+        self.command_processing_sensor = Mock()
+        self.plus5_sensors = {}
+        self.audio_out_signal_sensor = Mock()
+        self.unreachable_sensor = Mock()
+        self._update_output_video_sensors = Mock()
+        self.turn_on = Mock()
+        self.turn_off = Mock()
+        self.cancel_timer = Mock()
+        self.log_warning = Mock()
+        self.log_info = Mock()
+        self.run_in = Mock()
+        self.vertex2_tcp_port = 2220
+
+    @property
+    def args(self):
+        return {'vertex2_address': 'http://192.0.2.1'}
+
+    @property
+    def name(self):
+        return 'test_vertex'
+
+    def get_state(self, entity):
+        return 'on' if self.plugged else 'off'
+
+
+class PowerTests(unittest.TestCase):
+    def test_condition_closes_socket_before_cutting_power(self):
+        app = Harness()
+        connection = Mock()
+        app._interrupt_connection = connection
+        app.turn_off.side_effect = lambda entity: connection.close.assert_called_once()
+        app.on_change_power_plug_conditions(None, SmartCondition.Result.Failed)
+        connection.shutdown.assert_called_once_with(socket.SHUT_RDWR)
+        app.turn_off.assert_called_once_with('switch.vertex2')
+        self.assertFalse(app._vertex2_powered())
+        app.cancel_timer.assert_called_once_with('timer')
+        self.assertIsNone(app._command_processing_deadline)
+        app.command_processing_sensor.set_state.assert_called_once_with(state='off')
+        app.unreachable_sensor.set_state.assert_called_once_with(state='off')
+
+    def test_off_blocks_http_commands(self):
+        for plugged, allowed in [(False, True), (True, False)]:
+            app = Harness(plugged)
+            app._power_allowed = allowed
+            with patch('vertex2.get') as get:
+                app.send_cmd('reboot')
+            get.assert_not_called()
+
+    def test_no_plug_preserves_connectivity(self):
+        app = Harness(False)
+        app.power_plug_entity = None
+        self.assertTrue(app._vertex2_powered())
+
+    def test_delayed_status_is_ignored_after_power_off(self):
+        app = Harness(False)
+        app._apply_vertex2_status({})
+        app.log_info.assert_not_called()
+        app.run_in.assert_not_called()
+
+    def test_external_plug_off_suspends_connection(self):
+        app = Harness(False)
+        connection = Mock()
+        app._interrupt_connection = connection
+        app.on_power_plug_state_change('switch.vertex2', 'state', 'on', 'off', {})
+        connection.close.assert_called_once()
+        app._update_output_video_sensors.assert_called_once_with('', '')
+
+    def test_plug_on_requests_fresh_status(self):
+        app = Harness()
+        app.on_power_plug_state_change('switch.vertex2', 'state', 'off', 'on', {})
+        self.assertTrue(app._status_requested.is_set())
+        self.assertFalse(app._vertex2_ready())
+
+    def test_boot_commands_wait_and_are_replayed(self):
+        app = Harness()
+        with patch('vertex2.time.monotonic', return_value=100):
+            app._begin_startup()
+            with patch('vertex2.get') as get:
+                app.send_cmd('tx1plus5', 'on')
+                app.send_cmd('edid', 'audiotx1 on')
+            get.assert_not_called()
+            self.assertFalse(app._vertex2_ready())
+        with patch('vertex2.time.monotonic', return_value=130):
+            self.assertTrue(app._vertex2_ready())
+            with patch('vertex2.get') as get:
+                app._flush_startup_commands()
+            self.assertEqual(get.call_count, 2)
+            self.assertEqual(app._startup_commands, [])
+
+    def test_http_failure_after_boot_still_warns(self):
+        from requests.exceptions import ConnectionError
+        app = Harness()
+        app._startup_deadline = 130
+        with patch('vertex2.time.monotonic', return_value=130):
+            with patch('vertex2.get', side_effect=ConnectionError):
+                app.send_cmd('tx1plus5', 'on')
+        app.log_warning.assert_called_once()
+
+    def test_power_cut_discards_queued_commands(self):
+        app = Harness()
+        app._begin_startup()
+        app.send_cmd('tx1plus5', 'on')
+        app._suspend_vertex2()
+        self.assertEqual(app._startup_commands, [])
+
+    def test_listener_does_not_connect_during_boot(self):
+        app = Harness()
+        app._begin_startup()
+        worker = threading.Thread(target=app._interrupt_listener, args=(app._interrupt_stop,))
+        with patch('vertex2.socket.create_connection') as connect:
+            worker.start()
+            try:
+                self.assertFalse(app._interrupt_stop.wait(.1))
+                connect.assert_not_called()
+            finally:
+                app._interrupt_stop.set()
+                worker.join(2)
+        self.assertFalse(worker.is_alive())
+
+    def test_cut_power_unblocks_recv_without_warning(self):
+        app = Harness()
+        receiving = threading.Event()
+        closed = threading.Event()
+        connection = Mock()
+        connection.shutdown.side_effect = lambda how: closed.set()
+        def recv(size):
+            receiving.set()
+            closed.wait(2)
+            return b''
+        connection.recv.side_effect = recv
+        context = Mock(__enter__=Mock(return_value=connection), __exit__=Mock(return_value=False))
+        worker = threading.Thread(target=app._interrupt_listener, args=(app._interrupt_stop,))
+        with patch('vertex2.socket.create_connection', return_value=context):
+            worker.start()
+            try:
+                self.assertTrue(receiving.wait(2))
+                app.on_change_power_plug_conditions(None, SmartCondition.Result.Failed)
+                self.assertTrue(closed.is_set())
+            finally:
+                app._interrupt_stop.set()
+                closed.set()
+                worker.join(2)
+        self.assertFalse(worker.is_alive())
+        app.log_warning.assert_not_called()
+
+    def test_listener_waits_off_then_reconnects_when_on(self):
+        app = Harness(False)
+        attempted = threading.Event()
+        def connect(*args, **kwargs):
+            attempted.set()
+            app._interrupt_stop.set()
+            raise OSError('test stop')
+        worker = threading.Thread(target=app._interrupt_listener, args=(app._interrupt_stop,))
+        with patch('vertex2.socket.create_connection', side_effect=connect) as create:
+            worker.start()
+            try:
+                self.assertFalse(attempted.wait(.1))
+                create.assert_not_called()
+                app.plugged = True
+                self.assertTrue(attempted.wait(2))
+            finally:
+                app._interrupt_stop.set()
+                worker.join(2)
+        self.assertFalse(worker.is_alive())
+        app.log_warning.assert_not_called()
+
+
+class ReconfigurationTests(unittest.TestCase):
+    def test_reconnect_reapplies_current_plus5_and_selections(self):
+        app = Harness()
+        app.power_plug_conditions_evaluator = Mock()
+        app.power_plug_conditions_evaluator.evaluate.return_value = SmartCondition.Result.Succeeded
+        app.input_selection_evaluators = [Mock()]
+        app.output_selection_evaluators = [Mock()]
+        app.on_update_input_selection = Mock()
+        app.on_update_output_selection = Mock()
+        tx0, tx1 = Mock(), Mock()
+        tx0.evaluate.return_value = SmartCondition.Result.Failed
+        tx1.evaluate.return_value = SmartCondition.Result.Succeeded
+        app.output_config_conditions = {0: {'force_plus5': tx0}, 1: {'force_plus5': tx1}}
+        app.force_tx0_plus5 = Mock()
+        app.force_tx1_plus5 = Mock()
+        app._reconfigure_vertex2()
+        app.power_plug_conditions_evaluator.evaluate.assert_called_once_with(False)
+        app.on_update_input_selection.assert_called_once_with()
+        app.on_update_output_selection.assert_called_once_with()
+        app.force_tx0_plus5.assert_called_once_with(False)
+        app.force_tx1_plus5.assert_called_once_with(True)
+        for evaluator in (tx0, tx1):
+            evaluator.evaluate.assert_called_once_with(False)
+
+    def test_reconnect_stops_if_power_condition_no_longer_allows_device(self):
+        app = Harness()
+        app.power_plug_conditions_evaluator = Mock()
+        app.power_plug_conditions_evaluator.evaluate.return_value = SmartCondition.Result.Failed
+        app.on_update_output_selection = Mock()
+        app._reconfigure_vertex2()
+        app.turn_off.assert_called_once_with('switch.vertex2')
+        app.on_update_output_selection.assert_not_called()
+
+    def test_input_selection_without_matching_condition_does_not_send_command(self):
+        app = Harness()
+        evaluator = Mock()
+        evaluator.evaluate.return_value = SmartCondition.Result.Failed
+        app.input_selection_evaluators = [(evaluator, 'AppleTV')]
+        app.select_input = Mock()
+        app.on_update_input_selection()
+        app.select_input.assert_not_called()
+
+    def test_only_first_complete_status_requests_reconfiguration(self):
+        app = Harness()
+        fields = ('rx0', 'rx1', 'tx0', 'tx1', 'tx0sink', 'tx1sink',
+                  'aud0', 'aud1', 'audout', 'spd0', 'spd1')
+        response = ('\n'.join(f'{field.upper()}: no signal' for field in fields)
+                    + '\ninsel 0 0\n').encode('ascii')
+        connection = Mock()
+        responses = iter([response, response])
+        def recv(size):
+            try:
+                return next(responses)
+            except StopIteration:
+                app._interrupt_stop.set()
+                return b''
+        connection.recv.side_effect = recv
+        app.run_in.side_effect = lambda *args, **kwargs: app._status_requested.set()
+        context = Mock(__enter__=Mock(return_value=connection), __exit__=Mock(return_value=False))
+        with patch('vertex2.socket.create_connection', return_value=context):
+            app._interrupt_listener(app._interrupt_stop)
+        self.assertEqual([call.kwargs['reconfigure'] for call in app.run_in.call_args_list],
+                         [True, False])
+
+    def test_status_reconfigures_before_readback(self):
+        app = Harness()
+        app.input_selected_sensors = {i: Mock() for i in range(2)}
+        app.generate_input_selected_sensor_name = lambda i: i
+        app.get_input_config = Mock(return_value={'name': 'AppleTV'})
+        app.output_names = {}
+        app.output_selected_sensor = Mock()
+        app._refresh_selected_input_sensor = Mock()
+        sequence = Mock()
+        app._flush_startup_commands = sequence.flush
+        app._reconfigure_vertex2 = sequence.reconfigure
+        app._refresh_plus5_sensors = sequence.readback
+        app._apply_vertex2_status(dict(rx0='', rx1='', tx0='', tx1='', audout='',
+                                      insel='0 0', reconfigure=True))
+        self.assertEqual([call[0] for call in sequence.mock_calls],
+                         ['flush', 'reconfigure', 'readback'])
+        sequence.reset_mock()
+        app._apply_vertex2_status(dict(rx0='', rx1='', tx0='', tx1='', audout='', insel='0 0'))
+        sequence.reconfigure.assert_not_called()
+
+
+class Plus5Tests(unittest.TestCase):
+    def setUp(self):
+        self.app = Harness()
+        self.app.plus5_sensors = {i: Mock() for i in range(2)}
+
+    def test_reads_actual_settings_independently(self):
+        with patch('vertex2.get') as get:
+            get.return_value.json.return_value = {'tx0plus5': '0', 'tx1plus5': '1'}
+            self.app._refresh_plus5_sensors()
+        self.app.plus5_sensors[0].set_state.assert_called_once_with(state='off')
+        self.app.plus5_sensors[1].set_state.assert_called_once_with(state='on')
+
+    def test_unknown_value_does_not_confirm_off(self):
+        with patch('vertex2.get') as get:
+            get.return_value.json.return_value = {'tx0plus5': 'unexpected'}
+            self.app._refresh_plus5_sensors()
+        for sensor in self.app.plus5_sensors.values():
+            sensor.set_state.assert_called_once_with(state='unavailable')
+
+    def test_failed_or_malformed_read_invalidates_old_settings(self):
+        from requests.exceptions import HTTPError, Timeout
+        for error in (HTTPError('503'), Timeout(), ValueError('invalid JSON')):
+            with self.subTest(error=error):
+                for sensor in self.app.plus5_sensors.values():
+                    sensor.reset_mock()
+                with patch('vertex2.get', side_effect=error):
+                    self.app._refresh_plus5_sensors()
+                for sensor in self.app.plus5_sensors.values():
+                    sensor.set_state.assert_called_once_with(state='unavailable')
+
+    def test_power_off_during_read_discards_response(self):
+        with patch('vertex2.get') as get:
+            def result():
+                self.app.plugged = False
+                return {'tx0plus5': '1', 'tx1plus5': '1'}
+            get.return_value.json.side_effect = result
+            self.app._refresh_plus5_sensors()
+        for sensor in self.app.plus5_sensors.values():
+            sensor.set_state.assert_called_once_with(state='unavailable')
+
+    def test_command_requests_readback_without_optimistic_confirmation(self):
+        with patch('vertex2.get'):
+            self.app.send_cmd('tx0plus5', 'on')
+        self.assertTrue(self.app._status_requested.is_set())
+        for sensor in self.app.plus5_sensors.values():
+            sensor.set_state.assert_called_once_with(state='unavailable')
+
+    def test_power_cut_invalidates_settings(self):
+        self.app._suspend_vertex2()
+        for sensor in self.app.plus5_sensors.values():
+            sensor.set_state.assert_called_once_with(state='unavailable')
+
+
+class StatusChangeTests(unittest.TestCase):
+    def setUp(self):
+        self.app = app = Harness()
+        app.input_selected_sensors = {i: Mock() for i in range(2)}
+        app.generate_input_selected_sensor_name = lambda i: i
+        app.get_input_config = Mock(return_value={'name': 'AppleTV'})
+        app.output_names = {}
+        app.output_selected_sensor = Mock()
+        app._refresh_selected_input_sensor = Mock()
+        app._flush_startup_commands = Mock()
+        app._reconfigure_vertex2 = Mock()
+        app._refresh_plus5_sensors = Mock()
+        self.status = dict(rx0='no signal', rx1='no signal', tx0='', tx1='',
+                           audout='', insel='0 4')
+
+    def test_identical_status_is_silent_but_still_applied(self):
+        app = self.app
+        app._apply_vertex2_status(self.status)
+        app._apply_vertex2_status(dict(self.status, reconfigure=True))
+        app.log_info.assert_called_once()
+        app.run_in.assert_not_called()
+        self.assertEqual(app._update_output_video_sensors.call_count, 2)
+        app._reconfigure_vertex2.assert_called_once()
+
+    def test_changed_status_extends_active_window_only_once(self):
+        app = self.app
+        app._apply_vertex2_status(self.status)
+        changed = dict(self.status, tx0='4K50')
+        with patch('vertex2.time.monotonic', return_value=200):
+            app._apply_vertex2_status(changed)
+        self.assertEqual(app._command_processing_deadline, 230)
+        app._apply_vertex2_status(changed)
+        self.assertEqual(app.log_info.call_count, 2)
+        app.run_in.assert_called_once()
+        app._on_command_processing_timeout({'deadline': 123})
+        self.assertEqual(app._command_processing_deadline, 230)
+        app._on_command_processing_timeout({'deadline': 230})
+        app.command_processing_sensor.set_state.assert_called_with(state='off')
+
+    def test_change_without_active_command_logs_without_starting_window(self):
+        app = self.app
+        app._command_processing_deadline = None
+        app._apply_vertex2_status(self.status)
+        app._apply_vertex2_status(dict(self.status, audout='L-PCM'))
+        self.assertEqual(app.log_info.call_count, 2)
+        app.run_in.assert_not_called()
+        app.command_processing_sensor.set_state.assert_not_called()
+
+
+class CommandProcessingTests(unittest.TestCase):
+    def test_interrupt_spam_does_not_extend_command_deadline(self):
+        app = Harness()
+        app.command_processing_sensor.get_state.return_value = 'off'
+        with patch('vertex2.time.monotonic', return_value=100), patch('vertex2.get'):
+            app.send_cmd('mutetx0audio', 'on')
+        self.assertEqual(app._command_processing_deadline, 130)
+        app.run_in.reset_mock()
+        app.cancel_timer.reset_mock()
+        connection = Mock()
+        chunks = iter([b'int\n'] * 20)
+        def recv(size):
+            try:
+                return next(chunks)
+            except StopIteration:
+                app._interrupt_stop.set()
+                return b''
+        connection.recv.side_effect = recv
+        context = Mock(__enter__=Mock(return_value=connection), __exit__=Mock(return_value=False))
+        with patch('vertex2.socket.create_connection', return_value=context):
+            app._interrupt_listener(app._interrupt_stop)
+        self.assertEqual(app._command_processing_deadline, 130)
+        app.cancel_timer.assert_not_called()
+        app.run_in.assert_not_called()
+        app._on_command_processing_timeout({'deadline': 130})
+        app.command_processing_sensor.set_state.assert_called_with(state='off')
+
+    def test_new_command_extends_window_and_old_callback_is_ignored(self):
+        app = Harness()
+        with patch('vertex2.time.monotonic', return_value=100), patch('vertex2.get'):
+            app.send_cmd('mutetx0audio', 'on')
+        with patch('vertex2.time.monotonic', return_value=110), patch('vertex2.get'):
+            app.send_cmd('mutetx0audio', 'off')
+        self.assertEqual(app._command_processing_deadline, 140)
+        app.command_processing_sensor.set_state.reset_mock()
+        app._on_command_processing_timeout({'deadline': 130})
+        app.command_processing_sensor.set_state.assert_not_called()
+        app._on_command_processing_timeout({'deadline': 140})
+        app.command_processing_sensor.set_state.assert_called_once_with(state='off')
+
+    def test_queued_command_does_not_start_window_before_boot(self):
+        app = Harness()
+        app._begin_startup()
+        with patch('vertex2.get') as get:
+            app.send_cmd('mutetx0audio', 'on')
+        get.assert_not_called()
+        app.command_processing_sensor.set_state.assert_not_called()
+        app.run_in.assert_not_called()
+
+
+class SpamSensorTests(unittest.TestCase):
+    def setUp(self):
+        self.app = app = Harness()
+        app.no_signal_spam_sensor = Mock()
+        self.state = 'off'
+        app.no_signal_spam_sensor.get_state.side_effect = lambda: self.state
+        app.no_signal_spam_sensor.set_state.side_effect = lambda **kw: setattr(self, 'state', kw['state'])
+        app._spam_lock = threading.RLock()
+        app._spam_last_interrupt = None
+        app._spam_snapshot = None
+        app._spam_count = 0
+        app._spam_timer = None
+        app._spam_deadline = None
+        self.status = dict(rx0='no signal', rx1='no signal', tx0='', tx1='', audout='', insel='0 4')
+
+    def observe(self, at, interrupt=True, **changes):
+        status = dict(self.status, **changes)
+        if interrupt:
+            status['interrupt_at'] = at
+        with patch('vertex2.time.monotonic', return_value=at):
+            self.app._observe_no_signal_spam(status)
+
+    def test_three_identical_interrupts_detect_and_silence_clears(self):
+        self.observe(100)
+        self.observe(103)
+        self.assertEqual(self.state, 'off')
+        self.observe(106)
+        self.assertEqual(self.state, 'on')
+        self.app._expire_no_signal_spam({'deadline': 107})
+        self.assertEqual(self.state, 'on')
+        self.app._expire_no_signal_spam({'deadline': 110})
+        self.assertEqual(self.state, 'off')
+
+    def test_polls_do_not_count_and_changed_poll_breaks_streak(self):
+        for at in range(100, 110):
+            self.observe(at, interrupt=False)
+        self.assertEqual(self.state, 'off')
+        self.observe(112)
+        self.observe(115)
+        self.observe(116, interrupt=False, insel='1 4')
+        self.observe(118)
+        self.assertEqual(self.state, 'off')
+
+    def test_normal_transitions_and_nonperiodic_events_do_not_trigger(self):
+        self.observe(100)
+        self.observe(103, audout='L-PCM')
+        self.observe(106)
+        self.assertEqual(self.state, 'off')
+        self.observe(107)
+        self.observe(113)
+        self.assertEqual(self.state, 'off')
+
+    def test_signal_return_or_power_loss_clears(self):
+        for at in (100, 103, 106):
+            self.observe(at)
+        self.observe(107, tx0='4K50')
+        self.assertEqual(self.state, 'off')
+        for at in (110, 113, 116):
+            self.observe(at)
+        self.app._suspend_vertex2()
+        self.assertEqual(self.state, 'off')
+        self.app._expire_no_signal_spam({'deadline': 120})
+        self.assertEqual(self.state, 'off')
+
+    def test_unknown_input_is_not_no_signal(self):
+        for at in (100, 103, 106):
+            self.observe(at, rx0='unavailable')
+        self.assertEqual(self.state, 'off')
+
+
+if __name__ == '__main__':
+    unittest.main()