Tecna recipe reader tool WIP

src/components/tecna_marposs_provaset_t3.py

@@ -360,3 +360,38 @@ class TecnaMarpossProvasetT3(ModbusComponent):
             self.write_recipe(recipe, step, table=i)
         self.log.info(f"saved {min(len(recipes), max(self.max_program_number - 1, 0))} recipes")
         self._store_recipes_lock.release(1)
+
+    def read_recipe(self, recipe_number, step):
+        spec = {
+            "T0 - Pre-filling time": step.spec["pre_filling_time"],
+            "P0 - Pre-filling pressure": step.spec["pre_filling_pressure"],
+            "T1 - Filling time": step.spec["filling_time"],
+            "T2 - Settling time": step.spec["settling_time"],
+            "PR- - Min pressure tolerance %": step.spec["settling_pressure_min_percent"],
+            "PR+ - Max pressure tolerance % (P+)": step.spec["settling_pressure_max_percent"],
+            "T3 - Measure time": step.spec["test_time"],
+            "Q- Lower test leak limit": step.spec["test_pressure_qneg"],
+            "Q+ Upper test leak limit": step.spec["test_pressure_qpos"],
+            "FST - Discharge time": step.spec["flush_time"],
+            "FSL - Discharge limit": step.spec["flush_pressure"],
+        }
+        if self.model == "t3p":
+            spec.update({
+                "PREL - Nominal test pressure": step.spec["test_pressure"],
+            })
+        elif self.model == "t3l":
+            spec.update({
+                "Nominal peak pressure": step.spec["test_pressure"],
+            })
+        else:
+            raise NotImplementedError(f"tecna t3 model {self.model!r} not implemented.")
+
+        # SET RECIPE NUMBER
+        self.write("Active test program number", recipe_number)
+
+        read_recipe={}
+        # READ ALL PARAMETERS
+        for register_name, field_name in spec.items():
+            read_recipe[field_name] = self.read(register_name)
+
+        return read_recipe

src/scripts/save_tecna_recipes_to_csv.py

@@ -0,0 +1,137 @@
+#!/usr/bin/env python3
+import argparse
+import faulthandler
+import logging
+import os
+import platform
+import signal
+import sys
+import traceback
+import weakref
+from datetime import datetime
+from pathlib import Path
+from ui.diagnostics import Diagnostics
+from lib.helpers.single_process import SingleProcess
+
+if platform.system().lower() == "windows":
+    sys.path.append(f"{os.getcwd()}\src\components")
+else:
+    sys.path.append(f"{os.getcwd()}/src/components")
+
+app = None
+
+parser = argparse.ArgumentParser(prog='ST-TEN', description='Leak test system')
+parser.add_argument('-s', '--system-id')
+args, unspec = parser.parse_known_args()
+
+
+def quit_app(signalnum=None, handler=None):
+    logging.info(f"quitting app. signal: {signalnum!r}, handler: {handler!r}")
+    global app
+    if app is not None:
+        app.quit()
+    quit()
+
+
+# SETUP QUITTING ON CTRL+C
+signal.signal(signal.SIGINT, quit_app)
+
+# SETUP FAULTHANDLER
+faulthandler.enable(file=sys.stderr, all_threads=True)
+
+# SETUP LOGS
+logs_dir = Path(".") / "data" / "logs"
+os.makedirs(logs_dir, exist_ok=True)
+logging.basicConfig(
+    format="{asctime}:{name}:{levelname}:{message}",
+    datefmt="%Y-%m-%d_%H-%M-%S",
+    style="{",
+    level="DEBUG" if "--debug" in sys.argv else "INFO",
+    handlers=[
+        logging.StreamHandler(stream=sys.stderr),
+        logging.FileHandler(
+            logs_dir / f"{datetime.now().strftime('%Y-%m-%d_%H-%M-%S')}.log",
+            mode="a",
+            encoding="utf-8",
+            delay=False,
+            **({"errors": "surrogateescape"} if sys.version_info.major >= 3 and sys.version_info.minor >= 10 else {}),
+        ),
+    ],
+    force=True,
+    **({"encoding": "utf-8"} if sys.version_info.major >= 3 and sys.version_info.minor >= 10 else {}),
+    **({"errors": "surrogateescape"} if sys.version_info.major >= 3 and sys.version_info.minor >= 10 else {}),
+)
+
+try:
+    # IMPORT PROJECT ONLY AFTER SETTING UP SIGNAL, FAULTHANDLER AND LOGGING
+    from components import TecnaMarpossProvasetT3
+    from lib.db import Users
+    from lib.helpers import ConfigReader
+    from PyQt5.QtCore import QObject, QThread, pyqtSignal, pyqtSlot
+    from PyQt5.QtWidgets import QApplication, QMessageBox
+    from ui import About, Archive, Login, Main_Window, Test, Users_Management, Recipe_Selection, \
+        Barcode_Recipe_Selection
+
+
+
+    class Main(QObject):
+
+        def __init__(self, parent=None):
+            # print(f"MAIN   {int(QThread.currentThreadId())}", flush=True)
+            super().__init__()
+            try:
+                # READ CONFIG
+
+                system_id = args.system_id if "system_id" in args else None
+
+                self.config = ConfigReader(system_id=system_id)
+                logging.info(f"STARTING SESSION ON MACHINE {self.config['machine']['description']}")
+                # INIT COMPONENT
+                self.components_specs = {
+                    "tecna_t3": {"c": TecnaMarpossProvasetT3, "k": {"paused": True}},
+                }
+
+                self.components = {}
+                self.threads = {}
+                for component_name, spec in self.components_specs.items():
+                    self.components[component_name] = spec["c"](*spec.get("a", []), config=self.config,
+                                                                name=component_name, **spec.get("k", {}))
+                    if spec.get("t", True):
+                        self.threads[component_name] = QThread()
+                        self.threads[component_name].setTerminationEnabled(True)
+                        self.components[component_name].moveToThread(self.threads[component_name])
+
+                for component_name, thread in self.threads.items():
+                    component = self.components[component_name]
+                    thread.started.connect(component.start)
+                    thread.start()
+                    # DEBUGGER WORKAROUND
+                    QApplication.processEvents()
+                    QThread.msleep(1000)
+                    QApplication.processEvents()
+                    component.wait_completion()
+
+            except Exception as e:
+                logging.exception(traceback.format_exc())
+                QMessageBox.critical(None, "Errore", f"Errore di avvio del programma di collaudo:\n\n{e}")
+                quit()
+
+        def save_recipes(self):
+            self.components["tecna_t3"].read_recipes
+
+
+    if __name__ == "__main__":
+        app = QApplication(sys.argv)
+
+        with SingleProcess() as single_process_lock:
+            if not single_process_lock and "--no-lock" not in sys.argv:
+                logging.error(f"Program already opened, exiting...")
+                QMessageBox.critical(None, "ERRORE", "IL PROGRAMMA E' GIA' IN ESECUZIONE")
+                exit(0)
+
+            main = Main()
+            if "--no-gui" not in sys.argv:
+                app.exec()
+
+except Exception:
+    logging.exception(traceback.format_exc())