# Subject/Title: Blood Prime - SW - 02 - Q&R # # Functionalities: Testing of Blood Prime Screen Components # # Steps: # 1 Start Leahi Application and simulate TD standby mode and navigate to the Blood Prime Screen # 2 Test Blood Prime Progress Data and Blood flow component # 3 Test Saline component Different saline states, Progress bar, Start and stop button # 4 Test Blood Prime Pressure Ranges # - Set lower and upper bound for Arterial and test each limit # - Set lower and upper bound for Venous and test each limit # - Set lower and upper bound for trancememberance and test each limit import names, re from configuration import utility, config from leahi_dialin.ui import utils from leahi_dialin.ui.td_messaging import TD_Messaging from leahi_dialin.common.td_defs import TDOpModes, TDTreatmentStates from leahi_dialin.common.ui_defs import TXStates from leahi_dialin.common.msg_defs import MsgIds, MsgFieldPositions from leahi_dialin.protocols import CAN from leahi_dialin.utils import conversions td_simulator = TD_Messaging() can_interface = td_simulator.can_interface bloodFlowrate = None bloodPrimeState = None salineBolusStartState = None ART_LOW_VAL_MINUS_390 = -390 ART_HIGH_VAL_220 = 220 VENOUS_LOW_VAL_MINUS_90 = -90 VENOUS_HIGH_VAL_410 = 410 TMP_LOW_VAL_MINUS_640 = -640 TMP_HIGH_VAL_210 = 210 REJECTED = 1 ACCEPTED = 0 total_time_list = [90, 120] def handle_solution_infusion_request(message, timestamp=0.0): """ Called when the user requests to firmware from UI @return: None """ message = message["message"] index = MsgFieldPositions.START_POS_FIELD_1 state, index = conversions.bytearray_to_integer(message, index) global salineBolusStartState salineBolusStartState = state def test_pressure_displayed_in_blood_prime_screen_range_bar(): TREATMENT_PRESSURE_ADJUSTMENT_TD_TEST_VALUES = [ { "arterial_pressure": 100, "arterial_limit_low": -300, "arterial_limit_high": 120, "venous_pressure": 140, "venous_limit_low": -100, "venous_limit_high": 250, "pressure_limit_state": 2, "tmp_pressure": 250, "tmp_limit_low": -400, "tmp_limit_high": 200, }, { "arterial_pressure": -200, "arterial_limit_low": -150, "arterial_limit_high": 120, "venous_pressure": 140, "venous_limit_low": -100, "venous_limit_high": 250, "pressure_limit_state": 2, "tmp_pressure": 200, "tmp_limit_low": -300, "tmp_limit_high": 200, }, { "arterial_pressure": 110, "arterial_limit_low": 100, "arterial_limit_high": 120, "venous_pressure": -100, "venous_limit_low": -50, "venous_limit_high": 250, "pressure_limit_state": 2, "tmp_pressure": 250, "tmp_limit_low": -200, "tmp_limit_high": 400, }, { "arterial_pressure": -130, "arterial_limit_low": -140, "arterial_limit_high": 120, "venous_pressure": 200, "venous_limit_low": -100, "venous_limit_high": 150, "pressure_limit_state": 2, "tmp_pressure": 250, "tmp_limit_low": -100, "tmp_limit_high": 600, }, ] for index in range(0, len(TREATMENT_PRESSURE_ADJUSTMENT_TD_TEST_VALUES), 1): test.startSection( "Checking the value of the range bar in pressure on the Blood Prime screen - dataset # {}".format( index ) ) target_arterial_pressure = TREATMENT_PRESSURE_ADJUSTMENT_TD_TEST_VALUES[index][ "arterial_pressure" ] target_venous_pressure = TREATMENT_PRESSURE_ADJUSTMENT_TD_TEST_VALUES[index][ "venous_pressure" ] target_pressure_state = TREATMENT_PRESSURE_ADJUSTMENT_TD_TEST_VALUES[index][ "pressure_limit_state" ] target_arterial_min_limit = TREATMENT_PRESSURE_ADJUSTMENT_TD_TEST_VALUES[index][ "arterial_limit_low" ] target_arterial_max_limit = TREATMENT_PRESSURE_ADJUSTMENT_TD_TEST_VALUES[index][ "arterial_limit_high" ] target_venous_min_limit = TREATMENT_PRESSURE_ADJUSTMENT_TD_TEST_VALUES[index][ "venous_limit_low" ] target_venous_max_limit = TREATMENT_PRESSURE_ADJUSTMENT_TD_TEST_VALUES[index][ "venous_limit_high" ] target_tmp_pressure = TREATMENT_PRESSURE_ADJUSTMENT_TD_TEST_VALUES[index][ "tmp_pressure" ] target_tmp_min_limit = TREATMENT_PRESSURE_ADJUSTMENT_TD_TEST_VALUES[index][ "tmp_limit_low" ] target_tmp_max_limit = TREATMENT_PRESSURE_ADJUSTMENT_TD_TEST_VALUES[index][ "tmp_limit_high" ] td_simulator.td_pressure( H2_arterial_pressure=target_arterial_pressure, H14_venous_pressure=target_venous_pressure, limit_state=target_pressure_state, H2_arterial_min=target_arterial_min_limit, H2_arterial_max=target_arterial_max_limit, H14_venous_min=target_venous_min_limit, H14_venous_max=target_venous_max_limit, H2_arterial_long=0, H14_venous_long=0, tmp_pressure=target_tmp_pressure, tmp_min=target_tmp_min_limit, tmp_max=target_tmp_max_limit, ) # check the range bar's upper and lower bounds arterial_rangeBar = waitForObjectExists( names.o_TreatmentBloodPrime_arterialRangeBar ) if arterial_rangeBar is not None: test.compare( arterial_rangeBar.lowerBound, target_arterial_min_limit, "Checking lower bound of arterial pressure range bar", ) test.compare( arterial_rangeBar.upperBound, target_arterial_max_limit, "Checking upper bound of arterial pressure range bar", ) # check the marker value for the arterial pressure arterial_rangeBar_marker_text_object = waitForObjectExists( names.o_TreatmentBloodPrime_arterialRangeBar ) if arterial_rangeBar_marker_text_object is not None: actual_text = str(arterial_rangeBar_marker_text_object.pressure) expected_text = str(target_arterial_pressure) test.compare( actual_text, expected_text, "Expecting arterial pressure to be displayed", ) # check the range bar's upper and lower bounds venous_rangeBar = waitForObjectExists( names.o_TreatmentBloodPrime_venousRangeBar ) if venous_rangeBar is not None: test.compare( venous_rangeBar.lowerBound, target_venous_min_limit, "Checking lower bound of venous pressure range bar", ) test.compare( venous_rangeBar.upperBound, target_venous_max_limit, "Checking upper bound of venous pressure range bar", ) # check the marker value for the venous pressure venous_rangeBar_marker_text_object = waitForObjectExists( names.o_TreatmentBloodPrime_venousRangeBar ) if venous_rangeBar_marker_text_object is not None: actual_text = str(venous_rangeBar_marker_text_object.pressure) expected_text = str(target_venous_pressure) test.compare( actual_text, expected_text, "Expecting venous pressure to be displayed" ) # check the range bar's upper and lower bounds tmp_rangeBar = waitForObjectExists(names.o_TreatmentBloodPrime_tmpRangeBar) if tmp_rangeBar is not None: test.compare( tmp_rangeBar.lowerBound, target_tmp_min_limit, "Checking lower bound of trancememberance pressure range bar", ) test.compare( tmp_rangeBar.upperBound, target_tmp_max_limit, "Checking upper bound of trancememberance pressure range bar", ) # check the marker value for the trance memberance pressure tmp_rangeBar_marker_text_object = waitForObjectExists( names.o_TreatmentBloodPrime_tmpRangeBar ) if tmp_rangeBar_marker_text_object is not None: actual_text = str(tmp_rangeBar_marker_text_object.pressure) expected_text = str(target_tmp_pressure) test.compare( actual_text, expected_text, "Expecting trancememberance pressure to be displayed", ) test.endSection() def verify_arterial_and_venous_value_in_blood_prime_screen( accepted, art_low, art_high, ven_low, ven_high, tmp_low, tmp_high ): """ Method to verify Arterial low and high and Venous low and high value on Blood Prime screen @param accepted: (int) boolean accept/reject response @param arterial_low: (int) Arterial Pressure Limit Low (mmHg) @param arterial_high: (int) Arterial Pressure Limit High (mmHg) @param venous_low: (int) Venous Pressure Limit Low (mmHg) @param venous_high: (int) Venous Pressure Limit High (mmHg) @return: none """ test.startSection( "Verifying Arterial low and high and Venous low and high value on main treatment screen" ) if accepted == ACCEPTED: arterial_rangeBar = utility.get_object_from_names( names.o_TreatmentBloodPrime_arterialRangeBar, error_message="arterial rangeBar object is missing", ) arterial_low = arterial_rangeBar.lowerBound arterial_high = arterial_rangeBar.upperBound venous_rangeBar = utility.get_object_from_names( names.o_TreatmentBloodPrime_venousRangeBar, error_message="venous rangeBar object is missing", ) venous_low = venous_rangeBar.lowerBound venous_high = venous_rangeBar.upperBound tmp_RangeBar = utility.get_object_from_names( names.o_TreatmentBloodPrime_tmpRangeBar, error_message="trancememberance rangeBar object is missing", ) trancememberance_low = tmp_RangeBar.lowerBound trancememberance_high = tmp_RangeBar.upperBound test.compare( arterial_low, art_low, "Arterial low value should be '{}'".format(art_low) ) test.compare( arterial_high, art_high, "Arterial high value should be '{}'".format(art_high), ) test.compare( venous_low, ven_low, "Venous low value should be '{}'".format(ven_low) ) test.compare( venous_high, ven_high, "Venous high value should not be '{}'".format(ven_high), ) test.compare( trancememberance_low, tmp_low, "Trancememberance low value should be '{}'".format(tmp_low), ) test.compare( trancememberance_high, tmp_high, "Trancememberance high value should not be '{}'".format(tmp_high), ) else: if object.exists(pressure_text_obj(art_low)): test.fail("Arterial value {} should not exists".format(art_low)) if object.exists(pressure_text_obj(art_high)): test.fail("Arterial value {} should not exists".format(art_high)) if object.exists(pressure_text_obj(ven_low)): test.fail("Venous value {} should not exists".format(ven_low)) if object.exists(pressure_text_obj(ven_high)): test.fail("Venous value {} should not exists".format(ven_high)) if object.exists(pressure_text_obj(tmp_low)): test.fail("Trancememberance value {} should not exists".format(tmp_low)) if object.exists(pressure_text_obj(tmp_high)): test.fail("Trancememberance value {} should not exists".format(tmp_high)) test.endSection() def test_blood_prime_saline_delivery(): """ Verify the Different Saline States, ProgressBar, Start and Stop button """ test.startSection( "Verify the Different Saline States, ProgressBar, Start and Stop button" ) # handle sent messages from UI if can_interface is not None: channel_id = CAN.DenaliChannels.ui_to_td_ch_id message_id = MsgIds.MSG_ID_UI_SOLUTION_INFUSION_REQUEST.value can_interface.register_receiving_publication_function( channel_id, message_id, handle_solution_infusion_request ) # verify saline text salineTextObj = waitForObject(names.o_TreatmentBloodPrime_salineDescriptionText) test.compare( salineTextObj.text, config.TOTAL_SALINE_DELIVERED, "Testing 'Total Saline Delivered' text", ) # verify send start bolus req message mouseClick( waitForObject(names.o_TreatmentBloodPrime_startFluidButton_TouchRect, 3000) ) test.verify( waitFor(lambda: salineBolusStartState == 1, 10000), "Verifying FW received Saline Bolus Start State in Blood Priming screen", ) # change state to in progress td_simulator.td_tx_state( TDTreatmentStates.TREATMENT_BLOOD_PRIME_STATE.value, 0, 0, 0, 0, 0, 0, 0, TXStates.SALINE_BOLUS_STATE_IN_PROGRESS, # saline in progess 0, ) # verify isStarted property binding to vTDTreatmentStates.sbRunning treatmentSaline = waitForObject(names.o_TreatmentBloodPrime_treatmentSaline) test.compare( treatmentSaline.isStarted, True, "Testing vTDTreatmentStates.sbRunning" ) # verify text test.compare( salineTextObj.text, config.DELIVERING_SALINE, "Testing Saline Description text", ) # verify saline data bolusValueInTest = 75 td_simulator.td_saline( 100, # target_volume 150, # cumulative_volume bolusValueInTest, # bolus_volume 0, ) # verify saline text currentBolusTextObj = waitForObject(names.o_TreatmentBloodPrime_currentBolusVolume) test.compare( currentBolusTextObj.text, str(bolusValueInTest), "Testing current bolus value text", ) # verify progress bar progressBarObj = waitForObject( names.o_TreatmentBloodPrime_fluidProgressBar_ProgressBar ) test.compare( progressBarObj.value, bolusValueInTest, "Testing progress bar current bolus value", ) # verify send stop bolus req message mouseClick(waitForObject(names.o_TreatmentBloodPrime_startFluidButton_TouchRect)) test.verify( waitFor(lambda: salineBolusStartState == 0, 10000), "Verifying FW received Saline Bolus Start State in Blood Priming screen", ) # change state to idle td_simulator.td_tx_state( TDTreatmentStates.TREATMENT_BLOOD_PRIME_STATE.value, 0, 0, 0, 0, 0, 0, 0, TXStates.SALINE_BOLUS_STATE_IDLE, # saline in idle 0, ) # verify progress bar set off test.compare(progressBarObj.value, 0, "Testing progress bar bolus off") test.endSection() def handle_blood_flow_rate_request(message, timestamp=0.0): """ Called when the user requests to firmware from UI @return: None """ message = message["message"] index = MsgFieldPositions.START_POS_FIELD_1 state, index = conversions.bytearray_to_integer(message, index) global bloodFlowrate bloodFlowrate = state def handle_pause_blood_prime_request(message, timestamp=0.0): """ Called when the user requests to firmware from UI @return: None """ message = message["message"] index = MsgFieldPositions.START_POS_FIELD_1 state, index = conversions.bytearray_to_integer(message, index) global bloodPrimeState bloodPrimeState = state def test_blood_flow_rate(): """ Verifying Blood Prime Progress Data and Blood flow component """ test.startSection( "Verifying Blood Prime Progress Data and Blood flow component" ) td_simulator.td_blood_prime_progress(1000, 500, 0, 0) blood_prime_progress_maximum_volume = waitForObject( names.o_TreatmentBloodPrime_volumeProgress_ProgressCircle, 3000 ).maximum test.compare( blood_prime_progress_maximum_volume, 1000, "Comparison of Blood Prime Progress Maximum volume", ) blood_prime_progress_delivered_volume = waitForObject( names.o_TreatmentBloodPrime_volumeText_Text, 3000 ).text test.compare( blood_prime_progress_delivered_volume, "500 mL", "Comparison of Blood Prime Progress Delivered volume", ) td_simulator.td_treatment_set_points(70, 0, 0) if can_interface is not None: channel_id = CAN.DenaliChannels.ui_to_td_ch_id message_id = ( MsgIds.MSG_ID_UI_TREATMENT_SET_POINT_BLOOD_FLOW_CHANGE_REQUEST.value ) can_interface.register_receiving_publication_function( channel_id, message_id, handle_blood_flow_rate_request ) blood_flow_rate_value = waitForObject( names.o_TreatmentBloodPrime_flowRateValue_Text, 3000 ).text test.compare(blood_flow_rate_value, "70", "Comparison of Flow Rate Value") blood_flow_rate_unit = waitForObject( names.o_TreatmentBloodPrime_flowRateUnit_Text, 3000 ).text test.compare(blood_flow_rate_unit, "mL/min", "Comparison of Flow Rate Unit") mouseClick( waitForObject( names.o_TreatmentBloodPrime_incrementFlowRateButton_ArrowButton, 3000 ) ) test.verify( waitFor(lambda: bloodFlowrate == 80, 10000), "Verifying FW received Increased Blood flow Value in Blood Priming screen", ) td_simulator.td_treatment_set_points(70, 0, 0) mouseClick( waitForObject( names.o_TreatmentBloodPrime_decrementFlowRateButton_ArrowButton, 3000 ) ) test.verify( waitFor(lambda: bloodFlowrate == 60, 10000), "Verifying FW received Decreased Blood flow Value in Blood Priming screen", ) if can_interface is not None: channel_id = CAN.DenaliChannels.ui_to_td_ch_id message_id = MsgIds.MSG_ID_UI_BLOOD_PRIME_CMD_REQUEST.value can_interface.register_receiving_publication_function( channel_id, message_id, handle_pause_blood_prime_request ) mouseClick( waitForObject(names.o_TreatmentBloodPrime_pauseResumeButton_TouchRect, 3000) ) test.verify( waitFor(lambda: bloodPrimeState == 0, 10000), "Verifying FW received Pause State of Blood Flow Rate in Blood Priming screen", ) td_simulator.td_blood_prime_cmd_response(0, 2) blood_priming_reject_notification = utility.get_object_from_names( names.o_TreatmentBloodPrime_NotificationBar ) blood_priming_reject_notification_text = str(blood_priming_reject_notification.text) digit = re.sub(r"^.*?\[(\d+)\].*$", r"\1", blood_priming_reject_notification_text) if digit: digit_int = int(float(digit)) test.compare(2, digit_int, f"Rejection reason number comparison for reason {2}") test.log(f"Rejection reason text is {blood_priming_reject_notification_text}") td_simulator.td_tx_state( TDTreatmentStates.TREATMENT_BLOOD_PRIME_STATE.value, 2, 0, 0, 0, 0, 0, 0, 0, 0 ) blood_prime_progress_paused_state = waitForObject( names.o_TreatmentBloodPrime_pausedNotification, 3000 ).text test.compare( blood_prime_progress_paused_state, config.BLOOD_PRIME_PAUSED_NOTIFICATION, "Comparison of Blood Prime Progress Paused State", ) td_simulator.td_blood_prime_progress(1000, 500, 500, 250) blood_prime_progress_paused_minutes_countdown = waitForObject( names.o_TreatmentBloodPrime_TimeText_hour_Text, 3000 ).text test.compare( blood_prime_progress_paused_minutes_countdown, "04", "Comparison of Blood Prime Progress Paused Minutes Countdown", ) blood_prime_progress_paused_seconds_countdown = waitForObject( names.o_TreatmentBloodPrime_TimeText_minute_Text, 3000 ).text test.compare( blood_prime_progress_paused_seconds_countdown, "10", "Comparison of Blood Prime Progress Paused Seconds Countdown", ) mouseClick( waitForObject(names.o_TreatmentBloodPrime_pauseResumeButton_TouchRect, 3000) ) test.verify( waitFor(lambda: bloodPrimeState == 1, 10000), "Verifying FW received Resume State of Blood Flow Rate in Blood Priming screen", ) td_simulator.td_tx_state( TDTreatmentStates.TREATMENT_BLOOD_PRIME_STATE.value, 1, 0, 0, 0, 0, 0, 0, 0, 0 ) td_simulator.td_blood_prime_progress(1000, 500, 0, 0) mouseClick( waitForObject(names.o_TreatmentBloodPrime_pauseResumeButton_TouchRect, 3000) ) test.verify( waitFor(lambda: bloodPrimeState == 0, 10000), "Verifying FW received Pause State of Blood Flow Rate in Blood Priming screen", ) test.endSection() def main(): utils.tstStart(__file__) startApplication(utility.aut("-q")) td_simulator.td_operation_mode(TDOpModes.MODE_STAN.value) mouseClick(waitForObject(names.o_createTreatmentRect_TouchRect, 2000)) td_simulator.td_operation_mode(TDOpModes.MODE_TREA.value, 0) td_simulator.td_tx_state( TDTreatmentStates.TREATMENT_BLOOD_PRIME_STATE.value, 0, 0, 0, 0, 0, 0, 0, 0, 0 ) test_blood_flow_rate() test_blood_prime_saline_delivery() test_pressure_displayed_in_blood_prime_screen_range_bar() td_simulator.td_pressure( H2_arterial_pressure=0, H14_venous_pressure=0, limit_state=0, H2_arterial_min=ART_LOW_VAL_MINUS_390, H2_arterial_max=ART_HIGH_VAL_220, H14_venous_min=VENOUS_LOW_VAL_MINUS_90, H14_venous_max=VENOUS_HIGH_VAL_410, H2_arterial_long=0, H14_venous_long=0, tmp_pressure=0, tmp_min=TMP_LOW_VAL_MINUS_640, tmp_max=TMP_HIGH_VAL_210, ) verify_arterial_and_venous_value_in_blood_prime_screen( ACCEPTED, ART_LOW_VAL_MINUS_390, ART_HIGH_VAL_220, VENOUS_LOW_VAL_MINUS_90, VENOUS_HIGH_VAL_410, TMP_LOW_VAL_MINUS_640, TMP_HIGH_VAL_210, )