"""Manual, composite, and threaded EUT monitor implementations."""
from __future__ import annotations
import queue
import threading
from .events import (
EUT_EVENT_PHASES,
EUT_EVENT_STATUSES,
EUT_FUNCTIONAL_STATES,
make_eut_event,
validate_eut_event,
)
[docs]
class EUTMonitor:
"""Minimal interface for monitoring an EUT during one RF exposure.
Implementations must not access the measurement's ``MGraph``. Polling
should be non-blocking unless the monitor is wrapped in
:class:`ThreadedEUTMonitor`.
"""
[docs]
def start_exposure(self, context):
"""Prepare monitoring for one exposure.
Parameters
----------
context : mapping
Measurement context describing the exposure.
"""
[docs]
def start_phase(self, phase, context):
"""Notify the monitor that an exposure phase has started.
Parameters
----------
phase : str
Phase from :data:`EUT_EVENT_PHASES`.
context : mapping
Measurement context for the phase.
"""
if phase not in EUT_EVENT_PHASES:
raise ValueError("EUT monitor phase must be one of %s" % ", ".join(EUT_EVENT_PHASES))
[docs]
def poll_event(self):
"""Return one structured EUT event or ``None`` when nothing changed."""
return None
[docs]
def stop_exposure(self):
"""Stop monitoring the current exposure."""
[docs]
def close(self):
"""Release monitor resources."""
[docs]
class ManualEUTMonitor(EUTMonitor):
"""Always-available manual EUT status input.
Events may be submitted from a GUI thread or generated from a configured
keyboard callback. State after exposure and recovery method are accepted
as separate values so that the performance criterion remains evaluable.
Parameters
----------
poll_key : callable, optional
Non-blocking callback returning an integer key code or ``None``.
keylist : str, optional
Keys that record an operator-observed EUT failure.
stop_keys : str, optional
Keys that request measurement termination.
"""
[docs]
def __init__(self, poll_key=None, keylist="sS", stop_keys="qQ"):
self.poll_key = poll_key
self.keylist = str(keylist)
self.stop_keys = str(stop_keys)
self._events = queue.SimpleQueue()
self._state_lock = threading.Lock()
self._active = False
self._phase = "during_exposure"
[docs]
def start_exposure(self, context):
"""Enable manual input for a new exposure and discard stale events.
Parameters
----------
context : mapping
Measurement context describing the exposure.
"""
_ = context
with self._state_lock:
self._events = queue.SimpleQueue()
self._phase = "during_exposure"
self._active = True
@property
def is_active(self):
"""Whether manual observations are currently accepted."""
with self._state_lock:
return self._active
@property
def current_phase(self):
"""Current exposure phase for newly submitted observations."""
with self._state_lock:
return self._phase
[docs]
def start_phase(self, phase, context):
"""Select the phase assigned to subsequent manual observations.
Parameters
----------
phase : str
Phase from :data:`EUT_EVENT_PHASES`.
context : mapping
Measurement context for the phase.
"""
super().start_phase(phase, context)
with self._state_lock:
self._phase = phase
[docs]
def submit_event(self, event, *, source="manual_gui"):
"""Queue a manually supplied event for the active exposure.
Parameters
----------
event : mapping
EUT event fields accepted by :func:`validate_eut_event`.
source : str, optional
Operator interface that supplied the event.
Returns
-------
dict
Normalized event placed in the queue.
"""
with self._state_lock:
if not self._active:
raise RuntimeError("manual EUT input is only available during an active exposure")
phase = self._phase
event_queue = self._events
spec = dict(event)
spec["phase"] = phase
spec["source"] = source
normalized = validate_eut_event(spec)
event_queue.put(normalized)
return normalized
[docs]
def submit_status(self, status, reason="operator_observation", details=None,
*, source="manual_gui", **event_fields):
"""Queue a manual functional status in the current exposure phase.
Parameters
----------
status : str
Observation status from :data:`EUT_EVENT_STATUSES`.
reason : str, optional
Machine-readable reason for the observation.
details : object, optional
Supplemental operator information.
source : str, optional
Operator interface that supplied the status.
**event_fields : object
Additional fields accepted by :func:`make_eut_event`.
Returns
-------
dict
Normalized event placed in the queue.
"""
event = make_eut_event(
status,
reason,
details,
source=source,
**event_fields,
)
return self.submit_event(event, source=source)
[docs]
def submit_post_exposure_state(self, after_exposure_state, *, recovery,
reason="operator_post_exposure_observation",
details=None, source="manual_gui",
operating_mode_changed=None,
stored_data_lost=None):
"""Queue post-exposure state and recovery information.
Parameters
----------
after_exposure_state : str
Functional state observed after RF exposure.
recovery : str
Recovery mechanism consistent with the observed state.
reason : str, optional
Machine-readable reason for the observation.
details : object, optional
Supplemental operator information.
source : str, optional
Operator interface that supplied the observation.
operating_mode_changed : bool, optional
Whether the EUT operating mode changed unexpectedly.
stored_data_lost : bool, optional
Whether stored EUT data was lost.
Returns
-------
dict
Normalized post-exposure event placed in the queue.
"""
if self.current_phase != "post_exposure":
raise RuntimeError(
"post-exposure state can only be submitted during the post_exposure phase"
)
if after_exposure_state not in EUT_FUNCTIONAL_STATES:
raise ValueError(
"EUT after-exposure state must be one of %s"
% ", ".join(EUT_FUNCTIONAL_STATES)
)
valid_recoveries = {
"normal": ("not_required", "automatic", "operator", "reset"),
"degraded": ("failed",),
"failed": ("failed",),
"not_evaluated": ("not_evaluated",),
}[after_exposure_state]
if recovery not in valid_recoveries:
raise ValueError(
"recovery %r is inconsistent with after-exposure state %r; expected one of %s"
% (recovery, after_exposure_state, ", ".join(valid_recoveries))
)
status = {
"normal": "passed",
"degraded": "degraded",
"failed": "failed",
"not_evaluated": "not_evaluated",
}[after_exposure_state]
event = make_eut_event(
status,
reason,
details,
phase="post_exposure",
functional_state=after_exposure_state,
after_exposure_state=after_exposure_state,
recovery=recovery,
operating_mode_changed=operating_mode_changed,
stored_data_lost=stored_data_lost,
source=source,
)
return self.submit_event(event, source=source)
[docs]
def poll_event(self):
"""Return the next queued or keyboard-generated operator event.
Returns
-------
dict or None
Next event, or ``None`` when no operator input is available.
"""
with self._state_lock:
event_queue = self._events
phase = self._phase
active = self._active
if not active:
return None
try:
return event_queue.get_nowait()
except queue.Empty:
pass
key = self.poll_key() if callable(self.poll_key) else None
if not isinstance(key, int) or not 0 <= key <= 255:
return None
character = chr(key)
if character in self.keylist:
return make_eut_event(
"failed",
"operator_marked",
"EUT failure marked by operator",
phase=phase,
source="manual_keyboard",
)
if character in self.stop_keys:
return make_eut_event(
"not_evaluated",
"operator_stop",
"Measurement stop requested by operator",
phase=phase,
action="stop",
source="manual_keyboard",
)
return None
[docs]
def stop_exposure(self):
"""Disable manual input for the completed exposure."""
with self._state_lock:
self._active = False
# Compatibility name for callers that used the original keyboard-only class.
OperatorEUTMonitor = ManualEUTMonitor
[docs]
class CompositeEUTMonitor(EUTMonitor):
"""Combine mandatory manual input with optional automatic monitors.
Parameters
----------
manual_monitor : ManualEUTMonitor
Always-available operator interaction layer.
automatic_monitors : EUTMonitor or iterable of EUTMonitor, optional
Additional automatic observation sources.
"""
[docs]
def __init__(self, manual_monitor, automatic_monitors=()):
if not isinstance(manual_monitor, ManualEUTMonitor):
raise TypeError("manual_monitor must be a ManualEUTMonitor")
if isinstance(automatic_monitors, EUTMonitor):
automatic_monitors = (automatic_monitors,)
self.manual_monitor = manual_monitor
self.automatic_monitors = tuple(automatic_monitors)
for monitor in self.automatic_monitors:
if not isinstance(monitor, EUTMonitor):
raise TypeError("automatic monitors must implement EUTMonitor")
self._monitor_errors = queue.SimpleQueue()
self._disabled_monitors = set()
self._phase = "during_exposure"
def _monitor_source(self, monitor):
return "automatic:%s" % type(monitor).__name__
def _record_monitor_error(self, monitor, operation, exc):
self._disabled_monitors.add(monitor)
self._monitor_errors.put(make_eut_event(
"not_evaluated",
"monitor_error",
"%s during %s: %s" % (type(exc).__name__, operation, exc),
phase=self._phase,
action="continue",
source=self._monitor_source(monitor),
safety_action="none",
event_type="monitor_diagnostic",
))
[docs]
def start_exposure(self, context):
"""Start manual and automatic monitors for one exposure.
Parameters
----------
context : mapping
Measurement context describing the exposure.
"""
self._monitor_errors = queue.SimpleQueue()
self._disabled_monitors = set()
self._phase = "during_exposure"
self.manual_monitor.start_exposure(context)
for monitor in self.automatic_monitors:
try:
monitor.start_exposure(dict(context))
except Exception as exc:
self._record_monitor_error(monitor, "start_exposure", exc)
[docs]
def start_phase(self, phase, context):
"""Forward an exposure phase transition to all active monitors.
Parameters
----------
phase : str
Phase from :data:`EUT_EVENT_PHASES`.
context : mapping
Measurement context for the phase.
"""
super().start_phase(phase, context)
self._phase = phase
self.manual_monitor.start_phase(phase, context)
for monitor in self.automatic_monitors:
if monitor in self._disabled_monitors:
continue
try:
monitor.start_phase(phase, dict(context))
except Exception as exc:
self._record_monitor_error(monitor, "start_phase", exc)
[docs]
def poll_event(self):
"""Return the next manual, diagnostic, or automatic monitor event.
Returns
-------
dict or None
Next available event, or ``None`` when no monitor has an event.
"""
manual_event = self.manual_monitor.poll_event()
if manual_event is not None:
return manual_event
try:
return self._monitor_errors.get_nowait()
except queue.Empty:
pass
for monitor in self.automatic_monitors:
if monitor in self._disabled_monitors:
continue
try:
event = monitor.poll_event()
except Exception as exc:
self._record_monitor_error(monitor, "poll_event", exc)
return self._monitor_errors.get_nowait()
if event is not None:
spec = dict(event)
spec["phase"] = self._phase
if not spec.get("source"):
spec["source"] = self._monitor_source(monitor)
return validate_eut_event(spec)
return None
[docs]
def stop_exposure(self):
"""Stop every monitor while containing automatic-monitor failures."""
self.manual_monitor.stop_exposure()
for monitor in self.automatic_monitors:
try:
monitor.stop_exposure()
except Exception:
pass
[docs]
def close(self):
"""Release resources held by all component monitors."""
self.manual_monitor.close()
for monitor in self.automatic_monitors:
try:
monitor.close()
except Exception:
pass
[docs]
class ThreadedEUTMonitor(EUTMonitor):
"""Poll another EUT monitor in a dedicated background thread.
The wrapped monitor's ``poll_event`` may block only for a bounded time so
that ``stop_exposure`` can join the worker. ``start_exposure`` and
``stop_exposure`` themselves are called outside the worker and should
therefore return promptly.
Parameters
----------
monitor : EUTMonitor
Monitor polled by the background worker.
poll_interval : float, optional
Delay in seconds between polling calls.
join_timeout : float, optional
Maximum time in seconds allowed for worker shutdown.
"""
[docs]
def __init__(self, monitor, *, poll_interval=0.01, join_timeout=1.0):
if not isinstance(monitor, EUTMonitor):
required = ("start_exposure", "poll_event", "stop_exposure", "close")
if not all(callable(getattr(monitor, name, None)) for name in required):
raise TypeError("monitor does not implement the EUTMonitor interface")
self.monitor = monitor
self.poll_interval = float(poll_interval)
self.join_timeout = float(join_timeout)
if self.poll_interval < 0.0:
raise ValueError("poll_interval must be non-negative")
if self.join_timeout <= 0.0:
raise ValueError("join_timeout must be positive")
self._events = queue.SimpleQueue()
self._stop_requested = threading.Event()
self._monitor_lock = threading.Lock()
self._worker = None
@property
def is_running(self):
"""Whether the background polling worker is currently running."""
return self._worker is not None and self._worker.is_alive()
[docs]
def start_exposure(self, context):
"""Start the wrapped monitor and its background polling worker.
Parameters
----------
context : mapping
Measurement context describing the exposure.
"""
if self.is_running:
raise RuntimeError("EUT monitor exposure is already running")
self._events = queue.SimpleQueue()
self._stop_requested.clear()
self.monitor.start_exposure(dict(context))
self._worker = threading.Thread(
target=self._run,
name="mpylab-eut-monitor",
daemon=True,
)
self._worker.start()
def _run(self):
try:
while not self._stop_requested.is_set():
with self._monitor_lock:
event = self.monitor.poll_event()
if event is not None:
self._events.put(event)
if self.poll_interval:
self._stop_requested.wait(self.poll_interval)
except Exception as exc: # monitor failures must reach the measurement worker
self._events.put(make_eut_event(
"not_evaluated",
"monitor_error",
"%s: %s" % (type(exc).__name__, exc),
action="continue",
safety_action="none",
event_type="monitor_diagnostic",
))
[docs]
def start_phase(self, phase, context):
"""Forward a phase transition while serializing monitor access.
Parameters
----------
phase : str
Phase from :data:`EUT_EVENT_PHASES`.
context : mapping
Measurement context for the phase.
"""
super().start_phase(phase, context)
callback = getattr(self.monitor, "start_phase", None)
if callable(callback):
with self._monitor_lock:
callback(phase, dict(context))
[docs]
def poll_event(self):
"""Return one event queued by the worker without blocking.
Returns
-------
dict or None
Next queued event, or ``None`` when the queue is empty.
"""
try:
return self._events.get_nowait()
except queue.Empty:
return None
[docs]
def stop_exposure(self):
"""Stop and join the worker, then stop the wrapped monitor."""
self._stop_requested.set()
if self._worker is not None:
self._worker.join(self.join_timeout)
if self._worker.is_alive():
raise RuntimeError(
"EUT monitor worker did not stop; poll_event must use a bounded timeout"
)
self._worker = None
self.monitor.stop_exposure()
[docs]
def close(self):
"""Stop an active worker and release the wrapped monitor."""
if self.is_running:
self.stop_exposure()
self.monitor.close()
__all__ = [
"CompositeEUTMonitor",
"EUTMonitor",
"ManualEUTMonitor",
"OperatorEUTMonitor",
"ThreadedEUTMonitor",
]