DynaLab/src/dynalab/app.py

# Copyright (C) 2026 Hector van der Aa <hector@h3cx.dev>
# Copyright (C) 2026 Association Exergie <association.exergie@gmail.com>
# SPDX-License-Identifier: GPL-3.0-or-later
import logging
from pathlib import Path
from queue import Empty, Queue
from threading import Thread
import threading
import time
import traceback
import dearpygui.dearpygui as dpg
import sys

from dynalab.config import ConfigLoadError, load_config
from dynalab.data.json import json_thread_main
from dynalab.state import AppState
from dynalab.tags import WINDOW_DATA_INPUT_GAUGE_GRID
from dynalab.ui.error import build_error_modal, show_error_modal
import dynalab.ui.windows
from dynalab.ui.windows import (
    LOADING_WINDOW_HEIGHT,
    LOADING_WINDOW_WIDTH,
    build_loading,
    build_windows,
    center_loading_content,
)

import dpg_gauges as dpgg
from dpg_gauges.exceptions import GaugeNotFoundError


def _setup_logging(level: int = logging.INFO) -> None:
    logging.basicConfig(
        level=level,
        format="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
        datefmt="%H:%M:%S",
        stream=sys.stdout,
    )


def _asset_path(relative_path: str) -> str:
    path = Path(relative_path)
    candidates: list[Path] = []

    bundle_dir = getattr(sys, "_MEIPASS", None)
    if bundle_dir is not None:
        candidates.append(Path(bundle_dir) / path)

    candidates.extend(
        (
            Path.cwd() / path,
            Path(__file__).resolve().parents[2] / path,
        )
    )

    for candidate in candidates:
        if candidate.exists():
            return str(candidate)

    searched = ", ".join(str(candidate) for candidate in candidates)
    raise FileNotFoundError(f"Missing asset {relative_path!r}. Searched: {searched}")


def _init_worker(
    state: AppState, stop_event: threading.Event, message_queue: Queue[str]
) -> None:
    try:
        time.sleep(1.0)
        message_queue.put("Fetching DynaLab config")
        time.sleep(0.5)
        state.app_config = load_config()
        message_queue.put("Fetched DynaLab config")
        time.sleep(0.5)

        message_queue.put("Starting JSON data server")
        # Init and start json connector thread
        state.data_json_thread = Thread(
            target=json_thread_main,
            name="json-connector-network",
            daemon=True,
            args=(state,),
        )
        state.data_json_thread.start()
        time.sleep(0.5)
        message_queue.put("JSON data server started")
        time.sleep(0.5)
        message_queue.put("DynaLab Ready")
        time.sleep(1.0)

        stop_event.set()
    except ConfigLoadError as exc:
        state.init_error = str(exc)
        state.stop_init.set()
    except Exception as exc:
        state.init_error = f"Unexpected startup error: {exc}"
        state.stop_init.set()


def _load_image(path: Path, tag: str) -> None:
    width, height, channels, data = dpg.load_image(f"{path}")
    with dpg.texture_registry():
        dpg.add_static_texture(width=width, height=height, default_value=data, tag=tag)


def _maximize_viewport() -> None:
    maximize_viewport = getattr(dpg, "maximize_viewport", None)
    if callable(maximize_viewport):
        maximize_viewport()
        return

    dpg.set_viewport_height(720)
    dpg.set_viewport_width(1280)


def run() -> None:
    # Setup logging format and level
    _setup_logging(logging.DEBUG)

    # dpg setup
    dpg.create_context()

    try:
        # Switch callback management to manual
        dpg.configure_app(manual_callback_management=True)
        # Initialize viewport
        dpg.create_viewport(
            title="DynaLab", width=LOADING_WINDOW_WIDTH, height=LOADING_WINDOW_HEIGHT
        )

        # Load fonts at 2x size for HiDPI scaling
        with dpg.font_registry():
            app_font = dpg.add_font(
                _asset_path("assets/fonts/Inter-Regular.ttf"), 18 * 2
            )
            mono_font = dpg.add_font(
                _asset_path("assets/fonts/JetBrainsMono-Regular.ttf"),
                size=36,
                label="mono_font",
            )
        dpg.bind_font(app_font)

        _load_image(_asset_path("assets/images/logo@300.png"), "logo")

        build_loading()
        build_error_modal()

        # Open viewport
        dpg.setup_dearpygui()
        dpg.show_viewport()

        dpg.configure_item(
            "loading_window",
            pos=(0, 0),
            width=LOADING_WINDOW_WIDTH,
            height=LOADING_WINDOW_HEIGHT,
        )

        # Init global app state
        state: AppState = AppState()

        loading_msg_queue: Queue[str] = Queue()
        loading_stop: threading.Event = threading.Event()
        loading_thread = Thread(
            target=_init_worker,
            name="init-worker-thread",
            args=(state, loading_stop, loading_msg_queue),
            daemon=True,
        )
        loading_thread.start()

        while not loading_stop.is_set():
            # Fetch and execute callbacks
            if state.stop_init.is_set():
                show_error_modal(
                    state.init_error if state.init_error is not None else "Fatal error"
                )
            jobs = dpg.get_callback_queue()
            try:
                dpg.run_callbacks(jobs)
            except Exception:
                traceback.print_exc()

            center_loading_content()
            try:
                msg = loading_msg_queue.get_nowait()
            except Empty:
                pass
            else:
                dpg.set_value(item="loading_text", value=msg)
            # Render frame
            dpg.render_dearpygui_frame()

        dpg.delete_item("loading_window")
        # Call ui building function
        _maximize_viewport()
        build_windows(state)

        # Config main window size
        vp_w = dpg.get_viewport_client_width()
        vp_h = dpg.get_viewport_client_height()
        dpg.configure_item("main_window", pos=(0, 0), width=vp_w, height=vp_h)
        dpg.set_primary_window("main_window", True)

        # dpg UI loop
        while dpg.is_dearpygui_running():
            # Fetch and execute callbacks
            jobs = dpg.get_callback_queue()
            try:
                dpg.run_callbacks(jobs)
            except Exception:
                traceback.print_exc()

            try:
                new_signal = state.new_signal_queue.get_nowait()
            except Empty:
                pass
            else:
                dpgg.digital_gauge(
                    tag=f"{new_signal.id}",
                    label=f"{new_signal.name}",
                    parent=WINDOW_DATA_INPUT_GAUGE_GRID,
                    unit=new_signal.unit if new_signal.unit is not None else "",
                    value=0,
                    precision=1,
                    min_value=new_signal.min_value,
                    max_value=new_signal.max_value,
                )

            try:
                removed_signal = state.removed_signal_queue.get_nowait()
            except Empty:
                pass
            else:
                dpgg.delete_gauge(f"{removed_signal.id}")

            ctr: int = 0
            while not state.new_data_id.empty() or ctr < 5:
                try:
                    new_value_id = state.new_data_id.get_nowait()
                except Empty:
                    pass
                else:
                    try:
                        dpgg.set_value(
                            tag=f"{new_value_id}", value=state.live_data[new_value_id]
                        )
                    except GaugeNotFoundError:
                        pass

                ctr += 1

            # Render frame
            dpg.render_dearpygui_frame()
    finally:
        state.stop_event.set()
        dpg.destroy_context()