Index: leahi_dialin/td/modules/records.py =================================================================== diff -u -r28a21f1f2ca4ef02e6d54bae30b738fddb7bdb3a -r489ce5ef49041b863501b7fecd6e4716e7dcaa9f --- leahi_dialin/td/modules/records.py (.../records.py) (revision 28a21f1f2ca4ef02e6d54bae30b738fddb7bdb3a) +++ leahi_dialin/td/modules/records.py (.../records.py) (revision 489ce5ef49041b863501b7fecd6e4716e7dcaa9f) @@ -8,708 +8,421 @@ # @file records.py # # @author (last) Zoltan Miskolci -# @date (last) 22-Jun-2026 +# @date (last) 04-Aug-2026 # @author (original) Zoltan Miskolci -# @date (original) 22-Jun-2026 +# @date (original) 04-May-2026 # ############################################################################ # Module imports from logging import Logger -from functools import partial -from time import sleep # Project imports +from leahi_dialin.dd.modules.records_get_responses import DDRecordsGetHandlers +from leahi_dialin.dd.modules.records_get_requests import DDRecordsGetRequests +from leahi_dialin.dd.modules.records_set_responses import DDRecordsSetHandlers +from leahi_dialin.dd.modules.records_set_requests import DDRecordsSetRequests from leahi_dialin.common import td_enum_repository -from leahi_dialin.common.generic_defs import DataTypes +from leahi_dialin.common.msg_defs import RequestRejectReasons from leahi_dialin.common.msg_ids import MsgIds from leahi_dialin.common.override_templates import cmd_generic_override from leahi_dialin.protocols.CAN import CanMessenger, CanChannels -from leahi_dialin.utils.abstract_classes import AbstractSubSystem -from leahi_dialin.utils.base import publish -from leahi_dialin.utils.enums import DialinEnum -from leahi_dialin.utils.conversions import integer_to_bytearray, float_to_bytearray, byte_to_bytearray, unsigned_short_to_bytearray, string_to_bytearray +from leahi_dialin.utils.conversions import integer_to_bytearray -class TDRecords(AbstractSubSystem): +class TDRecords(DDRecordsGetHandlers, DDRecordsGetRequests, DDRecordsSetHandlers, DDRecordsSetRequests): """ - TD interface containing pressure related commands. + TD interface containing records related commands. """ def __init__(self, can_interface: CanMessenger, logger: Logger): """ - @param can_interface: The CanMessenger object + The main record class to handle all the subclasses. + + :param can_interface: (CanMessenger) The CanMessenger object + :param logger: (Logger) The Logger 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.td_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_TD_NVM_SYSTEM_RECORD_RESPONSE.value, - function = self._handler_system_record_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.td_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_TD_NVM_SERVICE_RECORD_RESPONSE.value, - function = self._handler_service_record_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.td_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_TD_NVM_CAL_RESPONSE.value, - function = self._handler_calibration_record_pres_sensor_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.td_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_TD_NVM_INSTITUTIONAL_RECORD_RESPONSE.value, - function = self._handler_institutional_record_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.td_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_TD_NVM_USAGE_INFO_RECORD_RESPONSE.value, - function = self._handler_usage_info_record_sync) + self.get_system_rec_timestamp = 0 #: The timestamp of the latest System Records message + self.get_system_rec_ack = False + self.get_system_rec_rr = RequestRejectReasons.REQUEST_REJECT_REASON_NONE - self.system_records_timestamp = 0 #: The timestamp of the latest System Records message - self.service_records_timestamp = 0 #: The timestamp of the latest Service Records message - self.calibration_records_timestamp = 0 #: The timestamp of the latest Calibration Records message - self.institutional_records_timestamp = 0 #: The timestamp of the latest Institutional Records message - self.usage_info_records_timestamp = 0 #: The timestamp of the latest Usage Information Records message + self.system_records = { } + + self.set_system_rec_rr_timestamp = 0 #: The timestamp of the latest Set System Reject Reason message + self.set_system_rec_ack = False + self.set_system_rec_rr = { } - self.system_records = { } #: The System Records data in dictionary format - self.service_records = { } #: The Service Records data in dictionary format - self.calibration_records = { } #: The Calibration Records data in dictionary format - self.institutional_records = { } #: The Institutional Records data in dictionary format - self.usage_info_records = { } #: The Usage Information Records data in dictionary format + self.get_service_rec_timestamp = 0 #: The timestamp of the latest Service Records message + self.get_service_rec_ack = False + self.get_service_rec_rr = RequestRejectReasons.REQUEST_REJECT_REASON_NONE - for sys_record in td_enum_repository.SystemRecordFields: - self.system_records[sys_record.name] = None + self.service_records = { } - for serv_record in td_enum_repository.ServiceRecordFields: - self.service_records[serv_record.name] = None + self.set_service_rec_rr_timestamp = 0 #: The timestamp of the latest Set Service Reject Reason message + self.set_service_rec_ack = False + self.set_service_rec_rr = { } - for cal_record in td_enum_repository.CalibrationRecordFields: - self.calibration_records[cal_record.name] = None + self.get_calibration_rec_timestamp = 0 #: The timestamp of the latest Calibration Records message + self.get_calibration_rec_ack = { } + self.get_calibration_rec_rr = { } - for inst_record in td_enum_repository.InstitutionalRecordFields: - self.institutional_records[inst_record.name] = None + self.calibration_records = { } - for usage_record in td_enum_repository.UsageInformationRecordFields: - self.usage_info_records[usage_record.name] = None + self.set_calibration_rec_rr_timestamp = 0 #: The timestamp of the latest Set Calibration Reject Reason message + self.set_calibration_rec_ack = { } + self.set_calibration_rec_rr = { } + self.get_institutional_rec_timestamp = 0 #: The timestamp of the latest Institutional Records message + self.get_institutional_rec_ack = { } + self.get_institutional_rec_rr = { } -# ================================================= CAN Message Handler Methods ================================================= - @publish(["msg_id_td_nvm_send_system_record", "system_records", "system_records_timestamp"]) - def _handler_system_record_sync(self, message, timestamp = 0.0): - """ - Handles published TD System Record data messages. TD System Records are captured - for reference. + self.institutional_records = { } - @param message: published data message - @return: none - """ - record_list = [] - for member in td_enum_repository.SystemRecordFields: - record_list.append((member.name, member.datatype(), member.multichar_length())) + self.set_institutional_rec_rr_timestamp = 0 #: The timestamp of the latest Set Institutional Reject Reason message + self.set_institutional_rec_ack = False + self.set_institutional_rec_rr = { } - self.process_into_dict(dict_to_update = self.system_records, - decoder_list = record_list, - message = message) + self.get_usage_info_rec_timestamp = 0 #: The timestamp of the latest Usage Information Records message + self.get_usage_info_rec_ack = False + self.get_usage_info_rec_rr = RequestRejectReasons.REQUEST_REJECT_REASON_NONE - self.system_records_timestamp = timestamp + self.usage_info_records = { } + self.set_usage_info_rec_rr_timestamp = 0 #: The timestamp of the latest Set Usage Information Reject Reason message + self.set_usage_info_rec_ack = False + self.set_usage_info_rec_rr = { } - @publish(["msg_id_td_nvm_send_service_record", "service_records", "service_records_timestamp"]) - def _handler_service_record_sync(self, message, timestamp = 0.0): - """ - Handles published TD Service Record data messages. TD Service Records are captured - for reference. + # Calling the Sub classes: TDRecordsGetHandlers, TDRecordsGetRequests, TDRecordsSetHandlers, TDRecordsSetRequests + # Doing it after initializeing the parameters will make those parameters accessable to the sub classes + super().__init__() - @param message: published data message - @return: none - """ - record_list = [] - for member in td_enum_repository.ServiceRecordFields: - record_list.append((member.name, member.datatype())) - self.process_into_dict(dict_to_update = self.service_records, - decoder_list = record_list, - message = message) - self.service_records_timestamp = timestamp +# ============================================================ Properties ============================================================ + @property + def system_records(self) -> dict: + """ + The System Records data + """ + return self._system_records + @system_records.setter + def system_records(self, value): + self._system_records = value - @publish(["msg_id_dd_nvm_cal_pressure_sensor_response", "calibration_records", "calibration_records_timestamp"]) - def _handler_calibration_record_pres_sensor_sync(self, message, timestamp = 0.0): - """ - Handles published TD Calibration Record Pressure Sensor Details data messages. - TD Calibration Records Pressure Sensor Details are captured for reference. - @param message: published data message - @return: none + @property + def service_records(self) -> dict: """ - # Get the sensor_id to identify the location of the payload - msg_list = [('sensor_id', DataTypes.U08)] - result = self.process_into_vars(decoder_list = msg_list, - message = message) - sensor_name = td_enum_repository.TDPressureSensorNames(result['sensor_id']).name + The Service Records data + """ + return self._service_records - record_list = [] - for member in td_enum_repository.CalibRecordSensorFields: - record_list.append((self.CALIB_RECORDS_PRESSURE_SENSORS, sensor_name, member.name, member.datatype())) + @service_records.setter + def service_records(self, value): + self._service_records = value - self.process_into_dict(dict_to_update = self.calibration_records, - decoder_list = record_list, - message = message, - start_from_byte = len(msg_list) * msg_list[0][1].size()) - self.calibration_records_timestamp = timestamp + @property + def calibration_records(self) -> dict: + """ + The Calibration Records data + """ + return self._calibration_records + @calibration_records.setter + def calibration_records(self, value): + self._calibration_records = value - @publish(["msg_id_td_nvm_send_institutional_record", "institutional_records", "institutional_records_timestamp"]) - def _handler_institutional_record_sync(self, message, timestamp = 0.0): - """ - Handles published TD Institutional Record data messages. TD Institutional Records are captured - for reference. - @param message: published data message - @return: none + @property + def institutional_records(self) -> dict: """ - record_list = [('record_id', DataTypes.U08)] - results = self.process_into_vars(decoder_list = record_list, - message = message) + The Institutional Records data + """ + return self._institutional_records - record_id = int(results['record_id']) - member = td_enum_repository.InstitutionalRecordFields(record_id) - - record_list = [] - record_list.append((member.name, member.datatype())) + @institutional_records.setter + def institutional_records(self, value): + self._institutional_records = value - self.process_into_dict(dict_to_update = self.institutional_records, - decoder_list = record_list, - message = message, - start_from_byte = len(record_list) * DataTypes.U08.size()) - self.institutional_records_timestamp = timestamp + @property + def usage_info_records(self) -> dict: + """ + The Usage Information Records data + """ + return self._usage_info_records + @usage_info_records.setter + def usage_info_records(self, value): + self._usage_info_records = value - @publish(["msg_id_td_nvm_send_usage_info_record", "usage_info_records", "usage_info_records_timestamp"]) - def _handler_usage_info_record_sync(self, message, timestamp = 0.0): - """ - Handles published TD Usage Information Record data messages. TD Usage Information Records are captured - for reference. - @param message: published data message - @return: none + @property + def get_system_rec_ack(self) -> bool: """ - record_list = [] - for member in td_enum_repository.UsageInformationRecordFields: - record_list.append((member.name, member.datatype())) + The Get System Records message acceptance + """ + return self._get_system_rec_ack - self.process_into_dict(dict_to_update = self.service_records, - decoder_list = record_list, - message = message) + @get_system_rec_ack.setter + def get_system_rec_ack(self, value): + self._get_system_rec_ack = value - self.usage_info_records_timestamp = timestamp - -# ================================================= Go to Service Mode Method ================================================= - def cmd_initiate_service_mode(self) -> int: + @property + def get_service_rec_ack(self) -> bool: """ - Constructs and sends a request to change to Service operation mode via CAN bus. - Constraints: - Must be logged into TD. - Transition from current to requested op mode must be legal. - - @return: 1 if successful, zero otherwise + The Get Service Records message acceptance """ - service_mode = td_enum_repository.TDOpModes.MODE_SERV + return self._get_service_rec_ack - payload = integer_to_bytearray(service_mode.value) + @get_service_rec_ack.setter + def get_service_rec_ack(self, value): + self._get_service_rec_ack = value - return cmd_generic_override( - payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_td_ch_id, - msg_id = MsgIds.MSG_ID_TD_SET_OPERATION_MODE_OVERRIDE_REQUEST, - entity_name = 'TD Operation Mode', - override_text = f'set to {service_mode.name}', - logger = self.logger, - can_interface = self.can_interface) + @property + def get_calibration_rec_ack(self) -> dict: + """ + The Get Calibration Records message acceptance + """ + return self._get_calibration_rec_ack + @get_calibration_rec_ack.setter + def get_calibration_rec_ack(self, value): + self._get_calibration_rec_ack = value -# ================================================= Request Records Methods ================================================= - def cmd_request_system_records(self) -> int: - """ - Constructs and sends a request for System Records. - Constraints: - Must be logged into TD. - Must be in Service mode. - @return: 1 if successful, zero otherwise + @property + def get_institutional_rec_ack(self) -> dict: """ - payload = byte_to_bytearray(td_enum_repository.RecordTypes.SYSTEM_RECORD.value) + The Get Institutional Records message acceptance + """ + return self._get_institutional_rec_ack - return cmd_generic_override( - payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_td_ch_id, - msg_id = MsgIds.MSG_ID_UI_TD_NVM_GET_RECORD_REQUEST, - entity_name = f'TD System Record request', - override_text = '', - logger = self.logger, - can_interface = self.can_interface) + @get_institutional_rec_ack.setter + def get_institutional_rec_ack(self, value): + self._get_institutional_rec_ack = value - def cmd_request_service_records(self) -> int: + @property + def get_usage_info_rec_ack(self) -> bool: """ - Constructs and sends a request for Service Records. - Constraints: - Must be logged into TD. - Must be in Service mode. - - @return: 1 if successful, zero otherwise + The Get Usage Information Records message acceptance """ - payload = byte_to_bytearray(td_enum_repository.RecordTypes.SERVICE_RECORD.value) + return self._get_usage_info_rec_ack - return cmd_generic_override( - payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_td_ch_id, - msg_id = MsgIds.MSG_ID_UI_TD_NVM_GET_RECORD_REQUEST, - entity_name = f'TD Service Record request', - override_text = '', - logger = self.logger, - can_interface = self.can_interface) + @get_usage_info_rec_ack.setter + def get_usage_info_rec_ack(self, value): + self._get_usage_info_rec_ack = value - def cmd_request_calibration_records(self) -> int: + @property + def get_system_rec_rr(self) -> dict: """ - Constructs and sends a request for Calibration Records. - Constraints: - Must be logged into TD. - Must be in Service mode. - - @return: 1 if successful, zero otherwise + The Get System Records Request reject reason data """ - payload = byte_to_bytearray(td_enum_repository.RecordTypes.CALIBRATION_RECORD.value) + return self._get_system_rec_rr - return cmd_generic_override( - payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_td_ch_id, - msg_id = MsgIds.MSG_ID_UI_TD_NVM_GET_RECORD_REQUEST, - entity_name = f'TD Calibration Record request', - override_text = '', - logger = self.logger, - can_interface = self.can_interface) + @get_system_rec_rr.setter + def get_system_rec_rr(self, value): + self._get_system_rec_rr = value - def cmd_request_institutional_records(self) -> int: + @property + def get_service_rec_rr(self) -> dict: """ - Constructs and sends a request for Institutional Records. - Constraints: - Must be logged into TD. - Must be in Service mode. - - @return: 1 if successful, zero otherwise + The Get Service Records Request reject reason data """ - payload = byte_to_bytearray(td_enum_repository.RecordTypes.INSTITUTIONAL_RECORD.value) + return self._get_service_rec_rr - return cmd_generic_override( - payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_td_ch_id, - msg_id = MsgIds.MSG_ID_UI_TD_NVM_GET_RECORD_REQUEST, - entity_name = f'TD Institutional Record request', - override_text = '', - logger = self.logger, - can_interface = self.can_interface) + @get_service_rec_rr.setter + def get_service_rec_rr(self, value): + self._get_service_rec_rr = value - def cmd_request_usage_information_records(self) -> int: + @property + def get_calibration_rec_rr(self) -> dict: """ - Constructs and sends a request for Usage Information Records. - Constraints: - Must be logged into TD. - Must be in Service mode. - - @return: 1 if successful, zero otherwise + The Get Calibration Records Request reject reason data """ - payload = byte_to_bytearray(td_enum_repository.RecordTypes.USAGE_INFORMATION_RECORD.value) + return self._get_calibration_rec_rr - return cmd_generic_override( - payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_td_ch_id, - msg_id = MsgIds.MSG_ID_UI_TD_NVM_GET_RECORD_REQUEST, - entity_name = f'TD Usage Information Record request', - override_text = '', - logger = self.logger, - can_interface = self.can_interface) + @get_calibration_rec_rr.setter + def get_calibration_rec_rr(self, value): + self._get_calibration_rec_rr = value - -# ================================================= Set Records Main Methods ================================================= - def cmd_set_system_records(self, - is_ro_featured: bool=False, - is_ro_featured_boost_pump: bool=False, - part_number: str='0000000000', - serial_number: str='00000000000000000000', - manufacturing_location: int=0, - manufacturing_date: int=0) -> int: + @property + def get_institutional_rec_rr(self) -> dict: """ - Constructs and sends a command for setting the System Records. - Constraints: - Must be logged into TD. - Must be in Service mode. - - @return: 1 if successful, zero otherwise + The Get Institutional Records Request reject reason data """ - if len(part_number) != 10: - raise ValueError("part_number must be 10 characters") - if len(serial_number) != 10: - raise ValueError("serial must be 20 characters") - - payload = integer_to_bytearray(is_ro_featured) - payload += integer_to_bytearray(is_ro_featured_boost_pump) - payload += string_to_bytearray(part_number) - payload += string_to_bytearray(serial_number) - payload += byte_to_bytearray(manufacturing_location) - payload += integer_to_bytearray(manufacturing_date) - payload += unsigned_short_to_bytearray(self.crc16(payload)) + return self._get_institutional_rec_rr - return cmd_generic_override( - payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_td_ch_id, - msg_id = MsgIds.MSG_ID_UI_TD_NVM_SET_SYSTEM_RECORD, - entity_name = f'New TD System Record', - override_text = 'being set', - logger = self.logger, - can_interface = self.can_interface) + @get_institutional_rec_rr.setter + def get_institutional_rec_rr(self, value): + self._get_institutional_rec_rr = value - def cmd_set_service_records(self, - is_hdf_online: bool=False, - water_recovery: int=0, - service_loc: int=0, - last_service_date: int=0, - service_interval_sec: int=0, - last_reset_time: int=0) -> int: + @property + def get_usage_info_rec_rr(self) -> dict: """ - Constructs and sends a command for setting the Service Records. - Constraints: - Must be logged into TD. - Must be in Service mode. - - @return: 1 if successful, zero otherwise + The Get Usage Information Records Request reject reason data """ - payload = integer_to_bytearray(is_hdf_online) - payload += integer_to_bytearray(water_recovery) - payload += byte_to_bytearray(service_loc) - payload += integer_to_bytearray(last_service_date) - payload += integer_to_bytearray(service_interval_sec) - payload += integer_to_bytearray(last_reset_time) - payload += unsigned_short_to_bytearray(self.crc16(payload)) + return self._get_usage_info_rec_rr - return cmd_generic_override( - payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_td_ch_id, - msg_id = MsgIds.MSG_ID_TD_NVM_SET_SERVICE_RECORD, - entity_name = f'New TD Service Record', - override_text = 'being set', - logger = self.logger, - can_interface = self.can_interface) + @get_usage_info_rec_rr.setter + def get_usage_info_rec_rr(self, value): + self._get_usage_info_rec_rr = value - def cmd_set_calibration_records(self, calibration_records: dict) -> int: + @property + def set_system_rec_ack(self) -> bool: """ - Constructs and sends a command for setting the Calibration Records. - Constraints: - Must be logged into TD. - Must be in Service mode. - - @param calibration_records: (Dictionary) The stored and structured Calibration Records - @return: 1 if successful, zero otherwise + The Set System Records message acceptance """ - msg_id_pairing = { - 'PRES': MsgIds.MSG_ID_UI_TD_NVM_SET_CAL_PRESSURE_SENSOR, - 'TEMP': MsgIds.MSG_ID_UI_TD_NVM_SET_CAL_TEMP_SENSOR, - 'ACCEL': MsgIds.MSG_ID_UI_TD_NVM_SET_CAL_ACCEL_SENSOR, - } - send_data = [] - checked = [] - for record_enum in td_enum_repository.CalibrationRecordFields: - record_parts = record_enum.name.split('__') - group = record_parts[0] + return self._set_system_rec_ack - # If a message already sent, then just go to the next value - if 'NUM_' in group or \ - group in checked or \ - record_parts[1] in checked: - continue + @set_system_rec_ack.setter + def set_system_rec_ack(self, value): + self._set_system_rec_ack = value - if group in ['PRES', 'TEMP']: - # Set the sensor name - if group == 'PRES': - sensor = td_enum_repository.DDPressureSensorNames.from_str(record_parts[1]) - elif group == 'TEMP': - sensor = td_enum_repository.DDTemperatureSensorNames.from_str(record_parts[1]) - else: - sensor = 'UNKNOWN' - # Set the arguments for the call - args = [] - kwargs = {} - fourth_order_coeff_entry_name = f'{group}__{sensor.name}__{td_enum_repository.CalibRecordSensorFields.FOURTH_ORDER_COEFF.name}' - third_order_coeff_entry_name = f'{group}__{sensor.name}__{td_enum_repository.CalibRecordSensorFields.THIRD_ORDER_COEFF.name}' - second_order_coeff_entry_name = f'{group}__{sensor.name}__{td_enum_repository.CalibRecordSensorFields.SECOND_ORDER_COEFF.name}' - gain_entry_name = f'{group}__{sensor.name}__{td_enum_repository.CalibRecordSensorFields.GAIN.name}' - offset_entry_name = f'{group}__{sensor.name}__{td_enum_repository.CalibRecordSensorFields.OFFSET.name}' - calibration_time_entry_name = f'{group}__{sensor.name}__{td_enum_repository.CalibRecordSensorFields.CALIBRATION_TIME.name}' + @property + def set_service_rec_ack(self) -> bool: + """ + The Set Service Records message acceptance + """ + return self._set_service_rec_ack - kwargs['sensor_enum'] = sensor - kwargs['msg_id'] = msg_id_pairing[group] - if fourth_order_coeff_entry_name in calibration_records and calibration_records[fourth_order_coeff_entry_name] is not None: - kwargs['forth_order_coeff'] = calibration_records[fourth_order_coeff_entry_name] + @set_service_rec_ack.setter + def set_service_rec_ack(self, value): + self._set_service_rec_ack = value - if third_order_coeff_entry_name in calibration_records and calibration_records[third_order_coeff_entry_name] is not None: - kwargs['third_order_coeff'] = calibration_records[third_order_coeff_entry_name] - if second_order_coeff_entry_name in calibration_records and calibration_records[second_order_coeff_entry_name] is not None: - kwargs['second_order_coeff'] = calibration_records[second_order_coeff_entry_name] + @property + def set_calibration_rec_ack(self) -> dict: + """ + The Set Calibration Records message acceptance + """ + return self._set_calibration_rec_ack - if gain_entry_name in calibration_records and calibration_records[gain_entry_name] is not None: - kwargs['gain'] = calibration_records[gain_entry_name] + @set_calibration_rec_ack.setter + def set_calibration_rec_ack(self, value): + self._set_calibration_rec_ack = value - if offset_entry_name in calibration_records and calibration_records[offset_entry_name] is not None: - kwargs['offset'] = calibration_records[offset_entry_name] - if calibration_time_entry_name in calibration_records and calibration_records[calibration_time_entry_name] is not None: - kwargs['calibration_time'] = calibration_records[calibration_time_entry_name] + @property + def set_institutional_rec_ack(self) -> dict: + """ + The Set Institutional Records message acceptance + """ + return self._set_institutional_rec_ack - checked.append(record_parts[1]) - # Skip sending if dictionary is provided but all entry for the sensor was None (except sensor_enum and MSG_ID, because those are neccesary) - if calibration_records != {} and len(kwargs) == 2: - continue + @set_institutional_rec_ack.setter + def set_institutional_rec_ack(self, value): + self._set_institutional_rec_ack = value - # Create the send data function - send_data.append(partial(self.cmd_set_calibration_records_sensor, *args, **kwargs)) - elif group in ['ACCEL']: - # Set the arguments for the call - args = [] - kwargs = {} - accel_x_offset_entry_name = f'{group}__{td_enum_repository.CalibRecordAccelerometerFields.ACCEL_X_OFFSET.name}' - accel_y_offset_entry_name = f'{group}__{td_enum_repository.CalibRecordAccelerometerFields.ACCEL_Y_OFFSET.name}' - accel_z_offset_entry_name = f'{group}__{td_enum_repository.CalibRecordAccelerometerFields.ACCEL_Z_OFFSET.name}' - calibration_time_entry_name = f'{group}__{td_enum_repository.CalibRecordAccelerometerFields.CALIBRATION_TIME.name}' + @property + def set_usage_info_rec_ack(self) -> bool: + """ + The Set Usage Information Records message acceptance + """ + return self._set_usage_info_rec_ack - kwargs['msg_id'] = msg_id_pairing[group] - if accel_x_offset_entry_name in calibration_records and calibration_records[accel_x_offset_entry_name] is not None: - kwargs['accel_x_offset'] = calibration_records[accel_x_offset_entry_name] + @set_usage_info_rec_ack.setter + def set_usage_info_rec_ack(self, value): + self._set_usage_info_rec_ack = value - if accel_y_offset_entry_name in calibration_records and calibration_records[accel_y_offset_entry_name] is not None: - kwargs['accel_y_offset'] = calibration_records[accel_y_offset_entry_name] - if accel_z_offset_entry_name in calibration_records and calibration_records[accel_z_offset_entry_name] is not None: - kwargs['accel_z_offset'] = calibration_records[accel_z_offset_entry_name] + @property + def set_system_rec_rr(self) -> dict: + """ + The Set System Records Request reject reason data + """ + return self._set_system_rec_rr - if calibration_time_entry_name in calibration_records and calibration_records[calibration_time_entry_name] is not None: - kwargs['calibration_time'] = calibration_records[calibration_time_entry_name] + @set_system_rec_rr.setter + def set_system_rec_rr(self, value): + self._set_system_rec_rr = value - checked.append(group) - # Skip sending if dictionary is provided but all entry for the sensor was None (except MSG_ID, because that is neccesary) - if calibration_records != {} and len(kwargs) == 1: - continue - # Store the function but do not execute it - send_data.append(partial(self.cmd_set_calibration_records_accelerometer, *args, **kwargs)) + @property + def set_service_rec_rr(self) -> dict: + """ + The Set Service Records Request reject reason data + """ + return self._set_service_rec_rr - # Execute the stored functions one by one - # Remove the ones that got 1 (successfully recieved) as response - # Retry the ones that are failed 2 more times - retry = 0 - while send_data != [] and retry < 3: - failed = [] - for func in send_data: - resp = func() - if resp == 1: - failed.append(func) - send_data = failed - retry += 1 - sleep(0.005) # Wait 50ms between sends + @set_service_rec_rr.setter + def set_service_rec_rr(self, value): + self._set_service_rec_rr = value - def cmd_set_institutional_records(self, institutional_records: dict={}) -> int: + @property + def set_calibration_rec_rr(self) -> dict: """ - Constructs and sends a command for setting the Institutional Records. - Constraints: - Must be logged into TD. - Must be in Service mode. + The Set Calibration Records Request reject reason data + """ + return self._set_calibration_rec_rr - @param institutional_records: (Dictionary) The stored and structured Institutional Records - @return: 1 if successful, zero otherwise + @set_calibration_rec_rr.setter + def set_calibration_rec_rr(self, value): + self._set_calibration_rec_rr = value + + + @property + def set_institutional_rec_rr(self) -> dict: """ - # Make a dictionary to store all the send functions for later send - send_data = [] - for record_enum in td_enum_repository.InstitutionalRecordFields: - # Set the enum_id to identify which parameter is being sent U08 - 1 byte - 0->255 !! - payload = byte_to_bytearray(record_enum.value) + The Set Institutional Records Request reject reason data + """ + return self._set_institutional_rec_rr - if record_enum in [td_enum_repository.InstitutionalRecordFields.CRC, td_enum_repository.InstitutionalRecordFields.NUM_OF_INSTITUTIONAL_RECORD_FIELDS]: - # Skip the Number of fields - continue - elif institutional_records == {}: - # Set default values if it's missing from the provided dictionary - if record_enum.datatype() in [DataTypes.U32, DataTypes.BOOL, DataTypes.S32, DataTypes.U16, DataTypes.U08, DataTypes.BOOL_U08]: - value = 0 - elif record_enum.datatype() in [DataTypes.F32]: - value = 0.0 - elif record_enum.name in institutional_records and institutional_records[record_enum.name] is not None: - # If record present in the dict and value is not None - value = institutional_records[record_enum.name] - else: - # If record is not present in the dict or it doesn't have value, skip sending it - continue + @set_institutional_rec_rr.setter + def set_institutional_rec_rr(self, value): + self._set_institutional_rec_rr = value - if record_enum.datatype() in [DataTypes.U32, DataTypes.BOOL, DataTypes.S32]: - payload += integer_to_bytearray(value) - elif record_enum.datatype() in [DataTypes.F32]: - payload += float_to_bytearray(value) - elif record_enum.datatype() in [DataTypes.U16]: - payload += unsigned_short_to_bytearray(value) - elif record_enum.datatype() in [DataTypes.U08, DataTypes.BOOL_U08]: - payload += byte_to_bytearray(value) - send_data.append(partial(cmd_generic_override, - payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_dd_ch_id, - msg_id = MsgIds.MSG_ID_UI_DD_NVM_SET_INSTITUTIONAL_RECORD, - entity_name = f'TD Institutional Record', - override_text = 'being set', - logger = self.logger, - can_interface = self.can_interface)) + @property + def set_usage_info_rec_rr(self) -> dict: + """ + The Set Usage Information Records Request reject reason data + """ + return self._set_usage_info_rec_rr - # Execute the stored functions one by one - # Remove the ones that got 1 (successfully recieved) as response - # Retry the ones that are failed 2 more times - retry = 0 - while send_data != [] and retry < 3: - failed = [] - for func in send_data: - resp = func() - if resp == 1: - failed.append(func) - send_data = failed - retry += 1 - sleep(0.005) # Wait 50ms between sends + @set_usage_info_rec_rr.setter + def set_usage_info_rec_rr(self, value): + self._set_usage_info_rec_rr = value - def cmd_set_usage_info_records(self, - ro_water_gen_total_l: float=0.0, - ro_water_gen_since_last_serv_l: float=0.0, - last_basic_flush_date: int=0, - last_heat_disinfect_date: int=0, - last_heat_active_cool_date: int=0, - last_filter_flush_date: int=0, - last_reset_time: int=0) -> int: + +# ================================================= Go to Service Mode Method ================================================= + def cmd_initiate_service_mode(self) -> int: """ - Constructs and sends a command for setting the Usage Information Records. + Constructs and sends a request to change to Service operation mode via CAN bus. Constraints: Must be logged into TD. - Must be in Service mode. + Transition from current to requested op mode must be legal. @return: 1 if successful, zero otherwise - """ - payload = float_to_bytearray(ro_water_gen_total_l) - payload += float_to_bytearray(ro_water_gen_since_last_serv_l) - payload += integer_to_bytearray(last_basic_flush_date) - payload += integer_to_bytearray(last_heat_disinfect_date) - payload += integer_to_bytearray(last_heat_active_cool_date) - payload += integer_to_bytearray(last_filter_flush_date) - payload += integer_to_bytearray(last_reset_time) - payload += unsigned_short_to_bytearray(self.crc16(payload)) + """ + service_mode = td_enum_repository.TDOpModes.MODE_SERV + payload = integer_to_bytearray(service_mode.value) + return cmd_generic_override( payload = payload, reset = None, - channel_id = CanChannels.dialin_to_td_ch_id, - msg_id = MsgIds.MSG_ID_TD_NVM_SET_USAGE_INFO_RECORD, - entity_name = f'New TD Usage Information Record', - override_text = 'being set', + channel_id = CanChannels.dialin_to_dd_ch_id, + msg_id = MsgIds.MSG_ID_TD_SET_OPERATION_MODE_OVERRIDE_REQUEST, + entity_name = 'TD Operation Mode', + override_text = f'set to {service_mode.name}', logger = self.logger, can_interface = self.can_interface) - - - -# ================================================= Set Records Support Methods ================================================= - def cmd_set_calibration_records_sensor(self, - sensor_enum: DialinEnum, - msg_id: MsgIds, - dialysate_pump_target_speed: float=0.0, - forth_order_coeff: float=0.0, - third_order_coeff: float=0.0, - second_order_coeff: float=0.0, - gain: float=0.0, - offset: float=0.0, - calibration_time: int = 0) -> int: - payload_sensor = integer_to_bytearray(sensor_enum.value) - payload_content = b'' - if msg_id == MsgIds.MSG_ID_UI_DD_NVM_SET_CAL_D12_PUMP: - payload_content += float_to_bytearray(dialysate_pump_target_speed) - payload_content += float_to_bytearray(forth_order_coeff) - payload_content += float_to_bytearray(third_order_coeff) - payload_content += float_to_bytearray(second_order_coeff) - payload_content += float_to_bytearray(gain) - payload_content += float_to_bytearray(offset) - payload_content += integer_to_bytearray(calibration_time) - payload_content += unsigned_short_to_bytearray(self.crc16(payload_content)) - payload = payload_sensor + payload_content - - sensor_type = '' - if msg_id == MsgIds.TBD: - sensor_type = 'Pressure Sensor' - elif msg_id == MsgIds.TBD: - sensor_type = 'Temperature Sensor' - elif msg_id == MsgIds.TBD: - sensor_type = 'Concentrate Pump' - elif msg_id == MsgIds.TBD: - sensor_type = 'Dialysate Pump' - - return cmd_generic_override(payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_td_ch_id, - msg_id = msg_id, - entity_name = f'TD {sensor_type} {sensor_enum.name} Calibration Record', - override_text = 'being set', - logger = self.logger, - can_interface = self.can_interface) - - - def cmd_set_calibration_records_accelerometer(self, - msg_id: MsgIds, - accel_x_offset: float=0.0, - accel_y_offset: float=0.0, - accel_z_offset: float=0.0, - calibration_time: int = 0) -> int: - - payload = float_to_bytearray(accel_x_offset) - payload += float_to_bytearray(accel_y_offset) - payload += float_to_bytearray(accel_z_offset) - payload += integer_to_bytearray(calibration_time) - payload += unsigned_short_to_bytearray(self.crc16(payload)) - - conc_type = '' - if msg_id == MsgIds.TBD: - conc_type = 'Acid' - elif msg_id == MsgIds.TBD: - conc_type = 'Bicarb' - - return cmd_generic_override(payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_td_ch_id, - msg_id = msg_id, - entity_name = f'TD {conc_type} Concentrate Calibration Record', - override_text = 'being set', - logger = self.logger, - can_interface = self.can_interface) - - - -# ================================================= Private Methods =================================================