import appdaemon.plugins.hass.hassapi as hass from smartswitch import SmartSwitch from ad_toolbox.eventhandler import EventHandler import ad_toolbox.smartcondition as SmartCondition import telnetlib import queue import threading from functools import wraps import datetime def queued_operation(method): """Keep compound operations and their retries together in the FIFO worker.""" @wraps(method) def enqueue(self, *args, **kwargs): if self._command_stop.is_set(): return None if threading.current_thread() is self._command_worker: return method(self, *args, **kwargs) self._command_queue.put((method, args, kwargs)) return None return enqueue class SmartProjector(SmartSwitch): WARM_UP_TIME = 35 def get_default_entity_state(self): return 'off' def on_initialize_smart_object(self): # Queue callbacks during initialization; start only once setup is complete. self._command_queue = queue.Queue() self._command_stop = threading.Event() self._command_worker = threading.Thread( target=self._process_commands, name=f"{self.name}-telnet", daemon=True) #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']) if 'press_enter_events' in self.args: self.event_handlers.append(EventHandler(self, self.args['press_enter_events'], self.on_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() self._command_worker.start() 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' @queued_operation 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") @queued_operation 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' @queued_operation 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 @queued_operation 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 @queued_operation 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 not self.attributes['warming_up']: 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 @queued_operation 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 on_press_enter_event(self, event_name, data, event_context): self.press_enter() @queued_operation def press_enter(self): if self.entity.get_state() == 'on': self.send_cmd("enter") @queued_operation def send_cmd(self,cmd, log_command = False): cmd = f"*{cmd}#" return self._send_cmd(cmd,log_command,3) def _process_commands(self): while not self._command_stop.is_set(): operation = self._command_queue.get() try: if operation is None or self._command_stop.is_set(): return method, args, kwargs = operation try: method(self, *args, **kwargs) except Exception as error: self.log_error(f"Projector operation {method.__name__} failed: {error}", dump_stack=True) finally: self._command_queue.task_done() def terminate(self): # Discard pending work and interrupt retry delays. An active exchange is # bounded by the connection and read timeouts (2 + 5 seconds). self._command_stop.set() self._command_queue.put(None) if self._command_worker.is_alive(): self._command_worker.join(timeout=8) super().terminate() def _send_cmd(self, cmd, log_command, try_count): invalid_result = {"*Unsupported Item#", "*Block Item#", "*Illegal format#"} for attempt in range(try_count + 1): if self._command_stop.is_set(): return None try: with telnetlib.Telnet(self.telnet_ip, self.telnet_port, timeout=2) as session: if log_command: self.log_info(f'Sending command "{cmd}"') session.write(cmd.encode('utf-8')) output = session.read_until(b"#", timeout=5).decode('ascii').strip() if self._command_stop.is_set(): return None 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 # read_until may return a partial response when its timeout expires. if output.endswith("#"): return output.strip("*").rstrip("#") reason = f"empty or incomplete response: {output!r}" except (OSError, EOFError) as error: if self._command_stop.is_set(): return None if isinstance(error, OSError) and error.errno == 113: raise reason = str(error) remaining = try_count - attempt self.log_info(f"Command {cmd} failed ({reason}); {remaining} retries left") if remaining and self._command_stop.wait(1): return None self.log_error(f"Could not send command {cmd}", dump_stack=True) return None