########################################################################### # # 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) 04-May-2026 # @author (original) Quang Nguyen # @date (original) 02-Sep-2020 # ############################################################################ # Module imports from logging import Logger import struct # Project imports from leahi_dialin.common.alarm_defs import AlarmList from leahi_dialin.common.constants import NO_RESET from leahi_dialin.common.generic_defs import DataTypes from leahi_dialin.common.msg_ids import MsgIds from leahi_dialin.common.override_templates import cmd_generic_broadcast_interval_override, cmd_generic_override from leahi_dialin.protocols.CAN import CanMessenger, CanMessage, 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 class DDAlarms(AbstractSubSystem): """ DD interface containing alarm related commands. """ _ALARM_ID_MAX_ALARMS = 500 START_POS_ALARM_ID = CanMessage.PAYLOAD_START_INDEX END_POS_ALARM_ID = START_POS_ALARM_ID + 2 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.dd_alarm_broadcast_ch_id, message_id = MsgIds.MSG_ID_ALARM_TRIGGERED.value, function = self._handler_alarm_triggered) self.dd_alarm_info_timestamp = 0.0 #: The timestamp of the last Alarm Information message self.dd_alarm_triggered_timestamp = 0.0 #: The timestamp of the last Alarm Triggered message self.dd_alarm_cleared_timestamp = 0.0 #: The timestamp of the last Alarm Cleared message self.dd_alarm_condition_cleared_timestamp = 0.0 #: The timestamp of the last Alarm Condition Cleared message 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 # ============================================================ Properties ============================================================ # ============================================================ Handlers ============================================================ @publish(["msg_id_dd_alarm_triggered", "alarm_states", "alarm_conditions", "alarm_data", "alarm_priorities", "alarm_ranks", "alarm_clear_top_only_flags", "dd_alarm_triggered_timestamp"]) def _handler_alarm_triggered(self, message, timestamp = 0.0): """ Handles published DD alarm activation messages. @param message: published DD 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') 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(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.dd_alarm_triggered_timestamp = timestamp # ============================================================ Support ============================================================ def clear_dialin_alarms(self): """ Clears the alarms states in Dialin. @return: none """ for x in range(self.alarm_states): self.alarm_states[x] = False def get_dialin_alarms(self): """ Get the active alarms in Dialin. @return: none """ active_alarms = [] for x in range(self.alarm_states): if self.alarm_states[x]: active_alarms.append(AlarmList(x)) return active_alarms # ============================================================ Overrides and Requests ============================================================ 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 DD. 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 = reset, channel_id = CanChannels.dialin_to_dd_ch_id, msg_id = MsgIds.MSG_ID_DD_ALARM_STATE_OVERRIDE_REQUEST, entity_name = f'DD {alarm_name} Alarm state', override_text = f'{state_name}', logger = self.logger, can_interface = self.can_interface)