Source code for mpylab.env.eut.session

"""RF-neutral lifecycle management for EUT monitoring."""

from __future__ import annotations

from collections.abc import Iterable, Mapping

from .events import validate_eut_event
from .monitors import CompositeEUTMonitor, EUTMonitor, ManualEUTMonitor




class EUTMonitoringSession:
    """Manage monitors and classify their events for one EUT exposure.

    The session deliberately has no access to RF hardware and does not apply
    retry or stop policies. Measurement kernels remain responsible for those
    safety and workflow decisions.

    Parameters
    ----------
    monitor : EUTMonitor
        Monitor, commonly a :class:`CompositeEUTMonitor`, whose lifecycle is
        managed by the session.

    Attributes
    ----------
    monitor : EUTMonitor
        Managed monitor instance.
    active : bool
        Whether an exposure is active.
    phase : str or None
        Current exposure phase, or ``None`` outside an exposure.
    """



    def __init__(self, monitor: EUTMonitor) -> None:
        if not isinstance(monitor, EUTMonitor):
            raise TypeError("monitor must implement EUTMonitor")
        self.monitor = monitor
        self.active = False
        self.phase = None
        self._events = []
        self._diagnostics = []




    @classmethod
    def from_monitors(
        cls,
        manual_monitor: ManualEUTMonitor,
        automatic_monitors: EUTMonitor | Iterable[EUTMonitor] = (),
    ) -> "EUTMonitoringSession":
        """Build a session with mandatory manual operator intervention.

        Parameters
        ----------
        manual_monitor : ManualEUTMonitor
            Always-available manual observation and intervention source.
        automatic_monitors : EUTMonitor or iterable of EUTMonitor, optional
            Additional automatic observation sources.

        Returns
        -------
        EUTMonitoringSession
            Session managing the combined monitor.
        """

        return cls(CompositeEUTMonitor(manual_monitor, automatic_monitors))


    @property
    def events(self) -> tuple[dict, ...]:
        """Return status events collected during the current exposure."""

        return tuple(dict(event) for event in self._events)

    @property
    def diagnostics(self) -> tuple[dict, ...]:
        """Return monitor diagnostics collected during the current exposure."""

        return tuple(dict(event) for event in self._diagnostics)



    def start_exposure(self, context: Mapping) -> None:
        """Start a new exposure and clear records from the previous one.

        Parameters
        ----------
        context : mapping
            Measurement-specific exposure context forwarded to every monitor.

        Raises
        ------
        RuntimeError
            If another exposure is already active.
        """

        if self.active:
            raise RuntimeError("an EUT monitoring exposure is already active")
        self._events = []
        self._diagnostics = []
        self.monitor.start_exposure(dict(context))
        self.active = True
        self.phase = "during_exposure"




    def start_phase(self, phase: str, context: Mapping) -> None:
        """Switch all monitors to another phase of the active exposure.

        Parameters
        ----------
        phase : str
            Exposure phase accepted by the EUT event schema.
        context : mapping
            Measurement-specific phase context forwarded to every monitor.

        Raises
        ------
        RuntimeError
            If no exposure is active.
        """

        if not self.active:
            raise RuntimeError("cannot change phase without an active exposure")
        self.monitor.start_phase(phase, dict(context))
        self.phase = phase




    def poll_event(self, diagnostic_context: Mapping | None = None):
        """Return the next status event and retain diagnostics separately.

        Parameters
        ----------
        diagnostic_context : mapping, optional
            Metadata added with ``setdefault`` to monitor diagnostics, for
            example the current frequency, position, or attempt number.

        Returns
        -------
        dict or None
            Next validated status event. Diagnostics and an empty monitor
            queue both return ``None``; diagnostics are available through
            :attr:`diagnostics` and :meth:`pop_diagnostics`.
        """

        if not self.active:
            return None
        candidate = self.monitor.poll_event()
        if candidate is None:
            return None
        event = validate_eut_event(candidate)
        if event.get("event_type") == "monitor_diagnostic":
            event = dict(event)
            for key, value in dict(diagnostic_context or {}).items():
                event.setdefault(key, value)
            self._diagnostics.append(event)
            return None
        self._events.append(event)
        return event




    def pop_diagnostics(self) -> list[dict]:
        """Remove and return all diagnostics collected since the last call.

        Returns
        -------
        list of dict
            Collected monitor diagnostics in arrival order.
        """

        diagnostics = [dict(event) for event in self._diagnostics]
        self._diagnostics = []
        return diagnostics




    def stop_exposure(self) -> None:
        """Stop the active exposure; repeated calls are harmless."""

        if not self.active:
            self.phase = None
            return
        try:
            self.monitor.stop_exposure()
        finally:
            self.active = False
            self.phase = None




    def close(self) -> None:
        """Stop an active exposure and release all monitor resources."""

        self.stop_exposure()
        self.monitor.close()




__all__ = ["EUTMonitoringSession"]