########################################################################### # # 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_requests.py # # @author (last) Zoltan Miskolci # @date (last) 04-Aug-2026 # @author (original) Zoltan Miskolci # @date (original) 04-Aug-2026 # ############################################################################ # Module imports 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_ids import MsgIds from leahi_dialin.common.override_templates import cmd_generic_override from leahi_dialin.protocols.CAN import CanChannels from leahi_dialin.utils.abstract_classes import AbstractSubSystem 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 DDRecordsSetRequests(AbstractSubSystem): """ Part of the DD Records """ def __init__(self, *args, **kwargs): """ The sub record class to handle all the set requests. """ # Passing the arguments over to the next class too super().__init__(*args, **kwargs) # ================================================= 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) 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) 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_PRESSURE_SENSOR_CAL_RECORD_REQUEST, 'TEMP': MsgIds.MSG_ID_UI_DD_NVM_SET_TEMPERATURE_SENSOR_CAL_RECORD_REQUEST, 'CONC': MsgIds.MSG_ID_UI_DD_NVM_SET_CONC_PUMP_CAL_RECORD_REQUEST, 'DIAL': MsgIds.MSG_ID_UI_DD_NVM_SET_D48_PUMP_CAL_RECORD_REQUEST, 'ACID': MsgIds.MSG_ID_UI_DD_NVM_SET_ACID_CONC_CAL_RECORD_REQUEST, 'BICARB': MsgIds.MSG_ID_UI_DD_NVM_SET_BICARB_CONC_CAL_RECORD_REQUEST, 'ACCEL': MsgIds.MSG_ID_UI_DD_NVM_SET_ACCEL_SENSOR_CAL_RECORD_REQUEST, 'BLOOD_LEAK': MsgIds.MSG_ID_UI_DD_NVM_SET_BLOOD_LEAK_SENSOR_CAL_RECORD_REQUEST, 'COND': MsgIds.MSG_ID_UI_DD_NVM_SET_CONDUCTIVITY_SENSOR_CAL_RECORD_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 = {} mapping = [ {'param_name': 'target_speed', 'entry_name': f'{group}__{sensor.name}__TARGET_SPEED' }, {'param_name': dd_enum_repository.CalibRecordSensorFields.GAIN.name.lower(), 'entry_name': f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordSensorFields.GAIN.name}' }, {'param_name': dd_enum_repository.CalibRecordSensorFields.OFFSET.name.lower(), 'entry_name': f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordSensorFields.OFFSET.name}' }, {'param_name': dd_enum_repository.CalibRecordSensorFields.CALIBRATION_TIME.name.lower(), 'entry_name': f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordSensorFields.CALIBRATION_TIME.name}' }, ] # Building the kwargs parameter list 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] for map in mapping: if map['entry_name'] in calibration_records and calibration_records[map['entry_name']] is not None: kwargs[map['param_name']] = calibration_records[map['entry_name']] checked.append(record_parts[1]) # Skip sending if dictionary is provided but all entry for the sensor was None (except sensor_enum and MSG_ID, because those are neccesary) if calibration_records != {} and len(kwargs) == 2: continue # Create the send data function send_data.append(partial(self.cmd_set_calibration_records_sensor, *args, **kwargs)) elif group in ['ACID', 'BICARB']: if group == 'ACID': conc = disp_enum_repository.AcidTypes.from_str(record_parts[1]) elif group == 'BICARB': conc = disp_enum_repository.BicarbTypes.from_str(record_parts[1]) else: conc = 'unknown' # Set the arguments for the call args = [] kwargs = {} mapping = [ {'param_name': dd_enum_repository.CalibRecordConcentrateFields.CONCENTRATE_MIX_RATIO.name.lower(), 'entry_name': f'{group}__{conc.name}__{dd_enum_repository.CalibRecordConcentrateFields.CONCENTRATE_MIX_RATIO.name}' }, {'param_name': dd_enum_repository.CalibRecordConcentrateFields.FULL_BOTTLE_VOLUME_ML.name.lower(), 'entry_name': f'{group}__{conc.name}__{dd_enum_repository.CalibRecordConcentrateFields.FULL_BOTTLE_VOLUME_ML.name}' }, {'param_name': dd_enum_repository.CalibRecordConcentrateFields.CONDUCTIVITY_USPCM.name.lower(), 'entry_name': f'{group}__{conc.name}__{dd_enum_repository.CalibRecordConcentrateFields.CONDUCTIVITY_USPCM.name}' }, {'param_name': dd_enum_repository.CalibRecordConcentrateFields.BOTTLE_TEMPERATURE_C.name.lower(), 'entry_name': f'{group}__{conc.name}__{dd_enum_repository.CalibRecordConcentrateFields.BOTTLE_TEMPERATURE_C.name}' }, {'param_name': dd_enum_repository.CalibRecordConcentrateFields.CALIBRATION_TIME.name.lower(), 'entry_name': f'{group}__{conc.name}__{dd_enum_repository.CalibRecordConcentrateFields.CALIBRATION_TIME.name}' }, ] # Building the kwargs parameter list kwargs['conc_enum'] = conc kwargs['msg_id'] = msg_id_pairing[group] for map in mapping: if map['entry_name'] in calibration_records and calibration_records[map['entry_name']] is not None: kwargs[map['param_name']] = calibration_records[map['entry_name']] checked.append(record_parts[1]) # Skip sending if dictionary is provided but all entry for the sensor was None (except conc_enum and MSG_ID, because those are neccesary) if calibration_records != {} and len(kwargs) == 2: continue # Store the function but do not execute it send_data.append(partial(self.cmd_set_calibration_records_concentrate, *args, **kwargs)) elif group in ['ACCEL']: # Set the arguments for the call args = [] kwargs = {} mapping = [ {'param_name': dd_enum_repository.CalibRecordAccelerometerFields.ACCEL_X_OFFSET.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.CalibRecordAccelerometerFields.ACCEL_X_OFFSET.name}' }, {'param_name': dd_enum_repository.CalibRecordAccelerometerFields.ACCEL_Y_OFFSET.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.CalibRecordAccelerometerFields.ACCEL_Y_OFFSET.name}' }, {'param_name': dd_enum_repository.CalibRecordAccelerometerFields.ACCEL_Z_OFFSET.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.CalibRecordAccelerometerFields.ACCEL_Z_OFFSET.name}' }, {'param_name': dd_enum_repository.CalibRecordAccelerometerFields.CALIBRATION_TIME.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.CalibRecordAccelerometerFields.CALIBRATION_TIME.name}' }, ] # Building the kwargs parameter list kwargs['msg_id'] = msg_id_pairing[group] for map in mapping: if map['entry_name'] in calibration_records and calibration_records[map['entry_name']] is not None: kwargs[map['param_name']] = calibration_records[map['entry_name']] checked.append(group) # Skip sending if dictionary is provided but all entry for the sensor was None (except MSG_ID, because that is neccesary) if calibration_records != {} and len(kwargs) == 1: continue # Store the function but do not execute it send_data.append(partial(self.cmd_set_calibration_records_accelerometer, *args, **kwargs)) elif group in ['BLOOD_LEAK']: # Set the arguments for the call args = [] kwargs = {} mapping = [ {'param_name': dd_enum_repository.CalibRecordBloodLeakFields.SET_POINT.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.CalibRecordBloodLeakFields.SET_POINT.name}' }, {'param_name': dd_enum_repository.CalibRecordBloodLeakFields.CALIBRATION_TIME.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.CalibRecordBloodLeakFields.CALIBRATION_TIME.name}' }, ] # Building the kwargs parameter list kwargs['msg_id'] = msg_id_pairing[group] for map in mapping: if map['entry_name'] in calibration_records and calibration_records[map['entry_name']] is not None: kwargs[map['param_name']] = calibration_records[map['entry_name']] checked.append(group) # Skip sending if dictionary is provided but all entry for the sensor was None (except MSG_ID, because that is neccesary) if calibration_records != {} and len(kwargs) == 1: continue # Store the function but do not execute it send_data.append(partial(self.cmd_set_calibration_records_blood_leak, *args, **kwargs)) elif group in ['COND']: sensor = dd_enum_repository.DDConductivitySensorNames.from_str(record_parts[1]) # Set the arguments for the call args = [] kwargs = {} mapping = [ {'param_name': dd_enum_repository.CalibRecordConductivityFields.K_HIGH.name.lower(), 'entry_name': f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordConductivityFields.K_HIGH.name}' }, {'param_name': dd_enum_repository.CalibRecordConductivityFields.K_LOW.name.lower(), 'entry_name': f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordConductivityFields.K_LOW.name}' }, {'param_name': dd_enum_repository.CalibRecordConductivityFields.ALPHA_HIGH.name.lower(), 'entry_name': f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordConductivityFields.ALPHA_HIGH.name}' }, {'param_name': dd_enum_repository.CalibRecordConductivityFields.ALPHA_LOW.name.lower(), 'entry_name': f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordConductivityFields.ALPHA_LOW.name}' }, {'param_name': dd_enum_repository.CalibRecordConductivityFields.ETA_HIGH.name.lower(), 'entry_name': f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordConductivityFields.ETA_HIGH.name}' }, {'param_name': dd_enum_repository.CalibRecordConductivityFields.ETA_LOW.name.lower(), 'entry_name': f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordConductivityFields.ETA_LOW.name}' }, {'param_name': dd_enum_repository.CalibRecordConductivityFields.ZETA_HIGH.name.lower(), 'entry_name': f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordConductivityFields.ZETA_HIGH.name}' }, {'param_name': dd_enum_repository.CalibRecordConductivityFields.ZETA_LOW.name.lower(), 'entry_name': f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordConductivityFields.ZETA_LOW.name}' }, {'param_name': dd_enum_repository.CalibRecordConductivityFields.BETA.name.lower(), 'entry_name': f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordConductivityFields.BETA.name}' }, {'param_name': dd_enum_repository.CalibRecordConductivityFields.DELTA.name.lower(), 'entry_name': f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordConductivityFields.DELTA.name}' }, {'param_name': dd_enum_repository.CalibRecordConductivityFields.RSRVD_1.name.lower(), 'entry_name': f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordConductivityFields.RSRVD_1.name}' }, {'param_name': dd_enum_repository.CalibRecordConductivityFields.RSRVD_2.name.lower(), 'entry_name': f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordConductivityFields.RSRVD_2.name}' }, {'param_name': dd_enum_repository.CalibRecordConductivityFields.CALIBRATION_TIME.name.lower(), 'entry_name': f'{group}__{sensor.name}__{dd_enum_repository.CalibRecordConductivityFields.CALIBRATION_TIME.name}' }, ] # Building the kwargs parameter list kwargs['sensor_enum'] = sensor kwargs['msg_id'] = msg_id_pairing[group] for map in mapping: if map['entry_name'] in calibration_records and calibration_records[map['entry_name']] is not None: kwargs[map['param_name']] = calibration_records[map['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 # Store the function but do not execute it send_data.append(partial(self.cmd_set_calibration_records_conductivity, *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 = [] checked = [] for record_enum in dd_enum_repository.InstitutionalRecordBasicFields: record_parts = record_enum.name.split('__') group = record_parts[0] if group in ['BASIC']: # Set the arguments for the call args = [] kwargs = {} mapping = [ {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.MIN_DIALYSATE_SYSTEM_FLOW_MLPM.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.MIN_DIALYSATE_SYSTEM_FLOW_MLPM.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.MAX_DIALYSATE_SYSTEM_FLOW_MLPM.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.MAX_DIALYSATE_SYSTEM_FLOW_MLPM.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.MIN_DIALYSATE_TEMP_C.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.MIN_DIALYSATE_TEMP_C.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.MAX_DIALYSATE_TEMP_C.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.MAX_DIALYSATE_TEMP_C.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.MIN_ACID_CONCENTRATE.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.MIN_ACID_CONCENTRATE.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.MAX_ACID_CONCENTRATE.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.MAX_ACID_CONCENTRATE.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.MIN_BICARB_CARTRIDGE_SIZE_G.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.MIN_BICARB_CARTRIDGE_SIZE_G.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.MAX_BICARB_CARTRIDGE_SIZE_G.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.MAX_BICARB_CARTRIDGE_SIZE_G.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.MIN_SODIUM_MEQPL.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.MIN_SODIUM_MEQPL.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.MAX_SODIUM_MEQPL.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.MAX_SODIUM_MEQPL.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.MIN_BICARBONATE_MEQPL.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.MIN_BICARBONATE_MEQPL.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.MAX_BICARBONATE_MEQPL.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.MAX_BICARBONATE_MEQPL.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.MIN_RO_REJECTION_RATIO_PCT.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.MIN_RO_REJECTION_RATIO_PCT.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.DISINFECTION_FREQUENCY.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.DISINFECTION_FREQUENCY.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.DISINFECTION_CYCLE_TIME.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.DISINFECTION_CYCLE_TIME.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.MIN_INLET_WATER_CIND_AKARN_LIMIT_USPCM.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.MIN_INLET_WATER_CIND_AKARN_LIMIT_USPCM.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.MAX_INLET_WATER_CIND_AKARN_LIMIT_USPCM.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.MAX_INLET_WATER_CIND_AKARN_LIMIT_USPCM.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.ACID_CONCENTRATE_JUG_SIZE_L.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.ACID_CONCENTRATE_JUG_SIZE_L.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.MIN_ACID_ALARM_LIMIT_PCT.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.MIN_ACID_ALARM_LIMIT_PCT.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.MIN_BICARB_ALARM_LIMIT_PCT.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.MIN_BICARB_ALARM_LIMIT_PCT.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.POST_TREAT_DRAIN_OPTION.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.POST_TREAT_DRAIN_OPTION.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.POST_TREAT_DRY_BICARB_OPTION.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.POST_TREAT_DRY_BICARB_OPTION.name}' }, {'param_name': dd_enum_repository.InstitutionalRecordBasicFields.CALIBRATION_TIME.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordBasicFields.CALIBRATION_TIME.name}' }, ] # Building the kwargs parameter list for map in mapping: if map['entry_name'] in institutional_records and institutional_records[map['entry_name']] is not None: kwargs[map['param_name']] = institutional_records[map['entry_name']] checked.append(group) # Skip sending if dictionary is provided if institutional_records != {}: continue # Store the function but do not execute it send_data.append(partial(self.cmd_set_institutional_records_basic, *args, **kwargs)) elif group in ['ADV']: # Set the arguments for the call args = [] kwargs = {} mapping = [ {'param_name': dd_enum_repository.InstitutionalRecordAdvancedFields.CALIBRATION_TIME.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordAdvancedFields.CALIBRATION_TIME.name}' }, ] # Building the kwargs parameter list for map in mapping: if map['entry_name'] in institutional_records and institutional_records[map['entry_name']] is not None: kwargs[map['param_name']] = institutional_records[map['entry_name']] checked.append(group) # Skip sending if dictionary is provided if institutional_records != {}: continue # Store the function but do not execute it send_data.append(partial(self.cmd_set_institutional_records_advanced, *args, **kwargs)) elif group in ['ADD']: # Set the arguments for the call args = [] kwargs = {} mapping = [ {'param_name': dd_enum_repository.InstitutionalRecordAdditionalFields.CALIBRATION_TIME.name.lower(), 'entry_name': f'{group}__{dd_enum_repository.InstitutionalRecordAdditionalFields.CALIBRATION_TIME.name}' }, ] # Building the kwargs parameter list for map in mapping: if map['entry_name'] in institutional_records and institutional_records[map['entry_name']] is not None: kwargs[map['param_name']] = institutional_records[map['entry_name']] checked.append(group) # Skip sending if dictionary is provided if institutional_records != {}: continue # Store the function but do not execute it send_data.append(partial(self.cmd_set_institutional_records_additional, *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_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) 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, target_speed: float=0.0, gain: float=0.0, offset: float=0.0, calibration_time: int=1) -> int: payload = b'' # Add sensor index when it's not Dialysate pump if msg_id not in [MsgIds.MSG_ID_UI_DD_NVM_SET_D12_PUMP_CAL_RECORD_REQUEST, MsgIds.MSG_ID_UI_DD_NVM_SET_D48_PUMP_CAL_RECORD_REQUEST]: if isinstance(sensor_enum, fp_enum_repository.FPPressureSensorNames): payload += byte_to_bytearray(sensor_enum.value + dd_enum_repository.DDPressureSensorNames.NUM_OF_PRESSURE_SENSORS.value) else: payload += byte_to_bytearray(sensor_enum.value) if msg_id == MsgIds.MSG_ID_UI_DD_NVM_SET_D12_PUMP_CAL_RECORD_REQUEST: payload += float_to_bytearray(target_speed) else: payload += float_to_bytearray(gain) payload += float_to_bytearray(offset) payload += integer_to_bytearray(calibration_time) # Get sensor type for the debug print sensor_type = '' if msg_id == MsgIds.MSG_ID_UI_DD_NVM_SET_PRESSURE_SENSOR_CAL_RECORD_REQUEST: sensor_type = 'Pressure Sensor' elif msg_id == MsgIds.MSG_ID_UI_DD_NVM_SET_TEMPERATURE_SENSOR_CAL_RECORD_REQUEST: sensor_type = 'Temperature Sensor' elif msg_id == MsgIds.MSG_ID_UI_DD_NVM_SET_CONC_PUMP_CAL_RECORD_REQUEST: sensor_type = 'Concentrate Pump' elif msg_id in [MsgIds.MSG_ID_UI_DD_NVM_SET_D12_PUMP_CAL_RECORD_REQUEST, MsgIds.MSG_ID_UI_DD_NVM_SET_D48_PUMP_CAL_RECORD_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, conc_enum: DialinEnum, msg_id: MsgIds, concentrate_mix_ratio: float=0.0, full_bottle_volume_ml: float=0.0, conductivity_uspcm: float=0.0, bottle_temperature_c: float=0.0, calibration_time: int=1) -> int: payload = byte_to_bytearray(conc_enum.value) payload += float_to_bytearray(concentrate_mix_ratio) payload += float_to_bytearray(full_bottle_volume_ml) payload += float_to_bytearray(conductivity_uspcm) payload += float_to_bytearray(bottle_temperature_c) payload += integer_to_bytearray(calibration_time) # Get concentrate type for the debug print conc_type = '' if msg_id == MsgIds.MSG_ID_UI_DD_NVM_SET_ACID_CONC_CAL_RECORD_REQUEST: conc_type = 'Acid' elif msg_id == MsgIds.MSG_ID_UI_DD_NVM_SET_BICARB_CONC_CAL_RECORD_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} {conc_enum.name} 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) return cmd_generic_override(payload = payload, reset = None, channel_id = CanChannels.dialin_to_dd_ch_id, msg_id = msg_id, entity_name = f'DD Accelerometer Calibration Record', override_text = 'being set', logger = self.logger, can_interface = self.can_interface) def cmd_set_calibration_records_blood_leak(self, msg_id: MsgIds, set_point: int=0, calibration_time: int=1) -> int: payload = integer_to_bytearray(set_point) payload += integer_to_bytearray(calibration_time) 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) def cmd_set_calibration_records_conductivity(self, msg_id: MsgIds, sensor_enum: DialinEnum, k_high: float=4.0, k_low: float=4.0, alpha_high: float=0.02, alpha_low: float=0.02, eta_high: float=1.0, eta_low: float=1.0, zeta_high: float=1.0, zeta_low: float=0.0, beta_high: float=1.0, delta_low: float=0.0, rsrvd_1: float=0.0, rsrvd_2: float=0.0, calibration_time: int=1) -> int: payload = integer_to_bytearray(sensor_enum.value) payload += float_to_bytearray(k_high) payload += float_to_bytearray(alpha_high) payload += float_to_bytearray(eta_high) payload += float_to_bytearray(zeta_high) payload += float_to_bytearray(k_low) payload += float_to_bytearray(alpha_low) payload += float_to_bytearray(eta_low) payload += float_to_bytearray(zeta_low) payload += float_to_bytearray(beta_high) payload += float_to_bytearray(delta_low) payload += float_to_bytearray(rsrvd_1) payload += float_to_bytearray(rsrvd_2) payload += integer_to_bytearray(calibration_time) return cmd_generic_override(payload = payload, reset = None, channel_id = CanChannels.dialin_to_dd_ch_id, msg_id = msg_id, entity_name = f'DD Conductivity {sensor_enum.name} Calibration Record', override_text = 'being set', logger = self.logger, can_interface = self.can_interface) def cmd_set_institutional_records_basic(self, min_dialysate_system_flow_mlpm: int=0, max_dialysate_system_flow_mlpm: int=0, min_dialysate_temp_c: float=0.0, max_dialysate_temp_c: float=0.0, min_acid_concentrate: int=0, max_acid_concentrate: int=0, min_bicarb_cartridge_size_g: int=0, max_bicarb_cartridge_size_g: int=0, min_sodium_meqpl: int=0, max_sodium_meqpl: int=0, min_bicarbonate_meqpl: int=0, max_bicarbonate_meqpl: int=0, min_ro_rejection_ratio_pct: float=0.0, disinfection_frequency: int=0, disinfection_cycle_time: float=0.0, min_inlet_water_cind_akarn_limit_uspcm: float=0.0, max_inlet_water_cind_akarn_limit_uspcm: float=0.0, acid_concentrate_jug_size_l: float=0.0, min_acid_alarm_limit_pct: float=0.0, min_bicarb_alarm_limit_pct: float=0.0, post_treat_drain_option: int=0, post_treat_dry_bicarb_option: int=0, calibration_time: int=1) -> int: payload = integer_to_bytearray(min_dialysate_system_flow_mlpm) payload += integer_to_bytearray(max_dialysate_system_flow_mlpm) payload += integer_to_bytearray(min_dialysate_temp_c) payload += integer_to_bytearray(max_dialysate_temp_c) payload += integer_to_bytearray(min_acid_concentrate) payload += integer_to_bytearray(max_acid_concentrate) payload += integer_to_bytearray(min_bicarb_cartridge_size_g) payload += integer_to_bytearray(max_bicarb_cartridge_size_g) payload += integer_to_bytearray(min_sodium_meqpl) payload += integer_to_bytearray(max_sodium_meqpl) payload += integer_to_bytearray(min_bicarbonate_meqpl) payload += integer_to_bytearray(max_bicarbonate_meqpl) payload += float_to_bytearray(min_ro_rejection_ratio_pct) payload += integer_to_bytearray(disinfection_frequency) payload += float_to_bytearray(disinfection_cycle_time) payload += float_to_bytearray(min_inlet_water_cind_akarn_limit_uspcm) payload += float_to_bytearray(max_inlet_water_cind_akarn_limit_uspcm) payload += float_to_bytearray(acid_concentrate_jug_size_l) payload += float_to_bytearray(min_acid_alarm_limit_pct) payload += float_to_bytearray(min_bicarb_alarm_limit_pct) payload += integer_to_bytearray(post_treat_drain_option) payload += integer_to_bytearray(post_treat_dry_bicarb_option) payload += integer_to_bytearray(calibration_time) 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_INSTIT_BASIC_RECORD_REQUEST, entity_name = f'DD Institutional Basic Records', override_text = 'being set', logger = self.logger, can_interface = self.can_interface) def cmd_set_institutional_records_advanced(self, calibration_time: int=1) -> int: payload = b'' payload += integer_to_bytearray(calibration_time) 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_INSTIT_ADVANCED_RECORD_REQUEST, entity_name = f'DD Institutional Advanced Records', override_text = 'being set', logger = self.logger, can_interface = self.can_interface) def cmd_set_institutional_records_additional(self, calibration_time: int=1) -> int: payload = b'' payload += integer_to_bytearray(calibration_time) 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_INSTIT_ADDITIONAL_RECORD_REQUEST, entity_name = f'DD Institutional Additional Records', override_text = 'being set', logger = self.logger, can_interface = self.can_interface)