st-ten-1/src/ui/test_leak/test_leak.py

import sys
import time
import weakref

from PyQt5.QtGui import QPixmap
from PyQt5.QtWidgets import QMessageBox, QDialog, QApplication
from PyQt5.QtCore import Qt
from ui import Dialog
from ui.test_instructions_reminder import Test_Instructions_Reminder
from ui.test_test import Test_Test

VALVE_TIME=0.5

class Test_Leak(Test_Test):
    def __init__(self, components=None, recipe=None, step=None, pieces=None, run_once=False, reset_on_start=True, enable_override=False,parent=None):
        super().__init__(components=components, recipe=recipe, step=step, pieces=pieces, run_once=run_once, reset_on_start=reset_on_start, enable_override=enable_override)
        self.get_connection = None
        self.io_ok = True
        self.blow_on = False
        self.parent=parent
        self.step=step
        self.recipe_written = False
        self.start_b.clicked.connect(self.start_test)
        self.stop_b.clicked.connect(self.stop_test)
        self.show_instruction_b.setVisible("show_instructions" in self.parent.config["hardware_config"].keys())
        self.show_instruction_b.clicked.connect(self.show_instruction)

    def show_instruction(self):
        dialog=Dialog()
        dialog.setCentralWidget(Test_Instructions_Reminder(recipe=self.parent.recipe,bench_name=self.parent.config.machine_id))
        dialog.show()

    def reset(self):
        self.components[self.tester_component].stop_test()
        super().reset()

    def stop_test(self):
        self.components[self.tester_component].stop_test()
        self.display_text(text="PROVA INTERROTTA", bg_color="yellow")
        time.sleep(1)
        self.start_b.setEnabled(True)
        self.stop_b.setEnabled(False)

    def start_test(self):
        # print extra labels
        if self.step.type == "leak_1":
            self.parent.print_extra_labels()

        # SELECT TEST CHANNEL
        if self.parent.config["hardware_config"].get("external_flush_blow", None) == "present":
            if self.parent.config["hardware_config"].get("dual_channel", None) == "present":
                chan_sel = self.step.spec["chan_sel"] # 0=CH1, 1=CH2
                self.set_digital_out("out_channel_select", chan_sel)
                self.set_digital_out("in_channel_select", chan_sel)
                time.sleep(VALVE_TIME)

                # SET LED INDICATORS
                if chan_sel == 0:
                    self.set_digital_out("ch1_led", True)
                else:
                    self.set_digital_out("ch2_led", True)

            self.blow_on=True
            self.display_text("SOFFIAGGIO IN CORSO...")
            self.set_digital_out("blow_led",True)
            self.set_digital_out("blow_on",True)
            self.set_digital_out("flush_on", True)
            blow_time=int(self.step.spec.get('ext_blow_time',3))
            self.set_digital_out("blow_led", True)
            time.sleep(blow_time)
            self.set_digital_out("blow_led", False)
            self.set_digital_out("blow_on", False)
            self.set_digital_out("flush_on", False)


        self.blow_on = False
        if not self.simulate:
            self.components[self.tester_component].start_test()


    def start(self, recipe=None, step=None, pieces=None):
        # TESTING
        if "--test-leak" in sys.argv:
            self.simulate = True
        else:
            self.simulate = False
        # /TESTING
        show = super().start(recipe=recipe, step=step, pieces=pieces)
        if show is False:
            return show

        if "leak_2" in [s.type for s in self.parent.cycle_steps]:
            if self.step.type=="leak_1":
                self.test_num_l.setText("1/2")
            else:
                self.test_num_l.setText("2/2")
        else:
            self.test_num_l.setText("1/1")

        self.recipe_pressure_l.setText(f"{self.step.spec['test_pressure']}")
        self.leak_min_l.setText(f"{self.step.spec['test_pressure_qneg']}")
        self.leak_max_l.setText(f"{self.step.spec['test_pressure_qpos']}")
        self.fill_time_l.setText(f"{self.step.spec['filling_time']}")
        self.settle_time_l.setText(f"{self.step.spec['settling_time']}")
        self.meas_time_l.setText(f"{self.step.spec['test_time']}")

        # SETUP TEST LOOP
        if self.step.spec.get("autotest", False):  # IF AUTOTESTING UPLOAD RECIPE EVERY TIME
            self.recipe_written = False

        if self.parent.config["hardware_config"].get("second_leak_test", "absent") == "present":  # IF SECOND LEAK TEST ENABLED UPLOAD RECIPE EVERY TIME
            self.recipe_written = False

        if not self.recipe_written:
            self.components[self.tester_component].write_recipe(self.recipe, self.step)
            self.recipe_written=True

        self.get_connection = self.components[self.tester_component].out.connect(self.get)
        self.components[self.tester_component].resume()
        if self.parent_assembly_widget is not None:
            self.display_text(text="ATTENDERE")
        self.start_b.setEnabled(False)
        self.stop_b.setEnabled(False)

        if self.step.spec.get("autotest", False) == "ok_check":
            self.display_text(text="AUTOTEST: RIMUOVERE FUGA CALIBRATA E PREMERE START PER INIZIARE LA PROVA TENUTA",
                              bg_color="blue", text_color="white")
            super().visualize(None, img=self.status_imgs_full["calibrated-leak-remove"])
        elif self.step.spec.get("autotest", False) == "ko_check":
            self.display_text(
                text="AUTOTEST: COLLEGARE TUBO-TUBO + FUGA CALIBRATA E PREMERE START PER INIZIARE LA PROVA TENUTA DI PROVA",
                bg_color="blue", text_color="white")
            super().visualize(None, img=self.status_imgs_full["calibrated-leak"])
        else:
            self.display_text(text="COLLEGARE GLI ATTACCHI PNEUMATICI E PREMERE START PER INIZIARE LA PROVA TENUTA")

        if self.simulate:
            QApplication.processEvents()
            time.sleep(2)

        # AUTO START SECOND TEST
        if step.type == "leak_2":
            self.recipe_written = False
            time.sleep(1)
            self.start_b.setEnabled(True)
            self.start_b.click()

        return show

    def stop(self):
        # disable test loop
        self.components[self.tester_component].stop_test()
        self.components[self.tester_component].pause()
        self.disconnect(self.get_connection)
        super().stop()
        self.start_b.setEnabled(False)
        self.stop_b.setEnabled(False)

    def get(self, data=None, override=False):
        if self.done:  # avoid processing if completed
            return
        if data is None or data[-1] is None:
            super().get(None, override=override)
            return
        data = data[-1]

        # TESTING
        if self.simulate:
            if "--fail-leak" in sys.argv:
                data[self.tester_component] = {
                    "Running test: active phase": "WAITING START",
                    "Running test: result": "-----TESTING----- fail",
                    "Running test: filling pressure":"5000",
                    "Running test: measured leak":"50",
                    "Running test: pressure at the end of settling":"4999",
                }
            else:
                data[self.tester_component] = {
                    "Running test: active phase": "WAITING START",
                    "Running test: result": "-----TESTING----- passed",
                    "Running test: filling pressure":"5000",
                    "Running test: measured leak":"5",
                    "Running test: pressure at the end of settling":"4999",
                }

        if "Running test: result" in data[self.tester_component]:
            # TEST ENDED
            result = data[self.tester_component]["Running test: result"]
            step=self.step.spec.get("autotest", "")
            if  step == "ok_check":
                ok = type(result) is str and "passed" in result.lower() # AUTOTEST - NO LEAK
                self.recipe_written=False
            elif  step == "ko_check":
                ok = type(result) is str and "passed" in result.lower() # AUTOTEST - LEAK
                self.recipe_written=False
            else:
                ok = type(result) is str and "passed" in result.lower() # NORMAL TEST

            # SET DIGITAL OUTPUTS

            if self.parent.config["hardware_config"].get("external_flush_blow", None) == "present":
                self.blow_on = True
                self.display_text("SCARICO ESTERNO IN CORSO...")
                self.set_digital_out("flush_led", True)
                self.set_digital_out("flush_on", True)
                time.sleep(VALVE_TIME)
                flush_time = int(self.step.spec.get('ext_flush_time',3))
                time.sleep(flush_time)
                self.set_digital_out("flush_led", False)
                #self.set_digital_out("flush_on", False)

            if self.parent.config["hardware_config"].get("dual_channel", None) == "present":
                self.set_digital_out("out_channel_select", False)
                self.set_digital_out("in_channel_select", False)
                self.set_digital_out("ch1_led", False)
                self.set_digital_out("ch2_led", False)
        else:
            #result = None
            ok = None

        results={"ok":ok}
        results.update(data)
        super().get([{
            "time": data.get("time", None),
            "results": results
            #"results": {
                #"ok": ok,
                #"result": result,
                #"data": data[self.tester_component],
            #},
        }], override=override, fail=ok is False)

    def visualize(self, data=None):
        if data is None:
            data = {}
        d = data.get("results", {}).get(self.tester_component, {})
        for k, l in {
            "Running test: active phase": self.test_phase_l,
            "Real time test pressure output": self.circuit_pressure_l,
            #"Real time differential pressure output": self.leak_l,
            "Running test: measured leak": self.leak_l,
            "Real time pressure line regulator": self.regulated_pressure_l,
            # "Active alarm flags": self._l,
            "Running test: test type": self.test_type_l,
            "Running test: sequence index": self.sequence_index_l,
        }.items():
            v = d.get(k, "-")
            if type(v) is float:
                v = round(v, 2)
            l.setText(str(v))
        if d.get("Running test: active phase", None) in {
            "WAITING START",
            "ATTESA START",
            "END TEST, WAITING THE START OF A NEW TEST"
            "FINE TEST",
            "STANDBY",
            "PRESSIONE BASSA",
            "PRESSIONE ALTA",
            "ERRORE"
        }:

            self.start_b.setEnabled(True)
            self.start_b.setDefault(True)
            self.start_b.setFocus()

            self.stop_b.setEnabled(False)
        else:
            if self.step is not None and not self.blow_on:
                if self.step.spec.get("autotest", False) is not True:
                    self.display_text(text="PROVA TENUTA IN CORSO")
                else:
                    self.display_text(text="AUTOTEST: PROVA TENUTA IN CORSO")
            self.start_b.setEnabled(False)
            self.stop_b.setEnabled(True)
            ok = data.get("results", {}).get("ok", None)

            super().visualize(data, img=self.status_imgs_full[ok])

    def display_text(self,text="", bg_color=None,text_color=None):
        if self.parent_assembly_widget is not None:
            self.parent_assembly_widget().set_text(text=text, bg_color=bg_color,text_color=text_color)
            QApplication.processEvents()
            time.sleep(0.3)
            QApplication.processEvents()

    def set_digital_out(self,out_name=None,state=1,component_name="digital_io_flush_blow"):
        if self.io_ok:
            bit = int(self.parent.config[component_name][out_name])
            ret = self.components[component_name].set_bit_verify(0,bit,state)
            if not ret:
                QMessageBox.critical(None, "ERRORE", f"ERRORE I/O DIGITALE - VERIFICARE CONNESSIONE USB")
                self.io_ok = False

    def save_last(self):
        if self.last is None:
            return