import appdaemon.plugins.hass.hassapi as hass from ad_toolbox.smartobject import SmartObject from ad_toolbox.eventhandler import EventHandler import ad_toolbox.smartcondition as SmartCondition from samsungtvws import SamsungTVWS from samsungtvws import exceptions as tv_exceptions import os import random import time import websocket class SamsungTheFrame(SmartObject): def on_initialize_smart_object(self): super().on_initialize_smart_object() self.log_info(f"Initializing TheFrame with ip {self.args['ip_address']}") #self.need_to_turn_off_when_hdmi_gone = False self.volume_connection = None self.requested_state = None self.requested_state_confirm_cb_handle = None self.power_plug = self.args['input_sensors']['power_plug'] self.power_plug_power = self.args['input_sensors']['power_plug_power'] self.is_on_entity = self.create_entity(self.args['output_sensors']['is_on']) self.is_art_mode_entity = self.create_entity(self.args['output_sensors']['is_art_mode']) auto_shutdown_state = self.get_state(self.args['output_sensors']['tv_auto_shutdown']) self.tv_auto_shutdown_entity = self.create_entity( self.args['output_sensors']['tv_auto_shutdown'], 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 = [] self.tv_states_evaluators = list() 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 key in self.args["tv_states"]: self.tv_states_evaluators.append((SmartCondition.Evaluator(self,self.args["tv_states"][key], condition_name = key, on_update_cb = self.on_update_tv_states),key)) self.on_update_tv_states() self.listen_state(self.update_sensors_cb,self.power_plug) self.listen_state(self.update_sensors_cb,self.power_plug_power) self.update_sensors() self.volume_event_handlers = [] for direction in ('up', 'down'): config = self.args.get(f'volume_{direction}_events') if config: self.volume_event_handlers.append(EventHandler( self, config, self.on_volume_event, direction)) def on_update_tv_states(self): for state_evaluator in self.tv_states_evaluators: if state_evaluator[0].evaluate(False) == SmartCondition.Result.Succeeded: state_evaluator[0].log_evaluation_result() new_state = state_evaluator[1] break self.request_tv_state(new_state) def on_update_tv_state_sensor(self, entity, attribute, old, new, kwargs): if new != old: self.request_tv_state(new) def connect(self, timeout=None): #self.log(f"Connecting to {self.args['ip_address']}") return SamsungTVWS(self.args['ip_address'],port = 8002, timeout=timeout, token_file = os.path.join(str(self.AD.app_dir),"theframe_token.txt")) def close_volume_connection(self): tv = self.volume_connection self.volume_connection = None if tv is not None: try: tv.close() except (OSError, websocket.WebSocketException): pass def terminate(self): self.close_volume_connection() super().terminate() def on_volume_event(self, event_name, event_data, direction): if self.is_on_entity.get_state() != 'on': self.close_volume_connection() return try: if self.volume_connection is None: self.volume_connection = self.connect(timeout=1) # ESPHome already limits repetitions to one per 300 ms. The # library's default one-second delay would queue button events. self.volume_connection.send_key( 'KEY_VOLUP' if direction == 'up' else 'KEY_VOLDOWN', key_press_delay=0) except (OSError, tv_exceptions.ConnectionFailure, tv_exceptions.ResponseError, websocket.WebSocketException) as error: self.log_warning(f"Volume {direction} failed: {error}") # Reconnect on the next press, without replaying a failed command. self.close_volume_connection() def has_power(self): return self.get_state(self.power_plug) == 'on' and float(self.get_state(self.power_plug_power)) > 35 def wake_up(self): if 'wake_up_service' in self.args: service = self.args['wake_up_service'] self.log_info(f"Calling service {service}") self.call_service(service) elif 'wol_switch' in self.args: self.log_info(f"Switching on {self.args['wol_switch']}") self.turn_on(self.args['wol_switch']) if 'mac_address' in self.args: wakeonlan.send_magic_packet(self.args['mac_address']) if 'ip_address' in self.args: wakeonlan.send_magic_packet(self.args['mac_address'],ip_address= self.args['ip_address'],port = 9) if 'broadcast_address' in self.args: wakeonlan.send_magic_packet(self.args['mac_address'],ip_address= self.args['broadcast_address']) def select_random_art(self): try: tv = self.connect() available_art = tv.art().available() if len(available_art): selected_art = random.choice(available_art) tv.art().select_image(selected_art['content_id'], show=True) except tv_exceptions.ResponseError as e: self.log_info(f"select_random_art with exception ResponseError {e}") except TimeoutError: self.log_info(f"select_random_art with exception TimeoutError") def request_tv_state(self,requested_state): valid_states = ['art_mode','tv_off','hdmi','power_off','manual'] assert requested_state in valid_states, f"{requested_state} is not a valid states. Valid states are {valid_states}" self.log_info(f"Requesting TV to switch to {requested_state}") if requested_state == 'hdmi' and self.is_on_entity.get_state() == 'off': self.log_info(f"setting {self.tv_auto_shutdown_entity.entity_id} to '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: self.log_info("Canceling previous confirm CB") self.cancel_timer(self.requested_state_confirm_cb_handle) self.requested_state_confirm_cb_handle = None if requested_state == 'manual': self.requested_state = None # 'manual' is not a real state else: self.requested_state = requested_state if self._process_state_request(): self.log_info(f"The TV was already in the right state") def _process_state_request(self): try: success, wait_time = self.update_sensors() 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.is_art_mode_entity.get_state() == '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': if not is_plugged: self.log_info("Powering on TV") self.turn_on(self.power_plug) self.requested_state_confirm_cb_handle = self.run_in(self.confirm_change_state_cb, 60) return False elif is_on: tv = self.connect() self.log_info("Switching to art mode") tv.shortcuts().power() self.requested_state_confirm_cb_handle = self.run_in(self.confirm_change_state_cb, 10) return False elif is_art_mode: tv = self.connect() self.log_info("Turning off TV") tv.hold_key('KEY_POWER',4) self.requested_state_confirm_cb_handle = self.run_in(self.confirm_change_state_cb, 10) return False elif self.requested_state == 'art_mode': if not is_art_mode: if not is_plugged: self.log_info("Powering on TV") self.turn_on(self.power_plug) self.requested_state_confirm_cb_handle = self.run_in(self.confirm_change_state_cb, 60) return False elif is_on: tv = self.connect() self.log("Switching to art mode") tv.shortcuts().power() self.requested_state_confirm_cb_handle = self.run_in(self.confirm_change_state_cb, 10) return False elif not self.has_power(): self.log("Waking up TV") self.wake_up() self.requested_state_confirm_cb_handle = self.run_in(self.confirm_change_state_cb, 60) return False elif self.requested_state == "hdmi": if not is_plugged: self.log_info("Powering on TV") self.turn_on(self.power_plug) self.requested_state_confirm_cb_handle = self.run_in(self.confirm_change_state_cb, 60) return False elif not self.has_power(): self.log_info("Waking up TV") self.wake_up() self.requested_state_confirm_cb_handle = self.run_in(self.confirm_change_state_cb, 60) return False elif not is_on: tv = self.connect() self.log_info("Turning on TV") tv.shortcuts().power() self.requested_state_confirm_cb_handle = self.run_in(self.confirm_change_state_cb, 5) return False else: tv = self.connect() for app in self.app_list: # {'id': '3201907018807', 'name': 'Netflix', 'running': True, 'version': '5.2.59020', 'visible': True} app_status = tv.rest_app_status(str(app)) if app_status['visible']: self.log_info(f"{app_status['name']} is active") self.log_info("Switching to Hdmi") tv.send_key('KEY_HDMI') self.requested_state_confirm_cb_handle = self.run_in(self.confirm_change_state_cb, 5) return False return True elif self.requested_state == 'power_off': if is_plugged: if is_on: tv = self.connect() self.log("Switching to art mode") tv.shortcuts().power() self.requested_state_confirm_cb_handle = self.run_in(self.confirm_change_state_cb, 10) return False elif not self.has_power(): self.log_info("Waking up TV") self.wake_up() self.requested_state_confirm_cb_handle = self.run_in(self.confirm_change_state_cb, 60) return False elif is_art_mode: #it's better to turn off while it's art mode to ensure the TV will wake up in art mode and support properly wake on lan self.select_random_art() time.sleep(5) self.log_info("Powering off TV") self.turn_off(self.power_plug) self.requested_state_confirm_cb_handle = self.run_in(self.confirm_change_state_cb, 5) return False return True except (BrokenPipeError,websocket.WebSocketConnectionClosedException): self.requested_state_confirm_cb_handle = self.run_in(self.confirm_change_state_cb, 5) return False def confirm_change_state_cb(self,*kwargs): self.requested_state_confirm_cb_handle = None if self._process_state_request(): self.log(f"The TV is now confirm to be in {self.requested_state}") self.requested_state = None # def on_art_mode_condition_change(self,prev_result,result): # if result == SmartCondition.Result.Succeeded: # self.request_tv_state('art_mode') # else: # 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') def update_sensors_cb(self,*args): success, wait_time = self.update_sensors() if not success: self.run_in(self.update_sensors_cb, wait_time) def update_sensors(self): is_art_mode = "off" is_on = "off" if self.has_power(): try: tv = self.connect() try: if tv.art().get_artmode() == 'on': is_art_mode = "on" else: is_on = "on" except KeyError as e: #sometimes the lib throw an exception in File "/samsungtvws/art.py", line 318, in get_artmode self.log_info(f"get_artmode() throw a KeyError Exception {e}") return False , 10 except websocket.WebSocketConnectionClosedException: self.log_info(f"update sensors failed with exception WebSocketConnectionClosedException") return False , 30 except ConnectionRefusedError: self.log_info(f"update sensors failed with exception ConnectionRefusedError") return False , 30 except tv_exceptions.ConnectionFailure: self.log_info(f"update sensors failed with exception ConnectionFailure") return False , 30 except OSError as error: self.log_info(f"update sensors failed with exception OsError {error}") return False , 30 except tv_exceptions.ResponseError as e: self.log_info(f"update sensors failed with exception ResponseError {e}") return False , 30 # except: # import traceback # exception_string = traceback.format_exc().splitlines()[-1] # self.disconnect() # self.log(f"update sensors failed with exception {exception_string}") # return False , 5 if self.is_art_mode_entity.get_state() != is_art_mode: self.log_info(f"Updating sensor: {self.is_art_mode_entity.entity_id} = '{is_art_mode}'") self.is_art_mode_entity.set_state(state = is_art_mode) if is_on == 'off': self.close_volume_connection() if self.is_on_entity.get_state() != is_on: self.log_info(f"Updating sensor: {self.is_on_entity.entity_id} = '{is_on}'") self.is_on_entity.set_state(state = is_on) if is_on == 'off' and self.tv_auto_shutdown_entity.get_state() != 'off': self.log_info(f"setting {self.tv_auto_shutdown_entity.entity_id} to 'off'") self.tv_auto_shutdown_entity.set_state(state = 'off') return True, 0