import os
import numpy as np
from scuq import si,quantities
from mpylab.tools import util, mgraph
from mpylab.tools.am_headroom import am_headroom_factor
from mpylab.env.tem.probe_orientation import (
CELL_AXES,
COMPONENT_MAGNITUDES,
format_probe_orientation_preflight,
map_probe_field_vector,
parse_probe_axis_map,
parse_probe_orientation,
resolve_probe_orientation,
)
from mpylab.env.univers.AmplifierTest import dBm2W
from mpylab.env.Measure import Measure
[docs]
class TestSusceptibility(Measure):
[docs]
def __init__(self, parent=None):
Measure.__init__(self, parent)
[docs]
def Init(self, names=None,
datafunc=None,
pin=None,
dwell_time=None,
e_target=None,
dotfile=None,
SearchPath=None,
leveler_par=None,
adjust_to_setting=None,
probe_orientations=None):
"""Configure the susceptibility measurement.
Raw field-probe readings are mapped once from probe coordinates into
the final ``(Ex, Ey, Ez)`` cell-coordinate order. Component selection
and custom ``datafunc`` callbacks operate on that mapped vector.
Parameters
----------
names : mapping, optional
Measurement-graph role to node-name mapping.
datafunc : callable, optional
Custom observer function receiving an already ordered three-axis
field vector. By default, ``adjust_to_setting`` selects the value.
pin : iterable of float, optional
Initial signal-generator levels in dBm.
dwell_time : float, optional
Exposure time in seconds; defaults to one second.
e_target : float, optional
Requested carrier field strength in volts per metre.
dotfile : path-like, optional
Measurement-graph DOT file.
SearchPath : iterable of path-like, optional
Search paths used to resolve graph configuration files.
leveler_par : mapping, optional
Explicit mpylab leveler parameters.
adjust_to_setting : {"x", "y", "z", "mag", "largest"}, optional
Field value used for leveling. Missing and legacy ``"auto"``
values select ``"y"``.
probe_orientations : mapping, optional
Direct or per-probe orientation configuration. Graph-node and
field-probe INI metadata are used when this is omitted.
"""
if names is None:
self.names = {
'sg': 'sg',
'a1': 'amp1',
'a2': 'amp2',
'fp': 'prb',
'tem': 'gtem'
}
else:
self.names = names
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.last_leveling_result = None
self.last_am_validation_metadata = {}
def __datafunc(data):
if self.adjust_to_setting == 'x':
return data[0] # x-coordinate
elif self.adjust_to_setting == 'y':
return data[1] # y-coordinate
elif self.adjust_to_setting == 'z':
return data[2] # z-coordinate
elif self.adjust_to_setting == 'largest':
return max(data)
elif self.adjust_to_setting == 'mag':
return np.linalg.norm(data)
if datafunc is None:
self.datafunc = __datafunc
else:
self.datafunc = datafunc
if pin is None:
self.pin = [quantities.Quantity(si.WATT, dBm2W(_dBm)) for _dBm in (-40, -30, -25)]
else:
self.pin = [quantities.Quantity(si.WATT, dBm2W(_dBm)) for _dBm in pin]
if dwell_time is None:
self.dwell_time = 1
else:
self.dwell_time = dwell_time
if e_target is None:
self.e_target = quantities.Quantity(si.VOLT / si.METER, 1)
else:
self.e_target = quantities.Quantity(si.VOLT / si.METER, e_target)
if dotfile is None:
self.dotfile = 'gtem.dot'
else:
self.dotfile = dotfile
if SearchPath is None:
self.SearchPath = ['.', os.path.abspath('conf')]
else:
self.SearchPath = SearchPath
self.mg = mgraph.MGraph(
self.dotfile,
themap=self.names.copy(),
SearchPaths=self.SearchPath,
condition_map=mgraph.FREQUENCY_CONDITION_MAP,
)
probe_name = self.mg.name.fp
resolved_orientation = resolve_probe_orientation(
self.mg,
probe_name,
config_probe_orientations=probe_orientations,
)
self.probe_orientation = resolved_orientation['orientation']
self.probe_orientation_source = resolved_orientation['source']
self.probe_orientation_candidates = resolved_orientation['candidates']
self.probe_data_kind = COMPONENT_MAGNITUDES
parsed_orientation = parse_probe_orientation(self.probe_orientation)
if parsed_orientation['axis_map'] is None:
raise ValueError(
"TEMField field-probe readings are component magnitudes; "
"they support signed axis permutations but not arbitrary "
"rotation matrices"
)
self.probe_orientation_preflight = format_probe_orientation_preflight(
probe_name,
self.probe_orientation,
source=self.probe_orientation_source,
)
self.probe_orientation_preflight += (
"\n data kind: component_magnitudes (axis signs ignored)"
)
if leveler_par is None:
self.leveler_par = {'mg': self.mg,
'actor': self.mg.name.sg,
'output': self.mg.name.tem,
'lpoint': self.mg.name.tem,
'observer': self.mg.name.fp,
'pin': self.pin,
'datafunc': self._probe_datafunc,
'min_actor': None}
else:
self.leveler_par = leveler_par
#self.ddict = self.mg.CreateDevices()
return 0
def _map_probe_field(self, data):
"""Map one raw probe reading into TEM-cell axis order."""
mapped = map_probe_field_vector(
data,
getattr(self, 'probe_orientation', None),
data_kind=COMPONENT_MAGNITUDES,
)
return [mapped[cell_axis] for cell_axis in CELL_AXES]
def _probe_datafunc(self, data):
"""Map a raw probe vector once, then select its cell-coordinate value."""
return self.datafunc(self._map_probe_field(data))
def _map_probe_waveform(self, ex, ey, ez):
"""Map component-magnitude waveforms into TEM-cell axis order."""
axis_map = parse_probe_orientation(
getattr(self, 'probe_orientation', None)
)['axis_map']
parsed = parse_probe_axis_map(axis_map)
probe_values = (ex, ey, ez)
return tuple(
probe_values[parsed[cell_axis]['probe_index']]
for cell_axis in CELL_AXES
)
[docs]
def init_measurement(self, am):
"""Initialize devices and AM configuration while keeping RF off."""
self.am = am
err = self.mg.CreateDevices()
err = self.mg.Init_Devices()
stat = self.mg.Zero_Devices()
#stat = self.mg.CmdDevices(True, 'ConfAM', {'source': 'INT1',
# 'freq': 1e3,
# 'depth': am,
# 'waveform': 'SINE',
# 'LFOut': 'OFF'})
stat = self.mg.CmdDevices(True, 'ConfAM', 'INT1',1e3,am*1e-2,'SINE','OFF')
stat = self.mg.RFOff_Devices()
return stat
[docs]
def reset_to_safe_actor_level(self):
"""Reset the generator to its lowest configured level with RF off."""
return self.set_level_protected(
self.mg,
self.pin[0],
output=self.mg.name.tem,
actor=self.mg.name.sg,
reason='TEMField RF-off frequency-transition reset',
)
[docs]
def rf_on(self):
try:
stat = self.mg.RFOn_Devices()
if stat == 0:
return True
else:
return False
except AttributeError:
return False
[docs]
def rf_off(self):
try:
stat = self.mg.RFOff_Devices()
if stat == 0:
return True
else:
return False
except AttributeError:
return False
[docs]
def am_on(self):
try:
stat = self.mg.CmdDevices(True, 'AMOn')
if stat == 0:
return True
else:
return False
except AttributeError:
return False
[docs]
def am_off(self):
try:
stat = self.mg.CmdDevices(True, 'AMOff')
if stat == 0:
return True
else:
return False
except AttributeError:
return False
[docs]
def adjust_level(self, target_field=None):
"""Level to the target field and return the full probe field vector.
The structured :class:`mpylab.tools.mgraph.LevelingResult` is retained
in :attr:`last_leveling_result`. Callers must only expose the EUT to
the test field when that result has status ``converged``.
"""
if target_field is None:
target_field = self.e_target
leveler = mgraph.Leveler(**self.leveler_par)
leveler.adjust_level(target_field)
self.last_leveling_result = getattr(leveler, 'last_result', None)
if self.last_leveling_result is None:
raise RuntimeError(
"The installed mpylab version does not provide a structured "
"leveling result. Install a newer mpylab version before running "
"a TEMField measurement."
)
res = self.mg.Read([self.mg.name.fp])
return self._map_probe_field(res[self.mg.name.fp])
def _read_probe_field(self):
"""Read the probe and return mapped ``(cell_x, cell_y, cell_z)`` data."""
result = self.mg.Read([self.mg.name.fp])
return self._map_probe_field(result[self.mg.name.fp])
[docs]
def set_am_ramp_level(self, actor_level):
"""Apply one protected signal-generator level during the AM ramp."""
return self.set_level_protected(
self.mg,
actor_level,
output=self.mg.name.tem,
actor=self.mg.name.sg,
reason='TEMField in-loop AM waveform ramp',
)
[docs]
def read_field(self):
"""Return the current mapped three-axis field-probe reading."""
return self._read_probe_field()
[docs]
def leveling_succeeded(self):
"""Return whether the most recent leveling operation converged."""
return (
self.last_leveling_result is not None
and self.last_leveling_result.status == 'converged'
)
[docs]
def do_measurement(self, f):
raise RuntimeError(
"do_measurement cannot perform the safety-monitored in-loop AM "
"ramp; use TEMFieldWorker"
)
[docs]
def quit_measurement(self):
try:
stat = self.mg.RFOff_Devices()
stat = self.mg.Quit_Devices()
except AttributeError:
pass
def __HandleUserInterrupt(self, dct, ignorelist='', handler=None):
if callable(handler):
return handler(dct, ignorelist=ignorelist)
else:
return self.stdUserInterruptHandler(dct, ignorelist=ignorelist)
[docs]
def stdUserInterruptHandler(self, dct, ignorelist=''):
key = self.UserInterruptTester()
if key and not chr(key) in ignorelist:
# empty key buffer
_k = self.UserInterruptTester()
while _k is not None:
_k = self.UserInterruptTester()
mg = self.mg
names = self.names
dwell_time = self.dwell_time
self.messenger(util.tstamp() + " RF Off...", [])
stat = mg.RFOff_Devices() # switch off after measure
msg1 = """The measurement has been interrupted by the user.\nHow do you want to proceed?\n\nContinue: go ahead...\nSuspend: Quit devices, go ahead later after reinit...\nInteractive: Go to interactive mode...\nQuit: Quit measurement..."""
but1 = ['Continue', 'Quit']
answer = self.messenger(msg1, but1)
# print answer
if answer == but1.index('Quit'):
self.messenger(util.tstamp() + " measurment terminated by user.", [])
raise UserWarning # to reach finally statement
self.messenger(util.tstamp() + " RF On...", [])
stat = mg.RFOn_Devices() # switch on just before measure