290 lines
14 KiB
Python
290 lines
14 KiB
Python
import appdaemon.plugins.hass.hassapi as hass
|
|
from ad_toolbox.smartobject import SmartObject
|
|
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(f"Initializing TheFrame with ip {self.args['ip_address']}")
|
|
#self.need_to_turn_off_when_hdmi_gone = False
|
|
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_sensor = self.args['output_sensors']['is_on']
|
|
self.is_art_mode_sensor = self.args['output_sensors']['is_art_mode']
|
|
self.tv_auto_shutdown = self.args['output_sensors']['tv_auto_shutdown']
|
|
|
|
if self.get_state(self.tv_auto_shutdown) == None:
|
|
self.set_state(self.tv_auto_shutdown,state = '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()
|
|
|
|
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):
|
|
#self.log(f"Connecting to {self.args['ip_address']}")
|
|
return SamsungTVWS(self.args['ip_address'],port = 8002, token_file = os.path.join(str(self.AD.app_dir),"theframe_token.txt"))
|
|
|
|
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(f"Calling service {service}")
|
|
self.call_service(service)
|
|
elif 'wol_switch' in self.args:
|
|
self.log(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(f"select_random_art with exception ResponseError {e}")
|
|
except TimeoutError:
|
|
self.log(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(f"Requesting TV to switch to {requested_state}")
|
|
|
|
if requested_state == 'hdmi' and self.get_state(self.is_on_sensor) == 'off':
|
|
self.log(f"setting {self.tv_auto_shutdown} to 'on'")
|
|
self.set_state(self.tv_auto_shutdown,state = 'on')
|
|
|
|
#if we already have a confirm callback, let's cancel it
|
|
if self.requested_state_confirm_cb_handle:
|
|
self.log("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(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.get_state(self.is_art_mode_sensor) == 'on'
|
|
is_on = self.get_state(self.is_on_sensor) == 'on'
|
|
is_plugged = self.get_state(self.power_plug) == 'on'
|
|
|
|
if self.requested_state == 'tv_off':
|
|
if not is_plugged:
|
|
self.log("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 is_art_mode:
|
|
tv = self.connect()
|
|
self.log("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("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("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("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("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(f"{app_status['name']} is active")
|
|
self.log("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("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("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.get_state(self.is_art_mode_sensor) == '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(f"get_artmode() throw a KeyError Exception {e}")
|
|
return False , 10
|
|
|
|
except websocket.WebSocketConnectionClosedException:
|
|
self.log(f"update sensors failed with exception WebSocketConnectionClosedException")
|
|
return False , 30
|
|
except ConnectionRefusedError:
|
|
self.log(f"update sensors failed with exception ConnectionRefusedError")
|
|
return False , 30
|
|
except tv_exceptions.ConnectionFailure:
|
|
self.log(f"update sensors failed with exception ConnectionFailure")
|
|
return False , 30
|
|
except OSError as error:
|
|
self.log(f"update sensors failed with exception OsError {error}")
|
|
return False , 30
|
|
except tv_exceptions.ResponseError as e:
|
|
self.log(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.get_state(self.is_art_mode_sensor) != is_art_mode:
|
|
self.log(f"Updating sensor: {self.is_art_mode_sensor} = '{is_art_mode}'")
|
|
self.set_state(self.is_art_mode_sensor,state = is_art_mode)
|
|
|
|
if self.get_state(self.is_on_sensor) != is_on:
|
|
self.log(f"Updating sensor: {self.is_on_sensor} = '{is_on}'")
|
|
self.set_state(self.is_on_sensor,state = is_on)
|
|
if is_on == 'off' and self.get_state(self.tv_auto_shutdown) != 'off':
|
|
self.log(f"setting {self.tv_auto_shutdown} to 'off'")
|
|
self.set_state(self.tv_auto_shutdown,state = 'off')
|
|
|
|
|
|
return True, 0 |