184 lines
8.2 KiB
Python
184 lines
8.2 KiB
Python
#!/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, Vision, 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},
|
|
"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)
|