Index: leahi_dialin/dd/modules/records.py =================================================================== diff -u -r28a21f1f2ca4ef02e6d54bae30b738fddb7bdb3a -r436742844a3fbbb6a3b0bc81dc87bc6385c3e368 --- leahi_dialin/dd/modules/records.py (.../records.py) (revision 28a21f1f2ca4ef02e6d54bae30b738fddb7bdb3a) +++ leahi_dialin/dd/modules/records.py (.../records.py) (revision 436742844a3fbbb6a3b0bc81dc87bc6385c3e368) @@ -8,827 +8,401 @@ # @file records.py # # @author (last) Zoltan Miskolci -# @date (last) 22-Jun-2026 +# @date (last) 04-Aug-2026 # @author (original) Zoltan Miskolci # @date (original) 04-May-2026 # ############################################################################ # Module imports from logging import Logger -from functools import partial -from time import sleep # Project imports -from leahi_dialin.common import dd_enum_repository, disp_enum_repository, fp_enum_repository -from leahi_dialin.common.generic_defs import DataTypes +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 dd_enum_repository +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 DDRecords(AbstractSubSystem): +class DDRecords(DDRecordsGetHandlers, DDRecordsGetRequests, DDRecordsSetHandlers, DDRecordsSetRequests): """ DD interface containing pressure 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: - # Data Request Response messages - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_SYSTEM_RECORD_RESPONSE.value, - function = self._handler_system_record_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_SERVICE_RECORD_RESPONSE.value, - function = self._handler_service_record_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_CAL_PRESSURE_SENSOR_RESPONSE.value, - function = self._handler_calibration_record_pres_sensor_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_CAL_TEMP_SENSOR_RESPONSE.value, - function = self._handler_calibration_record_temp_sensor_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_CAL_CONC_PUMP_RESPONSE.value, - function = self._handler_calibration_record_conc_pump_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_CAL_D12_PUMP_RESPONSE.value, - function = self._handler_calibration_record_dial_pump_d12_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_CAL_D48_PUMP_RESPONSE.value, - function = self._handler_calibration_record_dial_pump_d48_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_CAL_ACID_CONCENTRATE_RESPONSE.value, - function = self._handler_calibration_record_acid_conc_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_CAL_BICARB_CONCENTRATE_RESPONSE.value, - function = self._handler_calibration_record_bicarb_conc_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_CAL_ACCEL_SENSOR_RESPONSE.value, - function = self._handler_calibration_record_accelerometer_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_CAL_BLOOD_LEAK_SENSOR_RESPONSE.value, - function = self._handler_calibration_record_blood_leak_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_INSTITUTIONAL_RECORD_RESPONSE.value, - function = self._handler_institutional_record_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_USAGE_INFO_RECORD_RESPONSE.value, - function = self._handler_usage_info_record_sync) - # Set Data Response messages - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_SET_SYSTEM_RECORD_RESPONSE.value, - function = self._handler_system_rr_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_SET_SERVICE_RECORD_RESPONSE.value, - function = self._handler_service_rr_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_SET_CAL_PRESSURE_SENSOR_RESPONSE.value, - function = self._handler_cal_pressure_rr_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_SET_CAL_TEMP_SENSOR_RESPONSE.value, - function = self._handler_cal_temp_rr_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_SET_CAL_CONC_PUMP_RESPONSE.value, - function = self._handler_cal_conc_pump_rr_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_SET_CAL_D12_PUMP_RESPONSE.value, - function = self._handler_cal_d12_rr_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_SET_CAL_D48_PUMP_RESPONSE.value, - function = self._handler_cal_d48_rr_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_SET_CAL_ACID_CONCENTRATE_RESPONSE.value, - function = self._handler_cal_acid_rr_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_SET_CAL_BICARB_CONCENTRATE_RESPONSE.value, - function = self._handler_cal_bicarb_rr_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_SET_CAL_ACCEL_SENSOR_RESPONSE.value, - function = self._handler_cal_accel_rr_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_SET_CAL_BLOOD_LEAK_SENSOR_RESPONSE.value, - function = self._handler_cal_blood_leak_rr_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_SET_INSTITUTIONAL_RECORD_RESPONSE.value, - function = self._handler_institutional_rr_sync) - self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, - message_id = MsgIds.MSG_ID_DD_NVM_SET_USAGE_INFO_RECORD_RESPONSE.value, - function = self._handler_usage_info_rr_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 = { } + 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_timestamp = 0 #: The timestamp of the latest System Records message - self.system_set_rr_timestamp = 0 #: The timestamp of the latest Set System Reject Reason message - self.service_records_timestamp = 0 #: The timestamp of the latest Service Records message - self.service_set_rr_timestamp = 0 #: The timestamp of the latest Set Service Reject Reason message - self.calibration_records_timestamp = 0 #: The timestamp of the latest Calibration Records message - self.calibration_set_rr_timestamp = 0 #: The timestamp of the latest Set Calibration Reject Reason message - self.institutional_records_timestamp = 0 #: The timestamp of the latest Institutional Records message - self.institutional_set_rr_timestamp = 0 #: The timestamp of the latest Set Institutional Reject Reason message - self.usage_info_records_timestamp = 0 #: The timestamp of the latest Usage Information Records message - self.usage_info_set_rr_timestamp = 0 #: The timestamp of the latest Set Usage Information Reject Reason message + 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 - self.system_records = { } #: The System Records data in dictionary format - self.system_records_ack = { } #: The Set System Records message acceptance - self.system_records_rr = { } #: The Set System Records message reject reason + self.service_records = { } - self.service_records = { } #: The Service Records data in dictionary format - self.service_records_ack = { } #: The Set Service Records message acceptance - self.service_records_rr = { } #: The Set Service Records message reject reason + 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 = { } - self.calibration_records = { } #: The Calibration Records data in dictionary format - self.calibration_records_ack = { } #: The Set Calibration Records message acceptance - self.calibration_records_rr = { } #: The Set Calibration Records message reject reason + self.get_calibration_rec_timestamp = 0 #: The timestamp of the latest Calibration Records message + self.get_calibration_rec_ack = { } + self.get_calibration_rec_rr = { } - self.institutional_records = { } #: The Institutional Records data in dictionary format - self.institutional_records_ack = { } #: The Set Institutional Records message acceptance - self.institutional_records_rr = { } #: The Set Institutional Records message reject reason - + self.calibration_records = { } - self.usage_info_records = { } #: The Usage Information Records data in dictionary format - self.usage_info_ack = { } #: The Set Usage Information Records message acceptance - self.usage_info_rr = { } #: The Set Usage Information Records message reject reason + 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 = { } - for sys_record in dd_enum_repository.SystemRecordFields: - if 'num_' not in sys_record.name: - self.system_records[sys_record.name] = None + self.get_institutional_rec_timestamp = 0 #: The timestamp of the latest Institutional Records message + self.get_institutional_rec_ack = { } + self.get_institutional_rec_rr = { } - for serv_record in dd_enum_repository.ServiceRecordFields: - if 'num_' not in serv_record.name: - self.service_records[serv_record.name] = None + self.institutional_records = { } - for cal_record in dd_enum_repository.CalibrationRecordFields: - if 'num_' not in cal_record.name: - self.calibration_records[cal_record.name] = None + 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 = { } - for inst_record in dd_enum_repository.InstitutionalRecordFields: - if 'num_' not in inst_record.name: - self.institutional_records[inst_record.name] = None + 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 - for usage_record in dd_enum_repository.UsageInformationRecordFields: - if 'num_' not in usage_record.name: - self.usage_info_records[usage_record.name] = None + 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 = { } -# ================================================= CAN Message Handler Methods ================================================= - @publish(["msg_id_dd_nvm_send_system_record_response", "system_records", "system_records_timestamp"]) - def _handler_system_record_sync(self, message, timestamp = 0.0): - """ - Handles published DD System Record data messages. DD System Records are captured - for reference. + # Calling the Sub classes: DDRecordsGetHandlers, DDRecordsGetRequests, DDRecordsSetHandlers, DDRecordsSetRequests + # 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 dd_enum_repository.SystemRecordFields: - record_list.append((member.name, member.datatype(), member.multichar_length())) - self.process_into_dict(dict_to_update = self.system_records, - decoder_list = record_list, - message = message) - self.system_records_timestamp = timestamp - - - @publish(["msg_id_dd_nvm_send_service_record_response", "service_records", "service_records_timestamp"]) - def _handler_service_record_sync(self, message, timestamp = 0.0): +# ============================================================ Properties ============================================================ + @property + def system_records(self) -> dict: """ - Handles published DD Service Record data messages. DD Service Records are captured - for reference. - - @param message: published data message - @return: none + The System Records data """ - record_list = [] - for member in dd_enum_repository.ServiceRecordFields: - record_list.append((member.name, member.datatype())) + return self._system_records - self.process_into_dict(dict_to_update = self.service_records, - decoder_list = record_list, - message = message) + @system_records.setter + def system_records(self, value): + self._system_records = value - self.service_records_timestamp = timestamp - - @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): + @property + def service_records(self) -> dict: """ - Handles published DD Calibration Record Pressure Sensor Details data messages. - DD Calibration Records Pressure Sensor Details are captured for reference. - - @param message: published data message - @return: none + The Service Records data """ - # 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) - if int(result['sensor_id']) < dd_enum_repository.DDPressureSensorNames.NUM_OF_PRESSURE_SENSORS.value: - sensor_name = dd_enum_repository.DDPressureSensorNames(result['sensor_id']).name - else: - fp_sensor_id = int(result['sensor_id']) - dd_enum_repository.DDPressureSensorNames.NUM_OF_PRESSURE_SENSORS.value - sensor_name = fp_enum_repository.FPPressureSensorNames(fp_sensor_id).name + return self._service_records - record_list = [] - for member in dd_enum_repository.CalibRecordSensorFields: - if member == dd_enum_repository.CalibRecordSensorFields.NUM_OF_CALIBRATION_RECORD_SENSOR_FIELDS: - continue - entry_name = f'PRES__{sensor_name}__{member.name}' - record_list.append((entry_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 - - - @publish(["msg_id_dd_nvm_cal_temp_sensor_response", "calibration_records", "calibration_records_timestamp"]) - def _handler_calibration_record_temp_sensor_sync(self, message, timestamp = 0.0): + @property + def calibration_records(self) -> dict: """ - Handles published DD Calibration Record Temperature Sensor Details data messages. - DD Calibration Records Temperature Sensor Details are captured for reference. - - @param message: published data message - @return: none + The Calibration Records data """ - # 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 = dd_enum_repository.DDTemperatureSensorNames(result['sensor_id']).name + return self._calibration_records - record_list = [] - for member in dd_enum_repository.CalibRecordSensorFields: - if member == dd_enum_repository.CalibRecordSensorFields.NUM_OF_CALIBRATION_RECORD_SENSOR_FIELDS: - continue - entry_name = f'TEMP__{sensor_name}__{member.name}' - record_list.append((entry_name, member.datatype())) + @calibration_records.setter + def calibration_records(self, value): + self._calibration_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 - - - @publish(["msg_id_dd_nvm_cal_conc_pump_response", "calibration_records", "calibration_records_timestamp"]) - def _handler_calibration_record_conc_pump_sync(self, message, timestamp = 0.0): + @property + def institutional_records(self) -> dict: """ - Handles published DD Calibration Record Concentrate Pump Details data messages. - DD Calibration Records Concentrate Pump Details are captured for reference. - - @param message: published data message - @return: none + The Institutional Records data """ - # 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 = dd_enum_repository.DDConcentratePumpNames(result['sensor_id']).name + return self._institutional_records - record_list = [] - for member in dd_enum_repository.CalibRecordSensorFields: - if member == dd_enum_repository.CalibRecordSensorFields.NUM_OF_CALIBRATION_RECORD_SENSOR_FIELDS: - continue - entry_name = f'CONC__{sensor_name}__{member.name}' - record_list.append((entry_name, member.datatype())) + @institutional_records.setter + def institutional_records(self, value): + self._institutional_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 - - - @publish(["msg_id_dd_nvm_cal_d12_pump_response", "calibration_records", "calibration_records_timestamp"]) - def _handler_calibration_record_dial_pump_d12_sync(self, message, timestamp = 0.0): + @property + def usage_info_records(self) -> dict: """ - Handles published DD Calibration Record Dialysate Pump D12 Details data messages. - DD Calibration Records Dialysate Pump D12 Details are captured for reference. - - @param message: published data message - @return: none + The Usage Information Records data """ - record_list = [] - record_list.append((f'DIAL__{dd_enum_repository.DDDialysatePumpNames.D12_PUMP.name}__TARGET_SPEED', DataTypes.F32)) - record_list.append((f'DIAL__{dd_enum_repository.DDDialysatePumpNames.D12_PUMP.name}__CALIBRATION_TIME', DataTypes.U32)) - record_list.append((f'DIAL__{dd_enum_repository.DDDialysatePumpNames.D12_PUMP.name}__CRC', DataTypes.U16)) - # for member in dd_enum_repository.CalibRecordSensorFields: - # if member == dd_enum_repository.CalibRecordSensorFields.NUM_OF_CALIBRATION_RECORD_SENSOR_FIELDS: - # continue - # entry_name = f'DIAL__{dd_enum_repository.DDDialysatePumpNames.D12_PUMP.name}__{member.name}' - # record_list.append((entry_name, member.datatype())) + return self._usage_info_records - self.process_into_dict(dict_to_update = self.calibration_records, - decoder_list = record_list, - message = message) + @usage_info_records.setter + def usage_info_records(self, value): + self._usage_info_records = value - self.calibration_records_timestamp = timestamp - - @publish(["msg_id_dd_nvm_cal_d48_pump_response", "calibration_records", "calibration_records_timestamp"]) - def _handler_calibration_record_dial_pump_d48_sync(self, message, timestamp = 0.0): + @property + def get_system_rec_ack(self) -> bool: """ - Handles published DD Calibration Record Dialysate Pump D48 Details data messages. - DD Calibration Records Dialysate Pump D48 Details are captured for reference. - - @param message: published data message - @return: none + The Get System Records message acceptance """ - record_list = [] - for member in dd_enum_repository.CalibRecordSensorFields: - if member == dd_enum_repository.CalibRecordSensorFields.NUM_OF_CALIBRATION_RECORD_SENSOR_FIELDS: - continue - entry_name = f'DIAL__{dd_enum_repository.DDDialysatePumpNames.D48_PUMP.name}__{member.name}' - record_list.append((entry_name, member.datatype())) + return self._get_system_rec_ack - self.process_into_dict(dict_to_update = self.calibration_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.calibration_records_timestamp = timestamp - - @publish(["msg_id_dd_nvm_cal_acid_concentrate_sensor", "calibration_records", "calibration_records_timestamp"]) - def _handler_calibration_record_acid_conc_sync(self, message, timestamp = 0.0): + @property + def get_service_rec_ack(self) -> bool: """ - Handles published DD Calibration Record Acid Concentrate Details data messages. - DD Calibration Records Acid Concentrate Details are captured for reference. - - @param message: published data message - @return: none + The Get Service Records message acceptance """ - msg_list = [('sensor_id', DataTypes.U08)] - result = self.process_into_vars(decoder_list=msg_list, - message=message) - acid_type_name = disp_enum_repository.AcidTypes(result['sensor_id']).name + return self._get_service_rec_ack - record_list = [] - for member in dd_enum_repository.CalibRecordConcentrateFields: - if member == dd_enum_repository.CalibRecordConcentrateFields.NUM_OF_CALIBRATION_RECORD_CONCENTRATE_FIELDS: - continue - record_list.append((f'ACID__{acid_type_name}__{member.name}', member.datatype())) + @get_service_rec_ack.setter + def get_service_rec_ack(self, value): + self._get_service_rec_ack = 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 - - - @publish(["msg_id_dd_nvm_cal_bicarb_concentrate_sensor", "calibration_records", "calibration_records_timestamp"]) - def _handler_calibration_record_bicarb_conc_sync(self, message, timestamp = 0.0): + @property + def get_calibration_rec_ack(self) -> dict: """ - Handles published DD Calibration Record Bicarb Concentrate Details data messages. - DD Calibration Records Bicarb Concentrate Details are captured for reference. - - @param message: published data message - @return: none + The Get Calibration Records message acceptance """ - msg_list = [('sensor_id', DataTypes.U08)] - result = self.process_into_vars(decoder_list=msg_list, - message=message) - bicarb_type_name = disp_enum_repository.BicarbTypes(result['sensor_id']).name + return self._get_calibration_rec_ack - record_list = [] - for member in dd_enum_repository.CalibRecordConcentrateFields: - if member == dd_enum_repository.CalibRecordConcentrateFields.NUM_OF_CALIBRATION_RECORD_CONCENTRATE_FIELDS: - continue - record_list.append((f'BICARB__{bicarb_type_name}__{member.name}', member.datatype())) + @get_calibration_rec_ack.setter + def get_calibration_rec_ack(self, value): + self._get_calibration_rec_ack = 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 - - - @publish(["msg_id_dd_nvm_cal_accel_sensor_response", "calibration_records", "calibration_records_timestamp"]) - def _handler_calibration_record_accelerometer_sync(self, message, timestamp = 0.0): + @property + def get_institutional_rec_ack(self) -> dict: """ - Handles published DD Calibration Record Accelerometer Details data messages. - DD Calibration Records Accelerometer Details are captured for reference. - - @param message: published data message - @return: none + The Get Institutional Records message acceptance """ - record_list = [] - for member in dd_enum_repository.CalibRecordAccelerometerFields: - if member == dd_enum_repository.CalibRecordAccelerometerFields.NUM_OF_CALIBRATION_RECORD_ACCELEROMETER_FIELDS: - continue - record_list.append((f'ACCEL__{member.name}', member.datatype())) + return self._get_institutional_rec_ack - self.process_into_dict(dict_to_update = self.calibration_records, - decoder_list = record_list, - message = message) + @get_institutional_rec_ack.setter + def get_institutional_rec_ack(self, value): + self._get_institutional_rec_ack = value - self.calibration_records_timestamp = timestamp - - @publish(["msg_id_dd_nvm_cal_blood_leak_response", "calibration_records", "calibration_records_timestamp"]) - def _handler_calibration_record_blood_leak_sync(self, message, timestamp = 0.0): + @property + def get_usage_info_rec_ack(self) -> bool: """ - Handles published DD Calibration Record Blood Leak Details data messages. - DD Calibration Records Blood Leak Details are captured for reference. - - @param message: published data message - @return: none + The Get Usage Information Records message acceptance """ - record_list = [] - for member in dd_enum_repository.CalibRecordBloodLeakFields: - if member == dd_enum_repository.CalibRecordBloodLeakFields.NUM_OF_CALIBRATION_RECORD_BLOOD_LEAK_FIELDS: - continue - record_list.append((f'BLOOD_LEAK__{member.name}', member.datatype())) + return self._get_usage_info_rec_ack - self.process_into_dict(dict_to_update = self.calibration_records, - decoder_list = record_list, - message = message) + @get_usage_info_rec_ack.setter + def get_usage_info_rec_ack(self, value): + self._get_usage_info_rec_ack = value - self.calibration_records_timestamp = timestamp - - @publish(["msg_id_dd_nvm_institutional_record_response", "institutional_records", "institutional_records_timestamp"]) - def _handler_institutional_record_sync(self, message, timestamp = 0.0): + @property + def get_system_rec_rr(self) -> dict: """ - Handles published DD Institutional Record data messages. DD Institutional Records are captured - for reference. - - @param message: published data message - @return: none + The Get System Records Request reject reason data """ - msg_list = [('record_id', DataTypes.U08)] - results = self.process_into_vars(decoder_list = msg_list, - message = message) + return self._get_system_rec_rr - record_id = int(results['record_id']) - member = dd_enum_repository.InstitutionalRecordFields(record_id) - - record_list = [] - record_list.append((member.name, member.datatype())) + @get_system_rec_rr.setter + def get_system_rec_rr(self, value): + self._get_system_rec_rr = value - self.process_into_dict(dict_to_update = self.institutional_records, - decoder_list = record_list, - message = message, - start_from_byte = len(msg_list) * msg_list[0][1].size()) - self.institutional_records_timestamp = timestamp - - - @publish(["msg_id_dd_nvm_usage_info_record_response", "usage_info_records", "usage_info_records_timestamp"]) - def _handler_usage_info_record_sync(self, message, timestamp = 0.0): + @property + def get_service_rec_rr(self) -> dict: """ - Handles published DD Usage Information Record data messages. DD Usage Information Records are captured - for reference. - - @param message: published data message - @return: none + The Get Service Records Request reject reason data """ - record_list = [] - for member in dd_enum_repository.UsageInformationRecordFields: - record_list.append((member.name, member.datatype())) + return self._get_service_rec_rr - self.process_into_dict(dict_to_update = self.usage_info_records, - decoder_list = record_list, - message = message) + @get_service_rec_rr.setter + def get_service_rec_rr(self, value): + self._get_service_rec_rr = value - self.usage_info_records_timestamp = timestamp - -# ================================================= Reject Reason Message Handler Methods ================================================= - @publish(["msg_id_dd_nvm_set_system_record_response", "system_records_ack", "system_records_rr", "system_set_rr_timestamp"]) - def _handler_system_rr_sync(self, message, timestamp = 0.0): + @property + def get_calibration_rec_rr(self) -> dict: """ - Handles published DD Set System Records Reject Reason data messages. DD Set System Records Reject Reason are captured - for reference. - - @param message: published data message - @return: none + The Get Calibration Records Request reject reason data """ - msg_list = [] - msg_list.append(('self.system_records_ack', DataTypes.BOOL)) - msg_list.append(('self.system_records_rr', DataTypes.U32)) + return self._get_calibration_rec_rr - self.process_into_vars(decoder_list = msg_list, - message = message) + @get_calibration_rec_rr.setter + def get_calibration_rec_rr(self, value): + self._get_calibration_rec_rr = value - self.system_set_rr_timestamp = timestamp - - @publish(["msg_id_dd_nvm_set_service_record_response", "system_records_ack", "service_records_rr", "service_set_rr_timestamp"]) - def _handler_service_rr_sync(self, message, timestamp = 0.0): + @property + def get_institutional_rec_rr(self) -> dict: """ - Handles published DD Set Service Records Reject Reason data messages. DD Set Service Records Reject Reason are captured - for reference. - - @param message: published data message - @return: none + The Get Institutional Records Request reject reason data """ - msg_list = [] - msg_list.append(('self.service_records_ack', DataTypes.BOOL)) - msg_list.append(('self.service_records_rr', DataTypes.U32)) + return self._get_institutional_rec_rr - self.process_into_vars(decoder_list = msg_list, - message = message) + @get_institutional_rec_rr.setter + def get_institutional_rec_rr(self, value): + self._get_institutional_rec_rr = value - self.service_set_rr_timestamp = timestamp - - @publish(["msg_id_dd_nvm_set_institutional_record_response", "institutional_records_rr", "institutional_set_rr_timestamp"]) - def _handler_institutional_rr_sync(self, message, timestamp = 0.0): + @property + def get_usage_info_rec_rr(self) -> dict: """ - Handles published DD Set Institutional Records Reject Reason data messages. DD Set Institutional Records Reject Reason are captured - for reference. - - @param message: published data message - @return: none + The Get Usage Information Records Request reject reason data """ - msg_list = [] - msg_list.append(('ack', DataTypes.BOOL)) - msg_list.append(('rr', DataTypes.U32)) - msg_list.append(('id', DataTypes.U32)) + return self._get_usage_info_rec_rr - result = self.process_into_vars(decoder_list = msg_list, - message = message) - - if result['id'] not in self.institutional_records_rr: - self.institutional_records_rr[result['id']] = {} - self.institutional_records_rr[result['id']]['ack'] = result['ack'] - self.institutional_records_rr[result['id']]['rr'] = result['rr'] + @get_usage_info_rec_rr.setter + def get_usage_info_rec_rr(self, value): + self._get_usage_info_rec_rr = value - self.institutional_set_rr_timestamp = timestamp - - @publish(["msg_id_dd_nvm_set_usage_info_record_response", "usage_info_ack", "usage_info_rr", "usage_info_set_rr_timestamp"]) - def _handler_usage_info_rr_sync(self, message, timestamp = 0.0): + @property + def set_system_rec_ack(self) -> bool: """ - Handles published DD Set Usage Information Records Reject Reason data messages. DD Set Usage Information Records Reject Reason are captured - for reference. - - @param message: published data message - @return: none + The Set System Records message acceptance """ - msg_list = [] - msg_list.append(('self.usage_info_ack', DataTypes.BOOL)) - msg_list.append(('self.usage_info_rr', DataTypes.U32)) + return self._set_system_rec_ack - self.process_into_vars(decoder_list = msg_list, - message = message) + @set_system_rec_ack.setter + def set_system_rec_ack(self, value): + self._set_system_rec_ack = value - self.usage_info_set_rr_timestamp = timestamp - - @publish(["msg_id_dd_nvm_set_cal_pressure_sensor_response", "calibration_records_rr", "calibration_set_rr_timestamp"]) - def _handler_cal_pressure_rr_sync(self, message, timestamp = 0.0): + @property + def set_service_rec_ack(self) -> bool: """ - Handles published DD Set Calibration Records for Pressure Sensor Reject Reason data messages.DD Set Calibration Records for Pressure Sensor Reject Reason are captured - for reference. - - @param message: published data message - @return: none + The Set Service Records message acceptance """ - msg_list = [] - msg_list.append(('ack', DataTypes.BOOL)) - msg_list.append(('rr', DataTypes.U32)) - msg_list.append(('id', DataTypes.U32)) + return self._set_service_rec_ack - result = self.process_into_vars(decoder_list = msg_list, - message = message) - - if 'pres' not in self.calibration_records_rr: - self.calibration_records_rr['pres'] = {} - if result['id'] not in self.calibration_records_rr: - self.calibration_records_rr['pres'][result['id']] = {} - self.calibration_records_rr['pres'][result['id']]['ack'] = result['ack'] - self.calibration_records_rr['pres'][result['id']]['rr'] = result['rr'] + @set_service_rec_ack.setter + def set_service_rec_ack(self, value): + self._set_service_rec_ack = value - self.calibration_set_rr_timestamp = timestamp - - @publish(["msg_id_dd_nvm_set_cal_temp_sensor_response", "calibration_records_rr", "calibration_set_rr_timestamp"]) - def _handler_cal_temp_rr_sync(self, message, timestamp = 0.0): + @property + def set_calibration_rec_ack(self) -> dict: """ - Handles published DD Set Calibration Records for Temperature Sensor Reject Reason data messages.DD Set Calibration Records for Temperature Sensor Reject Reason are captured - for reference. - - @param message: published data message - @return: none + The Set Calibration Records message acceptance """ - msg_list = [] - msg_list.append(('ack', DataTypes.BOOL)) - msg_list.append(('rr', DataTypes.U32)) - msg_list.append(('id', DataTypes.U32)) + return self._set_calibration_rec_ack - result = self.process_into_vars(decoder_list = msg_list, - message = message) - - if 'temp' not in self.calibration_records_rr: - self.calibration_records_rr['temp'] = {} - if result['id'] not in self.calibration_records_rr: - self.calibration_records_rr['temp'][result['id']] = {} - self.calibration_records_rr['temp'][result['id']]['ack'] = result['ack'] - self.calibration_records_rr['temp'][result['id']]['rr'] = result['rr'] + @set_calibration_rec_ack.setter + def set_calibration_rec_ack(self, value): + self._set_calibration_rec_ack = value - self.calibration_set_rr_timestamp = timestamp - - @publish(["msg_id_dd_nvm_set_cal_conc_pump_response", "calibration_records_rr", "calibration_set_rr_timestamp"]) - def _handler_cal_conc_pump_rr_sync(self, message, timestamp = 0.0): + @property + def set_institutional_rec_ack(self) -> dict: """ - Handles published DD Set Calibration Records for Concentrate Pump Reject Reason data messages.DD Set Calibration Records for Concentrate Pump Reject Reason are captured - for reference. - - @param message: published data message - @return: none + The Set Institutional Records message acceptance """ - msg_list = [] - msg_list.append(('ack', DataTypes.BOOL)) - msg_list.append(('rr', DataTypes.U32)) - msg_list.append(('id', DataTypes.U32)) + return self._set_institutional_rec_ack - result = self.process_into_vars(decoder_list = msg_list, - message = message) - - if 'conc' not in self.calibration_records_rr: - self.calibration_records_rr['conc'] = {} - if result['id'] not in self.calibration_records_rr: - self.calibration_records_rr['conc'][result['id']] = {} - self.calibration_records_rr['conc'][result['id']]['ack'] = result['ack'] - self.calibration_records_rr['conc'][result['id']]['rr'] = result['rr'] + @set_institutional_rec_ack.setter + def set_institutional_rec_ack(self, value): + self._set_institutional_rec_ack = value - self.calibration_set_rr_timestamp = timestamp - - @publish(["msg_id_dd_nvm_set_cal_d12_pump_response", "calibration_records_rr", "calibration_set_rr_timestamp"]) - def _handler_cal_d12_rr_sync(self, message, timestamp = 0.0): + @property + def set_usage_info_rec_ack(self) -> bool: """ - Handles published DD Set Calibration Records for D12 Dialisate Pump Reject Reason data messages.DD Set Calibration Records for D12 Dialisate Pump Reject Reason are captured - for reference. - - @param message: published data message - @return: none + The Set Usage Information Records message acceptance """ - msg_list = [] - msg_list.append(('ack', DataTypes.BOOL)) - msg_list.append(('rr', DataTypes.U32)) + return self._set_usage_info_rec_ack - result = self.process_into_vars(decoder_list = msg_list, - message = message) - - if 'dial' not in self.calibration_records_rr: - self.calibration_records_rr['dial'] = {} - if 'd12_pump' not in self.calibration_records_rr: - self.calibration_records_rr['dial']['d12_pump'] = {} - self.calibration_records_rr['dial']['d12_pump']['ack'] = result['ack'] - self.calibration_records_rr['dial']['d12_pump']['rr'] = result['rr'] + @set_usage_info_rec_ack.setter + def set_usage_info_rec_ack(self, value): + self._set_usage_info_rec_ack = value - self.calibration_set_rr_timestamp = timestamp - - @publish(["msg_id_dd_nvm_set_cal_d48_pump_response", "calibration_records_rr", "calibration_set_rr_timestamp"]) - def _handler_cal_d48_rr_sync(self, message, timestamp = 0.0): + @property + def set_system_rec_rr(self) -> dict: """ - Handles published DD Set Calibration Records for D48 Dialisate Pump Reject Reason data messages.DD Set Calibration Records for D48 Dialisate Pump Reject Reason are captured - for reference. - - @param message: published data message - @return: none + The Set System Records Request reject reason data """ - msg_list = [] - msg_list.append(('ack', DataTypes.BOOL)) - msg_list.append(('rr', DataTypes.U32)) + return self._set_system_rec_rr - result = self.process_into_vars(decoder_list = msg_list, - message = message) - - if 'dial' not in self.calibration_records_rr: - self.calibration_records_rr['dial'] = {} - if 'd48_pump' not in self.calibration_records_rr: - self.calibration_records_rr['dial']['d48_pump'] = {} - self.calibration_records_rr['dial']['d48_pump']['ack'] = result['ack'] - self.calibration_records_rr['dial']['d48_pump']['rr'] = result['rr'] + @set_system_rec_rr.setter + def set_system_rec_rr(self, value): + self._set_system_rec_rr = value - self.calibration_set_rr_timestamp = timestamp - - @publish(["msg_id_dd_nvm_set_acid_concentrate_response", "calibration_records_rr", "calibration_set_rr_timestamp"]) - def _handler_cal_acid_rr_sync(self, message, timestamp = 0.0): + @property + def set_service_rec_rr(self) -> dict: """ - Handles published DD Set Calibration Records for Acid Concentrate Reject Reason data messages.DD Set Calibration Records for Acid Concentrate Reject Reason are captured - for reference. - - @param message: published data message - @return: none + The Set Service Records Request reject reason data """ - msg_list = [] - msg_list.append(('ack', DataTypes.BOOL)) - msg_list.append(('rr', DataTypes.U32)) - msg_list.append(('id', DataTypes.U32)) + return self._set_service_rec_rr - result = self.process_into_vars(decoder_list = msg_list, - message = message) + @set_service_rec_rr.setter + def set_service_rec_rr(self, value): + self._set_service_rec_rr = value - if 'acid' not in self.calibration_records_rr: - self.calibration_records_rr['acid'] = {} - if result['id'] not in self.calibration_records_rr: - self.calibration_records_rr['acid'][result['id']] = {} - self.calibration_records_rr['acid'][result['id']]['ack'] = result['ack'] - self.calibration_records_rr['acid'][result['id']]['rr'] = result['rr'] - self.calibration_set_rr_timestamp = timestamp - - - @publish(["msg_id_dd_nvm_set_cal_bicarb_concentrate_response", "calibration_records_rr", "calibration_set_rr_timestamp"]) - def _handler_cal_bicarb_rr_sync(self, message, timestamp = 0.0): + @property + def set_calibration_rec_rr(self) -> dict: """ - Handles published DD Set Calibration Records for Bicarb Concentrate Reject Reason data messages.DD Set Calibration Records for Bicarb Concentrate Reject Reason are captured - for reference. - - @param message: published data message - @return: none + The Set Calibration Records Request reject reason data """ - msg_list = [] - msg_list.append(('ack', DataTypes.BOOL)) - msg_list.append(('rr', DataTypes.U32)) - msg_list.append(('id', DataTypes.U32)) + return self._set_calibration_rec_rr - result = self.process_into_vars(decoder_list = msg_list, - message = message) + @set_calibration_rec_rr.setter + def set_calibration_rec_rr(self, value): + self._set_calibration_rec_rr = value - if 'bicarb' not in self.calibration_records_rr: - self.calibration_records_rr['bicarb'] = {} - if result['id'] not in self.calibration_records_rr: - self.calibration_records_rr['bicarb'][result['id']] = {} - self.calibration_records_rr['bicarb'][result['id']]['ack'] = result['ack'] - self.calibration_records_rr['bicarb'][result['id']]['rr'] = result['rr'] - self.calibration_set_rr_timestamp = timestamp - - - @publish(["msg_id_dd_nvm_set_cal_accel_sensor_response", "calibration_records_rr", "calibration_set_rr_timestamp"]) - def _handler_cal_accel_rr_sync(self, message, timestamp = 0.0): + @property + def set_institutional_rec_rr(self) -> dict: """ - Handles published DD Set Calibration Records for Accelerometer Sensor Reject Reason data messages.DD Set Calibration Records for Accelerometer Sensor Reject Reason are captured - for reference. - - @param message: published data message - @return: none + The Set Institutional Records Request reject reason data """ - msg_list = [] - msg_list.append(('ack', DataTypes.BOOL)) - msg_list.append(('rr', DataTypes.U32)) + return self._set_institutional_rec_rr - result = self.process_into_vars(decoder_list = msg_list, - message = message) + @set_institutional_rec_rr.setter + def set_institutional_rec_rr(self, value): + self._set_institutional_rec_rr = value - if 'accel' not in self.calibration_records_rr: - self.calibration_records_rr['accel'] = {} - self.calibration_records_rr['accel']['ack'] = result['ack'] - self.calibration_records_rr['accel']['rr'] = result['rr'] - self.calibration_set_rr_timestamp = timestamp - - - @publish(["msg_id_dd_nvm_set_cal_blood_leak_sensor_response", "calibration_records_rr", "calibration_set_rr_timestamp"]) - def _handler_cal_blood_leak_rr_sync(self, message, timestamp = 0.0): + @property + def set_usage_info_rec_rr(self) -> dict: """ - Handles published DD Set Calibration Records for Blood Leak Sensor Reject Reason data messages.DD Set Calibration Records for Blood Leak Sensor Reject Reason are captured - for reference. - - @param message: published data message - @return: none + The Set Usage Information Records Request reject reason data """ - msg_list = [] - msg_list.append(('ack', DataTypes.BOOL)) - msg_list.append(('rr', DataTypes.U32)) + return self._set_usage_info_rec_rr - result = self.process_into_vars(decoder_list = msg_list, - message = message) + @set_usage_info_rec_rr.setter + def set_usage_info_rec_rr(self, value): + self._set_usage_info_rec_rr = value - if 'blood_leak' not in self.calibration_records_rr: - self.calibration_records_rr['blood_leak'] = {} - self.calibration_records_rr['blood_leak']['ack'] = result['ack'] - self.calibration_records_rr['blood_leak']['rr'] = result['rr'] - self.calibration_set_rr_timestamp = timestamp - # ================================================= Go to Service Mode Method ================================================= def cmd_initiate_service_mode(self) -> int: """ @@ -852,628 +426,3 @@ override_text = f'set to {service_mode.name}', logger = self.logger, can_interface = self.can_interface) - - - -# ================================================= Request Records Methods ================================================= - def cmd_request_system_records(self) -> int: - """ - Constructs and sends a request for System Records. - Constraints: - Must be logged into DD. - Must be in Service mode. - - @return: 1 if successful, zero otherwise - """ - payload = byte_to_bytearray(dd_enum_repository.RecordTypes.SYSTEM_RECORD.value) - - return cmd_generic_override( - payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_dd_ch_id, - msg_id = MsgIds.MSG_ID_UI_DD_NVM_GET_RECORD_REQUEST, - entity_name = f'DD System Record request', - override_text = '', - logger = self.logger, - can_interface = self.can_interface) - - - def cmd_request_service_records(self) -> int: - """ - Constructs and sends a request for Service Records. - Constraints: - Must be logged into DD. - Must be in Service mode. - - @return: 1 if successful, zero otherwise - """ - payload = byte_to_bytearray(dd_enum_repository.RecordTypes.SERVICE_RECORD.value) - - return cmd_generic_override( - payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_dd_ch_id, - msg_id = MsgIds.MSG_ID_UI_DD_NVM_GET_RECORD_REQUEST, - entity_name = f'DD Service Record request', - override_text = '', - logger = self.logger, - can_interface = self.can_interface) - - - def cmd_request_calibration_records(self) -> int: - """ - Constructs and sends a request for Calibration Records. - Constraints: - Must be logged into DD. - Must be in Service mode. - - @return: 1 if successful, zero otherwise - """ - payload = byte_to_bytearray(dd_enum_repository.RecordTypes.CALIBRATION_RECORD.value) - - return cmd_generic_override( - payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_dd_ch_id, - msg_id = MsgIds.MSG_ID_UI_DD_NVM_GET_RECORD_REQUEST, - entity_name = f'DD Calibration Record request', - override_text = '', - logger = self.logger, - can_interface = self.can_interface) - - - def cmd_request_institutional_records(self) -> int: - """ - Constructs and sends a request for Institutional Records. - Constraints: - Must be logged into DD. - Must be in Service mode. - - @return: 1 if successful, zero otherwise - """ - payload = byte_to_bytearray(dd_enum_repository.RecordTypes.INSTITUTIONAL_RECORD.value) - - return cmd_generic_override( - payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_dd_ch_id, - msg_id = MsgIds.MSG_ID_UI_DD_NVM_GET_RECORD_REQUEST, - entity_name = f'DD Institutional Record request', - override_text = '', - logger = self.logger, - can_interface = self.can_interface) - - - def cmd_request_usage_information_records(self) -> int: - """ - Constructs and sends a request for Usage Information Records. - Constraints: - Must be logged into DD. - Must be in Service mode. - - @return: 1 if successful, zero otherwise - """ - payload = byte_to_bytearray(dd_enum_repository.RecordTypes.USAGE_INFORMATION_RECORD.value) - - return cmd_generic_override( - payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_dd_ch_id, - msg_id = MsgIds.MSG_ID_UI_DD_NVM_GET_RECORD_REQUEST, - entity_name = f'DD Usage Information Record request', - override_text = '', - logger = self.logger, - can_interface = self.can_interface) - - - -# ================================================= 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: - """ - Constructs and sends a command for setting the System Records. - Constraints: - Must be logged into DD. - Must be in Service mode. - - @return: 1 if successful, zero otherwise - """ - if len(part_number) != 10: - raise ValueError(f'part_number must be 10 characters, provided is: {len(part_number)}') - if len(serial_number) != 20: - raise ValueError(f'serial must be 20 characters, provided is: {len(serial_number)}') - - 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 cmd_generic_override( - payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_dd_ch_id, - msg_id = MsgIds.MSG_ID_UI_DD_NVM_SET_SYSTEM_RECORD_REQUEST, - entity_name = f'New DD System Record', - override_text = 'being set', - logger = self.logger, - can_interface = self.can_interface) - - - 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: - """ - Constructs and sends a command for setting the Service Records. - Constraints: - Must be logged into DD. - Must be in Service mode. - - @return: 1 if successful, zero otherwise - """ - 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 cmd_generic_override( - payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_dd_ch_id, - msg_id = MsgIds.MSG_ID_UI_DD_NVM_SET_SERVICE_RECORD_REQUEST, - entity_name = f'New DD Service Record', - override_text = 'being set', - logger = self.logger, - can_interface = self.can_interface) - - - def cmd_set_calibration_records(self, calibration_records: dict={}) -> int: - """ - Constructs and sends a command for setting the Calibration Records. - Constraints: - Must be logged into DD. - Must be in Service mode. - - @param calibration_records: (Dictionary) The stored and structured Calibration Records - @return: 1 if successful, zero otherwise - """ - msg_id_pairing = { - 'PRES': MsgIds.MSG_ID_UI_DD_NVM_SET_CAL_PRESSURE_SENSOR_REQUEST, - 'TEMP': MsgIds.MSG_ID_UI_DD_NVM_SET_CAL_TEMP_SENSOR_REQUEST, - 'CONC': MsgIds.MSG_ID_UI_DD_NVM_SET_CAL_CONC_PUMP_REQUEST, - 'DIAL': MsgIds.MSG_ID_UI_DD_NVM_SET_CAL_D48_PUMP_REQUEST, - 'ACID': MsgIds.MSG_ID_UI_DD_NVM_SET_CAL_ACID_CONCENTRATE_REQUEST, - 'BICARB': MsgIds.MSG_ID_UI_DD_NVM_SET_CAL_BICARB_CONCENTRATE_REQUEST, - 'ACCEL': MsgIds.MSG_ID_UI_DD_NVM_SET_CAL_ACCEL_SENSOR_REQUEST, - 'BLOOD_LEAK': MsgIds.MSG_ID_UI_DD_NVM_SET_CAL_BLOOD_LEAK_SENSOR_REQUEST, - } - # Make a dictionary to store all the send functions for later send - send_data = [] - checked = [] - for record_enum in dd_enum_repository.CalibrationRecordFields: - record_parts = record_enum.name.split('__') - group = record_parts[0] - - # 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 - - if group in ['PRES', 'TEMP', 'CONC', 'DIAL']: - # Set the sensor name - if group == 'PRES': - sensor = dd_enum_repository.DDPressureSensorNames.from_str(record_parts[1]) - if sensor is None: - sensor = fp_enum_repository.FPPressureSensorNames.from_str(record_parts[1]) - elif group == 'TEMP': - sensor = dd_enum_repository.DDTemperatureSensorNames.from_str(record_parts[1]) - elif group == 'CONC': - sensor = dd_enum_repository.DDConcentratePumpNames.from_str(record_parts[1]) - elif group == 'DIAL': - sensor = dd_enum_repository.DDDialysatePumpNames.from_str(record_parts[1]) - else: - sensor = 'UNKNOWN' - - # Set the arguments for the call - args = [] - kwargs = {} - target_speed_entry_name = f'{group}__{sensor.name}__TARGET_SPEED' - fourth_order_coeff_entry_name = f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordSensorFields.FOURTH_ORDER_COEFF.name}' - third_order_coeff_entry_name = f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordSensorFields.THIRD_ORDER_COEFF.name}' - second_order_coeff_entry_name = f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordSensorFields.SECOND_ORDER_COEFF.name}' - gain_entry_name = f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordSensorFields.GAIN.name}' - offset_entry_name = f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordSensorFields.OFFSET.name}' - calibration_time_entry_name = f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordSensorFields.CALIBRATION_TIME.name}' - - kwargs['sensor_enum'] = sensor - if sensor == dd_enum_repository.DDDialysatePumpNames.D12_PUMP: - kwargs['msg_id'] = MsgIds.MSG_ID_UI_DD_NVM_SET_CAL_D12_PUMP_REQUEST - else: - kwargs['msg_id'] = msg_id_pairing[group] - - if target_speed_entry_name in calibration_records and calibration_records[target_speed_entry_name] is not None: - kwargs['dialysate_pump_target_speed'] = calibration_records[target_speed_entry_name] - - if fourth_order_coeff_entry_name in calibration_records and calibration_records[fourth_order_coeff_entry_name] is not None: - kwargs['fourth_order_coeff'] = calibration_records[fourth_order_coeff_entry_name] - - 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] - - if gain_entry_name in calibration_records and calibration_records[gain_entry_name] is not None: - kwargs['gain'] = calibration_records[gain_entry_name] - - 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] - - 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 - - # Create the send data function - send_data.append(partial(self.cmd_set_calibration_records_sensor, *args, **kwargs)) - - elif group in ['ACID', 'BICARB']: - if group == 'ACID': - sensor = disp_enum_repository.AcidTypes.from_str(record_parts[1]) - elif group == 'BICARB': - sensor = disp_enum_repository.BicarbTypes.from_str(record_parts[1]) - else: - sensor = 'unknown' - - # Set the arguments for the call - args = [] - kwargs = {} - concentrate_mix_ratio_entry_name = f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordConcentrateFields.CONCENTRATE_MIX_RATIO.name}' - volume_ml_entry_name = f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordConcentrateFields.FULL_BOTTLE_VOLUME_ML.name}' - conductivity_uspcm_entry_name = f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordConcentrateFields.CONDUCTIVITY_USPCM.name}' - temperature_c_entry_name = f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordConcentrateFields.BOTTLE_TEMPERATURE_C.name}' - calibration_time_entry_name = f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordConcentrateFields.CALIBRATION_TIME.name}' - - kwargs['sensor_enum'] = sensor - kwargs['msg_id'] = msg_id_pairing[group] - if concentrate_mix_ratio_entry_name in calibration_records and calibration_records[concentrate_mix_ratio_entry_name] is not None: - kwargs['concentrate_mix_ratio'] = calibration_records[concentrate_mix_ratio_entry_name] - - if volume_ml_entry_name in calibration_records and calibration_records[volume_ml_entry_name] is not None: - kwargs['volume_ml'] = calibration_records[volume_ml_entry_name] - - if conductivity_uspcm_entry_name in calibration_records and calibration_records[conductivity_uspcm_entry_name] is not None: - kwargs['conductivity_uspcm'] = calibration_records[conductivity_uspcm_entry_name] - - if temperature_c_entry_name in calibration_records and calibration_records[temperature_c_entry_name] is not None: - kwargs['temperature_c'] = calibration_records[temperature_c_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] - - checked.append(record_parts[1]) - # 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_concentrate, *args, **kwargs)) - - elif group in ['ACCEL']: - # Set the arguments for the call - args = [] - kwargs = {} - accel_x_offset_entry_name = f'{group}__{dd_enum_repository.CalibRecordAccelerometerFields.ACCEL_X_OFFSET.name}' - accel_y_offset_entry_name = f'{group}__{dd_enum_repository.CalibRecordAccelerometerFields.ACCEL_Y_OFFSET.name}' - accel_z_offset_entry_name = f'{group}__{dd_enum_repository.CalibRecordAccelerometerFields.ACCEL_Z_OFFSET.name}' - calibration_time_entry_name = f'{group}__{dd_enum_repository.CalibRecordAccelerometerFields.CALIBRATION_TIME.name}' - - 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] - - 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] - - 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] - - 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)) - - elif group in ['BLOOD_LEAK']: - # Set the arguments for the call - args = [] - kwargs = {} - set_point_entry_name = f'{group}__{dd_enum_repository.CalibRecordBloodLeakFields.SET_POINT.name}' - calibration_time_entry_name = f'{group}__{dd_enum_repository.CalibRecordBloodLeakFields.CALIBRATION_TIME.name}' - - kwargs['msg_id'] = msg_id_pairing[group] - if set_point_entry_name in calibration_records and calibration_records[set_point_entry_name] is not None: - kwargs['set_point'] = calibration_records[set_point_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] - - 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_blood_leak, *args, **kwargs)) - - # 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 - - - def cmd_set_institutional_records(self, institutional_records: dict={}) -> int: - """ - Constructs and sends a command for setting the Institutional Records. - Constraints: - Must be logged into DD. - Must be in Service mode. - - @param institutional_records: (Dictionary) The stored and structured Institutional Records - @return: 1 if successful, zero otherwise - """ - # Make a dictionary to store all the send functions for later send - send_data = [] - for record_enum in dd_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) - - if record_enum in [dd_enum_repository.InstitutionalRecordFields.CRC, dd_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 - - 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_REQUEST, - entity_name = f'DD Institutional Record', - override_text = 'being set', - logger = self.logger, - can_interface = self.can_interface)) - - # 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 - - - 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: - """ - Constructs and sends a command for setting the Usage Information Records. - Constraints: - Must be logged into DD. - Must be in Service mode. - - @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)) - - return cmd_generic_override( - payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_dd_ch_id, - msg_id = MsgIds.MSG_ID_UI_DD_NVM_SET_USAGE_INFO_RECORD_REQUEST, - entity_name = f'New DD Usage Information Record', - override_text = 'being set', - 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, - fourth_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 = 1) -> int: - if isinstance(sensor_enum, fp_enum_repository.FPPressureSensorNames): - payload_sensor = byte_to_bytearray(sensor_enum.value + dd_enum_repository.DDPressureSensorNames.NUM_OF_PRESSURE_SENSORS.value) - else: - payload_sensor = byte_to_bytearray(sensor_enum.value) - payload_content = b'' - if msg_id == MsgIds.MSG_ID_UI_DD_NVM_SET_CAL_D12_PUMP_REQUEST: - payload_content += float_to_bytearray(dialysate_pump_target_speed) - else: - payload_content += float_to_bytearray(fourth_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)) - if msg_id in [MsgIds.MSG_ID_UI_DD_NVM_SET_CAL_D12_PUMP_REQUEST, MsgIds.MSG_ID_UI_DD_NVM_SET_CAL_D48_PUMP_REQUEST]: - payload = payload_content - else: - payload = payload_sensor + payload_content - - sensor_type = '' - if msg_id == MsgIds.MSG_ID_UI_DD_NVM_SET_CAL_PRESSURE_SENSOR_REQUEST: - sensor_type = 'Pressure Sensor' - elif msg_id == MsgIds.MSG_ID_UI_DD_NVM_SET_CAL_TEMP_SENSOR_REQUEST: - sensor_type = 'Temperature Sensor' - elif msg_id == MsgIds.MSG_ID_UI_DD_NVM_SET_CAL_CONC_PUMP_REQUEST: - sensor_type = 'Concentrate Pump' - elif msg_id in [MsgIds.MSG_ID_UI_DD_NVM_SET_CAL_D12_PUMP_REQUEST, MsgIds.MSG_ID_UI_DD_NVM_SET_CAL_D48_PUMP_REQUEST]: - sensor_type = 'Dialysate Pump' - - return cmd_generic_override(payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_dd_ch_id, - msg_id = msg_id, - entity_name = f'DD {sensor_type} {sensor_enum.name} Calibration Record', - override_text = 'being set', - logger = self.logger, - can_interface = self.can_interface) - - - def cmd_set_calibration_records_concentrate(self, - sensor_enum: DialinEnum, - msg_id: MsgIds, - concentrate_mix_ratio: float=0.0, - volume_ml: float=0.0, - conductivity_uspcm: float=0.0, - temperature_c: float=0.0, - calibration_time: int = 1) -> int: - - payload_sensor = byte_to_bytearray(sensor_enum.value) - payload_content = float_to_bytearray(concentrate_mix_ratio) - payload_content += float_to_bytearray(volume_ml) - payload_content += float_to_bytearray(conductivity_uspcm) - payload_content += float_to_bytearray(temperature_c) - payload_content += integer_to_bytearray(calibration_time) - payload_content += unsigned_short_to_bytearray(self.crc16(payload_content)) - payload = payload_sensor + payload_content - - conc_type = '' - if msg_id == MsgIds.MSG_ID_UI_DD_NVM_SET_CAL_ACID_CONCENTRATE_REQUEST: - conc_type = 'Acid' - elif msg_id == MsgIds.MSG_ID_UI_DD_NVM_SET_CAL_BICARB_CONCENTRATE_REQUEST: - conc_type = 'Bicarb' - - return cmd_generic_override(payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_dd_ch_id, - msg_id = msg_id, - entity_name = f'DD {conc_type} Concentrate 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 = 1) -> 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)) - - return cmd_generic_override(payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_dd_ch_id, - msg_id = msg_id, - entity_name = f'DD Accelerometer Calibration Record', - override_text = 'being set', - logger = self.logger, - can_interface = self.can_interface) - - - - def cmd_set_calibration_records_blood_leak(self, - msg_id: MsgIds, - set_point: int=0, - calibration_time: int = 1) -> int: - - payload = unsigned_short_to_bytearray(set_point) - payload += integer_to_bytearray(calibration_time) - payload += unsigned_short_to_bytearray(self.crc16(payload)) - - return cmd_generic_override(payload = payload, - reset = None, - channel_id = CanChannels.dialin_to_dd_ch_id, - msg_id = msg_id, - entity_name = f'DD Blood Leak Calibration Record', - override_text = 'being set', - logger = self.logger, - can_interface = self.can_interface) - - - -# ================================================= Private Methods =================================================