########################################################################### # # 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_requests.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.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.conversions import integer_to_bytearray class DDRecordsGetRequests(AbstractSubSystem): """ Part of the DD Records """ def __init__(self, *args, **kwargs): """ The sub record class to handle all the get handlers. """ # Passing the arguments over to the next class too super().__init__(*args, **kwargs) # ================================================= Request Records Methods ================================================= def cmd_request_system_records(self) -> int: """ Constructs and sends a request for System Records. Constraints: Must be logged into DD. Must be in Service mode. @return: 1 if successful, zero otherwise """ return cmd_generic_override( payload = None, reset = None, channel_id = CanChannels.dialin_to_dd_ch_id, msg_id = MsgIds.MSG_ID_UI_DD_NVM_GET_SYSTEM_RECORD_REQUEST, entity_name = f'DD System Record request', override_text = '', logger = self.logger, can_interface = self.can_interface) def cmd_request_service_records(self) -> int: """ Constructs and sends a request for Service Records. Constraints: Must be logged into DD. Must be in Service mode. @return: 1 if successful, zero otherwise """ return cmd_generic_override( payload = None, reset = None, channel_id = CanChannels.dialin_to_dd_ch_id, msg_id = MsgIds.MSG_ID_UI_DD_NVM_GET_SERVICE_RECORD_REQUEST, entity_name = f'DD Service Record request', override_text = '', logger = self.logger, can_interface = self.can_interface) def cmd_request_all_calibration_records(self) -> None: """ Constructs and sends a request for Calibration Records. Constraints: Must be logged into DD. Must be in Service mode. @return: None """ for pres_sensor in dd_enum_repository.DDPressureSensorNames: self.cmd_request_calibration_pressure_record(pres_sensor.value) for pres_sensor in fp_enum_repository.FPPressureSensorNames: self.cmd_request_calibration_pressure_record(pres_sensor.value + dd_enum_repository.DDPressureSensorNames.NUM_OF_PRESSURE_SENSORS.value) for temp_sensor in dd_enum_repository.DDTemperatureSensorNames: self.cmd_request_calibration_temperature_record(temp_sensor.value) for conc_pump in dd_enum_repository.DDConcentratePumpNames: self.cmd_request_calibration_concentrate_record(conc_pump.value) self.cmd_request_calibration_dialysate_d12_record() self.cmd_request_calibration_dialysate_d48_record() for acid_conc in disp_enum_repository.AcidTypes: self.cmd_request_calibration_acid_record(acid_conc.value) for bicarb_conc in disp_enum_repository.BicarbTypes: self.cmd_request_calibration_bicarb_record(bicarb_conc.value) self.cmd_request_calibration_accelerometer_record() self.cmd_request_calibration_blood_leak_record() for cond_sensor in dd_enum_repository.DDConductivitySensorNames: self.cmd_request_calibration_conductivity_record(cond_sensor.value) def cmd_request_calibration_pressure_record(self, sensor_id: int) -> int: """ Constructs and sends a request for Pressure Sensor Calibration Record. Constraints: Must be logged into DD. Must be in Service mode. @return: 1 if successful, zero otherwise """ payload = integer_to_bytearray(sensor_id) # Get the name of the sensor if sensor_id >= dd_enum_repository.DDPressureSensorNames.NUM_OF_PRESSURE_SENSORS.value: sensor_name = fp_enum_repository.FPPressureSensorNames(sensor_id).name else: sensor_name = dd_enum_repository.DDPressureSensorNames(sensor_id).name return cmd_generic_override( payload = payload, reset = None, channel_id = CanChannels.dialin_to_dd_ch_id, msg_id = MsgIds.MSG_ID_UI_DD_NVM_GET_PRESSURE_SENSOR_CAL_RECORD_REQUEST, entity_name = f'DD Calibration Record request for Pressure {sensor_name} sensor', override_text = '', logger = self.logger, can_interface = self.can_interface) def cmd_request_calibration_temperature_record(self, sensor_id: int) -> int: """ Constructs and sends a request for Temperature Sensor Calibration Record. Constraints: Must be logged into DD. Must be in Service mode. @return: 1 if successful, zero otherwise """ payload = integer_to_bytearray(sensor_id) # Get the name of the sensor sensor_name = dd_enum_repository.DDTemperatureSensorNames(sensor_id).name return cmd_generic_override( payload = payload, reset = None, channel_id = CanChannels.dialin_to_dd_ch_id, msg_id = MsgIds.MSG_ID_UI_DD_NVM_GET_TEMPERATURE_SENSOR_CAL_RECORD_REQUEST, entity_name = f'DD Calibration Record request for Temparature {sensor_name} sensor', override_text = '', logger = self.logger, can_interface = self.can_interface) def cmd_request_calibration_concentrate_record(self, pump_id: int) -> int: """ Constructs and sends a request for Concentrate Pump Calibration Record. Constraints: Must be logged into DD. Must be in Service mode. @return: 1 if successful, zero otherwise """ payload = integer_to_bytearray(pump_id) # Get the name of the sensor pump_name = dd_enum_repository.DDConcentratePumpNames(pump_id).name return cmd_generic_override( payload = payload, reset = None, channel_id = CanChannels.dialin_to_dd_ch_id, msg_id = MsgIds.MSG_ID_UI_DD_NVM_GET_CONC_PUMP_CAL_RECORD_REQUEST, entity_name = f'DD Calibration Record request for Concentrate {pump_name} pump', override_text = '', logger = self.logger, can_interface = self.can_interface) def cmd_request_calibration_dialysate_d12_record(self) -> int: """ Constructs and sends a request for Dialysate D12 Pump Calibration Record. Constraints: Must be logged into DD. Must be in Service mode. @return: 1 if successful, zero otherwise """ return cmd_generic_override( payload = None, reset = None, channel_id = CanChannels.dialin_to_dd_ch_id, msg_id = MsgIds.MSG_ID_UI_DD_NVM_GET_D12_PUMP_CAL_RECORD_REQUEST, entity_name = f'DD Calibration Record request for Dialysate D12 pump', override_text = '', logger = self.logger, can_interface = self.can_interface) def cmd_request_calibration_dialysate_d48_record(self) -> int: """ Constructs and sends a request for Dialysate D48 Pump Calibration Record. Constraints: Must be logged into DD. Must be in Service mode. @return: 1 if successful, zero otherwise """ return cmd_generic_override( payload = None, reset = None, channel_id = CanChannels.dialin_to_dd_ch_id, msg_id = MsgIds.MSG_ID_UI_DD_NVM_GET_D48_PUMP_CAL_RECORD_REQUEST, entity_name = f'DD Calibration Record request for Dialysate D48 pump', override_text = '', logger = self.logger, can_interface = self.can_interface) def cmd_request_calibration_acid_record(self, conc_id: int) -> int: """ Constructs and sends a request for Acid Concentrate Calibration Record. Constraints: Must be logged into DD. Must be in Service mode. @return: 1 if successful, zero otherwise """ payload = integer_to_bytearray(conc_id) # Get the name of the sensor conc_name = disp_enum_repository.AcidTypes(conc_id).name return cmd_generic_override( payload = payload, reset = None, channel_id = CanChannels.dialin_to_dd_ch_id, msg_id = MsgIds.MSG_ID_UI_DD_NVM_GET_ACID_CONC_CAL_RECORD_REQUEST, entity_name = f'DD Calibration Record request for Acid concentrate: {conc_name}', override_text = '', logger = self.logger, can_interface = self.can_interface) def cmd_request_calibration_bicarb_record(self, conc_id: int) -> int: """ Constructs and sends a request for Bicarb Concentrate Calibration Record. Constraints: Must be logged into DD. Must be in Service mode. @return: 1 if successful, zero otherwise """ payload = integer_to_bytearray(conc_id) # Get the name of the sensor conc_name = disp_enum_repository.BicarbTypes(conc_id).name return cmd_generic_override( payload = payload, reset = None, channel_id = CanChannels.dialin_to_dd_ch_id, msg_id = MsgIds.MSG_ID_UI_DD_NVM_GET_BICARB_CONC_CAL_RECORD_REQUEST, entity_name = f'DD Calibration Record request for Bicarb concentrate: {conc_name}', override_text = '', logger = self.logger, can_interface = self.can_interface) def cmd_request_calibration_accelerometer_record(self) -> int: """ Constructs and sends a request for Accelerometer Sensor Calibration Record. Constraints: Must be logged into DD. Must be in Service mode. @return: 1 if successful, zero otherwise """ return cmd_generic_override( payload = None, reset = None, channel_id = CanChannels.dialin_to_dd_ch_id, msg_id = MsgIds.MSG_ID_UI_DD_NVM_GET_ACCEL_SENSOR_CAL_RECORD_REQUEST, entity_name = f'DD Calibration Record request for Accelerometer sensor', override_text = '', logger = self.logger, can_interface = self.can_interface) def cmd_request_calibration_blood_leak_record(self) -> int: """ Constructs and sends a request for Blood Leak Sensor Calibration Record. Constraints: Must be logged into DD. Must be in Service mode. @return: 1 if successful, zero otherwise """ return cmd_generic_override( payload = None, reset = None, channel_id = CanChannels.dialin_to_dd_ch_id, msg_id = MsgIds.MSG_ID_UI_DD_NVM_GET_BLOOD_LEAK_SENSOR_CAL_RECORD_REQUEST, entity_name = f'DD Calibration Record request for Blood Leak sensor', override_text = '', logger = self.logger, can_interface = self.can_interface) def cmd_request_calibration_conductivity_record(self, sensor_id: int) -> int: """ Constructs and sends a request for Conductivity Sensor Calibration Record. Constraints: Must be logged into DD. Must be in Service mode. @return: 1 if successful, zero otherwise """ payload = integer_to_bytearray(sensor_id) # Get the name of the sensor sensor_name = dd_enum_repository.DDConductivitySensorNames(sensor_id).name return cmd_generic_override( payload = payload, reset = None, channel_id = CanChannels.dialin_to_dd_ch_id, msg_id = MsgIds.MSG_ID_UI_DD_NVM_GET_CONDUCTIVITY_SENSOR_CAL_RECORD_REQUEST, entity_name = f'DD Calibration Record request for Conductivity {sensor_name} sensor', override_text = '', logger = self.logger, can_interface = self.can_interface) def cmd_request_all_institutional_records(self) -> int: """ Constructs and sends a request for All Institutional Records. Constraints: Must be logged into DD. Must be in Service mode. @return: 1 if successful, zero otherwise """ resp_basic = self.cmd_request_institutional_basic_records() resp_adv = self.cmd_request_institutional_advanced_records() resp_add = self.cmd_request_institutional_additional_records() return (resp_basic * 100) + (resp_adv * 10) + resp_add def cmd_request_institutional_basic_records(self) -> int: """ Constructs and sends a request for Basic Institutional Records. Constraints: Must be logged into DD. Must be in Service mode. @return: 1 if successful, zero otherwise """ return cmd_generic_override( payload = None, reset = None, channel_id = CanChannels.dialin_to_dd_ch_id, msg_id = MsgIds.MSG_ID_UI_DD_NVM_GET_INSTIT_BASIC_RECORD_REQUEST, entity_name = f'DD Basic Institutional Record request', override_text = '', logger = self.logger, can_interface = self.can_interface) def cmd_request_institutional_advanced_records(self) -> int: """ Constructs and sends a request for Advanced Institutional Records. Constraints: Must be logged into DD. Must be in Service mode. @return: 1 if successful, zero otherwise """ return cmd_generic_override( payload = None, reset = None, channel_id = CanChannels.dialin_to_dd_ch_id, msg_id = MsgIds.MSG_ID_UI_DD_NVM_GET_INSTIT_ADVANCED_RECORD_REQUEST, entity_name = f'DD Advanced Institutional Record request', override_text = '', logger = self.logger, can_interface = self.can_interface) def cmd_request_institutional_additional_records(self) -> int: """ Constructs and sends a request for Additional Institutional Records. Constraints: Must be logged into DD. Must be in Service mode. @return: 1 if successful, zero otherwise """ return cmd_generic_override( payload = None, reset = None, channel_id = CanChannels.dialin_to_dd_ch_id, msg_id = MsgIds.MSG_ID_UI_DD_NVM_GET_INSTIT_ADDITIONAL_RECORD_REQUEST, entity_name = f'DD Additional Institutional Record request', override_text = '', logger = self.logger, can_interface = self.can_interface) def cmd_request_usage_information_records(self) -> int: """ Constructs and sends a request for Usage Information Records. Constraints: Must be logged into DD. Must be in Service mode. @return: 1 if successful, zero otherwise """ return cmd_generic_override( payload = None, reset = None, channel_id = CanChannels.dialin_to_dd_ch_id, msg_id = MsgIds.MSG_ID_UI_DD_NVM_GET_USAGE_INFO_RECORD_REQUEST, entity_name = f'DD Usage Information Record request', override_text = '', logger = self.logger, can_interface = self.can_interface)