import appdaemon.plugins.hass.hassapi as hass from requests import get from requests.exceptions import Timeout,ConnectionError from ad_toolbox.smartobject import SmartObject import ad_toolbox.smartcondition as SmartCondition import time import socket import threading import re from urllib.parse import urlparse class Vertex2(SmartObject): def on_initialize_smart_object(self): self._interrupt_stop = threading.Event() self._status_requested = threading.Event() self._tx0_recovery_attempted = False self._tx0_recovery_restore_handle = None self.video_tx_signal_sensor = self.create_entity( self.args['video_tx_signal_sensor'], state='unavailable', ) self.video_tx_signal_sensors = { index: self.create_entity( self.generate_output_sensor_name(self.args['video_tx_signal_sensor'], index), state='unavailable', ) for index in range(2) } self.allm_sensor = self.create_entity(self.args['allm_sensor'], state='unavailable') self.allm_sensors = { index: self.create_entity( self.generate_output_sensor_name(self.args['allm_sensor'], index), state='unavailable', ) for index in range(2) } self._tx_video_signal_values = {0: 'unavailable', 1: 'unavailable'} self.audio_out_signal_sensor = self.create_entity( self.args['audio_out_signal_sensor'], state='unavailable', ) self.input_names = {} if "input_config" in self.args: for key in self.args['input_config']: try: name = self.args['input_config'][key]['name'] except TypeError: name = self.args['input_config'][key] try: self.input_names[name] = int(key) except ValueError: pass #we can skip "default" self.output_names = {} if "output_config" in self.args: for key, config in self.args["output_config"].items(): name = config.get("name") if isinstance(config, dict) else config try: self.output_names[name] = int(key) except (TypeError, ValueError): pass default_output_config = self.get_output_config(None) self.output_selected_sensor = self.create_entity( self.args["output_selected_sensor"], state=default_output_config["name"], icon=default_output_config["icon"], ) self.input_selected_sensors = {} for i in range(2): sensor_name = self.generate_input_selected_sensor_name(i) input_config = self.get_input_config(None) self.input_selected_sensors[sensor_name] = self.create_entity( sensor_name, state=input_config['name'], icon=input_config['icon'], ) if "input_config" in self.args: self.listen_state(self.on_input_selected_sensor_change,sensor_name) default_input_config = self.get_input_config(None) self.input_selected_sensor = self.create_entity( self.args['input_selected_sensor'], state=default_input_config['name'], icon=default_input_config['icon'], ) self.unreachable_sensor = self.create_entity( self.args['unreachable_sensor'], state='off', ) self.power_plug_entity = None 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) if "mutetx0_conditions" in self.args: self.mutetx0_conditions = SmartCondition.Evaluator(self,self.args['mutetx0_conditions'], condition_name = "mutetx0_conditions",on_change_cb = self.on_mute_tx0) if "mutetx1_conditions" in self.args: self.mutetx1_conditions = SmartCondition.Evaluator(self,self.args['mutetx1_conditions'], condition_name = "mutetx1_conditions",on_change_cb = self.on_mute_tx1) self.input_selection_evaluators = list() if "input_selection" in self.args: for key in self.args["input_selection"]: self.input_selection_evaluators.append((SmartCondition.Evaluator(self,self.args["input_selection"][key], condition_name = key, on_update_cb = self.on_update_input_selection),key)) self.on_update_input_selection() self.output_selection_evaluators = [] if "output_selection" in self.args: for key, conditions in self.args["output_selection"].items(): self.output_selection_evaluators.append(( SmartCondition.Evaluator( self, conditions, condition_name=f"output_selection_{key}", on_update_cb=self.on_update_output_selection, trigger_callback_on_activation=False, ), key, )) 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, name=f'{self.name}_interrupt_listener', daemon=True, ) self.update_vertex2_states() self._interrupt_thread.start() def _interrupt_listener(self): address = urlparse(self.args['vertex2_address']) host = address.hostname status_fields = ('rx0', 'rx1', 'tx0', 'tx1', 'tx0sink', 'tx1sink', 'aud0', 'aud1', 'audout', 'spd0', 'spd1') commands = tuple(f'get status {field}' for field in status_fields) + ('get insel',) expected = set(status_fields) | {'insel'} while not self._interrupt_stop.is_set(): try: with socket.create_connection((host, self.vertex2_tcp_port), timeout=10) as connection: # 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) connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 5) connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 3) connection.settimeout(1) self.log_info(f'Connected to Vertex2 IP INTERRUPT at {host}:{self.vertex2_tcp_port}') self._status_requested.set() buffer = '' status = {} status_started = None interrupt_generation = 0 last_interrupt_at = None last_interrupt_interval = None request_generation = 0 request_interval = None previous_snapshot = None previous_snapshot_generation = None while not self._interrupt_stop.is_set(): if status_started is None and self._status_requested.is_set(): self._status_requested.clear() status = {} status_started = time.monotonic() request_generation = interrupt_generation request_interval = last_interrupt_interval for command in commands: connection.sendall((command + '\r').encode('ascii')) try: data = connection.recv(1024) except socket.timeout: if status_started is not None and time.monotonic() - status_started > 5: raise OSError('Vertex2 TCP status response timed out') continue if not data: raise OSError('Vertex2 closed the IP INTERRUPT connection') buffer += data.decode('ascii', errors='ignore') if len(buffer) > 4096: raise OSError('Vertex2 TCP response exceeded buffer limit') while '\n' in buffer: line, buffer = buffer.split('\n', 1) line = line.strip() if line == 'int': self.log_info('Vertex2 IP INTERRUPT received') now = time.monotonic() last_interrupt_interval = None if last_interrupt_at is None else now - last_interrupt_at last_interrupt_at = now interrupt_generation += 1 self._status_requested.set() elif status_started is not None: if line.startswith('insel '): status['insel'] = line.partition(' ')[2].strip() else: for field in status_fields: prefix = f'{field.upper()}:' if line.startswith(prefix): status[field] = line[len(prefix):].strip() break if expected <= status.keys(): snapshot = dict(status) 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']) status_started = None consecutive = (previous_snapshot_generation is not None and request_generation == previous_snapshot_generation + 1) unchanged = previous_snapshot == snapshot rx_has_signal = any( snapshot[field].strip().lower() not in ('', 'no signal', 'unavailable') for field in ('rx0', 'rx1') ) if (unchanged and consecutive and request_interval is not None and 2 <= request_interval < 3 and not rx_has_signal): if not self._tx0_recovery_attempted: self.run_in(self.trigger_tx0_stuck_workaround, 0, interrupt_interval=request_interval, snapshot=snapshot) elif not unchanged: self._tx0_recovery_attempted = False previous_snapshot = snapshot previous_snapshot_generation = request_generation if status_started is not None and time.monotonic() - status_started > 5: raise OSError('Vertex2 TCP status response timed out') except (OSError, ValueError) as error: if self._interrupt_stop.is_set(): break 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(): sensor.set_state(state='unavailable') self.allm_sensor.set_state(state='unavailable') for sensor in self.allm_sensors.values(): sensor.set_state(state='unavailable') self.audio_out_signal_sensor.set_state(state='unavailable') self.log(f'Vertex2 IP INTERRUPT unavailable: {error}', level='WARNING') self._interrupt_stop.wait(5) def trigger_tx0_stuck_workaround(self, kwargs): if self._tx0_recovery_attempted: return if self.output_selected_sensor.get_state() != self.get_output_config(0)['name']: return if self.unreachable_sensor.get_state() != 'off': return self._tx0_recovery_attempted = True snapshot_summary = ', '.join(f'{key}={value!r}' for key, value in kwargs['snapshot'].items()) self.log_warning( f"Vertex2 TX0 +5V workaround triggered: identical complete TCP status after " f"an IP INTERRUPT interval of {kwargs['interrupt_interval']:.2f}s " f"({snapshot_summary}); cycling TX0 +5V FORCE" ) self.force_tx0_plus5(False) self._tx0_recovery_restore_handle = self.run_in(self.restore_tx0_plus5_after_recovery, 1) def restore_tx0_plus5_after_recovery(self, kwargs): self._tx0_recovery_restore_handle = None if self.output_selected_sensor.get_state() == self.get_output_config(0)['name']: self.log_info('Vertex2 TX0 +5V workaround: restoring TX0 +5V FORCE') self.force_tx0_plus5(True) def terminate(self): interrupt_stop = getattr(self, '_interrupt_stop', None) if interrupt_stop is not None: interrupt_stop.set() restore_handle = getattr(self, '_tx0_recovery_restore_handle', None) if restore_handle is not None: self.cancel_timer(restore_handle) super().terminate() def generate_input_selected_sensor_name(self,i): return f"{self.args['input_selected_sensor']}_tx{i}" def generate_output_sensor_name(self, sensor_name, output_index): return f'{sensor_name}_tx{output_index}' def on_input_selected_sensor_change(self,entity, attribute, old, new, kwargs): if old != new: self.log(f"input_select switch from {old} to {new}") if old in self.input_names: input_id = self.input_names[old] config = self.args['input_config'][input_id] try: if config['reboot_when_leaving'] == True: self.reboot() except KeyError: pass def on_update_input_selection(self): for evaluator in self.input_selection_evaluators: if evaluator[0].evaluate(False) == SmartCondition.Result.Succeeded: evaluator[0].log_evaluation_result() input_selected = evaluator[1] break self.log_info("Input selection conditions changed") if input_selected != "None": self.select_input(input_selected) self.update_vertex2_states() def select_input(self,input_selected): try: input_value = int(input_selected) except ValueError: input_value = self.input_names[input_selected] self.log(f"Switching to input {input_value}") self.send_cmd("insel",f"{input_value} {input_value}") def on_update_output_selection(self): output_selected = None for evaluator, key in self.output_selection_evaluators: if evaluator.evaluate(False) == SmartCondition.Result.Succeeded: evaluator.log_evaluation_result() output_selected = key break if output_selected is None: self.log_warning("No output_selection condition succeeded; disabling TX0 and TX1 +5V FORCE") self.disable_output_selection() return self.select_output(output_selected) def select_output(self, output_selected): if str(output_selected) == "None": self.disable_output_selection() return try: output_index = int(output_selected) except (TypeError, ValueError): try: output_index = self.output_names[output_selected] except KeyError: self.log_error(f"Unknown output selection: {output_selected}") return if output_index not in (0, 1): self.log_error(f"Output selection must be 0 or 1, got: {output_index}") return output_config = self.get_output_config(output_index) self.output_selected_sensor.set_state( state=output_config["name"], attributes={"icon": output_config["icon"]}, ) self._refresh_selected_output_signal() self.log(f"Selecting TX{output_index} as the active output") if output_index == 0: self.force_tx1_plus5(False) self.set_edid_video_flags_to_tx0() self.force_tx0_plus5(True) else: self.force_tx0_plus5(False) self.set_edid_video_flags_to_tx1() self.force_tx1_plus5(True) def disable_output_selection(self): default_output_config = self.get_output_config(None) self.output_selected_sensor.set_state( state=default_output_config["name"], attributes={"icon": default_output_config["icon"]}, ) self._refresh_selected_output_signal() self.force_tx0_plus5(False) self.force_tx1_plus5(False) def on_filmmaker_mode_conditions_change(self,prev_result,result): if result == SmartCondition.Result.Succeeded: self.toggle_filmmaker_mode(True) else: self.toggle_filmmaker_mode(False) def on_mute_tx0(self,prev_result,result): if result == SmartCondition.Result.Succeeded: self.mute_tx0(True) else: self.mute_tx0(False) def on_mute_tx1(self,prev_result,result): if result == SmartCondition.Result.Succeeded: self.mute_tx1(True) else: self.mute_tx1(False) def on_change_power_plug_conditions(self,prev_result,result): if result == SmartCondition.Result.Succeeded: self.turn_on(self.power_plug_entity) else: self.turn_off(self.power_plug_entity) def send_cmd(self,cmd,value = ''): request = f'{self.args["vertex2_address"]}/cmd?{cmd}={value}' try: r = get(request,timeout=5) except (Timeout,ConnectionError): self.log(f"{request} is unreachable", level = "WARNING") def mute_tx0(self, mute): self.send_cmd("mutetx0audio","on" if mute else "off") def force_tx0_plus5(self, enabled): self.log(f"Setting TX0 +5V FORCE to {enabled}") self.send_cmd("tx0plus5", "on" if enabled else "off") def force_tx1_plus5(self, enabled): self.log(f"Setting TX1 +5V FORCE to {enabled}") self.send_cmd("tx1plus5", "on" if enabled else "off") def toggle_filmmaker_mode(self, value): self.log(f"Setting filmmaker mode to {value}") self.send_cmd("filmmaker","on" if value else "off") def set_edid_video_flags_to_tx0(self): self.log(f"Setting EDID Video Flags to TX0") self.send_cmd("edid","videotx0 on") def set_edid_video_flags_to_tx1(self): self.log(f"Setting EDID Video Flags to TX1") self.send_cmd("edid","videotx1 on") def mute_tx1(self, mute): self.send_cmd("mutetx1audio","on" if mute else "off") def reboot(self): self.log("Rebooting Vertex2") self.send_cmd("reboot") def get_input_config(self,input_number): if input_number == None: input_number = 'default' default_config = { 'name': input_number, "icon" : None } if "input_config" in self.args: try: config = self.args['input_config'][input_number] except KeyError: return default_config if isinstance(config,dict): return config else: return {"name" : config, "icon": None} return default_config def get_output_config(self, output_number): if output_number is None: output_number = "default" default_config = {"name": "None", "icon": None} config = self.args.get("output_config", {}).get(output_number, default_config) if isinstance(config, dict): return { "name": config.get("name", default_config["name"]), "icon": config.get("icon", default_config["icon"]), } 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': self.run_in(self._apply_vertex2_status, 0, rx0='', rx1='', tx0='', tx1='', audout='', insel='0 0') return self._status_requested.set() def _refresh_selected_output_signal(self): output_name = self.output_selected_sensor.get_state() output_index = self.output_names.get(output_name) self._refresh_selected_input_sensor(output_index) signal = self._tx_video_signal_values.get(output_index, 'unavailable') if self.video_tx_signal_sensor.get_state() != signal: self.video_tx_signal_sensor.set_state(state=signal) allm_state = self.allm_sensors[output_index].get_state() if output_index in self.allm_sensors else 'unavailable' if self.allm_sensor.get_state() != allm_state: self.allm_sensor.set_state(state=allm_state) def _refresh_selected_input_sensor(self, output_index=None): if output_index not in (0, 1): config = self.get_input_config(None) else: selected_name = self.input_selected_sensors[ self.generate_input_selected_sensor_name(output_index) ].get_state() input_number = self.input_names.get(selected_name) config = self.get_input_config(input_number) attributes = {} if config.get('icon') is not None: attributes['icon'] = config['icon'] if self.input_selected_sensor.get_state() != config['name']: self.input_selected_sensor.set_state(state=config['name'], attributes=attributes) elif attributes: self.input_selected_sensor.set_state(attributes=attributes) def _update_output_video_sensors(self, tx0_signal, tx1_signal): values = {0: tx0_signal.strip() or 'unavailable', 1: tx1_signal.strip() or 'unavailable'} self._tx_video_signal_values = values allm_states = {} for output_index, signal in values.items(): if self.video_tx_signal_sensors[output_index].get_state() != signal: self.video_tx_signal_sensors[output_index].set_state(state=signal) allm_state = 'unavailable' if signal == 'unavailable' else ( 'on' if re.search(r'\bALLM\b', signal, re.IGNORECASE) else 'off' ) allm_states[output_index] = allm_state if self.allm_sensors[output_index].get_state() != allm_state: self.allm_sensors[output_index].set_state(state=allm_state) selected_name = self.output_selected_sensor.get_state() selected_index = self.output_names.get(selected_name) selected_signal = values.get(selected_index, 'unavailable') selected_allm = allm_states.get(selected_index, 'unavailable') signal_changed = self.video_tx_signal_sensor.get_state() != selected_signal if signal_changed: self.video_tx_signal_sensor.set_state(state=selected_signal) if self.allm_sensor.get_state() != selected_allm: self.allm_sensor.set_state(state=selected_allm) def _apply_vertex2_status(self, kwargs): 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}" ) try: portselect = [int(value) for value in kwargs['insel'].split()] if len(portselect) != 2: raise ValueError('expected two input ports') except (KeyError, ValueError) as error: self.log_error(f'Invalid Vertex2 TCP input selection: {error}') self.unreachable_sensor.set_state(state='on') return if portselect[0] == 4: # 4 = copy TX1 portselect[0] = portselect[1] if portselect[1] == 4: # 4 = copy TX0 portselect[1] = portselect[0] input_signal = kwargs['rx0'].strip() rx1 = kwargs['rx1'].strip() if not input_signal or (input_signal == 'no signal' and rx1): input_signal = rx1 input_signal = input_signal or 'unavailable' audio_out_signal = kwargs['audout'].strip() or 'unavailable' for i in range(2): sensor = self.input_selected_sensors[self.generate_input_selected_sensor_name(i)] input_number = portselect[i] input_config = self.get_input_config(input_number) name = input_config['name'] attributes = {} if input_config.get('icon') is not None: attributes['icon'] = input_config['icon'] if sensor.get_state() != name: sensor.set_state(state=name, attributes=attributes) elif attributes: sensor.set_state(attributes=attributes) selected_output_index = self.output_names.get(self.output_selected_sensor.get_state()) self._refresh_selected_input_sensor(selected_output_index) self._update_output_video_sensors(kwargs['rx0'], kwargs['rx1']) 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')