########################################################################### # # 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 valves.py # # @author (last) Zoltan Miskolci # @date (last) 05-May-2026 # @author (original) Peman Montazemi # @date (original) 19-May-2020 # ############################################################################ # Module imports from logging import Logger from collections import OrderedDict # Project imports from leahi_dialin.common.constants import NO_RESET from leahi_dialin.common import fp_enum_repository from leahi_dialin.common.generic_defs import DataTypes from leahi_dialin.common.fp_defs import BETA_IOFP_VALVE_OFFSET 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 FPValves(AbstractSubSystem): """ Dialin API sub-class for FP valve 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.fp_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_FP_VALVES_STATES_DATA.value, function = self._handler_valves_sync) self.fp_valves_states_timestamp = 0.0 #: The timestamp of the last message self.valve_states_all = 0x0000 #: States of all the vales in binary format self.valves_sensed_states = OrderedDict() #: The Valve's Sensed State data in dictionary format self.m4_valv = {"id": fp_enum_repository.FPValveNames.M4_VALV.value, "state": fp_enum_repository.FPValveStates.VALVE_STATE_CLOSED.value } #: The M4 valve's value self.m12_valv = {"id": fp_enum_repository.FPValveNames.M12_VALV.value, "state": fp_enum_repository.FPValveStates.VALVE_STATE_CLOSED.value } #: The M12 valve's value self.p11_valv = {"id": fp_enum_repository.FPValveNames.P11_VALV.value, "state": fp_enum_repository.FPValveStates.VALVE_STATE_CLOSED.value } #: The P11 valve's value self.p33_valv = {"id": fp_enum_repository.FPValveNames.P33_VALV.value, "state": fp_enum_repository.FPValveStates.VALVE_STATE_CLOSED.value } #: The P33 valve's value self.p34_valv = {"id": fp_enum_repository.FPValveNames.P34_VALV.value, "state": fp_enum_repository.FPValveStates.VALVE_STATE_CLOSED.value } #: The P34 valve's value self.p37_valv = {"id": fp_enum_repository.FPValveNames.P37_VALV.value, "state": fp_enum_repository.FPValveStates.VALVE_STATE_CLOSED.value } #: The P37 valve's value self.p39_valv = {"id": fp_enum_repository.FPValveNames.P39_VALV.value, "state": fp_enum_repository.FPValveStates.VALVE_STATE_CLOSED.value } #: The P39 valve's value self.p6_valv = {"id": fp_enum_repository.FPValveNames.P6_VALV.value, "state": fp_enum_repository.FPValveStates.VALVE_STATE_CLOSED.value } #: The P6 valve's value #: States of all the vales in list format self.valve_states_enum = {} for i in range(0, fp_enum_repository.FPValveNames.NUM_OF_FP_VALVES.value): valve_name = fp_enum_repository.FPValveNames(i).name self.valve_states_enum[valve_name] = 0 self.valves_sensed_states[valve_name] = '' # ============================================================ Properties ============================================================ @property def fp_valves(self) -> dict: """ The Valves data """ return {fp_enum_repository.FPValveNames.M4_VALV.name: self.m4_valv['state'], fp_enum_repository.FPValveNames.M12_VALV.name: self.m12_valv['state'], fp_enum_repository.FPValveNames.P11_VALV.name: self.p11_valv['state'], fp_enum_repository.FPValveNames.P33_VALV.name: self.p33_valv['state'], fp_enum_repository.FPValveNames.P34_VALV.name: self.p34_valv['state'], fp_enum_repository.FPValveNames.P37_VALV.name: self.p37_valv['state'], fp_enum_repository.FPValveNames.P39_VALV.name: self.p39_valv['state'], fp_enum_repository.FPValveNames.P6_VALV.name: self.p6_valv['state']} # ============================================================ Handlers ============================================================ @publish([ "msg_id_fp_valves_states_data", "valve_states_all", "m4_valv", "m12_valv", "p11_valv", "p33_valv", "p34_valv", "p37_valv", "p39_valv", "p6_valv", "valve_states_enum", "fp_valves_states_timestamp" ]) def _handler_valves_sync(self, message, timestamp=0.0): """ Handles published FP valves states message. @param message: published FP valves states message @return: none """ io_valves = LocalVars('io_valves') fp_valves = LocalVars('fp_valves') msg_list = [] msg_list.append((io_valves, DataTypes.U08)) msg_list.append((fp_valves, DataTypes.U08)) self.process_into_vars(decoder_list = msg_list, message = message) # Extract each valve state from U08 valves states using bit-masking self.m4_valv["state"] = fp_enum_repository.FPValveStates(io_valves.value & 1) self.m12_valv["state"] = fp_enum_repository.FPValveStates(io_valves.value & 2) self.valve_states_all = io_valves[0] # Extract each valve state from U08 valves states using bit-masking self.p11_valv["state"] = fp_enum_repository.FPValveStates(fp_valves.value & 1) self.p33_valv["state"] = fp_enum_repository.FPValveStates(fp_valves.value & 2) self.p34_valv["state"] = fp_enum_repository.FPValveStates(fp_valves.value & 4) self.p37_valv["state"] = fp_enum_repository.FPValveStates(fp_valves.value & 8) self.p39_valv["state"] = fp_enum_repository.FPValveStates(fp_valves.value & 16) self.p6_valv["state"] = fp_enum_repository.FPValveStates(fp_valves.value & 32) self.valve_states_all += fp_valves[0] << 8 # Updating Valve_states_enum for i in range(0, fp_enum_repository.FPValveNames.NUM_OF_FP_VALVES.value): valve_name = fp_enum_repository.FPValveNames(i).name self.valve_states_enum[valve_name] = eval(f'self.{valve_name.lower()}["state"]') # Extract each valve sensed state start = len(msg_list) * DataTypes.U08.size() msg_list_2 = [] for valve_id in self.valves_sensed_states: msg_list.append((self.valves_sensed_states, valve_id, DataTypes.U08)) self.process_into_vars(decoder_list = msg_list_2, message = message, start_from_byte = start) self.fp_valves_states_timestamp = timestamp # ============================================================ Overrides and Requests ============================================================ def cmd_valve_broadcast_interval_override(self, ms: int, reset: int = NO_RESET) -> int: """ Constructs and sends the FP valve state override command. Constraints: Must be logged into FP. Given interval must be non-zero and a multiple of the FP general task interval (50 ms). @param ms: unsigned int - broadcast interval (in ms) @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_fp_ch_id, msg_id = MsgIds.MSG_ID_FP_VALVE_PUBLISH_INTERVAL_OVERRIDE_REQUEST, module_name = 'FP Valves', logger = self.logger, can_interface = self.can_interface) def cmd_valve_sensed_state_override(self, valve: int, state: bool, reset: int = NO_RESET) -> int: """ Constructs and sends the valve sensed state override command. Constraints: Must be logged into FP. Given valve ID must be one of the valve IDs listed below. @param valve: unsigned int - valve ID @param state: bool - valve state @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ rst = integer_to_bytearray(reset) ste = integer_to_bytearray(int(state)) vlv = integer_to_bytearray(valve) payload = rst + ste + vlv valve_name = fp_enum_repository.FPValveNames(valve).name.split('_')[0] return cmd_generic_override( payload = payload, reset = reset, channel_id = CanChannels.dialin_to_fp_ch_id, msg_id = MsgIds.MSG_ID_FP_VALVE_SENSED_STATE_OVERRIDE_REQUEST, entity_name = f'FP {valve_name} Valve sensed state', override_text = str(state), logger = self.logger, can_interface = self.can_interface) def cmd_valve_override(self, valve: int, state: int, reset: int = NO_RESET) -> int: """ Constructs and sends the FP valve state override command. Constraints: Must be logged into FP. Given valve ID must be one of the valve IDs listed below. @param valve: unsigned int - valve ID @param state: int - valve state @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ rst = integer_to_bytearray(reset) ste = integer_to_bytearray(int(state)) vlv = integer_to_bytearray(valve + BETA_IOFP_VALVE_OFFSET) payload = rst + ste + vlv valve_name = fp_enum_repository.FPValveNames(valve).name.split('_')[0] return cmd_generic_override( payload = payload, reset = reset, channel_id = CanChannels.dialin_to_fp_ch_id, msg_id = MsgIds.MSG_ID_FP_VALVE_CMD_STATE_OVERRIDE_REQUEST, entity_name = f'FP {valve_name} Valve state', override_text = str(state), logger = self.logger, can_interface = self.can_interface)