mpylab.env.eut.monitors module

Manual, composite, and threaded EUT monitor implementations.

class mpylab.env.eut.monitors.CompositeEUTMonitor(manual_monitor, automatic_monitors=())

Bases: 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.

close()

Release resources held by all component monitors.

poll_event()

Return the next manual, diagnostic, or automatic monitor event.

Returns:

Next available event, or None when no monitor has an event.

Return type:

dict or None

start_exposure(context)

Start manual and automatic monitors for one exposure.

Parameters:

context (mapping) – Measurement context describing the exposure.

start_phase(phase, context)

Forward an exposure phase transition to all active monitors.

Parameters:

  • phase (str) – Phase from EUT_EVENT_PHASES.

  • context (mapping) – Measurement context for the phase.

stop_exposure()

Stop every monitor while containing automatic-monitor failures.

class mpylab.env.eut.monitors.EUTMonitor

Bases: object

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 ThreadedEUTMonitor.

close()

Release monitor resources.

poll_event()

Return one structured EUT event or None when nothing changed.

start_exposure(context)

Prepare monitoring for one exposure.

Parameters:

context (mapping) – Measurement context describing the exposure.

start_phase(phase, context)

Notify the monitor that an exposure phase has started.

Parameters:

  • phase (str) – Phase from EUT_EVENT_PHASES.

  • context (mapping) – Measurement context for the phase.

stop_exposure()

Stop monitoring the current exposure.

class mpylab.env.eut.monitors.ManualEUTMonitor(poll_key=None, keylist='sS', stop_keys='qQ')

Bases: 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.

property current_phase

Current exposure phase for newly submitted observations.

property is_active

Whether manual observations are currently accepted.

poll_event()

Return the next queued or keyboard-generated operator event.

Returns:

Next event, or None when no operator input is available.

Return type:

dict or None

start_exposure(context)

Enable manual input for a new exposure and discard stale events.

Parameters:

context (mapping) – Measurement context describing the exposure.

start_phase(phase, context)

Select the phase assigned to subsequent manual observations.

Parameters:

  • phase (str) – Phase from EUT_EVENT_PHASES.

  • context (mapping) – Measurement context for the phase.

stop_exposure()

Disable manual input for the completed exposure.

submit_event(event, *, source='manual_gui')

Queue a manually supplied event for the active exposure.

Parameters:

  • event (mapping) – EUT event fields accepted by validate_eut_event().

  • source (str, optional) – Operator interface that supplied the event.

Returns:

Normalized event placed in the queue.

Return type:

dict

submit_post_exposure_state(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:

Normalized post-exposure event placed in the queue.

Return type:

dict

submit_status(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 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 make_eut_event().

Returns:

Normalized event placed in the queue.

Return type:

dict

mpylab.env.eut.monitors.OperatorEUTMonitor

alias of ManualEUTMonitor

class mpylab.env.eut.monitors.ThreadedEUTMonitor(monitor, *, poll_interval=0.01, join_timeout=1.0)

Bases: 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.

close()

Stop an active worker and release the wrapped monitor.

property is_running

Whether the background polling worker is currently running.

poll_event()

Return one event queued by the worker without blocking.

Returns:

Next queued event, or None when the queue is empty.

Return type:

dict or None

start_exposure(context)

Start the wrapped monitor and its background polling worker.

Parameters:

context (mapping) – Measurement context describing the exposure.

start_phase(phase, context)

Forward a phase transition while serializing monitor access.

Parameters:

  • phase (str) – Phase from EUT_EVENT_PHASES.

  • context (mapping) – Measurement context for the phase.

stop_exposure()

Stop and join the worker, then stop the wrapped monitor.