Index: leahi_dialin/common/msg_ids.py =================================================================== diff -u -rf0d5248dca73c374c16f48fd3da474cf3b2286f6 -rac344e50f31d776075a64f775b38244531f880d6 --- leahi_dialin/common/msg_ids.py (.../msg_ids.py) (revision f0d5248dca73c374c16f48fd3da474cf3b2286f6) +++ leahi_dialin/common/msg_ids.py (.../msg_ids.py) (revision ac344e50f31d776075a64f775b38244531f880d6) @@ -177,6 +177,9 @@ MSG_ID_TD_BLOOD_FLOW_ALPHA_Y_INTERCEPT_OVERRIDE_REQUEST = 0x8037 MSG_ID_TD_BLOOD_FLOW_WEAR_A_TERM_OVERRIDE_REQUEST = 0x8038 MSG_ID_TD_BLOOD_FLOW_WEAR_B_TERM_OVERRIDE_REQUEST = 0x8039 + MSG_ID_TD_SET_TEST_CONFIGURATION = 0x803A + MSG_ID_TD_GET_TEST_CONFIGURATION = 0x803B + MSG_ID_TD_RESET_ALL_TEST_CONFIGURATIONS = 0x803C MSG_ID_DD_TESTER_LOGIN_REQUEST = 0xA000 MSG_ID_DD_SOFTWARE_RESET_REQUEST = 0xA001 Index: leahi_dialin/common/test_config_defs.py =================================================================== diff -u -rdb4edbea32c4cf60d80710bd0eb56728706ac8c8 -rac344e50f31d776075a64f775b38244531f880d6 --- leahi_dialin/common/test_config_defs.py (.../test_config_defs.py) (revision db4edbea32c4cf60d80710bd0eb56728706ac8c8) +++ leahi_dialin/common/test_config_defs.py (.../test_config_defs.py) (revision ac344e50f31d776075a64f775b38244531f880d6) @@ -30,3 +30,8 @@ TEST_CONFIG_BETA_HW = 0 # Test configuration using Beta Hardware NUM_OF_TEST_CONFIGS = 1 # Number of Test Configs +@unique +class TDTestConfigOptions(DialinEnum): + TEST_CONFIG_FIRST = 0 # First test configuration (placeholder) + NUM_OF_TEST_CONFIGS = 1 # Number of Test Configs + Index: leahi_dialin/td/modules/td_test_configs.py =================================================================== diff -u --- leahi_dialin/td/modules/td_test_configs.py (revision 0) +++ leahi_dialin/td/modules/td_test_configs.py (revision ac344e50f31d776075a64f775b38244531f880d6) @@ -0,0 +1,176 @@ +########################################################################### +# +# 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 td_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 TDTestConfigOptions +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 TDTestConfig(AbstractSubSystem): + """ + + TD 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.td_test_configs = dict() + self.td_test_configs_response_timestamp = 0.0 + + self._reset_test_configs_record() + + if self.can_interface is not None: + channel_id = DenaliChannels.td_to_dialin_ch_id + msg_id = MsgIds.MSG_ID_TD_SEND_TEST_CONFIGURATION.value + self.can_interface.register_receiving_publication_function(channel_id, msg_id, + self._handler_td_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.td_test_configs[TDTestConfigOptions(config).name] + + + def cmd_set_test_config(self, config: int, reset: int = NO_RESET): + """ + Constructs and sends the TD test config + Constraints: + Must be logged into TD. + + @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_td_ch_id, + message_id=MsgIds.MSG_ID_TD_SET_TEST_CONFIGURATION.value, + payload=payload) + + if reset == NO_RESET: + self.logger.debug("Setting {}".format(TDTestConfigOptions(config).name)) + else: + self.logger.debug("Resetting {}".format(TDTestConfigOptions(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 TD test configs request + Constraints: + Must be logged into TD. + + @return: 1 if successful, zero otherwise + """ + message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_td_ch_id, + message_id=MsgIds.MSG_ID_TD_GET_TEST_CONFIGURATION.value) + + self.logger.debug('Getting TD 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 TD 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 TD test configs reset all + Constraints: + Must be logged into TD. + + @return: 1 if successful, zero otherwise + """ + message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_td_ch_id, + message_id=MsgIds.MSG_ID_TD_RESET_ALL_TEST_CONFIGURATIONS.value) + + self.logger.debug("Resetting all TD 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(['td_test_configs']) + def _handler_td_test_config_sync(self, message, timestamp=0.0): + """ + Handles published test configuration status messages. + + @param message: published TD test configurations message + @return: None + """ + payload = message['message'] + index = MsgFieldPositions.START_POS_FIELD_1 + + for config in TDTestConfigOptions.__members__: + if 'NUM_OF_TEST_CONFIGS' not in config: + config_value, index = bytearray_to_integer(payload, index, False) + self.td_test_configs[config] = config_value + + self.td_test_configs_response_timestamp = timestamp + + def _reset_test_configs_record(self): + """ + Resets the test configuration dictionary + + @return: None + """ + for config in TDTestConfigOptions.__members__: + # Loop through the list of the test configuration and set the values to 0xFFFFFFFF + if 'NUM_OF_TEST_CONFIGS' not in config: + self.td_test_configs[config] = 0xFFFFFFFF \ No newline at end of file