Index: leahi_dialin/common/msg_ids.py =================================================================== diff -u -re5772cd214690a61daa26c1c3c3ea9df7fd17ef5 -r22454fbcfdd5607b8138cd7412acfdbcdf3384b9 --- leahi_dialin/common/msg_ids.py (.../msg_ids.py) (revision e5772cd214690a61daa26c1c3c3ea9df7fd17ef5) +++ leahi_dialin/common/msg_ids.py (.../msg_ids.py) (revision 22454fbcfdd5607b8138cd7412acfdbcdf3384b9) @@ -244,9 +244,12 @@ MSG_ID_DD_STOP_PRE_GEN_DIALYSATE_MODE_OVERRIDE_REQUEST = 0xA040 MSG_ID_DD_STOP_GEN_DIALYSATE_MODE_OVERRIDE_REQUEST = 0xA041 MSG_ID_DD_SAFETY_SHUTDOWN_OVERRIDE_REQUEST = 0xA042 + # TODO need to align DD message IDs MSG_ID_DD_ALARM_STATE_OVERRIDE_REQUEST = 0xA043 # MSG_ID_DD_SEND_TEST_CONFIGURATION = 0xA002 - # TODO: need to add set, get, reset test configuration message IDs + MSG_ID_DD_SET_TEST_CONFIGURATION = 0xA043 + MSG_ID_DD_GET_TEST_CONFIGURATION = 0XA044 + MSG_ID_DD_RESET_ALL_TEST_CONFIGURATIONS = 0XA045 MSG_ID_DD_PISTON_PUMP_DATA_PUBLISH_OVERRIDE_REQUEST = 0xAF00 MSG_ID_DD_PISTON_PUMP_START_STOP_OVERRIDE_REQUEST = 0xAF01 Index: leahi_dialin/common/test_config_defs.py =================================================================== diff -u -re5772cd214690a61daa26c1c3c3ea9df7fd17ef5 -r22454fbcfdd5607b8138cd7412acfdbcdf3384b9 --- leahi_dialin/common/test_config_defs.py (.../test_config_defs.py) (revision e5772cd214690a61daa26c1c3c3ea9df7fd17ef5) +++ leahi_dialin/common/test_config_defs.py (.../test_config_defs.py) (revision 22454fbcfdd5607b8138cd7412acfdbcdf3384b9) @@ -20,6 +20,15 @@ @unique class DDTestConfigOptions(DialinEnum): - TEST_CONFIG_BETA_HW = 0 # Test configuration to denote Beta Hardware - TEST_CONFIG_DIENER_CONC_PUMP = 1 # Test configuration to use the Diener pump - TEST_CONFIG_BC_PRES_ALARM = 2 # Test configuration to enable balancing chamber pressure alarm + TEST_CONFIG_BETA_HW = 0 # Test configuration using Beta Hardware + TEST_CONFIG_FIRST = TEST_CONFIG_BETA_HW # First Test Configuration + TEST_CONFIG_DISABLE_BC_PRES_ALARM = 1 # Test configuration to disable inlet water temperature check + TEST_CONFIG_DIENER_CONC_PUMP = 2 # Test configuration recover treatment + NUM_OF_TEST_CONFIGS = 3 # Number of Test Configs + +@unique +class FPTestConfigOptions(DialinEnum): + TEST_CONFIG_BETA_HW = 0 # Test configuration using Beta Hardware + TEST_CONFIG_FIRST = TEST_CONFIG_BETA_HW # First Test Configuration + NUM_OF_TEST_CONFIGS = 1 # Number of Test Configs + Index: leahi_dialin/fp/modules/fp_test_configs.py =================================================================== diff -u --- leahi_dialin/fp/modules/fp_test_configs.py (revision 0) +++ leahi_dialin/fp/modules/fp_test_configs.py (revision 22454fbcfdd5607b8138cd7412acfdbcdf3384b9) @@ -0,0 +1,200 @@ +########################################################################### +# +# Copyright (c) 2020-2025 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 fp_test_configs.py +# +# @author (last) Jonny Paguio +# @date (last) 22-Aug-2025 +# @author (original) Jonny Paguio +# @date (original) 20-Aug-2025 +# +############################################################################ + +from logging import Logger + +from leahi_dialin.common.msg_defs import MsgIds, MsgFieldPositions +from leahi_dialin.common.test_config_defs import FPTestConfigOptions +from leahi_dialin.protocols.CAN import DenaliMessage, DenaliChannels +from .constants import NO_RESET +from leahi_dialin.utils.conversions import integer_to_bytearray, bytearray_to_integer +from leahi_dialin.utils.base import AbstractSubSystem, publish + + +class FPTestConfig(AbstractSubSystem): + """ + + FP Dialin API sub-class for setting and getting the test configurations. + """ + + def __init__(self, can_interface, logger: Logger): + """ + + @param can_interface: Denali CAN Messenger object + """ + super().__init__() + + self.can_interface = can_interface + self.logger = logger + self.fp_test_configs = dict() + self.fp_test_configs_response_timestamp = 0.0 + + self._reset_test_configs_record() + + if self.can_interface is not None: + channel_id = DenaliChannels.fp_to_dialin_ch_id + msg_id = MsgIds.MSG_ID_FP_SEND_TEST_CONFIGURATION.value + self.can_interface.register_receiving_publication_function(channel_id, msg_id, + self._handler_fp_test_config_sync) + + def cmd_get_test_config_status(self, config: int): + """ + Returns the status of a test config + + @param config: (int) Test config to set + @return: the status of a test config + """ + return self.fp_test_configs[FPTestConfigOptions(config).name] + # TODO + def cmd_set_recover_from_mode_fault_signal(self): + """ + Constructs and sends the FP test config the signal to recover from mode fault + Constraints: + Must be logged into FP. + + @return: 1 if successful, zero otherwise + """ + + message = DenaliMessage.build_message(channel_id=DenaliChannels.fp_to_dialin_ch_id, + message_id=MsgIds.MSG_ID_FP_SIGNAL_RECOVER_FROM_FAULT_MODE.value) + + self.logger.debug("Setting signal to recover from mode fault") + + # Send message + received_message = self.can_interface.send(message) + + # If there is no content... + if received_message is not None: + # response payload is OK or not OK + return received_message['message'][DenaliMessage.PAYLOAD_START_INDEX] + else: + self.logger.debug("Timeout!!!!") + return False + # TODO check if FP uses own test config messages or shares with DD + def cmd_set_test_config(self, config: int, reset: int = NO_RESET): + """ + Constructs and sends the FP test config + Constraints: + Must be logged into FP. + + @param config: (int) Test config to set + @param reset: (int) 1 to reset a previous override, 0 to override + @return: 1 if successful, zero otherwise + """ + reset_value = integer_to_bytearray(reset) + c = integer_to_bytearray(config) + payload = reset_value + c + + message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_fp_ch_id, + message_id=MsgIds.MSG_ID_FP_SET_TEST_CONFIGURATION.value, + payload=payload) + + if reset == NO_RESET: + self.logger.debug("Setting {}".format(FPTestConfigOptions(config).name)) + else: + self.logger.debug("Resetting {}".format(FPTestConfigOptions(config).name)) + + # Send message + received_message = self.can_interface.send(message) + + # If there is no content... + if received_message is not None: + # response payload is OK or not OK + return received_message['message'][DenaliMessage.PAYLOAD_START_INDEX] + else: + self.logger.debug("Timeout!!!!") + return False + + def cmd_request_test_config_status_from_fw(self): + """ + Constructs and sends the FP test configs request + Constraints: + Must be logged into FP. + + @return: 1 if successful, zero otherwise + """ + message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_fp_ch_id, + message_id=MsgIds.MSG_ID_FP_GET_TEST_CONFIGURATION.value) + + self.logger.debug('Getting FP test configuration record') + # Reset the test configs regardless of whether the message has been acknowledged or not. The reset might be out + # sync and reset the test configuration while the latest data has been received. If the test configuration is + # reset in Dialin but the message was not acknowledged, the user shall send the request again + self._reset_test_configs_record() + + received_message = self.can_interface.send(message) + + # If there is content... + if received_message is not None: + self.logger.debug("Received FW ACK after requesting FP test configuration record.") + # response payload is OK or not OK + return received_message['message'][DenaliMessage.PAYLOAD_START_INDEX] + else: + self.logger.debug("Timeout!!!!") + return False + + def cmd_reset_all_test_configs(self): + """ + Constructs and sends the FP test configs reset all + Constraints: + Must be logged into FP. + + @return: 1 if successful, zero otherwise + """ + message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_fp_ch_id, + message_id=MsgIds.MSG_ID_FP_RESET_ALL_TEST_CONFIGURATIONS.value) + + self.logger.debug("Resetting all FP test configurations") + + # Send message + received_message = self.can_interface.send(message) + + # If there is no content... + if received_message is not None: + # response payload is OK or not OK + return received_message['message'][DenaliMessage.PAYLOAD_START_INDEX] + else: + self.logger.debug("Timeout!!!!") + return False + + @publish(['fp_test_configs']) + def _handler_fp_test_config_sync(self, message, timestamp=0.0): + """ + Handles published test configuration status messages. + + @param message: published FP test configurations message + @return: None + """ + payload = message['message'] + index = MsgFieldPositions.START_POS_FIELD_1 + + for config in FPTestConfigOptions.__members__: + if 'NUM_OF_TEST_CONFIGS' not in config: + config_value, index = bytearray_to_integer(payload, index, False) + self.fp_test_configs[config] = config_value + + self.fp_test_configs_response_timestamp = timestamp + + def _reset_test_configs_record(self): + """ + Resets the test configuration dictionary + + @return: None + """ + for config in FPTestConfigOptions.__members__: + # Loop through the list of the test configuration and set the values to 0xFFFFFFFF + if 'NUM_OF_TEST_CONFIGS' not in config: + self.fp_test_configs[config] = 0xFFFFFFFF \ No newline at end of file