Index: dialin/common/msg_ids.py =================================================================== diff -u -r4a0c6600b52ec965a6625a428de986e7040ae2c9 -r9c740ce9dd2c7353eadc4d494f51a24157486ea7 --- dialin/common/msg_ids.py (.../msg_ids.py) (revision 4a0c6600b52ec965a6625a428de986e7040ae2c9) +++ dialin/common/msg_ids.py (.../msg_ids.py) (revision 9c740ce9dd2c7353eadc4d494f51a24157486ea7) @@ -370,6 +370,10 @@ MSG_ID_HD_DIAL_OUT_PUMP_HARD_STOP = 0x8090 MSG_ID_HD_BLOOD_PUMP_HARD_STOP = 0x8091 MSG_ID_HD_DIALIN_CHECK_IN = 0x8092 + MSG_ID_HD_SET_TEST_CONFIGURATION = 0x8093 + MSG_ID_HD_GET_TEST_CONFIGURATION = 0x8094 + MSG_ID_HD_SEND_TEST_CONFIGURATION = 0x8095 + MSG_ID_HD_RESET_ALL_TEST_CONFIGURATIONS = 0x8096 MSG_ID_DG_TESTER_LOGIN_REQUEST = 0xA000 MSG_ID_DG_ALARM_STATE_OVERRIDE = 0xA001 Index: dialin/common/test_config_defs.py =================================================================== diff -u -r4a0c6600b52ec965a6625a428de986e7040ae2c9 -r9c740ce9dd2c7353eadc4d494f51a24157486ea7 --- dialin/common/test_config_defs.py (.../test_config_defs.py) (revision 4a0c6600b52ec965a6625a428de986e7040ae2c9) +++ dialin/common/test_config_defs.py (.../test_config_defs.py) (revision 9c740ce9dd2c7353eadc4d494f51a24157486ea7) @@ -29,5 +29,6 @@ @unique class HDTestConfigOptions(DialinEnum): - pass - # TODo update + TEST_CONFIG_ENABLE_WET_CARTRIDGE_USE = 0 # Test configuration enable mixing with water. + # TEST_CONFIG_FIRST = TEST_CONFIG_ENABLE_MIXING_WITH_WATER, # Test configuration first configuration. + NUM_OF_TEST_CONFIGS = 1 Index: dialin/hd/hd_test_configs.py =================================================================== diff -u --- dialin/hd/hd_test_configs.py (revision 0) +++ dialin/hd/hd_test_configs.py (revision 9c740ce9dd2c7353eadc4d494f51a24157486ea7) @@ -0,0 +1,157 @@ + + +from logging import Logger +from ..common.msg_defs import MsgIds, MsgFieldPositions +from ..common.test_config_defs import HDTestConfigOptions +from ..protocols.CAN import DenaliMessage, DenaliChannels +from .constants import NO_RESET +from ..utils.conversions import integer_to_bytearray, bytearray_to_integer +from ..utils.base import AbstractSubSystem, publish + + +class HDTestConfig(AbstractSubSystem): + """ + + Hemodialysis (HD) 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.hd_test_configs = dict() + self.hd_test_configs_response_timestamp = 0.0 + + self._reset_test_configs_record() + + if self.can_interface is not None: + channel_id = DenaliChannels.dialin_to_hd_ch_id + msg_id = MsgIds.MSG_ID_HD_SEND_TEST_CONFIGURATION.value + self.can_interface.register_receiving_publication_function(channel_id, msg_id, + self._handler_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.hd_test_configs[HDTestConfigOptions(config).name] + + def cmd_set_test_config(self, config: int, reset: int = NO_RESET): + """ + Constructs and sends the HD test config + Constraints: + Must be logged into HD. + + @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_hd_ch_id, + message_id=MsgIds.MSG_ID_HD_SET_TEST_CONFIGURATION.value, + payload=payload) + + if reset == NO_RESET: + self.logger.debug("Setting {}".format(HDTestConfigOptions(config).name)) + else: + self.logger.debug("Resetting {}".format(HDTestConfigOptions(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 HD test configs request + Constraints: + Must be logged into HD. + + @return: 1 if successful, zero otherwise + """ + message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_hd_ch_id, + message_id=MsgIds.MSG_ID_HD_GET_TEST_CONFIGURATION.value) + + self.logger.debug('Getting HD test configuration record') + + received_message = self.can_interface.send(message) + + # If there is content... + if received_message is not None: + self._reset_test_configs_record() + self.logger.debug("Received FW ACK after requesting HD 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 HD test configs reset all + Constraints: + Must be logged into HD. + + @return: 1 if successful, zero otherwise + """ + message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_hd_ch_id, + message_id=MsgIds.MSG_ID_HD_RESET_ALL_TEST_CONFIGURATIONS.value) + + self.logger.debug("Resetting all HD 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(['hd_test_configs']) + def _handler_test_config_sync(self, message, timestamp=0.0): + """ + Handles published test configuration status messages. + + @param message: published HD test configurations message + @return: None + """ + payload = message['message'] + index = MsgFieldPositions.START_POS_FIELD_1 + + for config in HDTestConfigOptions.__members__: + if 'NUM_OF_TEST_CONFIGS' not in config: + config_value, index = bytearray_to_integer(payload, index, False) + self.hd_test_configs[config] = config_value + + self.hd_test_configs_response_timestamp = timestamp + + def _reset_test_configs_record(self): + """ + Resets the test configuration dictionary + + @return: None + """ + for config in HDTestConfigOptions.__members__: + # Loop through the list of the test configuration and set the values to 0xFFFFFFFF + if 'NUM_OF_TEST_CONFIGS' not in config: + self.hd_test_configs[config] = 0xFFFFFFFF