Source code for src.data_acquisition.event_manager.event_manager

from abc import ABC, abstractmethod
from logging import Logger
from typing import Generic, Optional, TypeVar
from unittest.mock import MagicMock

from ..types import ResultEventCallback
from .errors import EventManagerError, IncorrectMethodCallOrderError

T = TypeVar("T")




class EventManager(Generic[T], ABC):
    def __init__(self, *, logger: Optional[Logger] = None) -> None:
        self._callbacks: list[ResultEventCallback[T]] = []

        self._logger = logger if logger is not None else MagicMock()

        self._has_been_started = False



    def register_callback(self, callback: ResultEventCallback[T]) -> None:
        """
        :param callback: The callback function to be called when an event occurs that should end the current screen.
        """

        self._log(f"Registering callback: {callback}")

        self._callbacks.append(callback)


    def _log(self, message: str) -> None:
        self._logger.debug(f"{self.__class__.__name__}: {message}")



    def deregister_callback(self, callback: ResultEventCallback[T]) -> None:
        """
        :param callback: The callback function to be deregister.

        :raises EventManagerError: If the callback is not registered.
        """

        if callback not in self._callbacks:
            raise EventManagerError("Callback not registered.")

        self._log(f"Deregistering callback: {callback}")

        self._callbacks.remove(callback)


    def _trigger_callbacks(self, result: T) -> None:
        if len(self._callbacks) == 0:
            raise EventManagerError("No callback set.")

        self._log(f"Triggering {len(self._callbacks)} callbacks with result: {result}")

        for callback in self._callbacks:
            callback(result)



    def start(self) -> None:
        self._check_has_not_been_started()

        self._has_been_started = True

        self._log("Starting event manager.")

        self._start()


    def _check_has_not_been_started(self) -> None:
        if self._has_been_started:
            raise IncorrectMethodCallOrderError(
                "Event manager has already been started."
            )

        self._has_been_started = True

    @abstractmethod
    def _start(self) -> None:
        pass



    def stop(self) -> None:
        self._check_has_been_started()

        self._log("Stopping event manager.")

        self._stop()

        self._has_been_started = False


    def _check_has_been_started(self) -> None:
        if not self._has_been_started:
            raise IncorrectMethodCallOrderError(
                "Event manager has not been started yet."
            )

    @abstractmethod
    def _stop(self) -> None:
        pass



    @abstractmethod
    def clone(self) -> "EventManager[T]":
        """
        Method used to create a clone of the event manager, so that it can be used as a prototype and reused by multiple screens generated by a screen sequencer.

        :return: A new instance of the same type as this event manager, with the same configuration.
        """

        pass


    def _clone_registered_callbacks_from(
        self, event_manager: "EventManager[T]"
    ) -> None:
        for callback in event_manager._callbacks:
            self.register_callback(callback)