st-ten-1/src/main.py
sttentest 0735415dca dev
2025-09-17 14:40:04 +02:00

547 lines
27 KiB
Python

#!/usr/bin/env python3
import argparse
import faulthandler
import logging
import os
from lib.helpers.recipe_manager import backup_current_recipes
os.environ['PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION']="python"
import platform
import signal
import sys
import traceback
import weakref
from datetime import datetime
from pathlib import Path
if platform.system().lower() == "windows":
sys.path.append(f"{os.getcwd()}\src\components")
else:
sys.path.append(f"{os.getcwd()}/src/components")
sys.path.append(f"{os.getcwd()}")
from ui.diagnostics import Diagnostics
from lib.helpers.single_process import SingleProcess
from ui.logs_management.info import Logs_Management
app = None
parser = argparse.ArgumentParser(prog='ST-TEN', description='Leak test system')
parser.add_argument('-s', '--system-id')
parser.add_argument('-p', '--auto-select')
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 (ArchiveSynchronizer, Multicomp730424,
Os_Label_Printer, RemoteAPI,
TecnaMarpossProvasetT3, FurnessControlsLeakTester, TecnaScrewdriver, USB_586x, RFID_PN532,BrotherLabelPrinter,PipeCutterComponent)
from lib.db import Users
from lib.helpers import ConfigReader
from PyQt5.QtCore import QObject, QThread, pyqtSignal, pyqtSlot
from PyQt5.QtWidgets import QApplication, QMessageBox, QInputDialog, QLineEdit
import sip
from ui import About, Archive, Login, Main_Window, Test, Users_Management, Logs_Management, Recipes_Management , Recipe_Selection, \
Barcode_Recipe_Selection, LastCommit
if "--vision" in sys.argv:
from components import GalaxyCamera, NeoPixels, UVCCamera, HikrobotSmartCamera, Vision, VisionSaver
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.tag_loaded_recipe = None
self.do.connect(self._do)
try:
# READ CONFIG
system_id = args.system_id if "system_id" in args else None
auto_select = args.auto_select if "auto_select" in args else None
self.config = ConfigReader(system_id=system_id,auto_select=auto_select)
logging.info(f"STARTING SESSION ON MACHINE {self.config['machine']['description']}")
self.config["autotest_done"] = False
# INIT COMPONENT
self.components_specs = {
"archive_synchronizer": {"c": ArchiveSynchronizer},
"archive_synchronizer_extra": {"c": ArchiveSynchronizer},
"pipe_cutter": {"c": PipeCutterComponent, "k": {"paused": False}},
"label_printer": {"c": Os_Label_Printer, "t": False},
"extra_label_printer": {"c": Os_Label_Printer, "t": False},
"label_printer_2": {"c": Os_Label_Printer, "t": False},
"multicomp": {"c": Multicomp730424, "k": {"paused": True}},
"remote_api": {"c": RemoteAPI, "k": {"main": self}},
"screwdriver": {"c": TecnaScrewdriver, "k": {"paused": True}},
"tecna_t3": {"c": TecnaMarpossProvasetT3, "k": {"paused": True}},
"furness_controls": {"c": FurnessControlsLeakTester, "k": {"paused": True}},
"digital_io": {"c": USB_586x, "k": {"paused": True}},
"digital_io_flush_blow": {"c": USB_586x, "k": {"paused": True}},
"fixture_id": {"c": RFID_PN532, "k": {"paused": False, "lazy": False}},
}
# VISION COMPONENT IS OPTIONAL AND DISABLED BY DEFAULT
if "--vision" in sys.argv:
# merge dicts
self.components_specs = {**self.components_specs,
**{
"galaxy_camera": {"c": GalaxyCamera, "k": {"paused": True}},
"neo_pixels": {"c": NeoPixels, "t": False},
"uvc_camera": {"c": UVCCamera, "k": {"paused": True}},
"hikrobot_sc": {"c": HikrobotSmartCamera, "k": {"paused": True}},
"vision_saver": {"c": VisionSaver, "t": False},
"vision": {"c": Vision, "k": {"paused": True}}
}
}
for component_name in list(self.components_specs):
if self.config.get("hardware_config", {}).get(component_name, None) != "present":
self.components_specs.pop(component_name, None)
elif component_name not in self.components_specs:
raise AssertionError(f"{component_name!r} is not a valid component name")
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])
if "fixture_id" in self.components.keys():
self.components["fixture_id"].new_id_signal.connect(self.load_recipe_from_rfid)
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()
if component_name == "vision":
component.wait_completion(timeout=60)
else:
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()
# connect camera frames to vision
if "vision" in self.components and "uvc_camera" in self.components:
self.components["vision"].set_sources({"uvc_camera": self.components["uvc_camera"].out})
elif "vision" in self.components and "galaxy_camera" in self.components:
self.components["vision"].set_sources({"galaxy_camera": self.components["galaxy_camera"].out})
if "vision" in self.components and "hikrobot_sc" in self.components:
self.components["vision"].set_sources({"hikrobot_sc": self.components["hikrobot_sc"].out})
# connect tecna to screwdriver
if "screwdriver" in self.components and "tecna_t3" in self.components:
self.components["tecna_t3"].set_requestors({"screwdriver": self.components["screwdriver"].request})
self.components["screwdriver"].set_sources({"tecna_t3": self.components["tecna_t3"].out})
# GUI INIT
if "--no-gui" not in sys.argv:
self.main_window = Main_Window()
# CONNECT MAIN WINDOW ACTIONS
self.main_window.logout_a.triggered.connect(lambda checked, selfie=weakref.ref(self): selfie().logout())
self.main_window.archive_a.triggered.connect(
lambda checked, selfie=weakref.ref(self): selfie().main_window.open_dialog(
Archive(hide_cloud_image="vision_saver" not in selfie().components)))
if "--archive" in sys.argv:
self.main_window.archive_a.trigger()
if "--about" in sys.argv:
self.main_window.about_a.trigger()
# Keep the admin menu always visible
self.main_window.admin_m.menuAction().setVisible(True)
self.main_window.about_a.triggered.connect(
lambda checked, selfie=weakref.ref(self): selfie().main_window.open_dialog(About()))
self.main_window.last_commit_a.triggered.connect(
lambda checked, selfie=weakref.ref(self): selfie().main_window.open_dialog(LastCommit()))
self.main_window.download_a.triggered.connect(
lambda checked, selfie=weakref.ref(self): selfie().main_window.open_dialog(Logs_Management()))
self.main_window.quit_a.triggered.connect(quit_app)
self.main_window.users_management_a.triggered.connect(
lambda checked, selfie=weakref.ref(self): selfie().main_window.open_dialog(Users_Management()))
self.main_window.table_selection_a.triggered.connect(self.set_recipe_mode_table)
self.main_window.barcode_selection_a.triggered.connect(self.set_recipe_mode_barcode)
self.main_window.ristampa_etichetta_a.triggered.connect(self.reprint_label)
self.main_window.tag_a.triggered.connect(self.tag_write)
self.main_window.recipes_export_a.triggered.connect(self.trigger_recipe_backup)
if "pipe_cutter" in self.components.keys():
self.main_window.cut_a.setVisible(True)
self.main_window.cut_a.triggered.connect(self.cut_tube)
else:
self.main_window.cut_a.setVisible(False)
self.main_window.diagnostics_a.triggered.connect(
lambda checked, selfie=weakref.ref(self): selfie().main_window.open_dialog(Diagnostics(selfie())))
self.main_window.admin_enable_a.triggered.connect(
lambda checked, selfie=weakref.ref(self): selfie().enable_admin_privileges())
if "--users-management" in sys.argv:
self.main_window.users_management_a.trigger()
# CONFIG-SPECIFIC MENU ENTRY ACTIVATION
if "tecna_t3" in self.components and (
"--enable-saving-tecna-recipes" in sys.argv or self.config.get("tecna_t3", {}).get("saver",
None) == "present"):
self.main_window.save_tecna_recipes_a.triggered.connect(self.components["tecna_t3"].store_recipes)
self.main_window.save_tecna_recipes_a.setVisible(True)
if "--save-tecna-recipes" in sys.argv:
self.main_window.save_tecna_recipes_a.trigger()
else:
self.main_window.save_tecna_recipes_a.setVisible(False)
self.main_window.barcode_selection_a.setVisible(
self.config["hardware_config"]["barcode_recipe_selection"] == "present")
# 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()
# Always make the admin menu visible
self.main_window.admin_m.menuAction().setVisible(True)
if session is not None:
# Check if user has admin privileges (either permanent or temporary)
# Use session.is_admin instead of checking roles directly to be consistent with user.py
if session.is_admin:
self.main_window.tag_a.setVisible(True)
self.main_window.admin_enable_a.setVisible(False) # Hide admin enable action for admins
# Show admin features for users with admin privileges (permanent or temporary)
self.main_window.users_management_a.setVisible(True)
self.main_window.save_tecna_recipes_a.setVisible(True)
self.main_window.diagnostics_a.setVisible(True)
else:
# For non-admin users, only show the admin enable button
self.main_window.admin_enable_a.setVisible(True) # Show admin enable action for non-admins
self.main_window.users_management_a.setVisible(False) # Hide user management for non-admins
self.main_window.save_tecna_recipes_a.setVisible(False) # Hide tecna recipes for non-admins
self.main_window.diagnostics_a.setVisible(False) # Hide diagnostics for non-admins
self.main_window.tag_a.setVisible(False)
# open test
# Update background color based on admin privileges
self.update_window_backgrounds()
self.main_window.open_tab(Test(self.config, self.components, self))
self.main_window.centralWidget().request_autotest("login")
def logout(self):
# Users.logout()
# Keep the admin menu visible
self.main_window.admin_m.menuAction().setVisible(True)
# Get the current session (if any)
session = Users.get_session()
# Note: We no longer reset temp_admin here to preserve admin status across login/logout
# Reset background color to default
self.main_window.setStyleSheet("")
for window_name, window in self.main_window.windows.items():
if window is not None and not sip.isdeleted(window):
window.setStyleSheet("")
if type(self.main_window.centralWidget().centralWidget.widget) in (
Recipe_Selection, Barcode_Recipe_Selection):
# LOGOUT IMMEDIATELY IF NOT TESTING
Users.logout()
ArchiveSynchronizer.machine_status = "logged-out"
self.open_login()
else:
# ALWAYS REQUEST AUTOTEST BEFORE LOGOUT
self.main_window.centralWidget().request_autotest("logout")
def set_recipe_mode_table(self):
self.main_window.centralWidget().set_recipe_mode_table()
def set_recipe_mode_barcode(self):
self.main_window.centralWidget().set_recipe_mode_barcode()
def reprint_label(self):
self.main_window.centralWidget().reprint_label()
def tag_write(self):
if isinstance(self.main_window.centralWidget().centralWidget.widget, Barcode_Recipe_Selection):
barcode_data = self.main_window.centralWidget().centralWidget.widget.barcode_input_l.toPlainText().strip()
self.main_window.centralWidget().centralWidget.widget.tag_write(barcode_data)
def cut_tube(self):
self.main_window.centralWidget().cut_tube()
def enable_admin_privileges(self):
session = Users.get_session()
if session is None:
QMessageBox.warning(self.main_window, "Errore", "Nessun utente loggato")
return
# If user already has temporary admin privileges, toggle them off
if session.temp_admin:
# Use the Users class method to disable temp admin
Users.disable_temp_admin()
# Keep the admin menu visible
self.main_window.admin_m.menuAction().setVisible(True)
self.main_window.users_management_a.setVisible(False)
self.main_window.save_tecna_recipes_a.setVisible(False)
self.main_window.diagnostics_a.setVisible(False)
self.main_window.tag_a.setVisible(False)
self.main_window.admin_enable_a.setVisible(True) # Show admin enable action after removing temp admin privileges
# Call disable_temp_admin on the Test widget if it exists
current_widget = self.main_window.centralWidget()
if current_widget is not None and hasattr(current_widget, 'disable_temp_admin'):
current_widget.disable_temp_admin()
QMessageBox.information(
self.main_window,
"Successo",
"Privilegi di amministratore disabilitati"
)
# Update background color for all windows
self.update_window_backgrounds()
# Refresh the current UI component to reflect removed admin privileges
current_widget = self.main_window.centralWidget()
if current_widget is not None:
# If the current widget has a refresh method, call it
if hasattr(current_widget, 'refresh'):
current_widget.refresh()
# If the current widget has a crud attribute, refresh it
if hasattr(current_widget, 'crud'):
if callable(current_widget.crud):
crud = current_widget.crud()
else:
crud = current_widget.crud
if hasattr(crud, 'refresh'):
crud.refresh()
# Refresh all other open windows to reflect removed admin privileges
for window_name, window in self.main_window.windows.items():
if window is not None and not sip.isdeleted(window):
central_widget = window.centralWidget()
if central_widget is not None:
# If the central widget has a refresh method, call it
if hasattr(central_widget, 'refresh'):
central_widget.refresh()
# If the central widget has a crud attribute, refresh it
if hasattr(central_widget, 'crud'):
if callable(central_widget.crud):
crud = central_widget.crud()
else:
crud = central_widget.crud
if hasattr(crud, 'refresh'):
crud.refresh()
return
# If user is a permanent admin (not temporary), show a message
if "admin" in Users.parse_roles(session.user.roles):
QMessageBox.information(self.main_window, "Informazione", "Sei già un amministratore permanente")
return
password, ok = QInputDialog.getText(
self.main_window,
"Admin abilitazione",
"Inserisci la password di amministratore:",
QLineEdit.Password
)
if not ok or not password:
return
# Find an admin user to verify the password against
admin_users = [user for user in Users.get_users() if "admin" in Users.parse_roles(user.roles)]
if not admin_users:
QMessageBox.warning(self.main_window, "Errore", "Nessun utente amministratore trovato nel sistema")
return
# Use the Users class method to enable temp admin
if Users.enable_temp_admin(password):
self.main_window.admin_m.menuAction().setVisible(True)
self.main_window.tag_a.setVisible(True)
# Show admin features for users with temporary admin privileges
self.main_window.users_management_a.setVisible(True)
self.main_window.save_tecna_recipes_a.setVisible(True)
self.main_window.diagnostics_a.setVisible(True)
# Call enable_temp_admin on the Test widget if it exists
current_widget = self.main_window.centralWidget()
if current_widget is not None and hasattr(current_widget, 'enable_temp_admin'):
current_widget.enable_temp_admin()
QMessageBox.information(
self.main_window,
"Successo",
"Privilegi di amministratore abilitati temporaneamente"
)
# Update background color for all windows
self.update_window_backgrounds()
# Refresh the current UI component to reflect new admin privileges
current_widget = self.main_window.centralWidget()
if current_widget is not None:
# If the current widget has a refresh method, call it
if hasattr(current_widget, 'refresh'):
current_widget.refresh()
# If the current widget has a crud attribute, refresh it
if hasattr(current_widget, 'crud'):
if callable(current_widget.crud):
crud = current_widget.crud()
else:
crud = current_widget.crud
if hasattr(crud, 'refresh'):
crud.refresh()
# Refresh all other open windows to reflect new admin privileges
for window_name, window in self.main_window.windows.items():
if window is not None and not sip.isdeleted(window):
central_widget = window.centralWidget()
if central_widget is not None:
# If the central widget has a refresh method, call it
if hasattr(central_widget, 'refresh'):
central_widget.refresh()
# If the central widget has a crud attribute, refresh it
if hasattr(central_widget, 'crud'):
if callable(central_widget.crud):
crud = central_widget.crud()
else:
crud = central_widget.crud
if hasattr(crud, 'refresh'):
crud.refresh()
return
QMessageBox.warning(self.main_window, "Errore", "Password non valida")
def update_window_backgrounds(self):
"""Update the background color of all windows based on admin privileges"""
session = Users.get_session()
if session is None:
return
# Check if user has admin privileges (permanent or temporary)
# Use session.is_admin instead of checking roles directly to be consistent with user.py
has_admin = session.is_admin
# Set background color for main window
if has_admin:
self.main_window.setStyleSheet("background-color: #ffcccc;") # Light red background
else:
self.main_window.setStyleSheet("") # Reset to default
# Set background color for all other windows
for window_name, window in self.main_window.windows.items():
if window is not None and not sip.isdeleted(window):
if has_admin:
window.setStyleSheet("background-color: #ffcccc;") # Light red background
else:
window.setStyleSheet("") # Reset to default
@pyqtSlot(str)
def load_recipe_from_rfid(self, data):
self.tag_loaded_recipe = data
def trigger_recipe_backup(self):
"""
This method acts as a bridge to call the imported backup function.
"""
try:
# Use the imported function and pass the required objects
backup_current_recipes(config=self.config, logger=logging)
QMessageBox.information(
self.main_window,
"Esportazione Riuscita",
"Backup delle ricette creato con successo."
)
except Exception as e:
logging.exception("Error during recipe backup")
QMessageBox.critical(
self.main_window,
"Errore di Esportazione",
f"Si è verificato un errore durante il salvataggio: {e}"
)
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()
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)