########################################################################### # # Copyright (c) 2020-2024 Diality Inc. - All Rights Reserved. # # THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN # WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. # # @file records.py # # @author (last) Zoltan Miskolci # @date (last) 22-Jun-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.common.msg_defs 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 class DDRecords(AbstractSubSystem): """ DD interface containing pressure related commands. """ def __init__(self, can_interface: CanMessenger, logger: Logger): """ @param can_interface: The CanMessenger 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_USAGE_INFO_RECORD_RESPONSE.value, function = self._handler_usage_info_rr_sync) 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.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 = { } #: 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.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.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.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 for sys_record in dd_enum_repository.SystemRecordFields: if 'num_' not in sys_record.name: self.system_records[sys_record.name] = None for serv_record in dd_enum_repository.ServiceRecordFields: if 'num_' not in serv_record.name: self.service_records[serv_record.name] = None for cal_record in dd_enum_repository.CalibrationRecordFields: if 'num_' not in cal_record.name: self.calibration_records[cal_record.name] = None for inst_record in dd_enum_repository.InstitutionalRecordFields: if 'num_' not in inst_record.name: self.institutional_records[inst_record.name] = None for usage_record in dd_enum_repository.UsageInformationRecordFields: if 'num_' not in usage_record.name: self.usage_info_records[usage_record.name] = None # ================================================= 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. @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): """ Handles published DD Service Record data messages. DD Service Records are captured for reference. @param message: published data message @return: none """ record_list = [] for member in dd_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 @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 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 """ # Get the sensor_id to identify the location of the payload print(f'Pres cal msg recieved') print(message) 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 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())) 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): """ 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 """ # Get the sensor_id to identify the location of the payload print(f'Temp cal msg recieved') 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 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())) 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): """ 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 """ # 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 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())) 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): """ 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 """ record_list = [] record_list.append((f'DIAL__{dd_enum_repository.DDDialysatePumpNames.D12_PUMP.name}__TARGET_SPEED', DataTypes.F32)) 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())) self.process_into_dict(dict_to_update = self.calibration_records, decoder_list = record_list, message = message) 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): """ 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 """ 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())) self.process_into_dict(dict_to_update = self.calibration_records, decoder_list = record_list, message = message) 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): """ 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 """ 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 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())) self.process_into_dict(dict_to_update = self.calibration_records, decoder_list = record_list, message = message) 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): """ 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 """ 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 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())) self.process_into_dict(dict_to_update = self.calibration_records, decoder_list = record_list, message = message) 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): """ 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 """ 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())) self.process_into_dict(dict_to_update = self.calibration_records, decoder_list = record_list, message = message) 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): """ 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 """ 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())) self.process_into_dict(dict_to_update = self.calibration_records, decoder_list = record_list, message = message) 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): """ Handles published DD Institutional Record data messages. DD Institutional Records are captured for reference. @param message: published data message @return: none """ msg_list = [('record_id', DataTypes.U08)] results = self.process_into_vars(decoder_list = msg_list, message = message) record_id = int(results['record_id']) member = dd_enum_repository.InstitutionalRecordFields(record_id) record_list = [] record_list.append((member.name, member.datatype())) 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): """ Handles published DD Usage Information Record data messages. DD Usage Information Records are captured for reference. @param message: published data message @return: none """ record_list = [] for member in dd_enum_repository.UsageInformationRecordFields: record_list.append((member.name, member.datatype())) self.process_into_dict(dict_to_update = self.usage_info_records, decoder_list = record_list, message = message) 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): """ 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 """ msg_list = [] msg_list.append(('self.system_records_ack', DataTypes.BOOL)) msg_list.append(('self.system_records_rr', DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) 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): """ 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 """ msg_list = [] msg_list.append(('self.service_records_ack', DataTypes.BOOL)) msg_list.append(('self.service_records_rr', DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) self.service_set_rr_timestamp = timestamp @publish(["msg_id_dd_nvm_set_institutional_record_response", "institutional_records_ack", "institutional_records_rr", "id", "institutional_set_rr_timestamp"]) def _handler_institutional_rr_sync(self, message, timestamp = 0.0): """ 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 """ msg_list = [] msg_list.append(('self.institutional_records_ack', DataTypes.BOOL)) msg_list.append(('self.institutional_records_rr', DataTypes.U32)) msg_list.append(('self.id', DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) 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): """ 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 """ msg_list = [] msg_list.append(('self.usage_info_ack', DataTypes.BOOL)) msg_list.append(('self.usage_info_rr', DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) self.usage_info_set_rr_timestamp = timestamp # ================================================= Go to Service Mode Method ================================================= def cmd_initiate_service_mode(self) -> int: """ Constructs and sends a request to change to Service operation mode via CAN bus. Constraints: Must be logged into DD. Transition from current to requested op mode must be legal. @return: 1 if successful, zero otherwise """ service_mode = dd_enum_repository.DDOpModes.MODE_SERV payload = integer_to_bytearray(service_mode.value) return cmd_generic_override( payload = payload, reset = None, channel_id = CanChannels.dialin_to_dd_ch_id, msg_id = MsgIds.MSG_ID_DD_SET_OPERATION_MODE_OVERRIDE_REQUEST, entity_name = 'DD Operation Mode', 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['forth_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 print(f'kwargs: {kwargs}') # 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 print(f'kwargs: {kwargs}') # 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 print(f'kwargs: {kwargs}') # 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 print(f'kwargs: {kwargs}') # 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) 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)) 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: float=0.0, calibration_time: int = 1) -> int: payload = float_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 =================================================