########################################################################### # # Copyright (c) 2021-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 test_flush.py # # @author (last) Dara Navaei # @date (last) 18-Jan-2023 # @author (original) Dara Navaei # @date (original) 16-Apr-2021 # ############################################################################ import sys sys.path.append("..") from dialin.dg.dialysate_generator import DG from dialin.hd.hemodialysis_device import HD from dialin.dg.drain_pump import DrainPumpStates from time import sleep def get_flush_mode_info(): """ info = ('State, {}, Overall_elapsed_time, {}, State_elapsed_time, {}, Drain_vol, {:5.3f}, Top_alarm, {}, Op, {}, ' 'Sub, {}, History, {}, ' .format(dg.flush.flush_state, dg.flush.overall_elapsed_time, dg.flush.state_elapsed_time, dg.flush.flush_drain_line_volume_l, hd.alarms.alarm_top, dg.events.get_dg_events(1, 0), dg.events.get_dg_nth_event(2), dg.events.get_dg_events(2, 70))) """ info = ('State, {}, Overall_elapsed_time, {}, State_elapsed_time, {}, Drain_vol, {:5.3f}, Top_alarm, {}, ' .format(dg.flush.flush_state, dg.flush.overall_elapsed_time, dg.flush.state_elapsed_time, dg.flush.flush_drain_line_volume_l, hd.alarms.alarm_top)) return info def get_concentrate_pumps_info(): info = ('Bicarb_tgt_speed, {:5.3f}, Bicarb_speed, {:5.3f}, Acid_tgt_speed, {:5.3f}, Acid_speed, {:5.3f}, ' 'CPo, {:5.3f}, CD1, {:5.3f}, CD2, {:5.3f}, ' .format(dg.concentrate_pumps.concentrate_pump_cp2_current_set_speed, dg.concentrate_pumps.concentrate_pump_cp2_measured_speed, dg.concentrate_pumps.concentrate_pump_cp1_current_set_speed, dg.concentrate_pumps.concentrate_pump_cp1_measured_speed, dg.conductivity_sensors.conductivity_sensor_cpo, dg.conductivity_sensors.conductivity_sensor_cd1, dg.conductivity_sensors.conductivity_sensor_cd2)) return info def get_dg_valves_states(): info = ('VPi, {}, VSP, {}, VPd, {}, VBf, {}, VPo, {}, VDr, {}, VRc, {}, VRo, {}, VRd, {}, VRi, {}, VRf, {}, ' 'VRD1, {}, VRD2, {}, ' .format(dg.valves.valve_states_enum[dg.valves.VALVE_PRESSURE_INLET], dg.valves.valve_states_enum[dg.valves.VALVE_SAMPLING_PORT], dg.valves.valve_states_enum[dg.valves.VALVE_PRODUCTION_DRAIN], dg.valves.valve_states_enum[dg.valves.VALVE_BYPASS_FILTER], dg.valves.valve_states_enum[dg.valves.VALVE_PRESSURE_OUTLET], dg.valves.valve_states_enum[dg.valves.VALVE_DRAIN], dg.valves.valve_states_enum[dg.valves.VALVE_RECIRCULATE], dg.valves.valve_states_enum[dg.valves.VALVE_RESERVOIR_OUTLET], dg.valves.valve_states_enum[dg.valves.VALVE_RESERVOIR_DRAIN], dg.valves.valve_states_enum[dg.valves.VALVE_RESERVOIR_INLET], dg.valves.valve_states_enum[dg.valves.VALVE_RESERVOIR_FILL], dg.valves.valve_states_enum[dg.valves.VALVE_RESERVOIR_DRAIN_1], dg.valves.valve_states_enum[dg.valves.VALVE_RESERVOIR_DRAIN_2])) return info def get_drain_states_info(): info = ('Drain, {}, DAC, {}, Tgt_RPM, {}, Curr_RPM, {}, PRd, {:5.3f}, PDr, {:5.3f}, '. format(DrainPumpStates(dg.drain_pump.drain_pump_state).name, dg.drain_pump.dac_value, dg.drain_pump.target_drain_pump_rpm, dg.drain_pump.current_drain_pump_rpm, dg.pressures.drain_pump_inlet_pressure, dg.pressures.drain_pump_outlet_pressure)) return info def get_load_cells_info(): info = ('A1, {:5.3f}, A2, {:5.3f}, B1, {:5.3f}, B2, {:5.3f}, '. format(dg.load_cells.load_cell_A1, dg.load_cells.load_cell_A2, dg.load_cells.load_cell_B1, dg.load_cells.load_cell_B2)) return info def get_ro_info(): info = ('RO, {}, PPi, {:5.3f}, PPo, {:5.3f}, PWM, {:5.3f}, Flow, {:5.3f}, Feedback_PWM, {:5.3f},' .format(dg.ro_pump.ro_pump_state, dg.pressures.ro_pump_inlet_pressure, dg.pressures.ro_pump_outlet_pressure, dg.ro_pump.pwm_duty_cycle_pct, dg.ro_pump.measured_flow_rate_lpm, dg.ro_pump.feedback_duty_cycle_pct)) return info def get_uv_reactors_info(): info = ('Inlet_status, {}, Outlet_status, {}, ' .format(dg.uv_reactors.inlet_uv_reactor_state, dg.uv_reactors.outlet_uv_reactor_state)) return info def run_flush_mode(): complete_counter = 1 f = open("/home/fw/projects/leahi_dialin/tests/flush_mode.log", "w") dg.hd_proxy.cmd_start_stop_dg_flush() #dg.cmd_dg_software_reset_request() try: while True: flush = get_flush_mode_info() load_cell = get_load_cells_info() drain = get_drain_states_info() ro = get_ro_info() conc = get_concentrate_pumps_info() uv_reactors = get_uv_reactors_info() valves = get_dg_valves_states() var = flush + load_cell + drain + ro + conc + uv_reactors + valves + '\r' print(var) f.write(var) sleep(1) # If the mode came back to standby or standby solo if dg.dg_operation_mode == 3 or dg.dg_operation_mode == 4: if complete_counter == 1: dg.hd_proxy.cmd_start_stop_dg_flush() # Write a few more complete states to make sure the complete state items are recorded if complete_counter == 3: f.close() #events = dg.events.get_dg_events(2, 60) #for event in events: # print(event) break complete_counter += 1 except KeyboardInterrupt: dg.hd_proxy.cmd_start_stop_dg_flush(start=False) f.close() #events = dg.events.get_dg_events(2, 50) #for event in events: # print(event) if __name__ == "__main__": dg = DG(log_level='DEBUG') hd = HD(log_level='DEBUG') dg.cmd_log_in_to_dg() sleep(1) hd.cmd_log_in_to_hd() sleep(1) run_flush_mode()