C151/background_service.py at main · neo-empresarial/C151 · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
import threading
import time
import asyncio
from nicegui import ui, app
from src.common.utils import find_free_port
from src.services.services import camera_manager, engine, db_manager
from src.services.access_control import AccessController
from src.services.alert_manager import AlertManager
from src.services.ui_layout import UILayout
from src.common.logger import AppLogger
import logging

access_controller = AccessController()
alert_manager = AlertManager()
layout = UILayout(access_controller, db_manager)

def face_processing_loop():
    AppLogger.log("Loop de processamento iniciado.", "info")
    try:
        engine.start()
        engine.paused = False
        AppLogger.log("Engine iniciada.", "info")
        camera_manager.start()
        if camera_manager.cap is None or not camera_manager.cap.isOpened():
             AppLogger.log("AVISO - Falha ao abrir a câmera no reset inicial.", "warning")
        else:
             AppLogger.log("Câmera iniciada com sucesso.", "info")

    except Exception as e:
        AppLogger.log(f"Erro na inicialização: {e}", "error")
        pass

    first_pass = True
    while True:
        try:
            ret, frame = camera_manager.read()
            if first_pass:
                 if ret:
                      AppLogger.log("PRIMEIRO FRAME LIDO COM SUCESSO!", "info")
                 else:
                      AppLogger.log("FALHA AO LER PRIMEIRO FRAME - Verifique a camera", "error")
                 first_pass = False

            if ret:
                engine.update_frame(frame)
            else:
                pass

            results = engine.get_results()
            access_controller.process_result(results)

        except Exception as e:
            AppLogger.log(f"Erro no loop: {e}", "error")
            pass

        time.sleep(0.5)

@ui.page('/')
def main_page():
    layout.build()
    ui.timer(0.5, layout.update_visibility)

def main(timeout=None):
    AppLogger.setup()

    import sys
    if getattr(sys, 'frozen', False):
        try:
            import pyi_splash
            pyi_splash.close()
        except ImportError:
            pass

    proc_thread = threading.Thread(target=face_processing_loop, daemon=True)
    proc_thread.start()

    async def window_loop():
        await alert_manager.ensure_hidden_startup()
        while True:
            await asyncio.sleep(0.5)
            await alert_manager.manage_window(access_controller.denied)

    app.on_startup(window_loop)

    if timeout:
        async def register_auto_shutdown():
            async def auto_shutdown():
                await asyncio.sleep(timeout)
                if alert_manager.local_state_fullscreen:
                    AppLogger.log(f"Timeout reached ({timeout}s) but ALERT IS ACTIVE. Extension granted. Retrying in 5s...", "warning")
                    await asyncio.sleep(5)
                    await auto_shutdown()
                    return

                AppLogger.log(f"Timeout reached ({timeout}s). FORCING EXIT", "info")
                try:
                    app.shutdown()
                except Exception as e:
                    AppLogger.log(f"Error during shutdown: {e}", "error")
                await asyncio.sleep(0.5)
                import os
                os._exit(0)

            asyncio.create_task(auto_shutdown())

        app.on_startup(register_auto_shutdown)

    try:
        port = find_free_port()
        AppLogger.log(f"Starting Background Service UI on port {port}", "info")
        ui.run(
            title="Serviço de Biometria",
            port=port,
            show=False,
            reload=False,
            native=True,
            fullscreen=True,
        )
    except Exception as e:
        import traceback
        traceback.print_exc()
        AppLogger.log(f"FATAL ERROR in main: {e}", "error")

if __name__ in {"__main__", "__mp_main__"}:
    main()