diff --git a/apps/ad_toolbox b/apps/ad_toolbox index b997836..3e7ae75 160000 --- a/apps/ad_toolbox +++ b/apps/ad_toolbox @@ -1 +1 @@ -Subproject commit b997836401cbbd8f04572afaff75aab4217b7689 +Subproject commit 3e7ae7550b88f67c2a14aba31a3140055d200c04 diff --git a/apps/apps.yaml b/apps/apps.yaml index f5722f1..b88d066 100644 --- a/apps/apps.yaml +++ b/apps/apps.yaml @@ -71,3 +71,17 @@ haw_switch: off_events: - WAKEUP +nap_switch: + module: smartswitch + class: SmartSwitch + + entity: input_boolean.nap_switch + + auto_switch_off_after: 7200 # 2h + + toggle_events: + red_shelly_button: + event_name: state_changed + event_data: + entity_id: event.shelly_button_red_button + diff --git a/apps/global/homecinema.yaml b/apps/global/homecinema.yaml new file mode 100644 index 0000000..2379688 --- /dev/null +++ b/apps/global/homecinema.yaml @@ -0,0 +1,180 @@ +samsung_the_frame: + module: samsungtheframe + class: SamsungTheFrame + + ip_address: 10.0.0.58 + #broadcast_address: 10.0.0.255 + #mac_address: 98:06:3C:85:93:38 + #wol_switch: switch.theframewol + wake_up_service: "esphome/ir_bridge_turn_on_the_frame" + app_list: # https://github.com/tavicu/homebridge-samsung-tizen/issues/291 + - 3201907018807 # Netflix + - 3201910019365 # Prime + + virtual_sensors: + default_values: + binary_sensor.tv_auto_shutdown: false + binary_sensor.tv_on: false + sensors: + value_selector.samsung_the_frame_target_state: + power_off: + trigger_conditions: + - binary_sensor.is_everybody_sleeping or binary_sensor.is_living_room_sleeping or not binary_sensor.someone_home + - binary_sensor.projector_on or binary_sensor.projector_screen_down + hdmi: binary_sensor.vertex2_has_input_signal and not binary_sensor.projector_screen_down + manual: binary_sensor.tv_on and not binary_sensor.tv_auto_shutdown + tv_off: not binary_sensor.living_room_occupancy + art_mode: True + + tv_state: sensor.samsung_the_frame_target_state + + output_sensors: + is_on: binary_sensor.tv_on + is_art_mode: binary_sensor.tv_in_art_mode + tv_auto_shutdown: binary_sensor.tv_auto_shutdown + + input_sensors: + power_plug: switch.the_frame_power_plug + power_plug_power: sensor.the_frame_power_plug_power + +w4000i: + module: smartprojector_benq_over_telnet + class: SmartProjector + entity: switch.projector + + telnet_connection: + ip: 10.0.0.40 + port: 8000 + + power_plug: + entity: switch.living_room_projector_plug + power_sensor: sensor.living_room_projector_plug_power + power_threeshold: 5 + conditions: + trigger_conditions: + - binary_sensor.someone_home and not (binary_sensor.is_everybody_sleeping or binary_sensor.is_living_room_sleeping) + - sensor.living_room_projector_plug_power > 10 or binary_sensor.projector_on # don't remove this one or it will fry the projector + + virtual_sensors: + sensors: + value_selector.projector_icon: + projector_off: + conditions: not super + value: "'mdi:projector-off'" + projector_on: + conditions: True + value: "'mdi:projector'" + continuous_condition.projector_ready: + conditions: switch.living_room_projector_plug + time: 30 + + attributes_override: + icon: sensor.projector_icon + friendly_name: Projecteur + + press_enter_event: send_projector_enter + + projector_ready_conditions: binary_sensor.projector_ready + + #state_conditions: binary_sensor.projector_on + #cooling_down_sensor: binary_sensor.projector_cooling_down + + smart_conditions: + trigger_conditions: binary_sensor.vertex2_has_input_signal + disable_conditions: + - self == 'unavailable' + - self == 'off' and not binary_sensor.projector_screen_down + #- not binary_sensor.projector_screen_down + #- binary_sensor.projector_cooling_down + - switch.projector.warming_up + +bravia_theatre: + module: smartswitch + class: SmartSwitch + entity: media_player.bravia_theatre + + smart_conditions: binary_sensor.vertex2_has_input_signal + +projection_screen: + module: smartshutter + class: SmartShutter + entity: cover.living_room_projection_screen + close_position: 22 + + close_conditions: # close mean screen down + # when the projector is turn back on, it can happen that this condition is evaluated before projector_off_for_a_while is updated + # so I need to test both binary_sensor.projector_on and not binary_sensor.projector_off_for_a_while + trigger_conditions: binary_sensor.projector_on or not binary_sensor.projector_off_for_a_while + disable_conditions: + - binary_sensor.day_interval_night and not binary_sensor.projector_on + - binary_sensor.is_everybody_sleeping and not binary_sensor.projector_on + +vertex_2: + module: vertex2 + class: Vertex2 + priority: 5 + + input_config: + default: + name: None + icon: mdi:television-off + 0: + name: AppleTV + icon: mdi:apple + #reboot_when_leaving: True + 1: + name: Xbox + icon: mdi:microsoft-xbox + 2: + name: Playstation + icon: mdi:sony-playstation + # 3: + # name: Switch + # icon: mdi:nintendo-switch + + output_config: + default: + name: None + icon: mdi:television-off + 0: + name: TV + icon: mdi:television + 1: + name: Projector + icon: mdi:projector + + # vertex2_power_plug: + # entity: switch.fibaro_vertex_plug_51 + # # power_consumption: sensor.fibaro_vertex_plug_51_power + # conditions: + # trigger_conditions: binary_sensor.someone_home + # blocking_conditions: + # - binary_sensor.is_everybody_sleeping or binary_sensor.is_salon_sleeping + # - binary_sensor.vertex2_unreachable and not binary_sensor.tv_on and not binary_sensor.projector_on + + # mutetx0_conditions: + # trigger_conditions: + # - binary_sensor.is_sonos_amp_playing_tv and not media_player.piece_a_vivre.is_volume_muted + # - binary_sensor.is_hw_700_on + + #filmmaker_mode_conditions: sensor.vertex2_input_select_tx0 == 'AppleTV' + + output_selection: + Projector: binary_sensor.projector_on + TV: True + + # input_selection: + # None: binary_sensor.vertex2_has_input_signal # this will prevent input change if the current input is still active + # Switch: binary_sensor.is_playing_switch + # AppleTV: not binary_sensor.is_appletv_standy + # Playstation: switch.ps5_136_power + # Xbox: True + + vertex2_address: http://10.0.0.33 + vertex2_tcp_port: 2220 + + video_tx_signal_sensor: sensor.vertex2_video_signal + audio_out_signal_sensor: sensor.vertex2_audio_signal + output_selected_sensor: sensor.vertex2_output_selected + input_selected_sensor: sensor.vertex2_input_selected + unreachable_sensor: binary_sensor.vertex2_unreachable \ No newline at end of file diff --git a/apps/global/virtualsensors.yaml b/apps/global/virtualsensors.yaml index ede5605..02d3d7e 100644 --- a/apps/global/virtualsensors.yaml +++ b/apps/global/virtualsensors.yaml @@ -42,16 +42,17 @@ virtual_sensors: # is_apple_tv_on is not reliable as the the apple tv integration sometimes return "paused" when it's actually off, # you should use is_using_apple_tv instead for most automation binary_sensor.is_apple_tv_on: media_player.apple_tv != 'off' and media_player.apple_tv != 'idle' - binary_sensor.is_xbox_on: media_player.xbox != 'off' + binary_sensor.is_xbox_on: binary_sensor.yetanotherpedro #media_player.xbox != 'off' and media_player.xbox != 'unavailable' binary_sensor.is_playstation_on: switch.ps5_136_power retain_condition.vertex2_has_input_signal: retain_time: 10 conditions: - trigger_conditions: sensor.hdfury_vertex2_07_output_tx0 or sensor.hdfury_vertex2_07_audio_output - disable_conditions: sensor.hdfury_vertex2_07_output_tx0 == 'unavailable' - binary_sensor.is_using_apple_tv: binary_sensor.is_apple_tv_on and binary_sensor.vertex2_has_input_signal and select.hdfury_vertex2_07_port_select_tx0 == 0 - binary_sensor.is_using_xbox: binary_sensor.is_xbox_on and binary_sensor.vertex2_has_input_signal and select.hdfury_vertex2_07_port_select_tx0 == 1 - binary_sensor.is_using_playstation: binary_sensor.is_playstation_on and binary_sensor.vertex2_has_input_signal and select.hdfury_vertex2_07_port_select_tx0 == 2 + trigger_conditions: (sensor.vertex2_video_signal and sensor.vertex2_video_signal != 'no signal') or sensor.vertex2_audio_signal + disable_conditions: binary_sensor.vertex2_unreachable + binary_sensor.is_using_apple_tv: binary_sensor.is_apple_tv_on and binary_sensor.vertex2_has_input_signal #and select.hdfury_vertex2_07_port_select_tx0 == 0 + binary_sensor.is_using_xbox: binary_sensor.is_xbox_on and binary_sensor.vertex2_has_input_signal #and select.hdfury_vertex2_07_port_select_tx0 == 1 + binary_sensor.is_using_playstation: binary_sensor.is_playstation_on and binary_sensor.vertex2_has_input_signal #and select.hdfury_vertex2_07_port_select_tx0 == 2 + continuous_condition.is_vertex2_content_idle: time: 5 diff --git a/apps/rooms/living_room.yaml b/apps/rooms/living_room.yaml index 558716c..1b16679 100644 --- a/apps/rooms/living_room.yaml +++ b/apps/rooms/living_room.yaml @@ -86,29 +86,6 @@ light_living_room: off_events: - turn_off_all_lights - turn_off_all_living_room_lights - - -projection_screen: - module: smartshutter - class: SmartShutter - entity: cover.living_room_projection_screen - close_position: 22 - - close_conditions: # close mean screen down - # when the projector is turn back on, it can happen that this condition is evaluated before projector_off_for_a_while is updated - # so I need to test both binary_sensor.projector_on and not binary_sensor.projector_off_for_a_while - trigger_conditions: binary_sensor.projector_on or not binary_sensor.projector_off_for_a_while - disable_conditions: - - binary_sensor.day_interval_night and not binary_sensor.projector_on - - binary_sensor.is_everybody_sleeping and not binary_sensor.projector_on - -bravia_theatre: - module: smartswitch - class: SmartSwitch - entity: media_player.bravia_theatre - - smart_conditions: binary_sensor.vertex2_has_input_signal - # w4000i: # module: smartswitch # class: SmartSwitch @@ -116,55 +93,7 @@ bravia_theatre: # smart_conditions: binary_sensor.vertex2_has_input_signal -w4000i: - module: smartprojector_benq_over_telnet - class: SmartProjector - entity: switch.projector - telnet_connection: - ip: 10.0.0.40 - port: 8000 - - power_plug: - entity: switch.living_room_projector_plug - power_sensor: sensor.living_room_projector_plug_power - power_threeshold: 5 - conditions: - trigger_conditions: - - binary_sensor.someone_home and not (binary_sensor.is_everybody_sleeping or binary_sensor.is_living_room_sleeping) - - sensor.living_room_projector_plug_power > 10 or binary_sensor.projector_on # don't remove this one or it will fry the projector - - virtual_sensors: - sensors: - value_selector.projector_icon: - projector_off: - conditions: not super - value: "'mdi:projector-off'" - projector_on: - conditions: True - value: "'mdi:projector'" - continuous_condition.projector_ready: - conditions: switch.living_room_projector_plug - time: 30 - - attributes_override: - icon: sensor.projector_icon - friendly_name: Projecteur - - press_enter_event: send_projector_enter - - projector_ready_conditions: binary_sensor.projector_ready - - #state_conditions: binary_sensor.projector_on - #cooling_down_sensor: binary_sensor.projector_cooling_down - - smart_conditions: - trigger_conditions: binary_sensor.vertex2_has_input_signal - disable_conditions: - - self == 'unavailable' - #- not binary_sensor.projector_screen_down - #- binary_sensor.projector_cooling_down - - switch.projector.warming_up mezzanine_curtains: module: smartshutter @@ -191,42 +120,3 @@ living_room_sleep_switch: attributes: event_type: multi_press_1 -samsung_the_frame: - module: samsungtheframe - class: SamsungTheFrame - - ip_address: 10.0.0.58 - #broadcast_address: 10.0.0.255 - #mac_address: 98:06:3C:85:93:38 - #wol_switch: switch.theframewol - wake_up_service: "esphome/ir_bridge_turn_on_the_frame" - app_list: # https://github.com/tavicu/homebridge-samsung-tizen/issues/291 - - 3201907018807 # Netflix - - 3201910019365 # Prime - - virtual_sensors: - default_values: - binary_sensor.tv_auto_shutdown: false - binary_sensor.tv_on: false - sensors: - value_selector.samsung_the_frame_target_state: - power_off: - trigger_conditions: - - binary_sensor.is_everybody_sleeping or binary_sensor.is_living_room_sleeping or not binary_sensor.someone_home - - binary_sensor.projector_on or binary_sensor.projector_screen_down - #hdmi: binary_sensor.vertex2_has_input_signal and not binary_sensor.projector_screen_down - #manual: binary_sensor.tv_on and not binary_sensor.tv_auto_shutdown - manual: binary_sensor.tv_on #and not binary_sensor.tv_auto_shutdown - tv_off: not binary_sensor.living_room_occupancy - art_mode: True - - tv_state: sensor.samsung_the_frame_target_state - - output_sensors: - is_on: binary_sensor.tv_on - is_art_mode: binary_sensor.tv_in_art_mode - tv_auto_shutdown: binary_sensor.tv_auto_shutdown - - input_sensors: - power_plug: switch.the_frame_power_plug - power_plug_power: sensor.the_frame_power_plug_power \ No newline at end of file diff --git a/apps/vertex2.py b/apps/vertex2.py new file mode 100644 index 0000000..5e4edba --- /dev/null +++ b/apps/vertex2.py @@ -0,0 +1,488 @@ +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 +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.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) + + 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'], + 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') + 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 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.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.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='', audout='', insel='0 0') + return + self._status_requested.set() + + def _apply_vertex2_status(self, kwargs): + self.log_info( + f"Vertex2 TCP status: RX0={kwargs['rx0']!r}, RX1={kwargs['rx1']!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 = None if input_signal in ('unavailable', 'no signal') else 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) + + if self.video_tx_signal_sensor.get_state() != input_signal: + self.video_tx_signal_sensor.set_state(state=input_signal) + 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')