import appdaemon.plugins.hass.hassapi as hass from requests import get from requests.exceptions import Timeout,ConnectionError,RequestException from ad_toolbox.smartobject import SmartObject import ad_toolbox.smartcondition as SmartCondition import time import json import socket import threading import re from urllib.parse import urlparse class Vertex2(SmartObject): def on_initialize_smart_object(self): # HASS can call initialize() again without calling terminate() first. self._stop_interrupt_listener() self._interrupt_stop = threading.Event() self._status_requested = threading.Event() self._last_logged_status = None self._spam_lock = threading.RLock() self._spam_last_interrupt = None self._spam_snapshot = None self._spam_count = 0 self._spam_timer = None self._spam_deadline = None self.no_signal_spam_sensor = None if self.args.get('no_signal_spam_sensor'): self.no_signal_spam_sensor = self.create_entity( self.args['no_signal_spam_sensor'], state='off') self._connection_lock = threading.Lock() self._interrupt_connection = None self.startup_delay = max(0, float(self.args.get('startup_delay', 30))) self._startup_deadline = 0 self._startup_commands = [] self.power_plug_entity = self.args.get("vertex2_power_plug", {}).get("entity") self._power_allowed = self.power_plug_entity is None self._command_processing_lock = threading.Lock() self._command_processing_handle = None self._command_processing_deadline = None self.command_processing_timeout = max(0, float(self.args.get('command_processing_timeout', 30))) self.command_processing_sensor = None if self.args.get('command_processing_sensor'): self.command_processing_sensor = self.create_entity( self.args['command_processing_sensor'], state='off', ) 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.plus5_sensors = { index: self.create_entity(self.args[key], state='unavailable') for index, key in enumerate(('tx0_plus5_sensor', 'tx1_plus5_sensor')) if self.args.get(key) } 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', ) 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) if self.power_plug_entity is not None: self.listen_state(self.on_power_plug_state_change, self.power_plug_entity) 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) 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_config_conditions = {} for output_key, config in self.args.get('output_config', {}).items(): if not isinstance(config, dict): continue try: output_index = int(output_key) except (TypeError, ValueError): continue for condition_key in ('use_audio_out_audio_flags', 'force_plus5'): condition_config = config.get(condition_key) if condition_config is None: continue callback_options = ( {'on_change_cb': self.on_output_plus5_condition_change, 'trigger_callback_on_activation': True} if condition_key == 'force_plus5' else {'on_update_cb': self.on_output_config_conditions_update, 'trigger_callback_on_activation': False} ) self.output_config_conditions.setdefault(output_index, {})[condition_key] = SmartCondition.Evaluator( self, condition_config, condition_name=f'output_config_tx{output_index}_{condition_key}', pass_condition_name_to_cb=str(output_index), **callback_options, ) 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() self.vertex2_tcp_port = int(self.args.get('vertex2_tcp_port', 2220)) self._interrupt_thread = threading.Thread( target=self._interrupt_listener, args=(self._interrupt_stop,), name=f'{self.name}_interrupt_listener', daemon=True, ) self._initialize_input_select() self.update_vertex2_states() self._interrupt_thread.start() def _initialize_input_select(self): """Expose a real HA select, with device-confirmed state, through MQTT.""" self._input_select_config = self.args.get('input_select') if not self._input_select_config: return config = self._input_select_config self._select_namespace = config.get('mqtt_namespace', 'mqtt') self._select_topic = config.get('mqtt_topic', f'appdaemon/{self.name}/input').rstrip('/') self._select_uid = config.get('unique_id', f'{self.name}_input') self.listen_event(self._on_input_select_mqtt, 'MQTT_MESSAGE', namespace=self._select_namespace) for entity in (self.args['input_selected_sensor'], self.args['unreachable_sensor']): self.listen_state(self._on_input_select_state, entity, attribute='all') if self.power_plug_entity: self.listen_state(self._on_input_select_state, self.power_plug_entity) self._discover_input_select() def _publish_input_select(self, suffix, payload): if not isinstance(payload, str): payload = json.dumps(payload, ensure_ascii=False) self.call_service('mqtt/publish', topic=self._select_topic + suffix, payload=payload, retain=True, namespace=self._select_namespace) def _discover_input_select(self): config = self._input_select_config for topic in (self._select_topic + '/set', 'homeassistant/status'): self.call_service('mqtt/subscribe', topic=topic, namespace=self._select_namespace) availability = [{'topic': self._select_topic + '/availability'}] # The shared plugin's connection topic belongs to the MQTT transport, # regardless of the name originally chosen for that plugin. mqtt_config = self.get_plugin_config(namespace=self._select_namespace) or {} connection_topic = config.get('connection_topic') or mqtt_config.get('will_topic') if connection_topic: availability.append(dict( topic=connection_topic, payload_available=mqtt_config.get('birth_payload', 'online'), payload_not_available=mqtt_config.get('will_payload', 'offline'), )) payload = dict( name='Input', unique_id=self._select_uid, default_entity_id=config.get('entity', 'select.vertex2_input'), device=dict(identifiers=[f'appdaemon_{self.name}'], name='Vertex2', manufacturer='HDFury', model='Vertex2'), options=list(self.input_names), optimistic=False, command_topic=self._select_topic + '/set', state_topic=self._select_topic + '/state', json_attributes_topic=self._select_topic + '/attributes', availability=availability, availability_mode='all', ) self._select_discovery_payload = payload self._sync_input_select(force_discovery=True) self.log_info(f'Published MQTT discovery for {payload["default_entity_id"]}') def _on_input_select_state(self, entity, attribute, old, new, kwargs): self._sync_input_select() def _sync_input_select(self, force_discovery=False): if not getattr(self, '_input_select_config', None): return name = self.input_selected_sensor.get_state() valid = name in self.input_names config = self.get_input_config(self.input_names[name] if valid else None) icon = config.get('icon') or 'mdi:video-input-hdmi' # HA's MQTT JSON attributes do not override the entity's icon property. # Update discovery only when the icon changes (or on reconnect). discovery = self._select_discovery_payload if force_discovery or discovery.get('icon') != icon: discovery['icon'] = icon self.call_service('mqtt/publish', topic=f'homeassistant/select/{self._select_uid}/config', payload=json.dumps(discovery), retain=True, namespace=self._select_namespace) self._publish_input_select('/attributes', {'icon': icon}) self._publish_input_select('/state', name if valid else 'None') available = (valid and self._vertex2_powered() and self.unreachable_sensor.get_state() == 'off') self._publish_input_select('/availability', 'online' if available else 'offline') def _on_input_select_mqtt(self, event, data, kwargs): topic, payload = data.get('topic'), data.get('payload') if ((topic is None and data.get('state') == 'Connected') or (topic == 'homeassistant/status' and payload == 'online')): self._discover_input_select() elif topic == self._select_topic + '/set': if payload not in self.input_names: self.log_warning(f'Rejected unknown Vertex2 input: {payload!r}') return if not self._vertex2_ready(): self.log_warning('Vertex2 input command ignored: device is not ready') return # insel sets both TX ports; only a subsequent device status updates HA. self.select_input(payload) self.update_vertex2_states() def _reset_no_signal_spam(self, state='off'): if getattr(self, 'no_signal_spam_sensor', None) is None: return with self._spam_lock: if self._spam_timer is not None: self.cancel_timer(self._spam_timer) self._spam_timer = None self._spam_deadline = None self._spam_last_interrupt = None self._spam_snapshot = None self._spam_count = 0 if self.no_signal_spam_sensor.get_state() != state: self.no_signal_spam_sensor.set_state(state=state) def _observe_no_signal_spam(self, status): if getattr(self, 'no_signal_spam_sensor', None) is None: return with self._spam_lock: missing = ('', 'no signal') if any(status[field].strip().lower() not in missing for field in ('rx0', 'tx0')): self._reset_no_signal_spam() return snapshot = tuple(status[field].strip() for field in ('rx0', 'rx1', 'tx0', 'tx1', 'audout', 'insel')) if self._spam_snapshot is not None and snapshot != self._spam_snapshot: self._reset_no_signal_spam() at = status.get('interrupt_at') if at is None: if self.no_signal_spam_sensor.get_state() == 'unavailable': self.no_signal_spam_sensor.set_state(state='off') return # Polls and commands do not count as device interrupts. now = time.monotonic() if now - at >= 4: return # Do not revive a streak with a delayed callback. previous = self._spam_last_interrupt self._spam_last_interrupt = at snapshot = tuple(status[field].strip() for field in ('rx0', 'rx1', 'tx0', 'tx1', 'audout', 'insel')) self._spam_count = (self._spam_count + 1 if previous is not None and 2 <= at - previous <= 4 and snapshot == self._spam_snapshot else 1) self._spam_snapshot = snapshot state = 'on' if self._spam_count >= 3 else 'off' if self.no_signal_spam_sensor.get_state() != state: self.no_signal_spam_sensor.set_state(state=state) if self._spam_timer is not None: self.cancel_timer(self._spam_timer) self._spam_deadline = at + 4 self._spam_timer = self.run_in( self._expire_no_signal_spam, max(0, self._spam_deadline - now), deadline=self._spam_deadline) def _expire_no_signal_spam(self, kwargs): with self._spam_lock: if kwargs['deadline'] != self._spam_deadline: return self._spam_timer = None self._reset_no_signal_spam() def _interrupt_listener(self, interrupt_stop): 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 interrupt_stop.is_set(): if not self._vertex2_ready(): interrupt_stop.wait(0.5) continue try: with socket.create_connection((host, self.vertex2_tcp_port), timeout=10) as connection: with self._connection_lock: self._interrupt_connection = connection if interrupt_stop.is_set() or not self._vertex2_powered(): continue # 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 reconfigure = True pending_interrupt_at = None while not interrupt_stop.is_set() and self._vertex2_powered(): if status_started is None and self._status_requested.is_set(): self._status_requested.clear() status = {} status_started = time.monotonic() request_interrupt_at = pending_interrupt_at pending_interrupt_at = None 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 interrupt_stop.is_set() or not self._vertex2_powered(): break 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': pending_interrupt_at = time.monotonic() 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'], reconfigure=reconfigure, interrupt_at=request_interrupt_at) reconfigure = False status_started = None 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 interrupt_stop.is_set(): break if not self._vertex2_powered(): continue self._reset_no_signal_spam('unavailable') self._invalidate_plus5_sensors() 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_warning(f'Vertex2 IP INTERRUPT unavailable: {error}') interrupt_stop.wait(5) finally: with self._connection_lock: self._interrupt_connection = None def _close_interrupt_connection(self): connection_lock = getattr(self, "_connection_lock", None) if connection_lock is None: return with connection_lock: connection = self._interrupt_connection if connection is not None: try: connection.shutdown(socket.SHUT_RDWR) except OSError: pass connection.close() def _stop_interrupt_listener(self): self._reset_no_signal_spam('unavailable') interrupt_stop = getattr(self, '_interrupt_stop', None) if interrupt_stop is not None: interrupt_stop.set() self._close_interrupt_connection() interrupt_thread = getattr(self, '_interrupt_thread', None) if interrupt_thread is not None: # Allow the 10s connect timeout or 1s recv timeout to finish. interrupt_thread.join(timeout=12) if interrupt_thread.is_alive(): raise RuntimeError('Previous Vertex2 IP INTERRUPT listener did not stop') self._interrupt_thread = None processing_lock = getattr(self, '_command_processing_lock', None) if processing_lock is not None: with processing_lock: self._command_processing_deadline = None if self._command_processing_handle is not None: self.cancel_timer(self._command_processing_handle) self._command_processing_handle = None def terminate(self): if getattr(self, '_input_select_config', None): self._publish_input_select('/availability', 'offline') self._stop_interrupt_listener() 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_info(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): input_selected = None 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 is not None and str(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_info(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; clearing output selection") 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_info(f"Selecting TX{output_index} as the active output") self.apply_output_audio_settings(output_index) if output_index == 0: self.set_edid_video_flags_to_tx0() else: self.set_edid_video_flags_to_tx1() 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() def on_output_plus5_condition_change(self, output_index, previous_result, result): if result == SmartCondition.Result.Disabled: return enabled = result == SmartCondition.Result.Succeeded if int(output_index) == 0: self.force_tx0_plus5(enabled) elif int(output_index) == 1: self.force_tx1_plus5(enabled) 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 _vertex2_powered(self): return self._power_allowed and ( self.power_plug_entity is None or self.get_state(self.power_plug_entity) == 'on' ) def _vertex2_ready(self): return self._vertex2_powered() and time.monotonic() >= self._startup_deadline def _begin_startup(self): self._startup_deadline = time.monotonic() + self.startup_delay self.log_info(f'Waiting {self.startup_delay:g}s for Vertex2 to start') def _flush_startup_commands(self): with self._connection_lock: commands = self._startup_commands self._startup_commands = [] for cmd, value in commands: self.send_cmd(cmd, value) def _reconfigure_vertex2(self): # Conditions may not have changed while the device was powered off. # Evaluate them again instead of relying on change callbacks or old commands. if self.power_plug_entity is not None: result = self.power_plug_conditions_evaluator.evaluate(False) self.on_change_power_plug_conditions(None, result) if not self._vertex2_ready(): return self.log_info('Reapplying Vertex2 configuration after connection restored') if self.input_selection_evaluators: self.on_update_input_selection() if self.output_selection_evaluators: # Also reapplies audio/video EDID and audio mute for the selected output. self.on_update_output_selection() else: selected_output = self.output_names.get(self.output_selected_sensor.get_state()) if selected_output in (0, 1): self.apply_output_audio_settings(selected_output) for output_index, conditions in self.output_config_conditions.items(): evaluator = conditions.get('force_plus5') if evaluator is not None: self.on_output_plus5_condition_change( output_index, None, evaluator.evaluate(False), ) if 'filmmaker_mode_conditions' in self.args: self.on_filmmaker_mode_conditions_change( None, self.filmmaker_mode_conditions.evaluate(False), ) def _suspend_vertex2(self): self._reset_no_signal_spam() self._invalidate_plus5_sensors() self._close_interrupt_connection() with self._connection_lock: self._startup_commands.clear() self._startup_deadline = 0 self._status_requested.clear() self._update_output_video_sensors('', '') self.audio_out_signal_sensor.set_state(state='unavailable') self.unreachable_sensor.set_state(state='off') with self._command_processing_lock: self._command_processing_deadline = None if self._command_processing_handle is not None: self.cancel_timer(self._command_processing_handle) self._command_processing_handle = None if self.command_processing_sensor is not None: self.command_processing_sensor.set_state(state='off') def on_power_plug_state_change(self, entity, attribute, old, new, kwargs): if self._vertex2_powered(): if old != 'on' and new == 'on': self._begin_startup() self.update_vertex2_states() else: self._suspend_vertex2() def on_change_power_plug_conditions(self,prev_result,result): self._power_allowed = result == SmartCondition.Result.Succeeded if self._power_allowed: if self.get_state(self.power_plug_entity) != 'on': self._begin_startup() self.turn_on(self.power_plug_entity) else: # Close TCP before asking Home Assistant to cut the power. self._suspend_vertex2() self.turn_off(self.power_plug_entity) def _start_command_processing(self, only_if_active=False): # Only commands start a window; changed status may extend an active one. if self.command_processing_sensor is None: return with self._command_processing_lock: if only_if_active and self._command_processing_deadline is None: return if self._command_processing_handle is not None: self.cancel_timer(self._command_processing_handle) self._command_processing_deadline = time.monotonic() + self.command_processing_timeout if self.command_processing_sensor.get_state() != 'on': self.command_processing_sensor.set_state(state='on') self._command_processing_handle = self.run_in( self._on_command_processing_timeout, self.command_processing_timeout, deadline=self._command_processing_deadline, ) def _on_command_processing_timeout(self, kwargs): with self._command_processing_lock: # Ignore a callback already queued before a timer reset or shutdown. if kwargs['deadline'] != self._command_processing_deadline: return self._command_processing_handle = None self._command_processing_deadline = None self.command_processing_sensor.set_state(state='off') def send_cmd(self,cmd,value = ''): if not self._vertex2_powered(): return if not self._vertex2_ready(): with self._connection_lock: self._startup_commands.append((cmd, value)) return request = f'{self.args["vertex2_address"]}/cmd?{cmd}={value}' if cmd in ('tx0plus5', 'tx1plus5'): self._invalidate_plus5_sensors() self._start_command_processing() try: r = get(request, timeout=5) if cmd == 'edid' and value.startswith(('audiotx', 'audioout')): self.log_info( f'Vertex2 audio EDID command response: HTTP {r.status_code}, ' f'{r.text[:160]!r}' ) except (Timeout,ConnectionError): self.log_warning(f"{request} is unreachable") finally: if cmd in ('tx0plus5', 'tx1plus5'): # Read back the device setting even if no IP interrupt follows. self._status_requested.set() def _invalidate_plus5_sensors(self): for sensor in self.plus5_sensors.values(): if sensor.get_state() != 'unavailable': sensor.set_state(state='unavailable') def _refresh_plus5_sensors(self): if not self.plus5_sensors: return if not self._vertex2_ready(): self._invalidate_plus5_sensors() return try: # Same read-only endpoint used by the Vertex2 configuration page. response = get(f'{self.args["vertex2_address"].rstrip("/")}/ssi/confpage.ssi', timeout=5) response.raise_for_status() config = response.json() if not isinstance(config, dict): raise ValueError('expected a configuration object') except (RequestException, ValueError) as error: self._invalidate_plus5_sensors() self.log_warning(f'Vertex2 +5V readback failed: {error}') return if not self._vertex2_ready(): self._invalidate_plus5_sensors() return for index in range(2): value = config.get(f'tx{index}plus5') state = {'0': 'off', '1': 'on'}.get(str(value), 'unavailable') sensor = self.plus5_sensors.get(index) if sensor is not None and sensor.get_state() != state: sensor.set_state(state=state) def mute_tx0(self, mute): self.send_cmd("mutetx0audio","on" if mute else "off") def force_tx0_plus5(self, enabled): self.log_info(f"Setting TX0 +5V FORCE to {enabled}") self.send_cmd("tx0plus5", "on" if enabled else "off") def force_tx1_plus5(self, enabled): self.log_info(f"Setting TX1 +5V FORCE to {enabled}") self.send_cmd("tx1plus5", "on" if enabled else "off") def toggle_filmmaker_mode(self, value): self.log_info(f"Setting filmmaker mode to {value}") self.send_cmd("filmmaker","on" if value else "off") def on_output_config_conditions_update(self, output_index): try: output_index = int(output_index) except (TypeError, ValueError): return if self.output_names.get(self.output_selected_sensor.get_state()) == output_index: self.apply_output_audio_settings(output_index) def apply_output_audio_settings(self, output_index): conditions = self.output_config_conditions.get(output_index, {}) use_audio_out = conditions.get('use_audio_out_audio_flags') use_audio_out = ( use_audio_out is not None and use_audio_out.evaluate(False) == SmartCondition.Result.Succeeded ) if use_audio_out: self.set_edid_audio_flags_to_audio_out() elif output_index == 0: self.set_edid_audio_flags_to_tx0() elif output_index == 1: self.set_edid_audio_flags_to_tx1() self.mute_output_audio(output_index, use_audio_out) def mute_output_audio(self, output_index, mute): if output_index == 0: self.mute_tx0(mute) elif output_index == 1: self.mute_tx1(mute) def set_edid_audio_flags_to_tx0(self): self.log_info('Setting EDID Audio Flags to TX0 (edid=audiotx0 on)') self.send_cmd('edid', 'audiotx0 on') def set_edid_audio_flags_to_tx1(self): self.log_info('Setting EDID Audio Flags to TX1 (edid=audiotx1 on)') self.send_cmd('edid', 'audiotx1 on') def set_edid_audio_flags_to_audio_out(self): self.log_info('Setting EDID Audio Flags to AUDIO OUT (edid=audioout on)') self.send_cmd('edid', 'audioout on') def set_edid_video_flags_to_tx0(self): self.log_info(f"Setting EDID Video Flags to TX0") self.send_cmd("edid","videotx0 on") def set_edid_video_flags_to_tx1(self): self.log_info(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_info("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"]), "use_audio_out_audio_flags": config.get("use_audio_out_audio_flags"), "force_plus5": config.get("force_plus5"), } return {"name": config, "icon": None} def update_vertex2_states(self): if not self._vertex2_powered(): self._suspend_vertex2() 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): if not self._vertex2_powered(): return 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 self._observe_no_signal_spam(kwargs) # Ignore callback metadata (e.g. reconfigure) when comparing device state. snapshot = tuple(kwargs[field].strip() for field in ('rx0', 'rx1', 'tx0', 'tx1', 'audout', 'insel')) previous = self._last_logged_status if snapshot != previous: self._last_logged_status = snapshot 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}" ) if previous is not None: self._start_command_processing(only_if_active=True) 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') self._flush_startup_commands() if kwargs.get('reconfigure', False): self._reconfigure_vertex2() self._refresh_plus5_sensors()