# This Python file uses the following encoding: utf-8
import math
import time
import traceback
import numpy as np
from PySide6.QtCore import QObject, QMutex, QMutexLocker, QWaitCondition, Signal, Slot
from mpylab.env.eut import (
EUTMonitor,
EUTMonitoringSession,
ManualEUTMonitor,
ThreadedEUTMonitor,
evaluate_performance_criterion,
make_eut_event,
validate_eut_event_policy,
validate_performance_criterion,
)
from scuq import quantities, si
from .waveform_validation import (
AMWaveformLimits,
InLoopAMValidationResult,
analyse_am_waveform,
validate_am_fit,
validate_carrier_level,
validate_waveform_stability,
)
try:
from .TestSusceptibility import TestSusceptibility
except ImportError:
from TestSusceptibility import TestSusceptibility
[docs]
class TEMFieldWorker(QObject):
finished = Signal()
error = Signal(str)
log = Signal(str, object)
test_progress = Signal(int)
eut_progress = Signal(int)
frequency_done = Signal(object)
waveform = Signal(object)
rf_state_changed = Signal(bool)
am_state_changed = Signal(bool)
eut_phase_changed = Signal(str)
eut_event = Signal(object)
[docs]
def __init__(self, *, dwell_time, e_target, names, dotfile, searchpath,
adjust_to_setting, am, freqs, eut_monitor=None,
manual_eut_monitor=None, performance_criterion="A",
eut_event_policy=None, post_exposure_timeout=10.0,
probe_orientations=None,
am_waveform_limits=None, am_ramp_step_db=1.0,
am_ramp_settle_time=0.05):
"""Create a measurement worker.
Raw field-probe components are mapped once into TEM-cell coordinates
by the measurement layer before the worker evaluates or exports them.
Parameters
----------
dwell_time : float
Exposure time per frequency in seconds.
e_target : float
Requested carrier field strength in volts per metre.
names : mapping
Measurement-graph role to node-name mapping.
dotfile : path-like
Measurement-graph DOT file.
searchpath : iterable of path-like
Search paths used to resolve graph configuration files.
adjust_to_setting : {"x", "y", "z", "mag", "largest"}
Field value used for leveling and waveform analysis.
am : float
Sinusoidal AM depth in percent.
freqs : iterable of float
RF frequencies in hertz.
eut_monitor : mpylab.env.eut.EUTMonitor, optional
Automatic EUT monitor.
manual_eut_monitor : mpylab.env.eut.ManualEUTMonitor, optional
Operator-driven EUT monitor.
performance_criterion : str, optional
IEC 61000-4-20 EUT performance criterion.
eut_event_policy : mapping, optional
Actions for reported EUT event states.
post_exposure_timeout : float, optional
Maximum post-exposure monitoring time in seconds.
probe_orientations : mapping, optional
Direct or per-probe orientation checked against graph metadata.
am_waveform_limits : AMWaveformLimits or mapping, optional
Model-based compression, signal-quality, field, and stability
limits for the in-loop AM waveform validation.
am_ramp_step_db : float, optional
Maximum signal-generator power step during the AM ramp.
am_ramp_settle_time : float, optional
Settling time before each waveform acquisition, in seconds.
"""
super().__init__()
self.dwell_time = dwell_time
self.e_target = e_target
self.names = names
self.dotfile = dotfile
self.searchpath = searchpath
if adjust_to_setting in (None, "auto"):
adjust_to_setting = "y"
if adjust_to_setting not in ("x", "y", "z", "mag", "largest"):
raise ValueError(
"invalid field component selection: %r" % adjust_to_setting
)
self.adjust_to_setting = adjust_to_setting
self.am = am
self.freqs = list(freqs)
self.probe_orientations = probe_orientations
if am_waveform_limits is None:
am_waveform_limits = AMWaveformLimits()
elif isinstance(am_waveform_limits, dict):
am_waveform_limits = AMWaveformLimits(**am_waveform_limits)
if not isinstance(am_waveform_limits, AMWaveformLimits):
raise TypeError(
"am_waveform_limits must be AMWaveformLimits or a mapping"
)
self.am_waveform_limits = am_waveform_limits
self.am_ramp_step_db = float(am_ramp_step_db)
self.am_ramp_settle_time = float(am_ramp_settle_time)
if not math.isfinite(self.am_ramp_step_db) or self.am_ramp_step_db <= 0.0:
raise ValueError("am_ramp_step_db must be finite and positive")
if (
not math.isfinite(self.am_ramp_settle_time)
or self.am_ramp_settle_time < 0.0
):
raise ValueError(
"am_ramp_settle_time must be finite and non-negative"
)
if not 0.0 < float(self.am) < 100.0:
raise ValueError("am must be between 0 and 100 percent")
self._am_peak_factor = 1.0 + float(self.am) / 100.0
self.performance_criterion = validate_performance_criterion(
performance_criterion
)
self.eut_event_policy = validate_eut_event_policy(eut_event_policy)
if any(
self.eut_event_policy[status] == "retry"
for status in ("degraded", "failed", "not_evaluated")
):
raise ValueError("TEMField EUT event policies do not yet support retry")
self.post_exposure_timeout = float(post_exposure_timeout)
if (
not math.isfinite(self.post_exposure_timeout)
or self.post_exposure_timeout < 0.0
):
raise ValueError("post_exposure_timeout must be finite and non-negative")
if manual_eut_monitor is None:
manual_eut_monitor = ManualEUTMonitor()
if not isinstance(manual_eut_monitor, ManualEUTMonitor):
raise TypeError("manual_eut_monitor must be a ManualEUTMonitor")
self.manual_eut_monitor = manual_eut_monitor
automatic_monitors = self._prepare_automatic_monitors(eut_monitor)
self.eut_monitoring = EUTMonitoringSession.from_monitors(
self.manual_eut_monitor,
automatic_monitors,
)
self.eut_monitor = self.eut_monitoring.monitor
self.meas = None
self._devices_ready = False
self._stop = False
self._paused = False
self._rf_is_on = False
self._am_is_on = False
self._mutex = QMutex()
self._pause_cond = QWaitCondition()
[docs]
@Slot()
def run(self):
try:
self.meas = TestSusceptibility()
self.meas.Init(dwell_time=self.dwell_time,
e_target=self.e_target,
names=self.names,
dotfile=self.dotfile,
SearchPath=self.searchpath,
adjust_to_setting=self.adjust_to_setting,
probe_orientations=self.probe_orientations)
orientation_preflight = getattr(
self.meas, "probe_orientation_preflight", None
)
if orientation_preflight:
self.log.emit(
orientation_preflight,
"Field probe orientation: %s"
% getattr(
self.meas, "probe_orientation_source", "default"
),
)
self.meas.init_measurement(self.am)
self._devices_ready = True
self._rf_is_on = False
self.rf_state_changed.emit(False)
total = len(self.freqs)
for idx, f in enumerate(self.freqs, start=1):
if self._should_stop() or not self._wait_if_paused():
break
self.test_progress.emit(int(idx / total * 100) if total else 100)
self.log.emit(f"set freq to {f} MHz", f"Freq: {round(f * 1e-6, 2)} MHz")
self._set_am(False)
self._set_rf(False)
self.meas.mg.EvaluateConditions(context={"frequency": f})
self.meas.mg.SetFreq_Devices(f)
_safe_level, safe_protection = self.meas.reset_to_safe_actor_level()
if safe_protection.get("limited_by_amplifier_protection"):
raise RuntimeError(
"amplifier protection unexpectedly limited the RF-off "
"safe-level reset"
)
self._set_rf(True)
self.log.emit("adjust safe AM start level...", None)
e_field = self.meas.prepare_am_waveform_validation(self.am)
leveling_result = self.meas.last_leveling_result
leveling_data = leveling_result.as_dict()
short = (
f"Ex = {round(e_field[0].get_expectation_value_as_float(), 2)} V/m, "
f"Ey = {round(e_field[1].get_expectation_value_as_float(), 2)} V/m, "
f"Ez = {round(e_field[2].get_expectation_value_as_float(), 2)} V/m"
)
self.log.emit(f"E-Field: Ex = {e_field[0]}, Ey = {e_field[1]}, Ez = {e_field[2]},", short)
if leveling_result.status != "converged":
status = self._leveling_frequency_status(leveling_result.status)
self.log.emit(
self._leveling_failure_message(leveling_result),
status,
)
self._set_am(False)
self._set_rf(False)
self.frequency_done.emit({
"freq": f,
"e_field": e_field,
"status": status,
"leveling": leveling_data,
"headroom": None,
"am_validation": None,
"disturbance": self._disturbance_record(e_field),
})
continue
monitor_context, ramp_events, ramp_diagnostics = (
self._start_ramp_monitoring(f, e_field)
)
if self._set_am(True):
validation_result = self._run_am_waveform_validation(
f, ramp_events, ramp_diagnostics
)
else:
validation_result = InLoopAMValidationResult(
status="failed",
passed=False,
reason="AM could not be enabled at the safe start level",
factor=self._am_peak_factor,
target_field_v_per_m=float(self.e_target),
start_field_v_per_m=(
float(self.e_target) / self._am_peak_factor
),
limits=self.am_waveform_limits,
ramp_points=(),
stable_waveforms=(),
)
validation_data = validation_result.as_dict()
self._log_am_validation_diagnostics(f, validation_result)
if not validation_result.passed:
status = "AM waveform validation failed"
self._set_am(False)
self._set_rf(False)
self.eut_monitoring.stop_exposure()
self.eut_phase_changed.emit("idle")
self.frequency_done.emit({
"freq": f,
"e_field": e_field,
"status": status,
"leveling": leveling_data,
"headroom": validation_data,
"am_validation": validation_data,
"eut_events": ramp_events,
"eut_monitor_diagnostics": ramp_diagnostics,
"disturbance": self._disturbance_record(e_field),
})
continue
e_field = self.meas.read_field()
eut_result = self._run_eut_monitor(
f,
e_field,
monitoring_started=True,
initial_events=ramp_events,
initial_diagnostics=ramp_diagnostics,
context=monitor_context,
)
self.frequency_done.emit({
"freq": f,
"e_field": e_field,
"status": eut_result["status"],
"leveling": leveling_data,
"headroom": validation_data,
"am_validation": validation_data,
"eut_events": eut_result["events"],
"eut_monitor_diagnostics": eut_result["diagnostics"],
"performance_assessment": eut_result["assessment"],
"am_waveform_fit": eut_result["am_waveform_fit"],
"disturbance": self._disturbance_record(e_field),
})
if not eut_result["continue_measurement"]:
break
self._safe_finish()
if not self._should_stop():
self.log.emit("all frequencies processed", None)
self.test_progress.emit(100)
except Exception:
self.error.emit(traceback.format_exc())
self._safe_finish()
finally:
self.finished.emit()
@staticmethod
def _leveling_frequency_status(leveling_status):
if leveling_status == "protection_limited":
return "Leveling protection limit"
if leveling_status == "not_converged":
return "Leveling not converged"
return "Leveling failed"
def _disturbance_record(self, e_field):
"""Describe TEMField's disturbance using the generic result schema."""
target = getattr(self.meas, "e_target", self.e_target)
return {
"quantity_kind": "electric_field_strength",
"display_name": "Electric field strength",
"target": target,
"measured_components": {
axis: value
for axis, value in zip(("cell_x", "cell_y", "cell_z"), e_field)
},
"control_component": self.adjust_to_setting,
"coordinate_system": "tem_cell",
"probe_orientation": getattr(
self.meas, "probe_orientation", None
),
"probe_orientation_source": getattr(
self.meas, "probe_orientation_source", "default"
),
"probe_orientation_candidates": getattr(
self.meas, "probe_orientation_candidates", ()
),
"probe_data_kind": getattr(
self.meas, "probe_data_kind", "component_magnitudes"
),
}
@staticmethod
def _leveling_failure_message(result):
return (
"Leveling did not reach the target: status=%s, reason=%s, "
"target=%s, actual=%s, actor level=%s, relative error=%.6g"
% (
result.status,
result.reason,
result.target_value,
result.observed_value,
result.applied_level,
result.relative_error,
)
)
def _log_am_validation_diagnostics(self, frequency, result):
"""Write the in-loop waveform decision and model values to the log."""
final_fit = (
None if not result.stable_waveforms else result.stable_waveforms[-1]
)
message = (
"AM waveform validation: frequency=%.9g Hz, method=%s, "
"normative=%s, ramp_strategy=%s, factor=%.6g, "
"start_field=%.6g V/m, "
"target_field=%.6g V/m, ramp_points=%d, stable_waveforms=%d, "
"carrier_correction_attempts=%d, "
"result=%s, reason=%s, final_fit=%s, limits=%s"
% (
frequency,
result.method,
result.normative,
result.ramp_strategy,
result.factor,
result.start_field_v_per_m,
result.target_field_v_per_m,
len(result.ramp_points),
len(result.stable_waveforms),
result.carrier_correction_attempts,
result.status,
result.reason,
final_fit,
result.limits.as_dict(),
)
)
self.log.emit(message, "AM waveform: %s" % result.status)
def _start_ramp_monitoring(self, frequency, e_field):
"""Start EUT monitoring before AM is enabled and ramped."""
context = {
"frequency": frequency,
"e_field": e_field,
"target_efield": self.e_target,
"modulation_depth_percent": self.am,
"rf_on": True,
}
events = []
diagnostics = []
self.eut_monitoring.start_exposure(context)
# The shared EUT contract intentionally has a small fixed phase set.
# The AM ramp is part of ``during_exposure`` there; the dedicated UI
# signal below still distinguishes it operationally.
self.eut_monitoring.start_phase("during_exposure", context)
self.eut_phase_changed.emit("am_ramp")
return context, events, diagnostics
def _ramp_safety_failure(self, fit):
"""Return an immediate peak-safety failure during the AM ramp.
Signal-quality and compression limits apply only at the final carrier
target. Applying normalized residual limits at the deliberately low
start field would turn the probe's approximately absolute noise floor
into a level-dependent and intermittent rejection criterion.
"""
limits = self.am_waveform_limits
expected_peak = float(self.e_target) * self._am_peak_factor
if fit["fitted_peak"] > expected_peak * (
1.0 + limits.maximum_peak_relative_tolerance
):
return "fitted AM peak exceeds the safety limit"
return None
@staticmethod
def _next_am_ramp_watt(
current_watt,
measured_field_v_per_m,
target_field_v_per_m,
maximum_watt,
maximum_step_db,
):
"""Return the next protected AM-ramp power.
The local estimate follows ``E proportional to sqrt(P)``. Both the
configured step size and the bounded correction ceiling derived from
the safe CW start remain hard upper bounds.
"""
current = float(current_watt)
measured = float(measured_field_v_per_m)
target = float(target_field_v_per_m)
maximum = float(maximum_watt)
step_ratio = 10.0 ** (float(maximum_step_db) / 10.0)
if not all(
math.isfinite(value) and value > 0.0
for value in (current, measured, target, maximum, step_ratio)
):
raise ValueError("adaptive AM-ramp inputs must be finite and positive")
estimated = current * (target / measured) ** 2
return min(maximum, current * step_ratio, estimated)
@staticmethod
def _maximum_am_ramp_watt(
start_watt, peak_factor, carrier_relative_tolerance
):
"""Return the bounded actor-power ceiling for the adaptive AM ramp.
``start_watt * peak_factor**2`` is the ideal power needed to move the
fitted AM carrier from ``E_target / peak_factor`` to ``E_target``.
The additional divisor permits only the correction that would move a
carrier at the lower accepted boundary back to the nominal target.
Measured carrier and fitted-peak limits remain the final field guards.
"""
tolerance = float(carrier_relative_tolerance)
if not 0.0 <= tolerance < 1.0:
raise ValueError(
"carrier_relative_tolerance must be less than one"
)
return (
float(start_watt)
* float(peak_factor) ** 2
/ (1.0 - tolerance) ** 2
)
def _run_am_waveform_validation(self, frequency, events, diagnostics):
"""Ramp AM from below and validate the delivered waveform."""
limits = self.am_waveform_limits
target = float(self.e_target)
start_target = target / self._am_peak_factor
start_level = self.meas.last_leveling_result.applied_level.reduce_to(si.WATT)
start_watt = start_level.get_expectation_value_as_float()
maximum_watt = self._maximum_am_ramp_watt(
start_watt,
self._am_peak_factor,
limits.carrier_relative_tolerance,
)
ramp_points = []
stable_fits = []
carrier_correction_waveforms = []
carrier_correction_attempts = 0
failure = None
requested_watt = start_watt
for index in range(64):
if index:
requested = quantities.Quantity(si.WATT, requested_watt)
applied, protection = self.meas.set_am_ramp_level(requested)
if protection.get("limited_by_amplifier_protection"):
failure = "amplifier protection limited the AM ramp"
break
else:
applied = start_level
protection = {"limited_by_amplifier_protection": False}
if self.am_ramp_settle_time:
time.sleep(self.am_ramp_settle_time)
event = self._poll_eut_monitor(events, diagnostics)
if event is not None and (
event["status"] in ("degraded", "failed")
or event["safety_action"] == "rf_off"
or self.eut_event_policy.get(event["status"]) == "stop"
):
failure = "EUT event stopped the AM ramp: %s" % event["reason"]
break
waveform = self._emit_waveform(
phase="am_ramp", log_fit=True, frequency=frequency
)
fit = waveform.get("fit")
if fit is None:
failure = "field-probe waveform is unavailable or cannot be fitted"
break
fit = dict(fit)
fit.update({
"requested_actor_level_w": requested_watt,
"applied_actor_level": applied,
"protection": protection,
})
ramp_points.append(fit)
failure = self._ramp_safety_failure(fit)
if failure is not None:
break
# Aim at the nominal carrier. The acceptance interval is a final
# decision band, not an acquisition threshold: starting the final
# samples directly at its lower edge makes ordinary probe noise
# alternate between pass and fail.
if fit["offset"] < target:
lower_target = target * (
1.0 - limits.carrier_relative_tolerance
)
applied_watt = applied.reduce_to(
si.WATT
).get_expectation_value_as_float()
next_watt = self._next_am_ramp_watt(
applied_watt,
fit["offset"],
target,
maximum_watt,
self.am_ramp_step_db,
)
if next_watt <= applied_watt * (1.0 + 1.0e-12):
if fit["offset"] < lower_target:
failure = (
"AM ramp reached its bounded maximum actor level "
"before the accepted carrier-field interval"
)
break
# At the bounded ceiling, a value inside the acceptance
# interval is eligible for the final mean decision even
# if noise keeps this individual ramp fit below nominal.
else:
requested_watt = next_watt
continue
# The target-reaching ramp sample remains a ramp diagnostic. Take
# a fresh, explicitly labelled set for the final decision and UI.
# One downward closed-loop correction is allowed when an otherwise
# valid final set lies just above the carrier acceptance interval.
while True:
stable_fits = []
while len(stable_fits) < limits.required_stable_waveforms:
if self.am_ramp_settle_time:
time.sleep(self.am_ramp_settle_time)
event = self._poll_eut_monitor(events, diagnostics)
if event is not None and (
event["status"] in ("degraded", "failed")
or event["safety_action"] == "rf_off"
or self.eut_event_policy.get(event["status"]) == "stop"
):
failure = (
"EUT event stopped final AM validation: %s"
% event["reason"]
)
break
repeated = self._emit_waveform(
phase="am_validation", log_fit=True, frequency=frequency
).get("fit")
if repeated is None:
failure = "final field-probe waveform is unavailable"
break
repeated = dict(repeated)
repeated.update({
"requested_actor_level_w": requested_watt,
"applied_actor_level": applied,
"protection": protection,
})
stable_fits.append(repeated)
failure = self._ramp_safety_failure(repeated)
if failure is not None:
break
if failure is not None:
break
for stable_fit in stable_fits:
passed, reason = validate_am_fit(
stable_fit,
target,
self.am,
limits,
check_carrier=False,
)
if not passed:
failure = reason
break
if failure is not None:
break
passed, reason = validate_waveform_stability(stable_fits, limits)
if not passed:
failure = reason
break
passed, reason = validate_carrier_level(
stable_fits, target, limits
)
if passed:
break
mean_carrier = float(np.mean([
fit["offset"] for fit in stable_fits
]))
upper_target = target * (
1.0 + limits.carrier_relative_tolerance
)
if (
mean_carrier <= upper_target
or carrier_correction_attempts >= 1
):
failure = reason
break
carrier_correction_waveforms.extend(stable_fits)
applied_watt = applied.reduce_to(
si.WATT
).get_expectation_value_as_float()
corrected_watt = applied_watt * (
target / mean_carrier
) ** 2
if (
not math.isfinite(corrected_watt)
or corrected_watt <= 0.0
or corrected_watt >= applied_watt
):
failure = "invalid downward AM carrier correction"
break
requested_watt = corrected_watt
requested = quantities.Quantity(si.WATT, requested_watt)
applied, protection = self.meas.set_am_ramp_level(requested)
carrier_correction_attempts += 1
self.log.emit(
"AM carrier correction: frequency=%.9g Hz, attempt=%d, "
"mean_carrier=%.9g V/m, target=%.9g V/m, "
"requested_actor_level=%.9g W, applied_actor_level=%s"
% (
frequency,
carrier_correction_attempts,
mean_carrier,
target,
requested_watt,
applied,
),
"AM carrier correction",
)
if protection.get("limited_by_amplifier_protection"):
failure = "amplifier protection limited the AM carrier correction"
break
break
else:
failure = "AM ramp exceeded its iteration limit"
if failure is None and not stable_fits:
failure = "AM ramp did not reach the carrier-field target from below"
passed = failure is None
return InLoopAMValidationResult(
status="passed" if passed else "failed",
passed=passed,
reason=(
"waveform satisfies the in-loop AM validation limits"
if passed else failure
),
factor=self._am_peak_factor,
target_field_v_per_m=target,
start_field_v_per_m=start_target,
limits=limits,
ramp_points=tuple(ramp_points),
stable_waveforms=tuple(stable_fits),
carrier_correction_attempts=carrier_correction_attempts,
carrier_correction_waveforms=tuple(
carrier_correction_waveforms
),
)
@staticmethod
def _prepare_automatic_monitors(eut_monitor):
if eut_monitor is None:
return ()
if isinstance(eut_monitor, EUTMonitor):
monitors = (eut_monitor,)
else:
monitors = tuple(eut_monitor)
prepared = []
for monitor in monitors:
if not isinstance(monitor, EUTMonitor):
raise TypeError("automatic EUT monitors must implement EUTMonitor")
prepared.append(
monitor
if isinstance(monitor, ThreadedEUTMonitor)
else ThreadedEUTMonitor(monitor)
)
return tuple(prepared)
def _poll_eut_monitor(self, events, diagnostics):
event = self.eut_monitoring.poll_event()
new_diagnostics = self.eut_monitoring.diagnostics[len(diagnostics):]
for diagnostic in new_diagnostics:
diagnostics.append(diagnostic)
self.eut_event.emit(diagnostic)
self.log.emit(
"Automatic EUT monitor diagnostic: %s" % diagnostic["details"],
"EUT monitor diagnostic",
)
if event is None:
return None
self.eut_event.emit(event)
events.append(event)
return event
def _run_eut_monitor(
self,
frequency,
e_field,
*,
monitoring_started=False,
initial_events=None,
initial_diagnostics=None,
context=None,
):
if context is None:
context = {
"frequency": frequency,
"e_field": e_field,
"target_efield": self.e_target,
"modulation_depth_percent": self.am,
}
events = list(initial_events or ())
diagnostics = list(initial_diagnostics or ())
impaired = False
continue_measurement = True
if not monitoring_started:
self.eut_monitoring.start_exposure(context)
self.eut_monitoring.start_phase("during_exposure", context)
self.eut_phase_changed.emit("during_exposure")
start = now = time.monotonic()
end = start + self.dwell_time
last_waveform = start
am_fit_logged = False
am_waveform_fit = None
try:
while now < end:
if self._should_stop() or not self._wait_if_paused():
return {
"continue_measurement": False,
"status": "Stopped",
"events": events,
"diagnostics": diagnostics,
"assessment": None,
"am_waveform_fit": am_waveform_fit,
}
event = self._poll_eut_monitor(events, diagnostics)
if event is not None and event["status"] in ("degraded", "failed"):
impaired = True
if event is not None and event["safety_action"] == "rf_off":
self._set_am(False)
self._set_rf(False)
break
time.sleep(0.01)
now = time.monotonic()
percentage = (
round((now - start) / self.dwell_time, 2) * 100
if self.dwell_time else 100
)
self.eut_progress.emit(min(100, int(percentage)))
if now - last_waveform > 0.2:
waveform = self._emit_waveform(
phase="am",
log_fit=not am_fit_logged,
frequency=frequency,
)
if not am_fit_logged:
am_waveform_fit = waveform.get("fit")
am_fit_logged = True
last_waveform = now
if not events:
events.append(make_eut_event(
"passed",
"dwell_completed",
source="temfield_worker",
))
self.eut_progress.emit(100)
self._set_am(False)
self._set_rf(False)
post_event = None
if impaired:
self.eut_monitoring.start_phase(
"post_exposure",
{**context, "rf_on": False},
)
self.eut_phase_changed.emit("post_exposure")
deadline = time.monotonic() + self.post_exposure_timeout
while time.monotonic() <= deadline:
post_event = self._poll_eut_monitor(events, diagnostics)
if post_event is not None:
break
if self.post_exposure_timeout == 0.0:
break
time.sleep(min(0.01, max(0.0, deadline - time.monotonic())))
if post_event is None:
post_event = make_eut_event(
"not_evaluated" if impaired else "passed",
"post_exposure_timeout" if impaired else "post_exposure_completed",
phase="post_exposure",
after_exposure_state="not_evaluated" if impaired else "normal",
recovery="not_evaluated" if impaired else "not_required",
safety_action="none",
source="temfield_worker",
)
events.append(post_event)
self.eut_event.emit(post_event)
assessment = evaluate_performance_criterion(
events,
self.performance_criterion,
)
for event in events:
if event["status"] == "passed":
continue
if self.eut_event_policy[event["status"]] == "stop":
continue_measurement = False
status = (
"Passed (criterion %s)" % self.performance_criterion
if assessment["passed"] is True
else "Failed (criterion %s)" % self.performance_criterion
if assessment["passed"] is False
else "Not evaluated (criterion %s)" % self.performance_criterion
)
return {
"continue_measurement": continue_measurement,
"status": status,
"events": events,
"diagnostics": diagnostics,
"assessment": assessment,
"am_waveform_fit": am_waveform_fit,
}
finally:
self.eut_phase_changed.emit("idle")
self.eut_monitoring.stop_exposure()
def _emit_waveform(self, *, phase="unknown", log_fit=False, frequency=None):
"""Acquire, label, optionally fit, and emit one field waveform.
Parameters
----------
phase : str, optional
Acquisition phase such as ``cw`` or ``am``.
log_fit : bool, optional
Whether to fit the configured field component and log its AM
metrics.
frequency : float or None, optional
RF frequency in hertz used in the diagnostic message. It is
required when ``log_fit`` is true.
Returns
-------
dict
Signal payload containing raw waveform arrays, phase, selected
component, and an optional ``fit`` mapping. An empty mapping is
returned when no measurement object exists.
"""
if self.meas is None:
return {}
err, ts, ex, ey, ez = self.meas.get_waveform()
data = {
"err": err,
"t": ts,
"ex": ex,
"ey": ey,
"ez": ez,
"phase": phase,
"coordinate_system": "tem_cell",
"probe_orientation": getattr(
self.meas, "probe_orientation", None
),
"probe_orientation_source": getattr(
self.meas, "probe_orientation_source", "default"
),
"probe_data_kind": getattr(
self.meas, "probe_data_kind", "component_magnitudes"
),
}
if log_fit and err >= 0:
try:
fit = self._analyse_waveform(ts, ex, ey, ez)
except (TypeError, ValueError, RuntimeError, IndexError) as exc:
self.log.emit(
"AM waveform fit failed at %.9g Hz: %s"
% (frequency, exc),
"AM waveform fit failed",
)
else:
data["fit"] = fit
self.log.emit(
"AM waveform fit: frequency=%.9g Hz, component=%s, "
"offset=%.6g V/m, amplitude=%.6g V/m, "
"modulation_depth=%.6g %%, modulation_frequency=%.6g kHz, "
"minimum=%.6g V/m, maximum=%.6g V/m, RMSE=%.6g V/m, "
"normalized_RMSE=%.6g, R_squared=%.6g, "
"Rapp_peak_compression=%.6g dB, Rapp_knee=%.6g, "
"Rapp_smoothness=%.6g, samples=%d, sample_interval=%.6g ms, "
"duration=%.6g ms"
% (
frequency,
fit["component"],
fit["offset"],
fit["amplitude"],
fit["modulation_depth_percent"],
fit["frequency"],
fit["minimum"],
fit["maximum"],
fit["rmse"],
fit["normalized_rmse"],
fit["r_squared"],
fit["rapp_peak_compression_db"],
fit["rapp_knee"],
fit["rapp_smoothness"],
fit["sample_count"],
fit["sample_interval_ms"],
fit["duration_ms"],
),
"AM depth: %.2f %%" % fit["modulation_depth_percent"],
)
self.waveform.emit(data)
return data
def _analyse_waveform(self, ts, ex, ey, ez):
"""Fit the configured field component and return diagnostic metrics.
Parameters
----------
ts : array-like
Uniform sample times in milliseconds.
ex, ey, ez : array-like
Electric-field component samples in volts per metre.
Returns
-------
dict
Selected component, sinusoidal offset, amplitude, modulation
depth, frequency in kilohertz, extrema, normalized residual,
R-squared, fitted Rapp parameters, differential peak compression,
sample count, sample interval, and duration.
Raises
------
ValueError
If arrays are too short, mismatched, or non-finite.
RuntimeError
If the nonlinear sinusoidal fit cannot converge.
"""
t = np.asarray(ts, dtype=float)
components = tuple(np.asarray(values, dtype=float) for values in (ex, ey, ez))
if t.ndim != 1 or len(t) < 5:
raise ValueError("waveform needs at least five one-dimensional samples")
if any(values.shape != t.shape for values in components):
raise ValueError("waveform time and field arrays must have equal lengths")
if not np.all(np.isfinite(t)) or not all(
np.all(np.isfinite(values)) for values in components
):
raise ValueError("waveform contains non-finite values")
setting = self.adjust_to_setting
if setting in ("x", "y", "z"):
index = ("x", "y", "z").index(setting)
values = components[index]
component = setting
elif setting == "mag":
values = np.sqrt(sum(axis * axis for axis in components))
component = "magnitude"
elif setting == "largest":
index = int(np.argmax([np.max(np.abs(axis)) for axis in components]))
values = components[index]
component = ("x", "y", "z")[index]
else:
raise ValueError("invalid field component selection: %r" % setting)
result = analyse_am_waveform(t, values, self.am)
result["component"] = component
return result
def _safe_finish(self):
try:
if self.meas is not None:
self._set_am(False)
self._set_rf(False)
try:
self.meas.mg.CmdDevices(False, "Standby")
except AttributeError:
pass
self.meas.mg.Quit_Devices()
self._devices_ready = False
except Exception as exc:
self.error.emit(f"Could not shut down devices cleanly: {exc}")
try:
self.eut_monitoring.close()
except RuntimeError as exc:
self.error.emit(f"Could not shut down EUT monitors cleanly: {exc}")
def _set_rf(self, state):
if self.meas is None or not self._devices_ready:
return False
try:
status = self.meas.rf_on() if state else self.meas.rf_off()
except Exception as exc:
self.error.emit(f"RF {'On' if state else 'Off'} failed: {exc}")
return False
if status is True:
self._rf_is_on = state
self.rf_state_changed.emit(state)
self.log.emit("RF On" if state else "RF Off", None)
return status
def _set_am(self, state):
if self.meas is None or not self._devices_ready:
return False
try:
status = self.meas.am_on() if state else self.meas.am_off()
except Exception as exc:
self.error.emit(f"AM {'On' if state else 'Off'} failed: {exc}")
return False
if status is True:
self._am_is_on = state
self.am_state_changed.emit(state)
self.log.emit("AM On" if state else "AM Off", None)
return status
def _should_stop(self):
with QMutexLocker(self._mutex):
return self._stop
def _wait_if_paused(self):
self._mutex.lock()
try:
while self._paused and not self._stop:
self._pause_cond.wait(self._mutex)
return not self._stop
finally:
self._mutex.unlock()
[docs]
@Slot()
def stop(self):
with QMutexLocker(self._mutex):
self._stop = True
self._paused = False
self._pause_cond.wakeAll()
self.rf_off()
[docs]
@Slot(result=bool)
def toggle_pause(self):
with QMutexLocker(self._mutex):
if self._stop:
return False
self._paused = not self._paused
paused = self._paused
if not paused:
self._pause_cond.wakeAll()
if paused:
self.rf_off()
else:
self.rf_on()
if self._am_is_on:
self.am_on()
return paused
[docs]
@Slot()
def rf_on(self):
self._set_rf(True)
[docs]
@Slot()
def rf_off(self):
self._set_rf(False)
[docs]
@Slot()
def am_on(self):
self._set_am(True)
[docs]
@Slot()
def am_off(self):
self._set_am(False)