652 lines
29 KiB
Python
652 lines
29 KiB
Python
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):
|
|
# HASS can call initialize() again without calling terminate() first.
|
|
self._stop_interrupt_listener()
|
|
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)
|
|
|
|
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',):
|
|
condition_config = config.get(condition_key)
|
|
if not condition_config:
|
|
continue
|
|
self.output_config_conditions.setdefault(output_index, {})[condition_key] = SmartCondition.Evaluator(
|
|
self,
|
|
condition_config,
|
|
condition_name=f'output_config_tx{output_index}_{condition_key}',
|
|
on_update_cb=self.on_output_config_conditions_update,
|
|
pass_condition_name_to_cb=str(output_index),
|
|
trigger_callback_on_activation=False,
|
|
)
|
|
|
|
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,
|
|
args=(self._interrupt_stop,),
|
|
name=f'{self.name}_interrupt_listener',
|
|
daemon=True,
|
|
)
|
|
self.update_vertex2_states()
|
|
self._interrupt_thread.start()
|
|
|
|
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():
|
|
try:
|
|
with socket.create_connection((host, self.vertex2_tcp_port), timeout=10) as connection:
|
|
if interrupt_stop.is_set():
|
|
break
|
|
# 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 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 interrupt_stop.is_set():
|
|
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':
|
|
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 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')
|
|
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 _stop_interrupt_listener(self):
|
|
interrupt_stop = getattr(self, '_interrupt_stop', None)
|
|
if interrupt_stop is not None:
|
|
interrupt_stop.set()
|
|
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
|
|
restore_handle = getattr(self, '_tx0_recovery_restore_handle', None)
|
|
if restore_handle is not None:
|
|
self.cancel_timer(restore_handle)
|
|
self._tx0_recovery_restore_handle = None
|
|
|
|
def terminate(self):
|
|
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(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")
|
|
self.apply_output_audio_settings(output_index)
|
|
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_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)
|
|
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(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 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(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"]),
|
|
"use_audio_out_audio_flags": config.get("use_audio_out_audio_flags"),
|
|
}
|
|
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')
|