From e3db54583dc3e3ac6f1dc227931fe7986d90ca41 Mon Sep 17 00:00:00 2001 From: neo Date: Fri, 27 Sep 2024 19:11:49 +0200 Subject: [PATCH] Tecna recipe reader tool WIP --- src/components/tecna_marposs_provaset_t3.py | 35 +++++ src/scripts/save_tecna_recipes_to_csv.py | 137 ++++++++++++++++++++ 2 files changed, 172 insertions(+) create mode 100644 src/scripts/save_tecna_recipes_to_csv.py diff --git a/src/components/tecna_marposs_provaset_t3.py b/src/components/tecna_marposs_provaset_t3.py index cfba77b..e138835 100644 --- a/src/components/tecna_marposs_provaset_t3.py +++ b/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 diff --git a/src/scripts/save_tecna_recipes_to_csv.py b/src/scripts/save_tecna_recipes_to_csv.py new file mode 100644 index 0000000..4f46a48 --- /dev/null +++ b/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())