st-ten-1/src/main.py

#!/usr/bin/env python3
import argparse
import faulthandler
import logging
import os
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 ,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": BrotherLabelPrinter, "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)
                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

    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)