########################################################################### # # 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_set_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 DDRecordsSetHandlers(AbstractSubSystem): """ Part of the DD Records """ def __init__(self, *args, **kwargs): """ The sub record class to handle all the set handlers. """ if self.can_interface is not None: self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_DD_UI_NVM_SET_SYSTEM_RECORD_RESPONSE.value, function = self._handler_set_system_rec_rr_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_DD_UI_NVM_SET_SERVICE_RECORD_RESPONSE.value, function = self._handler_set_service_rec_rr_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_DD_UI_NVM_SET_PRESSURE_SENSOR_CAL_RECORD_RESPONSE.value, function = self._handler_set_calib_pressure_rec_rr_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_DD_UI_NVM_SET_TEMPERATURE_SENSOR_CAL_RECORD_RESPONSE.value, function = self._handler_set_calib_temp_rec_rr_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_DD_UI_NVM_SET_CONC_PUMP_CAL_RECORD_RESPONSE.value, function = self._handler_set_calib_conc_pump_rec_rr_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_DD_UI_NVM_SET_D12_PUMP_CAL_RECORD_RESPONSE.value, function = self._handler_set_calib_dial_d12_rec_rr_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_DD_UI_NVM_SET_D48_PUMP_CAL_RECORD_RESPONSE.value, function = self._handler_set_calib_dial_d48_rec_rr_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_DD_UI_NVM_SET_ACID_CONC_CAL_RECORD_RESPONSE.value, function = self._handler_set_calib_acid_rec_rr_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_DD_UI_NVM_SET_BICARB_CONC_CAL_RECORD_RESPONSE.value, function = self._handler_set_calib_bicarb_rec_rr_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_DD_UI_NVM_SET_ACCEL_SENSOR_CAL_RECORD_RESPONSE.value, function = self._handler_set_calib_accel_rec_rr_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_DD_UI_NVM_SET_BLOOD_LEAK_SENSOR_CAL_RECORD_RESPONSE.value, function = self._handler_set_calib_blood_leak_rec_rr_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_DD_UI_NVM_SET_CONDUCTIVITY_SENSOR_CAL_RECORD_RESPONSE.value, function = self._handler_set_calib_conductivity_rec_rr_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_DD_UI_NVM_SET_INSTIT_BASIC_RECORD_RESPONSE.value, function = self._handler_set_institutional_rec_basic_rr_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_DD_UI_NVM_SET_INSTIT_ADVANCED_RECORD_RESPONSE.value, function = self._handler_set_institutional_rec_advanced_rr_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_DD_UI_NVM_SET_INSTIT_ADDITIONAL_RECORD_RESPONSE.value, function = self._handler_set_institutional_rec_additional_rr_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.dd_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_DD_UI_NVM_SET_USAGE_INFO_RECORD_RESPONSE.value, function = self._handler_set_usage_info_rec_rr_sync) # Passing the arguments over to the next class too super().__init__(*args, **kwargs) # ================================================= Reject Reason Message Handler Methods ================================================= @publish([MsgIds.MSG_ID_DD_UI_NVM_SET_SYSTEM_RECORD_RESPONSE.name.lower(), "set_system_rec_ack", "set_system_rec_rr", "set_system_rec_rr_timestamp"]) def _handler_set_system_rec_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((type(self).set_system_rec_ack, DataTypes.BOOL)) for sys_rec in dd_enum_repository.SystemRecordFields: msg_list.append((self.set_system_rec_rr, sys_rec, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) # Convert to enum for sys_rec in self.set_system_rec_rr: if isinstance(self.set_system_rec_rr[sys_rec], int): self.set_system_rec_rr[sys_rec] = RequestRejectReasons(self.set_system_rec_rr[sys_rec]) self.set_system_rec_rr_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_SET_SERVICE_RECORD_RESPONSE.name.lower(), "set_service_rec_ack", "set_service_rec_rr", "set_service_rec_rr_timestamp"]) def _handler_set_service_rec_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((type(self).set_service_rec_ack, DataTypes.BOOL)) for ser_rec in dd_enum_repository.ServiceRecordFields: msg_list.append((self.set_service_rec_rr, ser_rec, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) # Convert to enum for ser_rec in self.set_service_rec_rr: if isinstance(self.set_service_rec_rr[ser_rec], int): self.set_service_rec_rr[ser_rec] = RequestRejectReasons(self.set_service_rec_rr[ser_rec]) self.set_service_rec_rr_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_SET_INSTIT_BASIC_RECORD_RESPONSE.name.lower(), "set_institutional_rec_ack", "set_institutional_rec_rr", "set_institutional_rec_rr_timestamp"]) def _handler_set_institutional_rec_basic_rr_sync(self, message, timestamp = 0.0): """ Handles published DD Set Basic 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((type(self).set_institutional_rec_ack, DataTypes.BOOL)) for inst_rec in dd_enum_repository.InstitutionalRecordBasicFields: msg_list.append((self.set_institutional_rec_rr, f'BASIC__{inst_rec}', DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) # Convert to enum for inst_rec in self.set_institutional_rec_rr: if self.set_institutional_rec_rr[inst_rec].startswith('BASIC__') and isinstance(self.set_institutional_rec_rr[inst_rec], int): self.set_institutional_rec_rr[inst_rec] = RequestRejectReasons(self.set_institutional_rec_rr[inst_rec]) self.set_institutional_rec_rr_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_SET_INSTIT_ADVANCED_RECORD_RESPONSE.name.lower(), "set_institutional_rec_ack", "set_institutional_rec_rr", "set_institutional_rec_rr_timestamp"]) def _handler_set_institutional_rec_advanced_rr_sync(self, message, timestamp = 0.0): """ Handles published DD Set Advanced 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((type(self).set_institutional_rec_ack, DataTypes.BOOL)) for inst_rec in dd_enum_repository.InstitutionalRecordAdvancedFields: msg_list.append((self.set_institutional_rec_rr, f'ADV__{inst_rec}', DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) # Convert to enum for inst_rec in self.set_institutional_rec_rr: if self.set_institutional_rec_rr[inst_rec].startswith('ADV__') and isinstance(self.set_institutional_rec_rr[inst_rec], int): self.set_institutional_rec_rr[inst_rec] = RequestRejectReasons(self.set_institutional_rec_rr[inst_rec]) self.set_institutional_rec_rr_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_SET_INSTIT_ADDITIONAL_RECORD_RESPONSE.name.lower(), "set_institutional_rec_ack", "set_institutional_rec_rr", "set_institutional_rec_rr_timestamp"]) def _handler_set_institutional_rec_additional_rr_sync(self, message, timestamp = 0.0): """ Handles published DD Set Additional 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((type(self).set_institutional_rec_ack, DataTypes.BOOL)) for inst_rec in dd_enum_repository.InstitutionalRecordAdditionalFields: msg_list.append((self.set_institutional_rec_rr, f'ADD__{inst_rec}', DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) # Convert to enum for inst_rec in self.set_institutional_rec_rr: if self.set_institutional_rec_rr[inst_rec].startswith('ADD__') and isinstance(self.set_institutional_rec_rr[inst_rec], int): self.set_institutional_rec_rr[inst_rec] = RequestRejectReasons(self.set_institutional_rec_rr[inst_rec]) self.set_institutional_rec_rr_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_SET_USAGE_INFO_RECORD_RESPONSE.name.lower(), "set_usage_info_rec_ack", "set_usage_info_rec_rr", "set_usage_info_rec_rr_timestamp"]) def _handler_set_usage_info_rec_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((type(self).set_usage_info_rec_ack, DataTypes.BOOL)) for info_rec in dd_enum_repository.UsageInformationRecordFields: msg_list.append((self.set_usage_info_rec_rr, info_rec, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) # Convert to enum for info_rec in self.set_usage_info_rec_rr: if isinstance(self.set_usage_info_rec_rr[info_rec], int): self.set_usage_info_rec_rr[info_rec] = RequestRejectReasons(self.set_usage_info_rec_rr[info_rec]) self.set_usage_info_rec_rr_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_SET_PRESSURE_SENSOR_CAL_RECORD_RESPONSE.name.lower(), "set_calibration_rec_ack", "set_calibration_rec_rr", "set_calibration_rec_rr_timestamp"]) def _handler_set_calib_pressure_rec_rr_sync(self, message, timestamp = 0.0): """ 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 """ acceptance = LocalVars('acceptance') global_rr = LocalVars('GLOBAL') 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((global_rr, DataTypes.U32)) msg_list.append((sensor_id, DataTypes.U32)) msg_list.append((gain, DataTypes.U32)) msg_list.append((offset, DataTypes.U32)) msg_list.append((calibration_time, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) # Get the 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 self.set_calibration_rec_ack[f'PRES__{sensor_name}'] = acceptance.value self.set_calibration_rec_rr[f'PRES__{sensor_name}__{global_rr.name}'] = RequestRejectReasons(global_rr.value) self.set_calibration_rec_rr[f'PRES__{sensor_name}__{gain.name}'] = RequestRejectReasons(gain.value) self.set_calibration_rec_rr[f'PRES__{sensor_name}__{offset.name}'] = RequestRejectReasons(offset.value) self.set_calibration_rec_rr[f'PRES__{sensor_name}__{calibration_time.name}'] = RequestRejectReasons(calibration_time.value) self.set_calibration_rec_rr_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_SET_TEMPERATURE_SENSOR_CAL_RECORD_RESPONSE.name.lower(), "set_calibration_rec_ack", "set_calibration_rec_rr", "set_calibration_rec_rr_timestamp"]) def _handler_set_calib_temp_rec_rr_sync(self, message, timestamp = 0.0): """ 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 """ acceptance = LocalVars('acceptance') global_rr = LocalVars('GLOBAL') 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((global_rr, DataTypes.U32)) msg_list.append((sensor_id, DataTypes.U32)) msg_list.append((gain, DataTypes.U32)) msg_list.append((offset, DataTypes.U32)) msg_list.append((calibration_time, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) # Get the Sensor Name sensor_name = dd_enum_repository.DDTemperatureSensorNames(sensor_id.value).name # Saving into the calibration data self.set_calibration_rec_ack[f'TEMP__{sensor_name}'] = acceptance.value self.set_calibration_rec_rr[f'TEMP__{sensor_name}__{global_rr.name}'] = RequestRejectReasons(global_rr.value) self.set_calibration_rec_rr[f'TEMP__{sensor_name}__{gain.name}'] = RequestRejectReasons(gain.value) self.set_calibration_rec_rr[f'TEMP__{sensor_name}__{offset.name}'] = RequestRejectReasons(offset.value) self.set_calibration_rec_rr[f'TEMP__{sensor_name}__{calibration_time.name}'] = RequestRejectReasons(calibration_time.value) self.set_calibration_rec_rr_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_SET_CONC_PUMP_CAL_RECORD_RESPONSE.name.lower(), "set_calibration_rec_ack", "set_calibration_rec_rr", "set_calibration_rec_rr_timestamp"]) def _handler_set_calib_conc_pump_rec_rr_sync(self, message, timestamp = 0.0): """ 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 """ acceptance = LocalVars('acceptance') global_rr = LocalVars('GLOBAL') pump_id = LocalVars('pump_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((global_rr, DataTypes.U32)) msg_list.append((pump_id, DataTypes.U32)) msg_list.append((gain, DataTypes.U32)) msg_list.append((offset, DataTypes.U32)) msg_list.append((calibration_time, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) # Get the Sensor Name pump_name = dd_enum_repository.DDConcentratePumpNames(pump_id.value).name # Saving into the calibration data self.set_calibration_rec_ack[f'CONC__{pump_name}'] = acceptance.value self.set_calibration_rec_rr[f'CONC__{pump_name}__{global_rr.name}'] = RequestRejectReasons(global_rr.value) self.set_calibration_rec_rr[f'CONC__{pump_name}__{gain.name}'] = RequestRejectReasons(gain.value) self.set_calibration_rec_rr[f'CONC__{pump_name}__{offset.name}'] = RequestRejectReasons(offset.value) self.set_calibration_rec_rr[f'CONC__{pump_name}__{calibration_time.name}'] = RequestRejectReasons(calibration_time.value) self.set_calibration_rec_rr_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_SET_D12_PUMP_CAL_RECORD_RESPONSE.name.lower(), "set_calibration_rec_ack", "set_calibration_rec_rr", "set_calibration_rec_rr_timestamp"]) def _handler_set_calib_dial_d12_rec_rr_sync(self, message, timestamp = 0.0): """ 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 """ acceptance = LocalVars('acceptance') global_rr = LocalVars('GLOBAL') 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((global_rr, DataTypes.U32)) msg_list.append((target_speed, DataTypes.U32)) msg_list.append((calibration_time, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) pump_name = dd_enum_repository.DDDialysatePumpNames.D12_PUMP.name # Saving into the calibration data self.set_calibration_rec_ack[f'DIAL__{pump_name}'] = acceptance.value self.set_calibration_rec_rr[f'DIAL__{pump_name}__{global_rr.name}'] = RequestRejectReasons(global_rr.value) self.set_calibration_rec_rr[f'DIAL__{pump_name}__{target_speed.name}'] = RequestRejectReasons(target_speed.value) self.set_calibration_rec_rr[f'DIAL__{pump_name}__{calibration_time.name}'] = RequestRejectReasons(calibration_time.value) self.set_calibration_rec_rr_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_SET_D48_PUMP_CAL_RECORD_RESPONSE.name.lower(), "set_calibration_rec_ack", "set_calibration_rec_rr", "set_calibration_rec_rr_timestamp"]) def _handler_set_calib_dial_d48_rec_rr_sync(self, message, timestamp = 0.0): """ 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 """ acceptance = LocalVars('acceptance') global_rr = LocalVars('GLOBAL') 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((global_rr, DataTypes.U32)) msg_list.append((gain, DataTypes.U32)) msg_list.append((offset, DataTypes.U32)) msg_list.append((calibration_time, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) pump_name = dd_enum_repository.DDDialysatePumpNames.D48_PUMP.name # Saving into the calibration data self.set_calibration_rec_ack[f'DIAL__{pump_name}'] = acceptance.value self.set_calibration_rec_rr[f'DIAL__{pump_name}__{global_rr.name}'] = RequestRejectReasons(global_rr.value) self.set_calibration_rec_rr[f'DIAL__{pump_name}__{gain.name}'] = RequestRejectReasons(gain.value) self.set_calibration_rec_rr[f'DIAL__{pump_name}__{offset.name}'] = RequestRejectReasons(offset.value) self.set_calibration_rec_rr[f'DIAL__{pump_name}__{calibration_time.name}'] = RequestRejectReasons(calibration_time.value) self.set_calibration_rec_rr_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_SET_ACID_CONC_CAL_RECORD_RESPONSE.name.lower(), "set_calibration_rec_ack", "set_calibration_rec_rr", "set_calibration_rec_rr_timestamp"]) def _handler_set_calib_acid_rec_rr_sync(self, message, timestamp = 0.0): """ 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 """ acceptance = LocalVars('acceptance') global_rr = LocalVars('GLOBAL') 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((global_rr, DataTypes.U32)) msg_list.append((conc_id, DataTypes.U32)) msg_list.append((mix_ratio, DataTypes.U32)) msg_list.append((volume, DataTypes.U32)) msg_list.append((conductivity, DataTypes.U32)) msg_list.append((temperature, DataTypes.U32)) 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 self.set_calibration_rec_ack[f'ACID__{acid_name}'] = acceptance.value self.set_calibration_rec_rr[f'ACID__{acid_name}__{global_rr.name}'] = RequestRejectReasons(global_rr.value) self.set_calibration_rec_rr[f'ACID__{acid_name}__{mix_ratio.name}'] = RequestRejectReasons(mix_ratio.value) self.set_calibration_rec_rr[f'ACID__{acid_name}__{volume.name}'] = RequestRejectReasons(volume.value) self.set_calibration_rec_rr[f'ACID__{acid_name}__{conductivity.name}'] = RequestRejectReasons(conductivity.value) self.set_calibration_rec_rr[f'ACID__{acid_name}__{temperature.name}'] = RequestRejectReasons(temperature.value) self.set_calibration_rec_rr[f'ACID__{acid_name}__{calibration_time.name}'] = RequestRejectReasons(calibration_time.value) self.set_calibration_rec_rr_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_SET_BICARB_CONC_CAL_RECORD_RESPONSE.name.lower(), "set_calibration_rec_ack", "set_calibration_rec_rr", "set_calibration_rec_rr_timestamp"]) def _handler_set_calib_bicarb_rec_rr_sync(self, message, timestamp = 0.0): """ 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 """ acceptance = LocalVars('acceptance') global_rr = LocalVars('GLOBAL') 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((global_rr, DataTypes.U32)) msg_list.append((conc_id, DataTypes.U32)) msg_list.append((mix_ratio, DataTypes.U32)) msg_list.append((volume, DataTypes.U32)) msg_list.append((conductivity, DataTypes.U32)) msg_list.append((temperature, DataTypes.U32)) 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.BicarbTypes(conc_id.value).name # Saving into the calibration data self.set_calibration_rec_ack[f'BICARB__{acid_name}'] = acceptance.value self.set_calibration_rec_rr[f'BICARB__{acid_name}__{global_rr.name}'] = RequestRejectReasons(global_rr.value) self.set_calibration_rec_rr[f'BICARB__{acid_name}__{mix_ratio.name}'] = RequestRejectReasons(mix_ratio.value) self.set_calibration_rec_rr[f'BICARB__{acid_name}__{volume.name}'] = RequestRejectReasons(volume.value) self.set_calibration_rec_rr[f'BICARB__{acid_name}__{conductivity.name}'] = RequestRejectReasons(conductivity.value) self.set_calibration_rec_rr[f'BICARB__{acid_name}__{temperature.name}'] = RequestRejectReasons(temperature.value) self.set_calibration_rec_rr[f'BICARB__{acid_name}__{calibration_time.name}'] = RequestRejectReasons(calibration_time.value) self.set_calibration_rec_rr_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_SET_ACCEL_SENSOR_CAL_RECORD_RESPONSE.name.lower(), "set_calibration_rec_ack", "set_calibration_rec_rr", "set_calibration_rec_rr_timestamp"]) def _handler_set_calib_accel_rec_rr_sync(self, message, timestamp = 0.0): """ 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 """ acceptance = LocalVars('acceptance') global_rr = LocalVars('GLOBAL') 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((global_rr, DataTypes.U32)) msg_list.append((accel_x, DataTypes.U32)) msg_list.append((accel_y, DataTypes.U32)) msg_list.append((accel_z, DataTypes.U32)) msg_list.append((calibration_time, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) # Saving into the calibration data self.set_calibration_rec_ack[f'ACCEL'] = acceptance.value self.set_calibration_rec_rr[f'ACCEL__{global_rr.name}'] = RequestRejectReasons(global_rr.value) self.set_calibration_rec_rr[f'ACCEL__{accel_x.name}'] = RequestRejectReasons(accel_x.value) self.set_calibration_rec_rr[f'ACCEL__{accel_y.name}'] = RequestRejectReasons(accel_y.value) self.set_calibration_rec_rr[f'ACCEL__{accel_z.name}'] = RequestRejectReasons(accel_z.value) self.set_calibration_rec_rr[f'ACCEL__{calibration_time.name}'] = RequestRejectReasons(calibration_time.value) self.set_calibration_rec_rr_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_SET_BLOOD_LEAK_SENSOR_CAL_RECORD_RESPONSE.name.lower(), "set_calibration_rec_ack", "set_calibration_rec_rr", "set_calibration_rec_rr_timestamp"]) def _handler_set_calib_blood_leak_rec_rr_sync(self, message, timestamp = 0.0): """ 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 """ acceptance = LocalVars('acceptance') global_rr = LocalVars('GLOBAL') 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((global_rr, DataTypes.U32)) msg_list.append((set_point, DataTypes.U32)) msg_list.append((calibration_time, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) # Saving into the calibration data self.set_calibration_rec_ack[f'BLOOD_LEAK'] = acceptance.value self.set_calibration_rec_rr[f'BLOOD_LEAK__{global_rr.name}'] = RequestRejectReasons(global_rr.value) self.set_calibration_rec_rr[f'BLOOD_LEAK__{set_point.name}'] = RequestRejectReasons(set_point.value) self.set_calibration_rec_rr[f'BLOOD_LEAK__{calibration_time.name}'] = RequestRejectReasons(calibration_time.value) self.set_calibration_rec_rr_timestamp = timestamp @publish([MsgIds.MSG_ID_DD_UI_NVM_SET_CONDUCTIVITY_SENSOR_CAL_RECORD_RESPONSE.name.lower(), "set_calibration_rec_ack", "set_calibration_rec_rr", "set_calibration_rec_rr_timestamp"]) def _handler_set_calib_conductivity_rec_rr_sync(self, message, timestamp = 0.0): """ Handles published DD Set Calibration Records for Conductivity Sensor Reject Reason data messages. DD Set Calibration Records for Conductivity Sensor Reject Reason are captured for reference. @param message: published data message @return: none """ acceptance = LocalVars('acceptance') global_rr = LocalVars('GLOBAL') 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((global_rr, DataTypes.U32)) msg_list.append((sensor_id, DataTypes.U32)) msg_list.append((k_high, DataTypes.U32)) msg_list.append((alpha_high, DataTypes.U32)) msg_list.append((eta_high, DataTypes.U32)) msg_list.append((zeta_high, DataTypes.U32)) msg_list.append((k_low, DataTypes.U32)) msg_list.append((alpha_low, DataTypes.U32)) msg_list.append((eta_low, DataTypes.U32)) msg_list.append((zeta_low, DataTypes.U32)) msg_list.append((beta, DataTypes.U32)) msg_list.append((delta, DataTypes.U32)) msg_list.append((rsrvd_1, DataTypes.U32)) msg_list.append((rsrvd_2, DataTypes.U32)) 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 self.set_calibration_rec_ack[f'COND__{sensor_name}'] = acceptance.value self.set_calibration_rec_rr[f'COND__{sensor_name}__{global_rr.name}'] = RequestRejectReasons(global_rr.value) self.set_calibration_rec_rr[f'COND__{sensor_name}__{k_high.name}'] = RequestRejectReasons(k_high.value) self.set_calibration_rec_rr[f'COND__{sensor_name}__{alpha_high.name}'] = RequestRejectReasons(alpha_high.value) self.set_calibration_rec_rr[f'COND__{sensor_name}__{eta_high.name}'] = RequestRejectReasons(eta_high.value) self.set_calibration_rec_rr[f'COND__{sensor_name}__{zeta_high.name}'] = RequestRejectReasons(zeta_high.value) self.set_calibration_rec_rr[f'COND__{sensor_name}__{k_low.name}'] = RequestRejectReasons(k_low.value) self.set_calibration_rec_rr[f'COND__{sensor_name}__{alpha_low.name}'] = RequestRejectReasons(alpha_low.value) self.set_calibration_rec_rr[f'COND__{sensor_name}__{eta_low.name}'] = RequestRejectReasons(eta_low.value) self.set_calibration_rec_rr[f'COND__{sensor_name}__{zeta_low.name}'] = RequestRejectReasons(zeta_low.value) self.set_calibration_rec_rr[f'COND__{sensor_name}__{beta.name}'] = RequestRejectReasons(beta.value) self.set_calibration_rec_rr[f'COND__{sensor_name}__{delta.name}'] = RequestRejectReasons(delta.value) self.set_calibration_rec_rr[f'COND__{sensor_name}__{rsrvd_1.name}'] = RequestRejectReasons(rsrvd_1.value) self.set_calibration_rec_rr[f'COND__{sensor_name}__{rsrvd_2.name}'] = RequestRejectReasons(rsrvd_2.value) self.set_calibration_rec_rr[f'COND__{sensor_name}__{calibration_time.name}'] = RequestRejectReasons(calibration_time.value) self.set_calibration_rec_rr_timestamp = timestamp