########################################################################### # # Copyright (c) 2020-2024 Diality Inc. - All Rights Reserved. # # THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN # WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. # # @file alarms.py # # @author (last) Zoltan Miskolci # @date (last) 05-May-2026 # @author (original) Peter Lucia # @date (original) 02-Apr-2020 # ############################################################################ # Module imports from functools import partial from logging import Logger # Project imports from leahi_dialin.common import td_enum_repository, AlarmList from leahi_dialin.common.constants import NO_RESET from leahi_dialin.common.generic_defs import DataTypes from leahi_dialin.common.msg_defs import MsgIds from leahi_dialin.common.override_templates import cmd_generic_broadcast_interval_override, cmd_generic_override from leahi_dialin.protocols.CAN import CanMessenger, CanChannels from leahi_dialin.utils.abstract_classes import AbstractSubSystem, LocalVars from leahi_dialin.utils.base import publish from leahi_dialin.utils.conversions import integer_to_bytearray, float_to_bytearray class TDAlarms(AbstractSubSystem): """ TD interface containing alarm related commands. """ def __init__(self, can_interface: CanMessenger, logger: Logger): """ @param can_interface: Can Messenger object """ super().__init__() self.can_interface = can_interface self.logger = logger if self.can_interface is not None: self.can_interface.register_receiving_publication_function(channel_id = CanChannels.td_alarm_broadcast_ch_id, message_id = MsgIds.MSG_ID_ALARM_STATUS_DATA.value, function = self._handler_alarms_status_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.td_alarm_broadcast_ch_id, message_id = MsgIds.MSG_ID_ALARM_TRIGGERED.value, function = self._handler_alarm_trigger) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.td_alarm_broadcast_ch_id, message_id = MsgIds.MSG_ID_ALARM_CLEARED.value, function = self._handler_alarm_clear) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.td_alarm_broadcast_ch_id, message_id = MsgIds.MSG_ID_ALARM_CONDITION_CLEARED.value, function = self._handler_alarm_condition_clear) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.td_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_TD_ALARM_INFORMATION_DATA.value, function = self._handler_alarm_information_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.td_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_TD_ACTIVE_ALARMS_LIST_REQUEST_RESPONSE.value, function = self._handler_alarm_list_response) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.td_to_dialin_ch_id, message_id = MsgIds.MSG_ID_TD_ALARM_PROPERTIES_RESPONSE.value, function = self._handler_alarm_table_response) self.td_alarm_status_timestamp = 0.0 #: The timestamp of the last Alarm Status message self.td_alarm_triggered_timestamp = 0.0 #: The timestamp of the last Alarm Triggered message self.td_alarm_cleared_timestamp = 0.0 #: The timestamp of the last Alarm Cleared message self.td_alarm_clr_condition_timestamp = 0.0 #: The timestamp of the last Alarm Clear Condition message self.td_alarm_information_timestamp = 0.0 #: The timestamp of the last Alarm Information message self.alarm_table_timestamp = 0.0 #: The timestamp of the last Alarm Table Properties message # composite alarm status based on latest TD alarm status broadcast message self.alarms_priority_state = 0 self.alarm_top = 0 self.alarms_silence_expires_in = 0 self.alarms_escalates_in = 0 self.alarms_flags = 0 self.alarm_states = [False] * 500 #: The States of all Alarms data in list format self.alarm_conditions = [False] * 500 #: The Conditions of all Alarms data in list format self.alarm_priorities = [0] * 500 #: The Priorities of all Alarms data in list format self.alarm_ranks = [0] * 500 #: The Ranks of all Alarms data in list format self.alarm_clear_top_only_flags = [False] * 500 #: The Clear Top only flags of all Alarms data in list format self.alarm_data = [0, 0] * 500 #: The all Alarms data in list format self.last_alarm_triggered = 0 #: The timestamp when did the last alarm triggered self.last_alarm_data_1 = 0.0 #: The last Alarm's data part 1 self.last_alarm_data_2 = 0.0 #: The last Alarm's data part 2 # alarm information self.alarm_volume = 0 self.alarm_audio_curr_hg = 0.0 self.alarm_audio_curr_lg = 0.0 self.alarm_backup_audio_curr = 0.0 self.safety_shutdown_active = False self.ac_power_lost = False self.alarm_table_button_blockers = [False] * td_enum_repository.AlarmResponseButtons.NUM_OF_TD_ALARM_RESPONSE_BUTTONS.value # Alarms list self.ui_alarm_list_accepted = True self.ui_alarm_list_reject_reason = 0 self.ui_alarm_list = [] self.ui_alarm_list_timestamp = 0.0 #: The timestamp of the last Alarm List # Alarm Table Properties self.alarm_table = {} def clear_dialin_alarms(self): """ Clears the alarms states in Dialin. @return: none """ for x in range(500): self.alarm_states[x] = False # ============================================================ Properties ============================================================ @property def alarms_priority_state(self): """ The current Top Alarm's priority value """ return self._alarms_priority_state @alarms_priority_state.setter def alarms_priority_state(self, value): self._alarms_priority_state = value @property def alarm_top(self): """ The current Top Alarm's value """ return self._alarm_top @alarm_top.setter def alarm_top(self, value): self._alarm_top = value @property def alarms_silence_expires_in(self): """ How many seconds the silence for the alarms expires in """ return self._alarms_silence_expires_in @alarms_silence_expires_in.setter def alarms_silence_expires_in(self, value): self._alarms_silence_expires_in = value @property def alarms_escalates_in(self): """ How many seconds the current Alarm escalates to the next priority """ return self._alarms_escalates_in @alarms_escalates_in.setter def alarms_escalates_in(self, value): self._alarms_escalates_in = value @property def alarms_flags(self): """ The Alarm flag's value """ return self._alarms_flags @alarms_flags.setter def alarms_flags(self, value): self._alarms_flags = value @property def alarm_volume(self): """ The Alarm's Volume level """ return self._alarm_volume @alarm_volume.setter def alarm_volume(self, value): self._alarm_volume = value @property def alarm_audio_curr_hg(self): """ The Alarm's Audio high current """ return self._alarm_audio_curr_hg @alarm_audio_curr_hg.setter def alarm_audio_curr_hg(self, value): self._alarm_audio_curr_hg = value @property def alarm_audio_curr_lg(self): """ The Alarm's Audio low current """ return self._alarm_audio_curr_lg @alarm_audio_curr_lg.setter def alarm_audio_curr_lg(self, value): self._alarm_audio_curr_lg = value @property def alarm_backup_audio_curr(self): """ The Alarm's Audio backup current """ return self._alarm_backup_audio_curr @alarm_backup_audio_curr.setter def alarm_backup_audio_curr(self, value): self._alarm_backup_audio_curr = value @property def safety_shutdown_active(self): """ The Safety Shutdown is active or not """ return self._safety_shutdown_active @safety_shutdown_active.setter def safety_shutdown_active(self, value): self._safety_shutdown_active = value @property def ac_power_lost(self): """ The AC Power is lost or not """ return self._ac_power_lost @ac_power_lost.setter def ac_power_lost(self, value): self._ac_power_lost = value @property def alarm_table_button_blockers(self): """ The Alarm Table's Button blocker data """ return self._alarm_table_button_blockers @alarm_table_button_blockers.setter def alarm_table_button_blockers(self, value): self._alarm_table_button_blockers = value @property def ui_alarm_list_accepted(self): """ The UI Alarm List request accepted """ return self._ui_alarm_list_accepted @ui_alarm_list_accepted.setter def ui_alarm_list_accepted(self, value): self._ui_alarm_list_accepted = value @property def ui_alarm_list_reject_reason(self): """ The UI Alarm List request reject reason """ return self._ui_alarm_list_reject_reason @ui_alarm_list_reject_reason.setter def ui_alarm_list_reject_reason(self, value): self._ui_alarm_list_reject_reason = value @property def ui_alarm_list(self): """ The UI Alarm List content """ return self._ui_alarm_list @ui_alarm_list.setter def ui_alarm_list(self, value): self._ui_alarm_list = value @property def alarm_table(self): """ The Alarm Table with Alarm Property details """ return self._alarm_table @alarm_table.setter def alarm_table(self, value): self._alarm_table = value # ============================================================ Handlers ============================================================ @publish(["msg_id_td_alarm_status_data", "alarms_priority_state", "alarm_top", "alarms_silence_expires_in", "alarms_escalates_in", "alarms_flags", "td_alarm_status_timestamp"]) def _handler_alarms_status_sync(self, message, timestamp=0.0): """ Handles published alarms status messages. alarms status data are captured for reference. @param message: published alarm status data message @return: none """ msg_list = [] msg_list.append((type(self).alarms_priority_state, DataTypes.U32)) msg_list.append((type(self).alarm_top, DataTypes.U32)) msg_list.append((type(self).alarms_escalates_in, DataTypes.U32)) msg_list.append((type(self).alarms_silence_expires_in, DataTypes.U32)) msg_list.append((type(self).alarms_flags, DataTypes.U16)) self.process_into_vars_2(decoder_list = msg_list, message = message) self.td_alarm_status_timestamp = timestamp @publish(["msg_id_td_alarm_triggered", "alarm_states", "alarm_conditions", "alarm_data", "alarm_priorities", "alarm_ranks", "alarm_clear_top_only_flags", "td_alarm_triggered_timestamp"]) def _handler_alarm_trigger(self, message, timestamp=0.0): """ Handles published TD alarm activation messages. @param message: published TD alarm activation message @return: none """ alarm_id = LocalVars('alarm_id') data_typ_1 = LocalVars('data_typ_1') data_1 = LocalVars('data_1') data_typ_2 = LocalVars('data_typ_2') data_2 = LocalVars('data_2') priority = LocalVars('priority') rank = LocalVars('rank') clr_top_only = LocalVars('clr_top_only') result = {} msg_list = [] msg_list.append((alarm_id, DataTypes.U32)) msg_list.append((data_typ_1, DataTypes.U32)) msg_list.append((data_1, 'data_typ_1')) msg_list.append((data_typ_2, DataTypes.U32)) msg_list.append((data_2, 'data_typ_2')) msg_list.append((priority, DataTypes.U32)) msg_list.append((rank, DataTypes.U32)) msg_list.append((clr_top_only, DataTypes.U32)) self.process_into_vars_2(decoder_list = msg_list, message = message) self.logger.debug("Alarm ID: %d %d %d" % (alarm_id.value, data_1.value, data_2.value)) self.alarm_states[alarm_id.value] = True self.alarm_conditions[alarm_id.value] = True self.alarm_priorities[alarm_id.value] = priority.value self.alarm_ranks[alarm_id.value] = rank.value self.alarm_clear_top_only_flags[alarm_id.value] = clr_top_only.value self.alarm_data[alarm_id.value] = [data_1.value, data_2.value] self.last_alarm_triggered = alarm_id.value self.last_alarm_data_1 = data_1.value self.last_alarm_data_2 = data_2.value self.td_alarm_triggered_timestamp = timestamp @publish(["msg_id_td_alarm_cleared", "alarm_states", "alarm_conditions", "TD_alarm_cleared_timestamp"]) def _handler_alarm_clear(self, message, timestamp=0.0): """ Handles published TD alarm clear messages. @param message: published TD alarm clear message @return: none """ alarm_id = LocalVars('alarm_id') msg_list = [] msg_list.append((alarm_id, DataTypes.U32)) self.process_into_vars_2(decoder_list = msg_list, message = message) self.alarm_states[alarm_id.value] = False self.alarm_conditions[alarm_id.value] = False self.td_alarm_cleared_timestamp = timestamp @publish(["msg_id_td_alarm_condition_cleared", "alarm_conditions", "alarm_conditions", "td_alarm_clr_condition_timestamp"]) def _handler_alarm_condition_clear(self, message, timestamp=0.0): """ Handles published TD alarm clear alarm condition messages. @param message: published TD alarm clear alarm condition message @return: none """ alarm_id = LocalVars('alarm_id') msg_list = [] msg_list.append((alarm_id, DataTypes.U32)) self.process_into_vars_2(decoder_list = msg_list, message = message) self.alarm_conditions[alarm_id.value] = False self.td_alarm_clr_condition_timestamp = timestamp @publish(["msg_id_td_alarm_info_data", "alarm_volume", "alarm_audio_curr_hg", "alarm_audio_curr_lg", "alarm_backup_audio_curr", "safety_shutdown_active", "ac_power_lost", "alarm_table_button_blockers", "td_alarm_information_timestamp"]) def _handler_alarm_information_sync(self, message, timestamp=0.0): """ Handles published TD alarm information broadcast messages. @param message: published TD alarm information message @return: none """ msg_list = [] msg_list.append((type(self).alarm_volume, DataTypes.U32)) msg_list.append((type(self).alarm_audio_curr_hg, DataTypes.F32)) msg_list.append((type(self).alarm_audio_curr_lg, DataTypes.F32)) msg_list.append((type(self).alarm_backup_audio_curr, DataTypes.U32)) msg_list.append((type(self).safety_shutdown_active, DataTypes.BOOL)) msg_list.append((type(self).ac_power_lost, DataTypes.BOOL)) msg_list.append((self.alarm_table_button_blockers, td_enum_repository.AlarmResponseButtons.RESUME.value, DataTypes.BOOL_U08)) msg_list.append((self.alarm_table_button_blockers, td_enum_repository.AlarmResponseButtons.RINSEBACK.value, DataTypes.BOOL_U08)) msg_list.append((self.alarm_table_button_blockers, td_enum_repository.AlarmResponseButtons.END_TREATMENT.value, DataTypes.BOOL_U08)) msg_list.append((self.alarm_table_button_blockers, td_enum_repository.AlarmResponseButtons.RESUME.value, DataTypes.BOOL_U08)) msg_list.append((self.alarm_table_button_blockers, td_enum_repository.AlarmResponseButtons.RINSEBACK.value, DataTypes.BOOL_U08)) msg_list.append((self.alarm_table_button_blockers, td_enum_repository.AlarmResponseButtons.END_TREATMENT.value, DataTypes.BOOL_U08)) self.process_into_vars_2(decoder_list = msg_list, message = message) self.td_alarm_information_timestamp = timestamp @publish(["msg_id_td_active_alarms_list_request_response", "accepted", "rejection_reason", "active_alarm_list"]) def _handler_alarm_list_response(self, message, timestamp=0.0): """ Handles published TD alarm list response message. @param message: published TD alarm list response message @return: none """ msg_list = [] msg_list.append((type(self).ui_alarm_list_accepted, DataTypes.U32)) msg_list.append((type(self).ui_alarm_list_reject_reason, DataTypes.U32)) msg_list.append((self.ui_alarm_list, DataTypes.U32)) # UI Alarm #1 msg_list.append((self.ui_alarm_list, DataTypes.U32)) # UI Alarm #2 msg_list.append((self.ui_alarm_list, DataTypes.U32)) # UI Alarm #3 msg_list.append((self.ui_alarm_list, DataTypes.U32)) # UI Alarm #4 msg_list.append((self.ui_alarm_list, DataTypes.U32)) # UI Alarm #5 msg_list.append((self.ui_alarm_list, DataTypes.U32)) # UI Alarm #6 msg_list.append((self.ui_alarm_list, DataTypes.U32)) # UI Alarm #7 msg_list.append((self.ui_alarm_list, DataTypes.U32)) # UI Alarm #8 msg_list.append((self.ui_alarm_list, DataTypes.U32)) # UI Alarm #9 msg_list.append((self.ui_alarm_list, DataTypes.U32)) # UI Alarm #10 self.process_into_vars_2(decoder_list = msg_list, message = message) self.ui_alarm_list_timestamp = timestamp @publish(["msg_id_td_active_alarms_properties_response", "active_table", "alarm_table_timestamp"]) def _handler_alarm_table_response(self, message, timestamp=0.0): """ Handles published TD alarm table properties messages. @param message: published TD alarm table properties message @return: none """ priority = LocalVars('priority') rank = LocalVars('rank') source = LocalVars('source') system_fault = LocalVars('system_fault') dd_fault = LocalVars('dd_fault') stops = LocalVars('stops') clear_immediately = LocalVars('clear_immediately') no_clear = LocalVars('no_clear') no_resume = LocalVars('no_resume') no_rinseback = LocalVars('no_rinseback') no_end_treatment = LocalVars('no_end_treatment') not_in_rinseback = LocalVars('not_in_rinseback') not_in_post_treatment = LocalVars('not_in_post_treatment') no_blood_recirc = LocalVars('no_blood_recirc') no_dial_recirc = LocalVars('no_dial_recirc') auto_resume = LocalVars('auto_resume') clear_alarm_only = LocalVars('clear_alarm_only') treatment_log = LocalVars('treatment_log') alarm_id = LocalVars('alarm_id') properties_list = [] properties_list.append((priority, DataTypes.U16)) properties_list.append((rank, DataTypes.U16)) properties_list.append((source, DataTypes.U08)) properties_list.append((system_fault, DataTypes.BOOL_U08)) properties_list.append((dd_fault, DataTypes.BOOL_U08)) properties_list.append((stops, DataTypes.BOOL_U08)) properties_list.append((clear_immediately, DataTypes.BOOL_U08)) properties_list.append((no_clear, DataTypes.BOOL_U08)) properties_list.append((no_resume, DataTypes.BOOL_U08)) properties_list.append((no_rinseback, DataTypes.BOOL_U08)) properties_list.append((no_end_treatment, DataTypes.BOOL_U08)) properties_list.append((not_in_rinseback, DataTypes.BOOL_U08)) properties_list.append((not_in_post_treatment, DataTypes.BOOL_U08)) properties_list.append((no_blood_recirc, DataTypes.BOOL_U08)) properties_list.append((no_dial_recirc, DataTypes.BOOL_U08)) properties_list.append((auto_resume, DataTypes.BOOL_U08)) properties_list.append((clear_alarm_only, DataTypes.BOOL_U08)) properties_list.append((treatment_log, DataTypes.BOOL_U08)) properties_list.append((alarm_id, DataTypes.U16)) self.process_into_vars_2(decoder_list = properties_list, message = message) # Save the result in the table for prop, _ in properties_list: if alarm_id.value not in self.alarm_table: self.alarm_table[alarm_id.value] = {} if prop != alarm_id.name: self.alarm_table[alarm_id.value][prop.name] = prop.value self.alarm_table_timestamp = timestamp # ============================================================ Overrides and Requests ============================================================ def cmd_alarm_status_broadcast_interval_override(self, ms: int = 250, reset: int = NO_RESET): """ Constructs and sends the alarm status broadcast interval override command Constraints: Must be logged into TD. Given interval must be non-zero and a multiple of the TD general task interval (50 ms). @param ms: integer - interval (in ms) to override with @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ return cmd_generic_broadcast_interval_override( ms = ms, reset = reset, channel_id = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_TD_ALARM_STATUS_PUBLISH_INTERVAL_OVERRIDE_REQUEST, module_name = 'TD Alarm Status', logger = self.logger, can_interface = self.can_interface) def cmd_alarm_info_broadcast_interval_override(self, ms: int = 1000, reset: int = NO_RESET): """ Constructs and sends the alarm information broadcast interval override command Constraints: Must be logged into TD. Given interval must be non-zero and a multiple of the TD general task interval (50 ms). @param ms: integer - interval (in ms) to override with @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ return cmd_generic_broadcast_interval_override( ms = ms, reset = reset, channel_id = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_TD_ALARM_INFO_PUBLISH_INTERVAL_OVERRIDE_REQUEST, module_name = 'TD Alarm Info', logger = self.logger, can_interface = self.can_interface) def cmd_alarm_state_override(self, alarm: int, state: int, reset: int = NO_RESET) -> int: """ Constructs and sends the alarm state override command Constraints: Must be logged into TD. Given alarm must be valid. If inactivating alarm, given alarm must be recoverable (clearable). @param alarm: integer - ID of alarm to override @param state: integer - 1 for alarm active, 0 for alarm inactive @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ rst = integer_to_bytearray(reset) sta = integer_to_bytearray(state) alm = integer_to_bytearray(alarm) payload = rst + sta + alm alarm_name = AlarmList(alarm).name state_name = 'Active' if state != 0 else 'Inactive' return cmd_generic_override( payload = payload, reset = NO_RESET, channel_id = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_TD_ALARM_STATE_OVERRIDE_REQUEST, entity_name = f'TD {alarm_name} Alarm state', override_text = f'{state_name}', logger = self.logger, can_interface = self.can_interface) def cmd_clear_all_alarms(self) -> int: """ Constructs and sends the clear all active alarms command. This will clear even non-recoverable alarms. Constraints: Must be logged into TD. @return: 1 if successful, zero otherwise """ key = integer_to_bytearray(-758926171) # 0xD2C3B4A5 payload = key return cmd_generic_override( payload = payload, reset = NO_RESET, channel_id = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_TD_ALARM_CLEAR_ALL_ALARMS_REQUEST, entity_name = f'TD Alarms', override_text = f'Cleared', logger = self.logger, can_interface = self.can_interface) def cmd_alarm_time_override(self, alarm: int, time_ms: int, reset: int = NO_RESET) -> int: """ Constructs and sends the alarm time override command Constraints: Must be logged into TD. Given alarm must be valid. @param alarm: integer - ID of alarm to override @param time_ms: integer - time (in ms) since alarm was activated @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ rst = integer_to_bytearray(reset) ms = integer_to_bytearray(time_ms) alm = integer_to_bytearray(alarm) payload = rst + ms + alm alarm_name = AlarmList(alarm).name return cmd_generic_override( payload = payload, reset = NO_RESET, channel_id = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_TD_ALARM_START_TIME_OVERRIDE_REQUEST, entity_name = f'TD {alarm_name} Alarms time', override_text = f'{time_ms} ms', logger = self.logger, can_interface = self.can_interface) def cmd_alarm_lamp_pattern_override(self, pattern: int, reset: int = NO_RESET) -> int: """ Constructs and sends the alarm lamp pattern override command. Constraints: Must be logged into TD. Given pattern must be one of the patterns listed below. @param pattern: integer - ID of alarm lamp pattern to override with @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise Patterns: \n 0 = off \n 1 = ok \n 2 = fault \n 3 = high \n 4 = medium \n 5 = low \n 6 = manual """ rst = integer_to_bytearray(reset) pat = integer_to_bytearray(pattern) payload = rst + pat if pattern == self.TD_ALARM_LAMP_PATTERN_OFF: str_pat = "Off" elif pattern == self.TD_ALARM_LAMP_PATTERN_OK: str_pat = "Ok" elif pattern == self.TD_ALARM_LAMP_PATTERN_FAULT: str_pat = "Fault" elif pattern == self.TD_ALARM_LAMP_PATTERN_HIGH: str_pat = "High" elif pattern == self.TD_ALARM_LAMP_PATTERN_MEDIUM: str_pat = "Medium" elif pattern == self.TD_ALARM_LAMP_PATTERN_LOW: str_pat = "Low" else: str_pat = "Manual" return cmd_generic_override( payload = payload, reset = NO_RESET, channel_id = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_TD_ALARM_AUDIO_LEVEL_OVERRIDE_REQUEST, entity_name = f'TD Alarms lamp pattern', override_text = f'{str_pat}', logger = self.logger, can_interface = self.can_interface) def cmd_alarm_audio_level_override(self, volume: int = 5, reset: int = NO_RESET): """ Constructs and sends the alarm audio volume override command Constraints: Must be logged into TD. Given volume must be an integer between 1 and 5. @param volume: integer - alarm audio volume level (1..5) @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ rst = integer_to_bytearray(reset) vol = integer_to_bytearray(volume) payload = rst + vol return cmd_generic_override( payload = payload, reset = NO_RESET, channel_id = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_TD_ALARM_AUDIO_LEVEL_OVERRIDE_REQUEST, entity_name = f'TD Alarms audio volume', override_text = f'{str(volume)}', logger = self.logger, can_interface = self.can_interface) def cmd_alarm_audio_current_hg_override(self, current: float = 0.0, reset: int = NO_RESET): """ Constructs and sends the alarm audio current (high gain) override command Constraints: Must be logged into TD. @param current: float - current (in mA) for high gain alarm audio @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ rst = integer_to_bytearray(reset) cur = float_to_bytearray(current) payload = rst + cur return cmd_generic_override( payload = payload, reset = NO_RESET, channel_id = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_TD_ALARM_AUDIO_CURRENT_HG_OVERRIDE_REQUEST, entity_name = f'TD Alarms audio high gain current', override_text = f'{str(current)} mA', logger = self.logger, can_interface = self.can_interface) def cmd_alarm_audio_current_lg_override(self, current: float = 0.0, reset: int = NO_RESET): """ Constructs and sends the alarm audio current (low gain) override command Constraints: Must be logged into TD. @param current: float - current (in mA) for low gain alarm audio @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ rst = integer_to_bytearray(reset) cur = float_to_bytearray(current) payload = rst + cur return cmd_generic_override( payload = payload, reset = NO_RESET, channel_id = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_TD_ALARM_AUDIO_CURRENT_LG_OVERRIDE_REQUEST, entity_name = f'TD Alarms audio low gain current', override_text = f'{str(current)} mA', logger = self.logger, can_interface = self.can_interface) def cmd_alarm_backup_audio_current_override(self, current: float = 0.0, reset: int = NO_RESET): """ Constructs and sends the backup alarm audio current override command Constraints: Must be logged into TD. @param current: float - current (in mA) for backup alarm audio @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ rst = integer_to_bytearray(reset) cur = float_to_bytearray(current) payload = rst + cur return cmd_generic_override( payload = payload, reset = NO_RESET, channel_id = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_TD_BACKUP_ALARM_AUDIO_CURRENT_OVERRIDE_REQUEST, entity_name = f'TD Alarms backup audio current', override_text = f'{str(current)} mA', logger = self.logger, can_interface = self.can_interface) def cmd_request_alarm_table(self) -> int: """ Constructs and sends the Alarm details request for selected alarm. Constraints: Must be logged into TD. @param alarm_id: (Integer) The Alarm's ID @return: 1 if successful, zero otherwise """ response = 1 execute_list = [] for alarm_enum in AlarmList: execute_list.append(partial(self.cmd_request_alarm_details_for_alarm, alarm_enum.value)) # Execute the stored functions one by one # Remove the ones that got 1 (successfully recieved) as response # Retry the ones that are failed 2 more times retry = 0 while execute_list != [] and retry < 3: failed = [] for func in execute_list: resp = func() if resp == 1: failed.append(func) execute_list = failed retry += 1 response = 1 if failed == [] else 0 return response def cmd_request_alarm_details_for_alarm(self, alarm_id: int) -> int: """ Constructs and sends the Alarm details request for selected alarm. Constraints: Must be logged into TD. @param alarm_id: (Integer) The Alarm's ID @return: 1 if successful, zero otherwise """ payload = integer_to_bytearray(alarm_id) return cmd_generic_override( payload = payload, reset = None, channel_id = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_TD_GET_ALARM_PROPERTIES_REQUEST, entity_name = f'TD Alarms Details for {AlarmList(alarm_id)} ({alarm_id})', override_text = f'requested', logger = self.logger, can_interface = self.can_interface)