st-ten-1/src/test/uvc_camera/test_uvc_camera.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
from PyQt5.QtCore import Qt, pyqtSignal
from PyQt5 import uic
from PyQt5.QtGui import QPixmap
from PyQt5.QtWidgets import QMainWindow
from src.components import Vision

# 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="DEBUG",
    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 {}),
)

class Main_Window(QMainWindow):

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.ui = uic.loadUi("src/test/uvc_camera/test_uvc_camera.ui", self)

    def open_window(self, show=True, maximized=False):
        w = QMainWindow()
        w.setAttribute(Qt.WA_DeleteOnClose)
        w.setWindowFlags(Qt.Window)
        if show:
            if maximized:
                w.showMaximized()
            else:
                w.show()
        return w

try:
    # IMPORT PROJECT ONLY AFTER SETTING UP SIGNAL, FAULTHANDLER AND LOGGHING
    from components import (ArchiveSynchronizer, GalaxyCamera, Multicomp730424,
                            NeoPixels, Os_Label_Printer, RemoteAPI,
                            TecnaMarpossProvasetT3, TecnaScrewdriver,
                            UVCCamera, VisionSaver)
    from lib.db import Users
    from lib.helpers import ConfigReader
    from PyQt5.QtCore import QObject, QThread, pyqtSignal
    from PyQt5.QtWidgets import QApplication, QMessageBox, QMainWindow

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


        def __init__(self, parent=None):
            super().__init__()
            try:
                # READ CONFIG
                self.config = ConfigReader()
                self.config["autotest_done"]=False
                # INIT COMPONENT
                self.components_specs = {
                    "uvc_camera": {"c": UVCCamera, "k": {"paused": False}},
                    "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()

                    component.wait_completion(timeout=60)

            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})
            self.components["uvc_camera"].out.connect(self.update_gui)
            self.components["uvc_camera"].resume()
            self.main_window = Main_Window()
            self.main_window.btn_save_image.clicked.connect(self.save_frame)

            # SHOW MAIN WINDOW
            self.main_window.show()

        def update_gui(self,frame):
            if frame[0]["uvc_camera"] is not None:
                self.frame = frame[0]["uvc_camera"]
                render = QPixmap.fromImage(self.components["vision"].render(
                    frame[0]["uvc_camera"],
                    detections=None,
                    vision_results=None,
                    mask=False,
                    offset=None,
                ))
                self.main_window.image_l.setPixmap(render.scaled(self.main_window.image_l.width(), self.main_window.image_l.height(), Qt.KeepAspectRatio, Qt.SmoothTransformation))

        def save_frame(self):
            self.components["vision_saver"].save(frame=self.frame, location="tmp")


    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)