st-ten-1/src/main.py

#!/usr/bin/env python3
import faulthandler
import logging
import os
# import pdb
import signal
import sys
import traceback
import weakref
from datetime import datetime
from pathlib import Path

app = None


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-%dT%H-%M-%S%z",
    style="{",
    level="INFO",
    handlers=[
        logging.StreamHandler(stream=sys.stderr),
        logging.FileHandler(
            logs_dir / f"{datetime.now().isoformat().replace(':', ';')}.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 LOGGHING
    from components import (ArchiveSynchronizer, GalaxyCamera, NeoPixels,
                            Os_Label_Printer, RemoteAPI,
                            TecnaMarpossProvasetT3P, VisionSaver)
    from lib.db import Users
    from lib.helpers import ConfigReader
    from PyQt5.QtCore import QObject, QThread, pyqtSignal
    from PyQt5.QtWidgets import QApplication, QMessageBox
    from ui import About, Archive, Login, Main_Window, Test, Users_Management

    class Main(QObject):
        do = pyqtSignal(dict)

        @staticmethod
        def _do(config):
            return config["f"](*config.get("a", []), **config.get("k", {}))

        def __init__(self, parent=None):
            # print(f"MAIN   {int(QThread.currentThreadId())}", flush=True)
            super().__init__()
            self.do.connect(self._do)
            try:
                # READ CONFIG
                self.config = ConfigReader()
                # INIT COMPONENT
                self.components_specs = {
                    # "archive_synchronizer": {"c": ArchiveSynchronizer},
                    "galaxy_camera": {"c": GalaxyCamera, "k": {"paused": True}},
                    "label_printer": {"c": Os_Label_Printer, "t": False},
                    "neo_pixels": {"c": NeoPixels, "t": False},
                    "remote_api": {"c": RemoteAPI, "k": {"main": self}},
                    "tecna_t3": {"c": TecnaMarpossProvasetT3P, "k": {"paused": True}},
                    "vision_saver": {"c": VisionSaver, "t": False},

                }

                for key in self.components_specs.copy().keys():
                    if key in self.config["hardware_config"].keys() and self.config["hardware_config"][key] != "present":
                        self.components_specs.pop(key)
                if self.config["hardware_config"]["vision"] is "present":
                    from src.components.vision import Vision
                    self.components_specs["vision"]= {"c": Vision, "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()
                    if "--debugger-workaround" in sys.argv:
                        QApplication.processEvents()
                        QThread.msleep(1000)
                        QApplication.processEvents()
                    if component_name == "vision":
                        component.wait_ready(timeout=60)
                    else:
                        component.wait_ready()
            except Exception as e:
                logging.exception(traceback.format_exc())
                QMessageBox.critical(None, "Errore Banco", f"Non e stato possibile connettersi al banco:\n\n{e}")
                quit()
            # connect camera frames to vision
            if "vision" in self.components and "galaxy_camera" in self.components:
                self.components["vision"].set_sources({"galaxy_camera": self.components["galaxy_camera"].out})
            # GUI INIT
            if "--no-gui" not in sys.argv:
                # self.main_window = Main_Window(self.bench)
                self.main_window = Main_Window()
                # CONNECT MAIN WINDOW ACTIONS
                self.main_window.archive_a.triggered.connect(lambda checked, self=weakref.ref(self): self().main_window.open_dialog(Archive()))
                if "--archive" in sys.argv:
                    self.main_window.archive_a.trigger()
                # self.main_window.autotests_archive_a.triggered.connect(lambda checked, self=weakref.ref(self): self().main_window.open_dialog(Autotests_Archive()))
                # if "--autotests-archive" in sys.argv:
                #     self.main_window.autotests_archive_a.trigger()
                self.main_window.about_a.triggered.connect(lambda checked, self=weakref.ref(self): self().main_window.open_dialog(About()))
                if "--about" in sys.argv:
                    self.main_window.about_a.trigger()
                self.main_window.admin_m.menuAction().setVisible(False)  # admin menu should not be visible before an admin logs in
                self.main_window.quit_a.triggered.connect(quit_app)
                self.main_window.users_management_a.triggered.connect(lambda checked, self=weakref.ref(self): self().main_window.open_dialog(Users_Management()))
                if "--users-management" in sys.argv:
                    self.main_window.users_management_a.trigger()
                # self.main_window.recipes_management_a.triggered.connect(lambda checked, self=weakref.ref(self): self().main_window.open_dialog(Recipes_Management()))
                # if "--recipes-management" in sys.argv:
                #     self.main_window.recipes_management_a.trigger()
                # self.main_window.steps_management_a.triggered.connect(lambda checked, self=weakref.ref(self): self().main_window.open_dialog(Steps_Management()))
                # if "--steps-management" in sys.argv:
                #     self.main_window.steps_management_a.trigger()
                # OPEN LOGIN TAB
                self.open_login()
                # SHOW MAIN WINDOW
                if "--panel" in sys.argv:
                    self.main_window.show()
                elif "--maximized" in sys.argv:
                    self.main_window.showMaximized()
                elif "--full-screen" in sys.argv:
                    self.main_window.showFullScreen()
                else:
                    self.main_window.showFullScreen()

        def open_login(self):
            tab = Login()
            tab.successful_login.connect(self.logged_in)
            self.main_window.open_tab(tab)

        def logged_in(self):
            session = Users.get_session()
            if session is not None:
                if session.is_admin:
                    self.main_window.admin_m.menuAction().setVisible(True)
                else:
                    self.main_window.admin_m.menuAction().setVisible(False)
                # open test
                self.main_window.open_tab(Test(self.config.machine_id, self.components))

    if __name__ == "__main__":
        app = QApplication(sys.argv)
        main = Main()
        if "--no-gui" not in sys.argv:
            app.exec()
        if "--interact" in sys.argv:
            import code
            import readline
            variables = globals().copy()
            variables.update(locals())
            shell = code.InteractiveConsole(variables)
            shell.interact()
except Exception:
    logging.exception(traceback.format_exc())
    # extype, value, tb = sys.exc_info()
    # pdb.post_mortem(tb)