import appdaemon.plugins.hass.hassapi as hass from smartswitch import SmartSwitch import ad_toolbox.smartcondition as SmartCondition import telnetlib import time import datetime class SmartProjector(SmartSwitch): WARM_UP_TIME = 35 def get_default_entity_state(self): return 'off' def on_initialize_smart_object(self): #those variable are use withing get_default_entity_state() self.telnet_ip = self.args['telnet_connection']['ip'] self.telnet_port = self.args['telnet_connection']['port'] self.update_state_cb_handle = None self.warming_up_done_cb_handle = None self.power_plug_entity = None self.power_plug_sensor = None self.attributes = { 'warming_up' : False, 'picture_mode' : None} if "power_plug" in self.args: self.power_plug_entity = self.args["power_plug"]["entity"] self.power_plug_conditions_evaluator = SmartCondition.Evaluator(self,self.args["power_plug"]["conditions"],condition_name = 'power_plug',on_change_cb = self.on_change_power_plug_conditions,unavaibility_result = SmartCondition.Result.Unavailable) if "power_sensor" in self.args["power_plug"]: self.power_threeshold = self.args["power_plug"]["power_threeshold"] self.power_plug_sensor = self.get_entity(self.args["power_plug"]["power_sensor"]) super().on_initialize_smart_object() self.entity.set_state(attributes = self.attributes) self.listen_event(self.on_turn_on_service_call,"call_service",domain = "switch", service = "turn_on", service_data = {'entity_id' : self.entity_id}) self.listen_event(self.on_turn_off_service_call,"call_service",domain = "switch", service = "turn_off", service_data = {'entity_id' : self.entity_id}) if 'press_enter_event' in self.args: self.listen_event(lambda event_name, data, kwargs: self.press_enter(),self.args['press_enter_event']) self.projector_ready_conditions = SmartCondition.Evaluator(self,self.args["projector_ready_conditions"],condition_name = 'projector_ready_conditions') # self.send_cmd("menu=?",True) self.lamp_mode_evaluators = list() if "lamp_mode" in self.args: for key in self.args["lamp_mode"]: self.lamp_mode_evaluators.append((SmartCondition.Evaluator(self,self.args["lamp_mode"][key], condition_name = f"lamp_mode][{key}", on_update_cb = self.on_update_lamp_mode,constants = self.constants, templates_library = self.templates_library),key)) self.on_update_lamp_mode() self.picture_mode_evaluators = list() if "picture_mode" in self.args: for key in self.args["picture_mode"]: self.picture_mode_evaluators.append((SmartCondition.Evaluator(self,self.args["picture_mode"][key], condition_name = f"picture_mode][{key}", on_update_cb = self.on_update_picture_mode,constants = self.constants, templates_library = self.templates_library),key)) self.on_update_picture_mode() if self.power_plug_sensor != None: self.power_plug_sensor.listen_state(self.on_power_change) self.reset_update_state_cb() #self.entity.listen_state(self.on_state_change) self.update_state() def on_power_change(self, entity, attribute, old, new, kwargs): self.update_state() def on_change_power_plug_conditions(self,prev_result,result): if result != SmartCondition.Result.Unavailable: if result == SmartCondition.Result.Succeeded: self.turn_on(self.power_plug_entity) else: self.turn_off(self.power_plug_entity) def on_turn_on_service_call(self, event_name, data, kwargs): self.log_info("on_turn_on_service_call") self.switch_on() def on_turn_off_service_call(self, event_name, data, kwargs): self.log_info("on_turn_off_service_call") self.switch_off() def reset_update_state_cb(self): # when we use a power sensor, the change of power consuption is responsible of triggering the update if self.power_plug_sensor == None: if self.update_state_cb_handle != None: self.cancel_timer(self.update_state_cb_handle) self.update_state_cb_handle = self.run_every(self.update_state,datetime.datetime.now(), 30) def is_available(self): return self.entity.get_state() != 'unavailable' def switch_on(self): if self.is_available(): self.log(f"Turn on {self.entity_id}") # when we turn on the projector, pow=? will return OFF during the warm up period (a few seconds) # it can result in a switch hickup if update_state is called right after we turn on the projector self.reset_update_state_cb() #send_cmd can take 1-2s when the projector is off, so it's better to optimisticaly set the state and correct afterwards self.set_warming_up_attribute(True) self.entity.set_state(state = 'on',attributes = self.attributes) cmd_result = self.send_cmd('pow=on') if cmd_result is not None and cmd_result != "POW=ON": # Got a definitive bad response — revert self.set_warming_up_attribute(False) self.entity.set_state(state = 'off',attributes = self.attributes) # If cmd_result is None (projector dropped connection during power-on sequence), # keep optimistic 'on' state — update_state() will correct after warm-up else: self.log_warning("Turn On ignored as the projector is unavailable") def switch_off(self): if self.is_available(): self.log(f"Turn off {self.entity_id}") if self.send_cmd('pow=off') == "POW=OFF": self.set_warming_up_attribute(False) self.entity.set_state(state = 'off',attributes = self.attributes) else: self.log_warning("Turn Off ignored as the projector is unavailable") def get_projector_state(self): if self.projector_ready_conditions.evaluate(False) == SmartCondition.Result.Succeeded: if self.power_plug_sensor != None: if float(self.power_plug_sensor.get_state()) > self.power_threeshold: return 'on' else: return 'off' else: try: cmd_result = self.send_cmd('pow=?') if cmd_result == "POW=ON": return 'on' if cmd_result == "POW=OFF": return 'off' self.log_info(f"pow=? returned and invalid result : {cmd_result}, the projector will stay in the previous state") return self.entity.get_state() except OSError as error: # telenetlib raise an OsError [Errno 113] Host is unreachable when the projector is unreachable (most likely unplugged) if error.errno == 113: return 'unavailable' else: raise else: return 'off' def on_update_picture_mode(self): for evaluator in self.picture_mode_evaluators: if evaluator[0].evaluate(False) == SmartCondition.Result.Succeeded: evaluator[0].log_evaluation_result() if evaluator[1] != "None": cmd_result = self.send_cmd("appmod=?") if cmd_result == 'APPMOD=FIMMAKER': self.log_info(f"Can't switch picture mode to {evaluator[1]} because the projector is in Filmmaker mode") else: cmd_result = self.send_cmd(f"appmod={evaluator[1]}",True) self.attributes['picture_mode'] = cmd_result.split("=")[1] if cmd_result else None self.entity.set_state(attributes = self.attributes) break def on_update_lamp_mode(self): for evaluator in self.lamp_mode_evaluators: if evaluator[0].evaluate(False) == SmartCondition.Result.Succeeded: evaluator[0].log_evaluation_result() if evaluator[1] != "None": result = self.send_cmd(f"lampm={evaluator[1]}",True) break def update_state(self,kwargs = None): new_state = self.get_projector_state() if new_state != self.entity.get_state(): self.set_warming_up_attribute(new_state == 'on') if new_state == 'on': # the projector is not responding during power on sequence if self.attributes['warming_up'] == 'off': cmd_result = self.send_cmd("appmod=?") if cmd_result: self.attributes['picture_mode'] = cmd_result.split("=")[1] else: self.attributes['picture_mode'] = "unavailable" else: self.attributes['picture_mode'] = "unavailable" self.entity.set_state(state = new_state,attributes = self.attributes) def set_warming_up_attribute(self, new_value): if new_value != self.attributes['warming_up']: # let's cancel previous CB if self.warming_up_done_cb_handle != None: self.cancel_timer(self.warming_up_done_cb_handle) self.warming_up_done_cb_handle = None if new_value: self.warming_up_done_cb_handle = self.run_in(self.on_warm_up_done,self.WARM_UP_TIME) self.attributes['warming_up'] = new_value def on_warm_up_done(self,*kwargs): self.warming_up_done_cb_handle = None self.attributes['warming_up'] = False self.entity.set_state(state = self.get_projector_state(),attributes = self.attributes) def press_enter(self): self.send_cmd("enter") def send_cmd(self,cmd, log_command = False): cmd = f"*{cmd}#" return self._send_cmd(cmd,log_command,3) def _send_cmd(self,cmd,log_command,try_count): invalid_result = [ "*Unsupported Item#", "*Block Item#", "*Illegal format#" ] should_retry = False with telnetlib.Telnet(self.telnet_ip,self.telnet_port) as session: if log_command: self.log_info(f'Sending command "{cmd}"') try: session.write(bytes(cmd,'utf-8')) output = session.read_until(b"#",timeout = 5).decode('ascii') except (ConnectionResetError,EOFError): self.log_warning(f"Connection Reset While sending command {cmd}. {try_count} try left") if try_count <= 0: self.log_error(f"Could not send command {cmd}",dump_stack = True) return None else: should_retry = True # opening session recursivly inside the with doesn't sound like a good idea if should_retry: if try_count == 1: # I know calling sleep() is in callback is evil # but this is a very small sleep time and working with event would make the code much much more complexe time.sleep(1) return self._send_cmd(cmd,log_command,try_count - 1) if log_command: self.log_info(f"Result = {output}") if output in invalid_result: self.log_error(f"Command rejected. {cmd} returned {output}",dump_stack = True) return None if not output: if try_count > 0: self.log_warning(f"Command output is empty. {cmd}. {try_count} try left") return self._send_cmd(cmd,log_command,try_count - 1) else: self.log_error(f"Command output is empty. {cmd}. {try_count} try left",dump_stack = True) return None return output.strip("*").rstrip("#")