########################################################################### # # 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_get_responses.py # # @author (last) Zoltan Miskolci # @date (last) 04-Aug-2026 # @author (original) Zoltan Miskolci # @date (original) 04-Aug-2026 # ############################################################################ # 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 RequestRejectReasons from leahi_dialin.common.msg_ids import MsgIds from leahi_dialin.protocols.CAN import CanChannels from leahi_dialin.utils.abstract_classes import AbstractSubSystem, LocalVars from leahi_dialin.utils.base import publish class DDRecordsGetHandlers(AbstractSubSystem): """ Part of the DD Records """ def __init__(self, *args, **kwargs): """ The sub record class to handle all the get handlers. """ 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_UI_NVM_GET_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_UI_NVM_GET_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_UI_NVM_GET_PRESSURE_SENSOR_CAL_RECORD_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_UI_NVM_GET_TEMPERATURE_SENSOR_CAL_RECORD_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_UI_NVM_GET_CONC_PUMP_CAL_RECORD_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_UI_NVM_GET_D12_PUMP_CAL_RECORD_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_UI_NVM_GET_D48_PUMP_CAL_RECORD_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_UI_NVM_GET_ACID_CONC_CAL_RECORD_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_UI_NVM_GET_BICARB_CONC_CAL_RECORD_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_UI_NVM_GET_ACCEL_SENSOR_CAL_RECORD_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_UI_NVM_GET_BLOOD_LEAK_SENSOR_CAL_RECORD_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_UI_NVM_GET_CONDUCTIVITY_SENSOR_CAL_RECORD_RESPONSE.value, function = self._handler_calibration_record_cond_sensor_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_DD_UI_NVM_GET_INSTIT_BASIC_RECORD_RESPONSE.value, function = self._handler_institutional_basic_record_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_DD_UI_NVM_GET_INSTIT_ADVANCED_RECORD_RESPONSE.value, function = self._handler_institutional_adv_record_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_DD_UI_NVM_GET_INSTIT_ADDITIONAL_RECORD_RESPONSE.value, function = self._handler_institutional_add_record_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_DD_UI_NVM_GET_USAGE_INFO_RECORD_RESPONSE.value, function = self._handler_usage_info_record_sync) # Passing the arguments over to the next class too super().__init__(*args, **kwargs) # ================================================= CAN Message Handler Methods ================================================= @publish([MsgIds.MSG_ID_DD_UI_NVM_GET_SYSTEM_RECORD_RESPONSE.name.lower(), "get_system_rec_ack", "get_system_rec_rr", "system_records", "get_system_rec_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 """ msg_list = [] msg_list.append((type(self).get_system_rec_ack, DataTypes.BOOL)) msg_list.append((type(self).get_system_rec_rr, DataTypes.U32)) for member in dd_enum_repository.SystemRecordFields: if member is not dd_enum_repository.SystemRecordFields.NUM_OF_SYSTEM_RECORD_FIELDS: msg_list.append((self.system_records, member.name, member.datatype(), member.multichar_length())) self.process_into_vars(decoder_list = msg_list, message = message) # Convert to enum if isinstance(self.get_system_rec_rr, int): self.get_system_rec_rr = RequestRejectReasons(self.get_system_rec_rr) self.get_system_rec_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_GET_SERVICE_RECORD_RESPONSE.name.lower(), "get_service_rec_ack", "get_service_rec_rr", "service_records", "get_service_rec_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 """ msg_list = [] msg_list.append((type(self).get_service_rec_ack, DataTypes.BOOL)) msg_list.append((type(self).get_service_rec_rr, DataTypes.U32)) for member in dd_enum_repository.ServiceRecordFields: if member is not dd_enum_repository.ServiceRecordFields.NUM_OF_SERVICE_RECORD_FIELDS: msg_list.append((self.system_records, member.name, member.datatype())) self.process_into_vars(decoder_list = msg_list, message = message) # Convert to enum if isinstance(self.get_service_rec_rr, int): self.get_service_rec_rr = RequestRejectReasons(self.get_service_rec_rr) self.get_service_rec_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_GET_PRESSURE_SENSOR_CAL_RECORD_RESPONSE.name.lower(), "get_calibration_rec_ack", "get_calibration_rec_rr", "calibration_records", "get_calibration_rec_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 """ acceptance = LocalVars('acceptance') reject_reason = LocalVars('reject_reason') sensor_id = LocalVars('sensor_id') gain = LocalVars(dd_enum_repository.CalibRecordSensorFields.GAIN.name) offset = LocalVars(dd_enum_repository.CalibRecordSensorFields.OFFSET.name) calibration_time = LocalVars(dd_enum_repository.CalibRecordSensorFields.CALIBRATION_TIME.name) msg_list = [] msg_list.append((acceptance, DataTypes.BOOL)) msg_list.append((reject_reason, DataTypes.U32)) msg_list.append((sensor_id, DataTypes.U32)) msg_list.append((gain, DataTypes.F32)) msg_list.append((offset, DataTypes.F32)) msg_list.append((calibration_time, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) # Get the Pressure Sensor Name if sensor_id.value < dd_enum_repository.DDPressureSensorNames.NUM_OF_PRESSURE_SENSORS.value: sensor_name = dd_enum_repository.DDPressureSensorNames(sensor_id.value).name else: sensor_name = fp_enum_repository.FPPressureSensorNames(sensor_id.value - dd_enum_repository.DDPressureSensorNames.NUM_OF_PRESSURE_SENSORS.value).name # Saving into the calibration data if acceptance.value: self.calibration_records[f'PRES__{sensor_name}__{gain.name}'] = gain.value self.calibration_records[f'PRES__{sensor_name}__{offset.name}'] = offset.value self.calibration_records[f'PRES__{sensor_name}__{calibration_time.name}'] = calibration_time.value else: self.get_calibration_rec_ack[f'PRES__{sensor_name}'] = acceptance.value self.get_calibration_rec_rr[f'PRES__{sensor_name}'] = RequestRejectReasons(reject_reason.value) self.get_calibration_rec_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_GET_TEMPERATURE_SENSOR_CAL_RECORD_RESPONSE.name.lower(), "get_calibration_rec_ack", "get_calibration_rec_rr", "calibration_records", "get_calibration_rec_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 """ acceptance = LocalVars('acceptance') reject_reason = LocalVars('reject_reason') sensor_id = LocalVars('sensor_id') gain = LocalVars(dd_enum_repository.CalibRecordSensorFields.GAIN.name) offset = LocalVars(dd_enum_repository.CalibRecordSensorFields.OFFSET.name) calibration_time = LocalVars(dd_enum_repository.CalibRecordSensorFields.CALIBRATION_TIME.name) msg_list = [] msg_list.append((acceptance, DataTypes.BOOL)) msg_list.append((reject_reason, DataTypes.U32)) msg_list.append((sensor_id, DataTypes.U32)) msg_list.append((gain, DataTypes.F32)) msg_list.append((offset, DataTypes.F32)) msg_list.append((calibration_time, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) # Get the Temperature Sensor Name sensor_name = dd_enum_repository.DDTemperatureSensorNames(sensor_id.value).name # Saving into the calibration data if acceptance.value: self.calibration_records[f'TEMP__{sensor_name}__{gain.name}'] = gain.value self.calibration_records[f'TEMP__{sensor_name}__{offset.name}'] = offset.value self.calibration_records[f'TEMP__{sensor_name}__{calibration_time.name}'] = calibration_time.value else: self.get_calibration_rec_ack[f'TEMP__{sensor_name}'] = acceptance.value self.get_calibration_rec_rr[f'TEMP__{sensor_name}'] = RequestRejectReasons(reject_reason.value) self.get_calibration_rec_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_GET_CONC_PUMP_CAL_RECORD_RESPONSE.name.lower(), "get_calibration_rec_ack", "get_calibration_rec_rr", "calibration_records", "get_calibration_rec_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 """ acceptance = LocalVars('acceptance') reject_reason = LocalVars('reject_reason') sensor_id = LocalVars('sensor_id') gain = LocalVars(dd_enum_repository.CalibRecordSensorFields.GAIN.name) offset = LocalVars(dd_enum_repository.CalibRecordSensorFields.OFFSET.name) calibration_time = LocalVars(dd_enum_repository.CalibRecordSensorFields.CALIBRATION_TIME.name) msg_list = [] msg_list.append((acceptance, DataTypes.BOOL)) msg_list.append((reject_reason, DataTypes.U32)) msg_list.append((sensor_id, DataTypes.U32)) msg_list.append((gain, DataTypes.F32)) msg_list.append((offset, DataTypes.F32)) msg_list.append((calibration_time, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) # Get the Concentrate Pump Name pump_name = dd_enum_repository.DDConcentratePumpNames(sensor_id.value).name # Saving into the calibration data if acceptance.value: self.calibration_records[f'CONC__{pump_name}__{gain.name}'] = gain.value self.calibration_records[f'CONC__{pump_name}__{offset.name}'] = offset.value self.calibration_records[f'CONC__{pump_name}__{calibration_time.name}'] = calibration_time.value else: self.get_calibration_rec_ack[f'CONC__{pump_name}'] = acceptance.value self.get_calibration_rec_rr[f'CONC__{pump_name}'] = RequestRejectReasons(reject_reason.value) self.get_calibration_rec_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_GET_D12_PUMP_CAL_RECORD_RESPONSE.name.lower(), "get_calibration_rec_ack", "get_calibration_rec_rr", "calibration_records", "get_calibration_rec_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 """ acceptance = LocalVars('acceptance') reject_reason = LocalVars('reject_reason') target_speed = LocalVars('TARGET_SPEED') calibration_time = LocalVars(dd_enum_repository.CalibRecordSensorFields.CALIBRATION_TIME.name) msg_list = [] msg_list.append((acceptance, DataTypes.BOOL)) msg_list.append((reject_reason, DataTypes.U32)) msg_list.append((target_speed, DataTypes.F32)) msg_list.append((calibration_time, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) # Get the Dialysate Pump Name pump_name = dd_enum_repository.DDDialysatePumpNames.D12_PUMP.name # Saving into the calibration data if acceptance.value: self.calibration_records[f'DIAL__{pump_name}__{target_speed.name}'] = target_speed.value self.calibration_records[f'DIAL__{pump_name}__{calibration_time.name}'] = calibration_time.value else: self.get_calibration_rec_ack[f'DIAL__{pump_name}'] = acceptance.value self.get_calibration_rec_rr[f'DIAL__{pump_name}'] = RequestRejectReasons(reject_reason.value) self.get_calibration_rec_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_GET_D48_PUMP_CAL_RECORD_RESPONSE.name.lower(), "get_calibration_rec_ack", "get_calibration_rec_rr", "calibration_records", "get_calibration_rec_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 """ acceptance = LocalVars('acceptance') reject_reason = LocalVars('reject_reason') gain = LocalVars(dd_enum_repository.CalibRecordSensorFields.GAIN.name) offset = LocalVars(dd_enum_repository.CalibRecordSensorFields.OFFSET.name) calibration_time = LocalVars(dd_enum_repository.CalibRecordSensorFields.CALIBRATION_TIME.name) msg_list = [] msg_list.append((acceptance, DataTypes.BOOL)) msg_list.append((reject_reason, DataTypes.U32)) msg_list.append((gain, DataTypes.F32)) msg_list.append((offset, DataTypes.F32)) msg_list.append((calibration_time, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) # Get the Dialysate Pump Name pump_name = dd_enum_repository.DDDialysatePumpNames.D48_PUMP.name # Saving into the calibration data if acceptance.value: self.calibration_records[f'DIAL__{pump_name}__{gain.name}'] = gain.value self.calibration_records[f'DIAL__{pump_name}__{offset.name}'] = offset.value self.calibration_records[f'DIAL__{pump_name}__{calibration_time.name}'] = calibration_time.value else: self.get_calibration_rec_ack[f'DIAL__{pump_name}'] = acceptance.value self.get_calibration_rec_rr[f'DIAL__{pump_name}'] = RequestRejectReasons(reject_reason.value) self.get_calibration_rec_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_GET_ACID_CONC_CAL_RECORD_RESPONSE.name.lower(), "get_calibration_rec_ack", "get_calibration_rec_rr", "calibration_records", "get_calibration_rec_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 """ acceptance = LocalVars('acceptance') reject_reason = LocalVars('reject_reason') conc_id = LocalVars('conc_id') mix_ratio = LocalVars(dd_enum_repository.CalibRecordConcentrateFields.CONCENTRATE_MIX_RATIO.name) volume = LocalVars(dd_enum_repository.CalibRecordConcentrateFields.FULL_BOTTLE_VOLUME_ML.name) conductivity = LocalVars(dd_enum_repository.CalibRecordConcentrateFields.CONDUCTIVITY_USPCM.name) temperature = LocalVars(dd_enum_repository.CalibRecordConcentrateFields.BOTTLE_TEMPERATURE_C.name) calibration_time = LocalVars(dd_enum_repository.CalibRecordConcentrateFields.CALIBRATION_TIME.name) msg_list = [] msg_list.append((acceptance, DataTypes.BOOL)) msg_list.append((reject_reason, DataTypes.U32)) msg_list.append((conc_id, DataTypes.U32)) msg_list.append((mix_ratio, DataTypes.F32)) msg_list.append((volume, DataTypes.F32)) msg_list.append((conductivity, DataTypes.F32)) msg_list.append((temperature, DataTypes.F32)) msg_list.append((calibration_time, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) # Get the Acid Concentrate's Name acid_name = disp_enum_repository.AcidTypes(conc_id.value).name # Saving into the calibration data if acceptance.value: self.calibration_records[f'ACID__{acid_name}__{mix_ratio.name}'] = mix_ratio.value self.calibration_records[f'ACID__{acid_name}__{volume.name}'] = volume.value self.calibration_records[f'ACID__{acid_name}__{conductivity.name}'] = conductivity.value self.calibration_records[f'ACID__{acid_name}__{temperature.name}'] = temperature.value self.calibration_records[f'ACID__{acid_name}__{calibration_time.name}'] = calibration_time.value else: self.get_calibration_rec_ack[f'ACID__{acid_name}'] = acceptance.value self.get_calibration_rec_rr[f'ACID__{acid_name}'] = RequestRejectReasons(reject_reason.value) self.get_calibration_rec_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_GET_BICARB_CONC_CAL_RECORD_RESPONSE.name.lower(), "get_calibration_rec_ack", "get_calibration_rec_rr", "calibration_records", "get_calibration_rec_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 """ acceptance = LocalVars('acceptance') reject_reason = LocalVars('reject_reason') conc_id = LocalVars('conc_id') mix_ratio = LocalVars(dd_enum_repository.CalibRecordConcentrateFields.CONCENTRATE_MIX_RATIO.name) volume = LocalVars(dd_enum_repository.CalibRecordConcentrateFields.FULL_BOTTLE_VOLUME_ML.name) conductivity = LocalVars(dd_enum_repository.CalibRecordConcentrateFields.CONDUCTIVITY_USPCM.name) temperature = LocalVars(dd_enum_repository.CalibRecordConcentrateFields.BOTTLE_TEMPERATURE_C.name) calibration_time = LocalVars(dd_enum_repository.CalibRecordConcentrateFields.CALIBRATION_TIME.name) msg_list = [] msg_list.append((acceptance, DataTypes.BOOL)) msg_list.append((reject_reason, DataTypes.U32)) msg_list.append((conc_id, DataTypes.U32)) msg_list.append((mix_ratio, DataTypes.F32)) msg_list.append((volume, DataTypes.F32)) msg_list.append((conductivity, DataTypes.F32)) msg_list.append((temperature, DataTypes.F32)) msg_list.append((calibration_time, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) # Get the Bicarb Concentrate's Name bicarb_name = disp_enum_repository.AcidTypes(conc_id.value).name # Saving into the calibration data if acceptance.value: self.calibration_records[f'BICARB__{bicarb_name}__{mix_ratio.name}'] = mix_ratio.value self.calibration_records[f'BICARB__{bicarb_name}__{volume.name}'] = volume.value self.calibration_records[f'BICARB__{bicarb_name}__{conductivity.name}'] = conductivity.value self.calibration_records[f'BICARB__{bicarb_name}__{temperature.name}'] = temperature.value self.calibration_records[f'BICARB__{bicarb_name}__{calibration_time.name}'] = calibration_time.value else: self.get_calibration_rec_ack[f'BICARB__{bicarb_name}'] = acceptance.value self.get_calibration_rec_rr[f'BICARB__{bicarb_name}'] = RequestRejectReasons(reject_reason.value) self.get_calibration_rec_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_GET_ACCEL_SENSOR_CAL_RECORD_RESPONSE.name.lower(), "get_calibration_rec_ack", "get_calibration_rec_rr", "calibration_records", "get_calibration_rec_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 """ acceptance = LocalVars('acceptance') reject_reason = LocalVars('reject_reason') accel_x = LocalVars(dd_enum_repository.CalibRecordAccelerometerFields.ACCEL_X_OFFSET.name) accel_y = LocalVars(dd_enum_repository.CalibRecordAccelerometerFields.ACCEL_Y_OFFSET.name) accel_z = LocalVars(dd_enum_repository.CalibRecordAccelerometerFields.ACCEL_Z_OFFSET.name) calibration_time = LocalVars(dd_enum_repository.CalibRecordAccelerometerFields.CALIBRATION_TIME.name) msg_list = [] msg_list.append((acceptance, DataTypes.BOOL)) msg_list.append((reject_reason, DataTypes.U32)) msg_list.append((accel_x, DataTypes.F32)) msg_list.append((accel_y, DataTypes.F32)) msg_list.append((accel_z, DataTypes.F32)) msg_list.append((calibration_time, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) # Saving into the calibration data if acceptance.value: self.calibration_records[f'ACCEL__{accel_x.name}'] = accel_x.value self.calibration_records[f'ACCEL__{accel_y.name}'] = accel_y.value self.calibration_records[f'ACCEL__{accel_z.name}'] = accel_z.value self.calibration_records[f'ACCEL__{calibration_time.name}'] = calibration_time.value else: self.get_calibration_rec_ack[f'ACCEL'] = acceptance.value self.get_calibration_rec_rr[f'ACCEL'] = RequestRejectReasons(reject_reason.value) self.get_calibration_rec_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_GET_BLOOD_LEAK_SENSOR_CAL_RECORD_RESPONSE.name.lower(), "get_calibration_rec_ack", "get_calibration_rec_rr", "calibration_records", "get_calibration_rec_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 """ acceptance = LocalVars('acceptance') reject_reason = LocalVars('reject_reason') set_point = LocalVars(dd_enum_repository.CalibRecordBloodLeakFields.SET_POINT.name) calibration_time = LocalVars(dd_enum_repository.CalibRecordBloodLeakFields.CALIBRATION_TIME.name) msg_list = [] msg_list.append((acceptance, DataTypes.BOOL)) msg_list.append((reject_reason, DataTypes.U32)) msg_list.append((set_point, DataTypes.F32)) msg_list.append((calibration_time, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) # Saving into the calibration data if acceptance.value: self.calibration_records[f'BLOOD_LEAK__{set_point.name}'] = set_point.value self.calibration_records[f'BLOOD_LEAK__{calibration_time.name}'] = calibration_time.value else: self.get_calibration_rec_ack[f'BLOOD_LEAK'] = acceptance.value self.get_calibration_rec_rr[f'BLOOD_LEAK'] = RequestRejectReasons(reject_reason.value) self.get_calibration_rec_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_GET_CONDUCTIVITY_SENSOR_CAL_RECORD_RESPONSE.name.lower(), "get_calibration_rec_ack", "get_calibration_rec_rr", "calibration_records", "get_calibration_rec_timestamp"]) def _handler_calibration_record_cond_sensor_sync(self, message, timestamp = 0.0): """ Handles published DD Calibration Record Conductivity Sensors Details data messages. DD Calibration RecordsConductivity Sensors Details are captured for reference. @param message: published data message @return: none """ acceptance = LocalVars('acceptance') reject_reason = LocalVars('reject_reason') sensor_id = LocalVars('sensor_id') k_high = LocalVars(dd_enum_repository.CalibRecordConductivityFields.K_HIGH.name) alpha_high = LocalVars(dd_enum_repository.CalibRecordConductivityFields.ALPHA_HIGH.name) eta_high = LocalVars(dd_enum_repository.CalibRecordConductivityFields.ETA_HIGH.name) zeta_high = LocalVars(dd_enum_repository.CalibRecordConductivityFields.ZETA_HIGH.name) k_low = LocalVars(dd_enum_repository.CalibRecordConductivityFields.K_LOW.name) alpha_low = LocalVars(dd_enum_repository.CalibRecordConductivityFields.ALPHA_LOW.name) eta_low = LocalVars(dd_enum_repository.CalibRecordConductivityFields.ETA_LOW.name) zeta_low = LocalVars(dd_enum_repository.CalibRecordConductivityFields.ZETA_LOW.name) beta = LocalVars(dd_enum_repository.CalibRecordConductivityFields.BETA.name) delta = LocalVars(dd_enum_repository.CalibRecordConductivityFields.DELTA.name) rsrvd_1 = LocalVars(dd_enum_repository.CalibRecordConductivityFields.RSRVD_1.name) rsrvd_2 = LocalVars(dd_enum_repository.CalibRecordConductivityFields.RSRVD_2.name) calibration_time = LocalVars(dd_enum_repository.CalibRecordBloodLeakFields.CALIBRATION_TIME.name) msg_list = [] msg_list.append((acceptance, DataTypes.BOOL)) msg_list.append((reject_reason, DataTypes.U32)) msg_list.append((sensor_id, DataTypes.U32)) msg_list.append((k_high, DataTypes.F32)) msg_list.append((alpha_high, DataTypes.F32)) msg_list.append((eta_high, DataTypes.F32)) msg_list.append((zeta_high, DataTypes.F32)) msg_list.append((k_low, DataTypes.F32)) msg_list.append((alpha_low, DataTypes.F32)) msg_list.append((eta_low, DataTypes.F32)) msg_list.append((zeta_low, DataTypes.F32)) msg_list.append((beta, DataTypes.F32)) msg_list.append((delta, DataTypes.F32)) msg_list.append((rsrvd_1, DataTypes.F32)) msg_list.append((rsrvd_2, DataTypes.F32)) msg_list.append((calibration_time, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) # Get the Conductivity Sensor Name sensor_name = dd_enum_repository.DDConductivitySensorNames(sensor_id.value).name # Saving into the calibration data if acceptance.value: self.calibration_records[f'COND__{sensor_name}__{k_high.name}'] = k_high.value self.calibration_records[f'COND__{sensor_name}__{alpha_high.name}'] = alpha_high.value self.calibration_records[f'COND__{sensor_name}__{eta_high.name}'] = eta_high.value self.calibration_records[f'COND__{sensor_name}__{zeta_high.name}'] = zeta_high.value self.calibration_records[f'COND__{sensor_name}__{k_low.name}'] = k_low.value self.calibration_records[f'COND__{sensor_name}__{alpha_low.name}'] = alpha_low.value self.calibration_records[f'COND__{sensor_name}__{eta_low.name}'] = eta_low.value self.calibration_records[f'COND__{sensor_name}__{zeta_low.name}'] = zeta_low.value self.calibration_records[f'COND__{sensor_name}__{beta.name}'] = beta.value self.calibration_records[f'COND__{sensor_name}__{delta.name}'] = delta.value self.calibration_records[f'COND__{sensor_name}__{rsrvd_1.name}'] = rsrvd_1.value self.calibration_records[f'COND__{sensor_name}__{rsrvd_2.name}'] = rsrvd_2.value self.calibration_records[f'COND__{sensor_name}__{calibration_time.name}'] = calibration_time.value else: self.get_calibration_rec_ack[f'COND__{sensor_name}'] = acceptance.value self.get_calibration_rec_rr[f'COND__{sensor_name}'] = RequestRejectReasons(reject_reason.value) self.get_calibration_rec_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_GET_INSTIT_BASIC_RECORD_RESPONSE.name.lower(), "get_institutional_rec_ack", "get_institutional_rec_rr", "institutional_records", "get_institutional_rec_timestamp"]) def _handler_institutional_basic_record_sync(self, message, timestamp = 0.0): """ Handles published DD Institutional Record Basic details data messages. DD Institutional Records are captured for reference. @param message: published data message @return: none """ acceptance = LocalVars('acceptance') reject_reason = LocalVars('reject_reason') msg_list = [] msg_list.append((acceptance, DataTypes.BOOL)) msg_list.append((reject_reason, DataTypes.U32)) for member in dd_enum_repository.InstitutionalRecordBasicFields: if member is not dd_enum_repository.InstitutionalRecordBasicFields.NUM_OF_INSTITUTIONAL_RECORD_BASIC_FIELDS: msg_list.append((self.institutional_records, f'BASIC__{member.name}', member.datatype())) self.process_into_vars(decoder_list = msg_list, message = message) # Convert to enum if isinstance(self.get_institutional_rec_rr, int): self.get_institutional_rec_rr['BASIC'] = RequestRejectReasons(self.get_institutional_rec_rr) self.get_institutional_rec_ack['BASIC'] = acceptance.value self.get_institutional_rec_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_GET_INSTIT_ADVANCED_RECORD_RESPONSE.name.lower(), "get_institutional_rec_ack", "get_institutional_rec_rr", "institutional_records", "get_institutional_rec_timestamp"]) def _handler_institutional_adv_record_sync(self, message, timestamp = 0.0): """ Handles published DD Institutional Record Advanced details data messages. DD Institutional Records are captured for reference. @param message: published data message @return: none """ acceptance = LocalVars('acceptance') reject_reason = LocalVars('reject_reason') msg_list = [] msg_list.append((acceptance, DataTypes.BOOL)) msg_list.append((reject_reason, DataTypes.U32)) for member in dd_enum_repository.InstitutionalRecordAdvancedFields: if member is not dd_enum_repository.InstitutionalRecordAdvancedFields.NUM_OF_INSTITUTIONAL_RECORD_ADVANCED_FIELDS: msg_list.append((self.institutional_records, f'ADV__{member.name}', member.datatype())) self.process_into_vars(decoder_list = msg_list, message = message) # Convert to enum if isinstance(self.get_institutional_rec_rr, int): self.get_institutional_rec_rr['ADV'] = RequestRejectReasons(self.get_institutional_rec_rr) self.get_institutional_rec_ack['ADV'] = acceptance.value self.get_institutional_rec_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_GET_INSTIT_ADDITIONAL_RECORD_RESPONSE.name.lower(), "get_institutional_rec_ack", "get_institutional_rec_rr", "institutional_records", "get_institutional_rec_timestamp"]) def _handler_institutional_add_record_sync(self, message, timestamp = 0.0): """ Handles published DD Institutional Record Additional details data messages. DD Institutional Records are captured for reference. @param message: published data message @return: none """ acceptance = LocalVars('acceptance') reject_reason = LocalVars('reject_reason') msg_list = [] msg_list.append((acceptance, DataTypes.BOOL)) msg_list.append((reject_reason, DataTypes.U32)) for member in dd_enum_repository.InstitutionalRecordAdditionalFields: if member is not dd_enum_repository.InstitutionalRecordAdditionalFields.NUM_OF_INSTITUTIONAL_RECORD_ADDITIONAL_FIELDS: msg_list.append((self.institutional_records, f'ADD__{member.name}', member.datatype())) self.process_into_vars(decoder_list = msg_list, message = message) # Convert to enum if isinstance(self.get_institutional_rec_rr, int): self.get_institutional_rec_rr['ADD'] = RequestRejectReasons(self.get_institutional_rec_rr) self.get_institutional_rec_ack['ADD'] = acceptance.value self.get_institutional_rec_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_GET_BICARB_CONC_CAL_RECORD_RESPONSE.name.lower(), "get_usage_info_rec_ack", "get_usage_info_rec_rr", "usage_info_records", "get_usage_info_rec_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 """ msg_list = [] msg_list.append((type(self).get_usage_info_rec_ack, DataTypes.BOOL)) msg_list.append((type(self).get_usage_info_rec_rr, DataTypes.U32)) for member in dd_enum_repository.UsageInformationRecordFields: if member is not dd_enum_repository.UsageInformationRecordFields.NUM_OF_USAGE_INFO_RECORD_FIELDS: msg_list.append((self.usage_info_records, member.name, member.datatype())) self.process_into_vars(decoder_list = msg_list, message = message) # Convert to enum if isinstance(self.get_usage_info_rec_rr, int): self.get_usage_info_rec_rr = RequestRejectReasons(self.get_usage_info_rec_rr) self.get_usage_info_rec_timestamp = timestamp