mpylab.env.eut_monitor module

Compatibility imports for mpylab.env.eut.

New code should import EUT monitoring APIs from mpylab.env.eut.

class mpylab.env.eut_monitor.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_monitor.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_monitor.EUTMonitoringSession(monitor: EUTMonitor)

Bases: object

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 CompositeEUTMonitor, whose lifecycle is managed by the session.

monitor

Managed monitor instance.

Type:

EUTMonitor

active

Whether an exposure is active.

Type:

bool

phase

Current exposure phase, or None outside an exposure.

Type:

str or None

close() → None

Stop an active exposure and release all monitor resources.

property diagnostics: tuple[dict, ...]

Return monitor diagnostics collected during the current exposure.

property events: tuple[dict, ...]

Return status events collected during the current exposure.

classmethod from_monitors(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:

Session managing the combined monitor.

Return type:

EUTMonitoringSession

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

Next validated status event. Diagnostics and an empty monitor queue both return None; diagnostics are available through diagnostics and pop_diagnostics().

Return type:

dict or None

pop_diagnostics() → list[dict]

Remove and return all diagnostics collected since the last call.

Returns:

Collected monitor diagnostics in arrival order.

Return type:

list of dict

start_exposure(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.

start_phase(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.

stop_exposure() → None

Stop the active exposure; repeated calls are harmless.

class mpylab.env.eut_monitor.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_monitor.OperatorEUTMonitor

alias of ManualEUTMonitor

class mpylab.env.eut_monitor.RandomEUTMonitor(*, seed=None, status_weights=None, recovery_weights=None, delay=0.0, source='random_simulation', operating_mode_changed=False, stored_data_lost=False)

Bases: EUTMonitor

Generate reproducible random EUT observations without hardware.

This class is both a simulation helper and a compact template for custom monitors. Its poll_event() method never blocks. Replace _make_exposure_event() with a camera, communication, or process-data check when adapting the class to real EUT monitoring.

Parameters:

  • seed – Seed used by an instance-local random-number generator. Equal seeds and configurations produce equal event sequences.

  • status_weights – Relative weights for passed, degraded, failed, and not_evaluated observations during exposure.

  • recovery_weights – Relative weights for automatic, operator, reset, and failed recovery after a degraded or failed observation.

  • delay – Non-negative delay in seconds before an event becomes available.

  • source – Source label stored in every generated event.

  • operating_mode_changed – Simulated observations needed for performance criterion B. Both default to False.

  • stored_data_lost – Simulated observations needed for performance criterion B. Both default to False.

Notes

This simulation must not be used as evidence of real EUT performance. The mandatory manual monitor remains active when this monitor is passed to Measure_Immunity as an automatic eut_monitor.

poll_event()

Return the scheduled random event after its configured delay.

Returns:

Scheduled event, or None while no event is due.

Return type:

dict or None

start_exposure(context)

Generate and schedule a random event for a new exposure.

Parameters:

context (mapping) – Measurement context describing the exposure.

start_phase(phase, context)

Generate a post-exposure event when that phase begins.

Parameters:

  • phase (str) – Phase from EUT_EVENT_PHASES.

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

stop_exposure()

Discard any random event still pending for the exposure.

class mpylab.env.eut_monitor.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.

class mpylab.env.eut_monitor.VirtualEUTMonitor(events=())

Bases: EUTMonitor

Emit a configurable sequence of events without EUT hardware.

Each sequence item belongs to one exposure. A normal event dictionary is assigned to during_exposure for compatibility. A phase mapping may instead provide separate during_exposure, post_exposure, and recovery events. None emits no event in that phase. Event dictionaries accept the normal make_eut_event() fields plus an optional non-negative delay in seconds.

Parameters:

events (iterable of mapping or None, optional) – Per-exposure virtual event specifications.

poll_event()

Return the active virtual event once its delay has elapsed.

Returns:

Configured event, or None while no event is due.

Return type:

dict or None

start_exposure(context)

Activate the next configured virtual exposure specification.

Parameters:

context (mapping) – Measurement context describing the exposure.

start_phase(phase, context)

Activate the configured event for the requested exposure phase.

Parameters:

  • phase (str) – Phase from EUT_EVENT_PHASES.

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

stop_exposure()

Discard the active virtual exposure and any pending event.

mpylab.env.eut_monitor.evaluate_performance_criterion(events, criterion='A')

Evaluate structured EUT observations against criterion A, B, or C.

A diagnostic retry never removes a known degradation or failure from a criterion-A result. An earlier not_evaluated attempt may, however, be superseded by a later evaluated attempt because it contains no functional observation. Criteria B and C require an explicit normal post-exposure state whenever an impairment was observed.

Parameters:

  • events (iterable of mapping) – Structured EUT observations, including any repeated attempts.

  • criterion (str, optional) – EMC performance criterion "A", "B", or "C".

Returns:

Evaluation result with status, reasons, and observation metadata.

Return type:

dict

mpylab.env.eut_monitor.make_eut_event(status, reason, details=None, *, timestamp=None, action='continue', phase='during_exposure', functional_state=None, after_exposure_state=None, recovery=None, operating_mode_changed=None, stored_data_lost=None, safety_action=None, source=None, event_type='status')

Return a validated, pickle-friendly EUT status event.

action is advisory. The measurement kernel remains responsible for all RF safety actions and for deciding whether a run continues or stops. safety_action is independent of that measurement-flow recommendation. It defaults to "rf_off" for failures and unevaluable states, and to "none" for passed or merely degraded observations.

Parameters:

  • status (str) – Observation status from EUT_EVENT_STATUSES.

  • reason (str) – Machine-readable reason for the observation.

  • details (object, optional) – Pickle-friendly supplemental information.

  • timestamp (float, optional) – Unix timestamp. The current time is used when omitted.

  • action (str, optional) – Advisory measurement-flow action.

  • phase (str, optional) – Exposure phase to which the observation belongs.

  • functional_state (str, optional) – Observed functional state. It is inferred from status by default.

  • after_exposure_state (str, optional) – Functional state observed after RF exposure.

  • recovery (str, optional) – Recovery mechanism observed after exposure.

  • operating_mode_changed (bool, optional) – Whether the EUT operating mode changed unexpectedly.

  • stored_data_lost (bool, optional) – Whether stored EUT data was lost.

  • safety_action (str, optional) – Requested safety action, inferred from status when omitted.

  • source (str, optional) – Identifier of the monitor that produced the event.

  • event_type (str, optional) – Event category from EUT_EVENT_TYPES.

Returns:

Normalized, validated EUT event.

Return type:

dict

mpylab.env.eut_monitor.validate_eut_event(event)

Validate and normalize an event returned by an EUT monitor.

Parameters:

event (mapping) – Event fields accepted by make_eut_event().

Returns:

Normalized EUT event.

Return type:

dict

mpylab.env.eut_monitor.validate_eut_event_policy(policy=None)

Return a complete measurement-level EUT event policy.

Parameters:

policy (mapping, optional) – Overrides for DEFAULT_EUT_EVENT_POLICY.

Returns:

Complete normalized policy.

Return type:

dict

mpylab.env.eut_monitor.validate_performance_criterion(criterion='A')

Return a normalized EMC immunity performance criterion.

Parameters:

criterion (str) – Performance criterion "A", "B", or "C". Input is normalized to uppercase.

Returns:

Normalized criterion.

Return type:

str