Исходный код EJIO.ctk_extender.event.controller

## /ctk_extender/event/controller.py

import customtkinter as ctk
from collections.abc import Callable
from threading import Lock
from typing import Any, cast, NamedTuple
from loguru import logger

from EJIO.utils import StaticClass, Queue
from .base import Event

__all__ = [
    'EventController',
]


class _ListenerInfo(NamedTuple):
    """Внутренняя структура для хранения метаданных подписчика."""
    callback: Callable[[Any], None]
    sender: Any | None
    predicate: Callable[[Any], bool] | None




class EventController(StaticClass):
    """
    Потокобезопасный контроллер событий.
    Поддерживает строгую фильтрацию событий по отправителю (sender) или кастомному предикату.
    Вызывает обработчики строго в главном UI-потоке customtkinter.
    """
    __slots__ = ()

    _lock: Lock = Lock()
    _queue: Queue[Event] = Queue()
    _listeners: dict[type[Event], list[_ListenerInfo]] = {}
    _root_widget: ctk.CTk | None = None
    _polling_interval_ms: int = 16



    @classmethod
    def initialize(cls, root_widget: ctk.CTk, polling_interval_ms: int = 16) -> None:
        """
        Инициализирует контроллер и запускает бесконечный цикл опроса очереди.

        :param root_widget: Корневой виджет ctk приложения.
        :param polling_interval_ms: Интервал между обработкой событий.
        """
        with cls._lock:
            cls._root_widget = root_widget
            cls._polling_interval_ms = polling_interval_ms

        cls._root_widget.after(cls._polling_interval_ms, cls._process_queue)
        logger.info("EventController for ctk_extender successfully initialized.")




    @classmethod
    def subscribe[E: Event](
            cls,
            event_type: type[E],
            callback: Callable[[E], None],
            *,
            sender: Any | None = None,
            predicate: Callable[[E], bool] | None = None
    ) -> None:
        """
        Подписывает обработчик на определенный класс события с возможностью фильтрации.

        :param event_type: Класс события (наследник Event).
        :param callback: Функция обратного вызова, принимающая экземпляр этого события.
        :param sender: Опционально: фильтровать события строго от этого объекта-отправителя.
        :param predicate: Опционально: кастомная функция-фильтр, возвращающая True/False.
        """
        with cls._lock:
            if event_type not in cls._listeners:
                cls._listeners[event_type] = []

            info = _ListenerInfo(
                callback=cast(Callable[[Any], None], callback),
                sender=sender,
                predicate=cast(Callable[[Any], bool], predicate)
            )
            cls._listeners[event_type].append(info)
            logger.debug(f"Subscriber added on {event_type.__name__} (Filter at sender: {sender is not None})")




    @classmethod
    def unsubscribe[E: Event](cls, event_type: type[E], callback: Callable[[E], None]) -> None:
        """
        Отписывает обработчик от указанного класса события.

        :param event_type: Класс события (наследник Event).
        :param callback: Функция обратного вызова, привязанная к этому событию.
        """
        with cls._lock:
            if event_type in cls._listeners:
                # Удаляем все записи, связанные с этим коллбэком
                cls._listeners[event_type] = [
                    info for info in cls._listeners[event_type] if info.callback != callback
                ]




    @classmethod
    def emit(cls, event: Event) -> None:
        """
        Потокобезопасно отправляет объект события в очередь.

        :param event: Событие которое нужно отправить в очередь.
        """
        cls._queue.push(event)


    @classmethod
    def _process_queue(cls) -> None:
        """Внутренний воркер опроса очереди. Выполняется в главном потоке."""
        if cls._root_widget is None:
            return

        while not cls._queue.empty:
            try:
                event = cls._queue.pop()
                event_type = type(event)

                with cls._lock:
                    listeners = cls._listeners.get(event_type, []).copy()

                for info in listeners:
                    try:
                        # --- Шаг фильтрации 1: Проверка по конкретному отправителю ---
                        if info.sender is not None and event.sender is not info.sender:
                            continue

                        # --- Шаг фильтрации 2: Проверка по кастомному предикату ---
                        if info.predicate is not None and not info.predicate(event):
                            continue

                        # Если все фильтры пройдены — вызываем обработчик
                        info.callback(event)
                    except Exception as e:
                        logger.exception(f"Critical error in event handler {event_type.__name__}: {e}")

                # Явно уничтожаем локальную ссылку на событие, освобождая память в текущем такте
                del event

            except IndexError:
                break

        cls._root_widget.after(cls._polling_interval_ms, cls._process_queue)