Measurement lifecycle

The Measure base class owns the common lifecycle of a measurement: logging, user interaction, autosave and resume, device initialization, RF-off, finalization, and pickle restoration. Measurement environments such as MSC, TEMCell, and AmplifierTest derive from it.

Persistence helpers

mpylab.env.Measure.dump_pickle(obj, pfile, protocol=4)

Serialize an mpylab object to an open binary file.

Parameters:

  • obj – Object to serialize.

  • pfile – Writable binary file-like object.

  • protocol – Pickle protocol to use. The default is the stable protocol 4, which is smaller and faster for measurement data than the legacy protocol 2 while remaining supported by all mpylab Python versions.

The cyclic garbage collector is paused during serialization because large evaluated measurement objects otherwise trigger costly collections. Its previous state is always restored. Reference counting remains active.

mpylab.env.Measure.load_pickle_compat(fname)

Load current and legacy mpylab pickle files.

The compatibility path supports old pickle files that still reference the former mpy package name and Python 2 style byte strings. Gzip-compressed files may use .gz or the historical .zip suffix; the latter does not denote a standard ZIP archive.

Parameters:

fname (path-like) – Pickle or gzip-compressed pickle file to load.

Returns:

Deserialized object graph.

Return type:

object

Only load trusted pickle files. Loading a Python pickle can execute code.

Base class

class mpylab.env.Measure.Measure(SearchPaths=None)

Base class for measurements.

Parameters:

SearchPaths (iterable of path-like, optional) – Directories searched for measurement configuration files. The current working directory is used by default.

_finalize_measurement_devices(mg, do_rfoff=True, do_quit=True)

Finalize measurement devices in a fail-safe way.

Returns:

last device status seen (0 by default).

Return type:

int

_init_measurement_devices(mg, do_zero=False, do_rfoff=False)

Initialize measurement devices with optional safe defaults.

Returns:

status from init path (0 means success).

Return type:

int

do_autosave(name_or_obj=None, depth=None, prefixes=None)

Serialize the measurement state using pickle.

Assuming a calling sequence like so:

script -> method of measurement class -> do_autosave

depth=1 stores the command issued in the script as self.ascmd. If the requested depth is too large, the outermost command is used.

Measurement methods may additionally set self.autosave_resume to a structured restart description. ascmd remains available as the human-readable and legacy restart instruction.

Parameters:

  • name_or_obj (str or binary file-like, optional) – Path-like destination or writable binary stream. self.asname is used by default. Path destinations are replaced atomically; caller-owned streams remain open.

  • depth (int, optional) – Number of caller frames used to derive ascmd.

  • prefixes (iterable of str, optional) – Command prefixes considered when deriving the restart command.

set_autosave(name)

Set the autosave filename used by do_autosave().

Parameters:

name (path-like or None) – Autosave destination, or None to disable file output.

set_autosave_interval(interval)

Set the minimum interval between automatic saves.

Parameters:

interval (float) – Minimum interval in seconds.

set_autosave_resume(measurement, method, description, parameter_source='current_configuration')

Store a structured restart instruction alongside ascmd.

Parameters:

  • measurement (str) – Stable measurement-family identifier.

  • method (str) – Method used to continue the measurement.

  • description (str) – Data-set description to resume.

  • parameter_source (str, optional) – Source from which restart parameters must be reconstructed.

Returns:

Pickle-friendly structured restart instruction.

Return type:

dict

wait(delay, dct, uitester, intervall=0.1)

Wait while repeatedly polling an interrupt callback.

Parameters:

  • delay (float) – Number of seconds to wait.

  • dct (mapping) – Namespace passed to legacy callbacks accepting one argument.

  • uitester (callable) – Interrupt callback accepting either dct or no arguments.

  • intervall (float, optional) – Delay in seconds between callback invocations.

The two underscored device-lifecycle methods are documented because derived measurement classes are expected to use them. Other underscored members remain implementation details.

TEM uniform-area collections

The TEM verification and immunity starters accept a named uniform_areas mapping and register all entries in the TEMCell history. A single TEMCell.Measure_Verification call continues to select exactly one stored name or one complete area definition through its uniform_area parameter. The bilingual German and English guides describe consecutive verification datasets, later immunity selection, and interpolation constraints.

class mpylab.env.Measure.AmplifierProtectionError(message)

Report that a requested RF level violates amplifier protection.

Parameters:

message (str) – Human-readable protection failure details.

EUT monitoring and immunity results

The German and English guides describe event semantics, performance criteria, disturbance-neutral records, and the optional JSON interchange document. The task-oriented implementation guides are available in German and English.

mpylab.env.eut.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.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

class mpylab.env.eut.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 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.ManualEUTMonitor(poll_key=None, keylist='sS', stop_keys='qQ')

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

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

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.EUTMonitoringSession(monitor: EUTMonitor)

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.ThreadedEUTMonitor(monitor, *, poll_interval=0.01, join_timeout=1.0)

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.VirtualEUTMonitor(events=())

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.

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

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.

mpylab.env.immunity_result.build_immunity_document(*, application, eut_description, performance_criterion, records, created_at)

Build a disturbance-neutral, JSON-compatible immunity document.

Parameters:

  • application (str) – Name of the application or measurement workflow.

  • eut_description (str) – Description of the equipment under test.

  • performance_criterion (str) – Performance criterion used to assess the EUT.

  • records (iterable) – Immunity result records to include.

  • created_at (str) – Timestamp associated with the document.

Returns:

JSON-compatible document following the mpylab immunity schema.

Return type:

dict

mpylab.env.immunity_result.write_immunity_document(path, document)

Write an immunity document.

Parameters:

  • path (path-like) – Destination filename.

  • document (mapping) – JSON-compatible immunity document.

Returns:

Resolved path of the written file.

Return type:

pathlib.Path

mpylab.env.immunity_result.load_immunity_document(path)

Load and minimally validate a structured immunity-result document.

Parameters:

path (path-like) – Input filename.

Returns:

Validated immunity document.

Return type:

dict

Raises:

ValueError – If the schema, schema version, or records collection is invalid.