# -*- coding: utf-8 -*- import names import squish from leahi_dialin.ui import utils from configuration import utility,config,navigation from configuration import config from leahi_dialin.ui.td_messaging import TD_Messaging from leahi_dialin.common.ui_defs import TXStates from leahi_dialin.common.td_defs import TDOpModes,TDTreatmentStates from leahi_dialin.common.msg_defs import MsgIds, MsgFieldPositions from leahi_dialin.utils import conversions import can from leahi_dialin.protocols import CAN td =TD_Messaging() ArtWindow = None Venouspressurewindow ="0" venouspressurelimitasymetric ="0" tmppressurelimut ="0" salineBolusvalue ="0" def handle_saline_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 saline saline = state def handle_treatment_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 state,index = conversions.bytearray_to_integer( message, index) global dialysateflow dialysateflow = state val,self.index = conversions.bytearray_to_integer( message, self. Index) global dialysatetemperature dialysatetemperature = val def handle_pressure_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 ArtWindow ArtWindow = state test.log("Inside function") test.log(str(ArtWindow)) test.log(str(state)) state,index = conversions.bytearray_to_integer( message, index) global Venouspressurewindow Venouspressurewindow= state state,index = conversions.bytearray_to_integer( message, index) global venouspressurelimitasymetric venouspressurelimitasymetric = state state,index = conversions.bytearray_to_integer( message, index) global tmppressurelimut tmppressurelimut = state def handle_bloodflow_request(): message = message['message'] index = MsgFieldPositions.START_POS_FIELD_1 state,index = conversions.bytearray_to_integer( message, index) global bloodflow bloodflow = state def main(): can_interface = td.can_interface global Venouspressurewindow if can_interface is not None: channel_id = CAN.DenaliChannels.ui_to_td_ch_id message_id = MsgIds.MSG_ID_UI_PRESSURE_LIMITS_CHANGE_REQUEST.value can_interface.register_receiving_publication_function(channel_id, message_id, handle_pressure_request) utils.tstStart(__file__) startApplication(config.AUT_NAME) td.td_operation_mode(TDOpModes.MODE_STAN.value) # mouseClick(waitForObject(names.o_createTreatmentRect_TouchRect)) # td.td_operation_mode(TDOpModes.MODE_PRET.value) # utility.verify_create_treatment_parameters() td.td_tx_state( TDTreatmentStates.TREATMENT_DIALYSIS_STATE.value, 0 , 0 , 0 , 0 , 0 , 0 , 0 , 0 , 0) td.td_pressure(H2_arterial_pressure = 40, H14_venous_pressure = 50 , limit_state = 0, H2_arterial_min = -390, H2_arterial_max = 220 , H14_venous_min = -90 , H14_venous_max = 410 , H2_arterial_long = 0 , H14_venous_long = 0 , tmp_pressure = 43 , tmp_min = -640 , tmp_max = 210) # mouseClick(waitForObject(names.o_treatmentHome_editButton_IconButton)) pressureEdit = waitForObject(names.o_treatmentHome_treatmentPressures_TreatmentPressures) parent = utility.findObjectById(pressureEdit, "_editButton") mouseClick(parent) test.log("Check pressure edit values") #check the values in edit pressure venouswindow_value = waitForObject(names.o_venousWindow_LabelUnitValueAdjuster) test.log(str(venouswindow_value.value)) tmpwindow_value = waitForObject(names.o_tmpWindow_LabelUnitValueAdjuster) test.log(str(tmpwindow_value.value)) utility.set_value_based_on_target(names.o_arterialWindow_LabelUnitValueAdjuster,80) arterialwindow_value = waitForObject(names.o_arterialWindow_LabelUnitValueAdjuster) Arterial_Value =arterialwindow_value.value utility.set_value_based_on_target(names.o_venousWindow_LabelUnitValueAdjuster,60) utility.set_value_based_on_target(names.o_tmpWindow_LabelUnitValueAdjuster,100) #rejection reason is 1 payload = conversions.integer_to_bytearray(0) payload += conversions.integer_to_bytearray(1) td.cmd_send_general_response(message_id =MsgIds.MSG_ID_TD_PRESSURE_LIMITS_CHANGE_RESPONSE.value, reason=0, accepted=1, is_pure_data = False, has_parameters = True, parameters_payload= payload) rejectionReason = utility.get_object_from_names(names.o_notification_bar) test.compare(rejectionReason.text, "[1] Request is not allowed in the current operating mode", "Rejection Reason text should be :" + str("[1] Request is not allowed in the current operating mode")) # handle sent messages from UI mouseClick(waitForObject(names.o_confirmButton_TouchRect)) #rejection reason 0 payload = conversions.integer_to_bytearray(1) payload += conversions.integer_to_bytearray(0) td.cmd_send_general_response(message_id =MsgIds.MSG_ID_TD_PRESSURE_LIMITS_CHANGE_RESPONSE.value, reason=0, accepted=1, is_pure_data = False, has_parameters = True, parameters_payload= payload) global ArtWindow test.verify(waitFor( lambda: ArtWindow == 10, 5000), "Testing lambda this should fail") test.verify(waitFor( lambda: ArtWindow == arterialwindow_value.value, 5000), "Testing FW received Arterial Window Value") test.log("Post waitfor() Verify ArtWindow: ") test.log(str(ArtWindow)) test.log(str(arterialwindow_value.value)) test.verify(Venouspressurewindow == venouswindow_value.value ) pressurepopup = waitFor(lambda: not object.exists(names.o_HeaderBar_WiFi_Popup), 3000) test.compare(pressurepopup, True, "Popup is not present") test.startSection("Saline data") td.td_saline( 100 , # target_volume 150 , # cumulative_volume 75 , # bolus_volume 0 ) mouseClick(waitForObject(names.o_treatmentHome_editButton_IconButton)) fluidbolusvolume_value = waitForObject(names.o_valueAdjuster_ValueAdjuster) test.log(str(fluidbolusvolume_value.value)) utility.set_value_based_on_target(names.o_valueAdjuster_ValueAdjuster,200) payload = conversions.integer_to_bytearray(0) payload += conversions.integer_to_bytearray(1) td.cmd_send_general_response(message_id =MsgIds.MSG_ID_TD_BOLUS_VOLUME_CHANGE_RESPONSE.value, reason=0, accepted=1, is_pure_data = False, has_parameters = True, parameters_payload= payload) rejectionReason = utility.get_object_from_names(names.o_notification_bar) test.compare(rejectionReason.text, "[1] Request is not allowed in the current operating mode", "Rejection Reason text should be :" + str("[1] Request is not allowed in the current operating mode")) #rejection reason 0 mouseClick(waitForObject(names.o_confirmButton_TouchRect)) payload = conversions.integer_to_bytearray(1) payload += conversions.integer_to_bytearray(0) td.cmd_send_general_response(message_id =MsgIds.MSG_ID_TD_BOLUS_VOLUME_CHANGE_RESPONSE.value, reason=0, accepted=1, is_pure_data = False, has_parameters = True, parameters_payload= payload) # 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_BOLUS_VOLUME_CHANGE_REQUEST.value can_interface.register_receiving_publication_function(channel_id, message_id, handle_saline_request) # test.log(str(salineBolusvalue)) salinepopup = waitFor(lambda: not object.exists(names.o_HeaderBar_WiFi_Popup), 3000) test.compare(salinepopup, True, "Popup is not present") #check the recieved values test.startSection("Treatment Parameters") td.td_treatment_set_points( blood_flow = 42, dialysate_flow = 43, dialysate_temp = 49) test.log("Sample") treatmentEdit = waitForObject(names.o_treatmentHome_treatmentFlows_TreatmentFlows) treatmentEditbutton = utility.findObjectById(treatmentEdit, "_editButton") mouseClick(treatmentEditbutton) bloodFlow = waitForObject(names.o_bloodFlowRate_LabelUnitValueAdjuster) bloodflowvalue = utility.findObjectById(bloodFlow, "_valueAdjuster") test.log(str(bloodflowvalue.value)) test.log(str(waitForObject(names.o_bloodFlowRate_LabelUnitValueAdjuster).valueTextColor.name)) utility.set_value_based_on_target(names.o_bloodFlowRate_LabelUnitValueAdjuster,"50") test.log(str(bloodflowvalue.value)) test.log(str(waitForObject(names.o_bloodFlowRate_LabelUnitValueAdjuster).valueTextColor.name)) dialysateFlow = waitForObject(names.o_dialysateFlowRate_LabelUnitValueAdjuster) dialysateflowvalue = utility.findObjectById(dialysateFlow, "_valueAdjuster") test.log(str(dialysateflowvalue.value)) test.log(str(waitForObject(names.o_dialysateFlowRate_LabelUnitValueAdjuster).valueTextColor.name)) utility.set_value_based_on_target(names.o_dialysateFlowRate_LabelUnitValueAdjuster,"50") test.log(str(dialysateflowvalue.value)) test.log(str(waitForObject(names.o_dialysateFlowRate_LabelUnitValueAdjuster).valueTextColor.name)) dialysateTemp = waitForObject(names.o_dialysateTemperature_LabelUnitValueAdjuster) dialysatetempvalue = utility.findObjectById(dialysateTemp, "_valueAdjuster") test.log(str(dialysatetempvalue.value)) test.log(str(waitForObject(names.o_dialysateTemperature_LabelUnitValueAdjuster).valueTextColor.name)) utility.set_value_based_on_target(names.o_dialysateTemperature_LabelUnitValueAdjuster,"38") test.log(str(dialysatetempvalue.value)) test.log(str(waitForObject(names.o_dialysateTemperature_LabelUnitValueAdjuster).valueTextColor.name)) #rejection reason is 1 payload = conversions.integer_to_bytearray(0) payload += conversions.integer_to_bytearray(1) td.cmd_send_general_response(message_id =MsgIds.MSG_ID_TD_TREATMENT_SET_POINTS_CHANGE_RESPONSE.value, reason=0, accepted=1, is_pure_data = False, has_parameters = True, parameters_payload= payload) rejectionReason = utility.get_object_from_names(names.o_notification_bar).text test.compare(rejectionReason, "[1] Request is not allowed in the current operating mode", "Rejection Reason text should be :" + str("[1] Request is not allowed in the current operating mode")) #rejection reason is 0 mouseClick(waitForObject(names.o_confirmButton_TouchRect)) payload = conversions.integer_to_bytearray(1) payload += conversions.integer_to_bytearray(0) td.cmd_send_general_response(message_id =MsgIds.MSG_ID_TD_TREATMENT_SET_POINTS_CHANGE_RESPONSE.value, reason=0, accepted=1, is_pure_data = False, has_parameters = True, parameters_payload= payload) if can_interface is not None: channel_id = CAN.DenaliChannels.ui_to_td_ch_id message_id = MsgIds.MSG_ID_UI_TREATMENT_SET_POINTS_CHANGE_REQUEST.value can_interface.register_receiving_publication_function(channel_id, message_id, handle_treatment_request) # test.log(str(salineBolusvalue)) treatmentpopup = waitFor(lambda: not object.exists(names.o_HeaderBar_WiFi_Popup), 3000) test.compare(treatmentpopup, True, "Popup is not present") test.endSection() test.startSection("Main treatment screen check the set points") td.td_treatment_set_points( blood_flow = 70, dialysate_flow = 70, dialysate_temp = 36) bloodflowUpArrow = waitForObject(names.o_blood_flow_value) bloodflowUpArrowbutton = utility.findObjectById(bloodflowUpArrow,"_upArrowIcon") mouseClick(bloodflowUpArrowbutton) 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_bloodflow_request) #check the value incremented in blood flow bloodflowdownArrowbutton = utility.findObjectById(bloodflowUpArrow,"_downArrowIcon") mouseClick(bloodflowdownArrowbutton) #check the decremented value dialysateflowuparrow = waitForObject(names.o_dial_flow_value) dialysateflowuparrowbutton = utility.findObjectById(dialysateflowuparrow,"_upArrowIcon") mouseClick(dialysateflowuparrowbutton) if can_interface is not None: channel_id = CAN.DenaliChannels.ui_to_td_ch_id message_id = MsgIds.MSG_ID_TD_TREATMENT_SET_POINT_DIALYSATE_FLOW_CHANGE_RESPONSE.value can_interface.register_receiving_publication_function(channel_id, message_id, handle_bloodflow_request) #check the value incremented in dialysate flow value dialysateflowdownArrowbutton = utility.findObjectById(dialysateflowuparrow,"_downArrowIcon") mouseClick(dialysateflowdownArrowbutton) #check the decremented value dialysatetempuparrow = waitForObject(names.o_dial_tmp_value) dialysatetempuparrowbutton = utility.findObjectById(dialysatetempuparrow,"_upArrowIcon") mouseClick(dialysatetempuparrowbutton) if can_interface is not None: channel_id = CAN.DenaliChannels.ui_to_td_ch_id message_id = MsgIds.MSG_ID_TD_TREATMENT_SET_POINT_DIALYSATE_TEMPERATURE_CHANGE_RESPONSE.value can_interface.register_receiving_publication_function(channel_id, message_id, handle_bloodflow_request) #check the value incremented in dialysate flow value dialysatetempdownArrowbutton = utility.findObjectById(dialysatetempuparrow,"_downArrowIcon") mouseClick(dialysatetempdownArrowbutton) #rejection reason is 1 payload = conversions.integer_to_bytearray(0) payload += conversions.integer_to_bytearray(1) td.cmd_send_general_response(message_id =MsgIds.MSG_ID_TD_TREATMENT_SET_POINT_BLOOD_FLOW_CHANGE_RESPONSE.value, reason=0, accepted=1, is_pure_data = False, has_parameters = True, parameters_payload= payload) rejectionReason = utility.get_object_from_names(names.o_treatmentHome_notification_NotificationBarSmall) test.compare(rejectionReason.text, "[1] Request is not allowed in the current operating mode", "Rejection Reason text should be :" + str("[1] Request is not allowed in the current operating mode")) #rejection reason is 0 payload = conversions.integer_to_bytearray(1) payload += conversions.integer_to_bytearray(0) td.cmd_send_general_response(message_id =MsgIds.MSG_ID_TD_TREATMENT_SET_POINT_BLOOD_FLOW_CHANGE_RESPONSE.value, reason=0, accepted=1, is_pure_data = False, has_parameters = True, parameters_payload= payload) #need to verify the response td.td_treatment_set_points( blood_flow = 94, dialysate_flow = 94, dialysate_temp = 28) bloodflowUpArrow = waitForObject(names.o_blood_flow_value) bloodflowUpArrowbutton = utility.findObjectById(bloodflowUpArrow,"_upArrowIcon") mouseClick(bloodflowUpArrowbutton) 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_bloodflow_request) #check the value incremented in blood flow dialysateflowuparrow = waitForObject(names.o_dial_flow_value) dialysateflowuparrowbutton = utility.findObjectById(dialysateflowuparrow,"_upArrowIcon") mouseClick(dialysateflowuparrowbutton) if can_interface is not None: channel_id = CAN.DenaliChannels.ui_to_td_ch_id message_id = MsgIds.MSG_ID_TD_TREATMENT_SET_POINT_DIALYSATE_FLOW_CHANGE_RESPONSE.value can_interface.register_receiving_publication_function(channel_id, message_id, handle_bloodflow_request) #check the value incremented in dialysate flow value dialysatetempuparrow = waitForObject(names.o_dial_tmp_value) dialysatetempuparrowbutton = utility.findObjectById(dialysatetempuparrow,"_upArrowIcon") mouseClick(dialysatetempuparrowbutton) if can_interface is not None: channel_id = CAN.DenaliChannels.ui_to_td_ch_id message_id = MsgIds.MSG_ID_TD_TREATMENT_SET_POINT_DIALYSATE_TEMPERATURE_CHANGE_RESPONSE.value can_interface.register_receiving_publication_function(channel_id, message_id, handle_bloodflow_request) #check the value incremented in dialysate flow value test.endSection() test.startSection("Treatment Duration Edit") td.td_treatment_time( tx_duration_s = 51 , tx_elapsed_s = 5 , tx_remaining_s = 6 ) td.td_ultrafiltration(10,0.53,0.37,0) td.td_param_ranges(min_tx_time_s = 50, max_tx_time_s = 300, min_uf_volume_mL = 0.0, max_uf_volume_mL = 1790, min_dial_rate_mLH = 0, max_dial_rate_mLH = 0) treatmentEdit = waitForObject(names.o_treatment_duration) treatmentEditbutton = utility.findObjectById(treatmentEdit,"_editTimeIcon") mouseClick(treatmentEditbutton) ufvolumeremoved = waitForObject(names.o_ufVolumeRemoved_LabelUnitText).value test.log(str(ufvolumeremoved)) ufvolumegoal =waitForObject(names.o_ufVolumeGoal_LabelUnitText) test.log(str(ufvolumegoal)) treatmenttimeelapsed =waitForObject(names.o_timeElapsed_LabelUnitText) test.log(str(treatmenttimeelapsed)) utility.set_value_based_on_target(names.o_newTreatmentDuration_LabelUnitValueAdjuster,51) utility.set_value_based_on_target(names.o_newTreatmentDuration_LabelUnitValueAdjuster,300) #rejection reason 0 # payload = conversions.integer_to_bytearray(1) # payload += conversions.integer_to_bytearray(0) # td.cmd_send_general_response(message_id =MsgIds.MSG_ID_TD_DURATION_CONFIRM_RESPONSE.value, # reason=0, # accepted=1, # is_pure_data = False, # has_parameters = True, # parameters_payload= payload) # # #need to verify the response # # #rejection reason is 1 td.td_duration_validate_response(vRejectionReason =1, vDuration= 78, vUFVolume= 0.9, vUFRate= 0.6) rejectionReason = utility.get_object_from_names(names.o_notification_bar) test.compare(rejectionReason.text, "[1] Request is not allowed in the current operating mode", "Rejection Reason text should be :" + str("[1] Request is not allowed in the current operating mode")) td.td_duration_validate_response(vRejectionReason =0, vDuration= 78, vUFVolume= 0.9, vUFRate= 0.6) #need to check the request #rejection reason is 1 payload = conversions.integer_to_bytearray(0) payload += conversions.integer_to_bytearray(1) td.cmd_send_general_response(message_id =MsgIds.MSG_ID_TD_DURATION_CONFIRM_RESPONSE.value, reason=0, accepted=1, is_pure_data = False, has_parameters = True, parameters_payload= payload) rejectionReason = utility.get_object_from_names(names.o_notification_bar).text test.compare(rejectionReason, "[1] Request is not allowed in the current operating mode", "Rejection Reason text should be :" + str("[1] Request is not allowed in the current operating mode")) #rejection reason is 0 mouseClick(waitForObject(names.o_confirmButton_TouchRect)) payload = conversions.integer_to_bytearray(1) payload += conversions.integer_to_bytearray(0) td.cmd_send_general_response(message_id =MsgIds.MSG_ID_TD_DURATION_CONFIRM_RESPONSE.value, reason=0, accepted=1, is_pure_data = False, has_parameters = True, parameters_payload= payload) # if can_interface is not None: # channel_id = CAN.DenaliChannels.ui_to_td_ch_id # message_id = MsgIds.MSG_ID_UI_TREATMENT_SET_POINTS_CHANGE_REQUEST.value # can_interface.register_receiving_publication_function(channel_id, # message_id, # handle_treatment_request) treatmentpopup = waitFor(lambda: not object.exists(names.o_HeaderBar_WiFi_Popup), 3000) test.compare(treatmentpopup, True, "Popup is not present")