# -*- coding: utf-8 -*-
"""This is :mod:`mpylab.env.Measure` with :class:`mpylab.env.Measure.Measure` being the base class for e.g. :class:`mpylab.env.msc.MSC.MSC`
:author: Hans Georg Krauthäuser (main author)
:license: GPL-3 or higher
"""
from typing import Any
import gc
import gzip
import inspect
import os
from pathlib import Path
import pickle
import sys
import tempfile
import time
import re
#try:
# import mpylab.tools.unixcrt as crt
#except ImportError:
# class CRT:
# def unbuffer_stdin(self):
# pass
#
# def restore_stdin(self):
# pass
#
# crt = CRT()
from mpylab.tools import util, calling, mgraph
from mpylab.tools.keyboard import KeyboardUnavailableError, anykeyevent, keypress
from mpylab.tools.logging_utils import LogError, tstamp
from mpylab.tools.runtime_introspection import get_var_from_nearest_outerframe, interactive
from mpylab.tools.sequences import flatten, issequence
from mpylab.tools.regular_expressions import FP
from mpylab.env.ui.ui_adapter import TUIAdapter
from mpylab.tools.uconv import UConv
from scuq.quantities import Quantity
from scuq.si import WATT
LEGACY_PICKLE_MODULE_MAP = {
"mpy.env.msc.MSC": "mpylab.env.msc.MSC",
"mpy.tools.mgraph": "mpylab.tools.mgraph",
}
MEASUREMENT_PICKLE_PROTOCOL = 4
def dump_pickle(obj, pfile, protocol=MEASUREMENT_PICKLE_PROTOCOL):
"""Serialize an mpylab object to an open binary file.
:param obj: Object to serialize.
:param pfile: Writable binary file-like object.
:param 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.
"""
gc_was_enabled = gc.isenabled()
if gc_was_enabled:
gc.disable()
try:
pickle.dump(obj, pfile, protocol=protocol)
finally:
if gc_was_enabled:
gc.enable()
class MpyLabCompatUnpickler(pickle.Unpickler):
"""Unpickler for legacy mpylab pickle files."""
def find_class(self, module, name):
"""Resolve renamed legacy modules before loading a pickled class.
Parameters
----------
module : str
Module name stored in the pickle.
name : str
Class or global name stored in the pickle.
Returns
-------
object
Resolved class or global object.
"""
module = LEGACY_PICKLE_MODULE_MAP.get(module, module)
return super().find_class(module, name)
def _open_pickle_file(fname, mode="rb"):
name = os.fspath(fname)
if name.endswith((".gz", ".zip")):
return gzip.open(name, mode)
return open(name, mode)
def _flush_pickle_file(pfile):
pfile.flush()
try:
os.fsync(pfile.fileno())
except (AttributeError, OSError):
pass
def _atomic_dump_pickle_path(obj, destination):
destination = Path(destination)
suffix = destination.suffix
fd, temporary_name = tempfile.mkstemp(
prefix=f".{destination.name}.",
suffix=suffix,
dir=destination.parent,
)
os.close(fd)
temporary_path = Path(temporary_name)
try:
with _open_pickle_file(temporary_path, "wb") as pfile:
dump_pickle(obj, pfile)
_flush_pickle_file(pfile)
os.replace(temporary_path, destination)
finally:
if temporary_path.exists():
temporary_path.unlink()
def 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
-------
object
Deserialized object graph.
"""
with _open_pickle_file(fname, "rb") as pfile:
return MpyLabCompatUnpickler(pfile, encoding="latin1").load()
def _restore_logfile_from_pickle(logfilename):
"""Open a usable logfile when restoring a measurement from pickle.
The stored logfile path may refer to another machine. In that case create
an explicit restore log in the current working directory, and only fall
back to a persistent temporary file if the current directory is not
writable.
"""
if not logfilename:
return None, None, None, "none", None
original_logfilename = os.fspath(logfilename)
restore_name = "restored-%s" % (Path(original_logfilename).name or "mpylab.log")
attempts = (
("original", Path(original_logfilename)),
("restored_cwd", Path(restore_name)),
)
for status, path in attempts:
try:
logfile = open(path, "a+")
except (OSError, TypeError, ValueError):
continue
actual_logfilename = str(path if status == "original" else path.resolve())
message = None
if status != "original":
message = (
"Original logfile %r could not be reopened. "
"Logging restored to %r."
) % (original_logfilename, actual_logfilename)
logfile.write(message + "\n")
logfile.flush()
return logfile, actual_logfilename, original_logfilename, status, message
try:
tmp = tempfile.NamedTemporaryFile(
mode="a+",
prefix="mpylab-restored-",
suffix=".log",
delete=False,
)
except (OSError, TypeError, ValueError) as exc:
message = (
"Original logfile %r could not be reopened and no restore logfile "
"could be created: %s"
) % (original_logfilename, exc)
return None, original_logfilename, original_logfilename, "unavailable", message
message = (
"Original logfile %r could not be reopened. "
"Logging restored to temporary file %r."
) % (original_logfilename, tmp.name)
tmp.write(message + "\n")
tmp.flush()
return tmp, tmp.name, original_logfilename, "temporary", message
try:
import pyttsx3
_tts = pyttsx3.init()
_tts.setProperty('volume', 1.0)
vs = _tts.getProperty('voices')
for v in vs:
if 'en_GB' in v.languages: # take first british speaker
_tts.setProperty('voice', v.id)
break
#import festival
#festival.execCommand("(voice_en1_mbrola)")
#_tts = festival
#_tts.say = _tts.sayText
#def __runAndWait():
# pass
#_tts.runAndWait = __runAndWait
except ImportError:
#festival = None
pyttsx3 = None
_tts = None
def parse_quantity(s: str):
"""Parse a simple quantity string representation.
Parameters
----------
s : str
Text in the form ``Quantity(UNIT, value)``.
Returns
-------
tuple of (str, float)
Unit name and numeric value.
"""
pattern = rf'Quantity\(\s*([A-Za-z]+)\s*,\s*({FP})\s*\)'
match = re.fullmatch(pattern, s)
if not match:
raise ValueError(f"Ungültiges Format: {s}")
unit = match.group(1)
value = float(match.group(2))
return unit, value
[docs]
class Measure(object):
"""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.
"""
[docs]
def __init__(self, SearchPaths=None):
"""constructor"""
if SearchPaths is None:
SearchPaths = [os.getcwd()]
self.SearchPaths = SearchPaths
self.asname = None
self.ascmd = None
self.autosave_resume = None
self.autosave = False
self.autosave_interval = 3600
self.lastautosave = time.time()
self.logger = [self.stdlogger]
self.logfile = None
self.logfilename = None
self._setup_ui_adapter()
def _setup_ui_adapter(self):
"""Create/recreate UI adapter and bind legacy interaction hooks."""
self.ui = TUIAdapter(
messenger=self.stdUserMessenger,
logger=self.logger,
interrupt_tester=self.stdUserInterruptTester,
pre_user_event=self.stdPreUserEvent,
post_user_event=self.stdPostUserEvent,
interactive_runner=self.stdInteractiveSession,
)
self.messenger = self.ui.ask
self.UserInterruptTester = self.ui.check_interrupt
self.PollKey = self.ui.poll_key
self.PreUserEvent = self.ui.pre_user_event
self.PostUserEvent = self.ui.post_user_event
[docs]
def set_autosave_resume(self, 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
-------
dict
Pickle-friendly structured restart instruction.
"""
self.autosave_resume = {
'schema_version': 1,
'measurement': measurement,
'method': method,
'description': description,
'parameter_source': parameter_source,
}
return self.autosave_resume
def __setstate__(self, dct):
"""used instead of __init__ when instance is created from pickle file"""
logfile, logfilename, original_logfilename, restore_status, restore_message = (
_restore_logfile_from_pickle(dct.get('logfilename'))
)
self.__dict__.update(dct)
self.asname = getattr(self, 'asname', None)
self.ascmd = getattr(self, 'ascmd', None)
self.autosave_resume = getattr(self, 'autosave_resume', None)
self.autosave = getattr(self, 'autosave', False)
self.autosave_interval = getattr(self, 'autosave_interval', 3600)
self.lastautosave = getattr(self, 'lastautosave', time.time())
self.logfilename = logfilename
self.original_logfilename = original_logfilename
self.logfile_restore_status = restore_status
self.logfile_restore_message = restore_message
self.logfile = logfile
self.logger = [self.stdlogger]
self._setup_ui_adapter()
def __getstate__(self):
"""prepare a dict for pickling"""
odict = self.__dict__.copy()
odict.pop('logfile', None)
odict.pop('logger', None)
odict.pop('messenger', None)
odict.pop('UserInterruptTester', None)
odict.pop('PollKey', None)
odict.pop('PreUserEvent', None)
odict.pop('PostUserEvent', None)
odict.pop('ui', None)
return odict
@staticmethod
def _invoke_wait_handler(handler, dct):
"""Call a wait/interrupt handler in a backward-compatible way.
Supported styles:
- ``handler(dct)`` (legacy flow handler)
- ``handler()`` (poll-key style)
"""
if not callable(handler):
return None
try:
sig = inspect.signature(handler)
params = list(sig.parameters.values())
has_var_positional = any(p.kind == inspect.Parameter.VAR_POSITIONAL for p in params)
positional = [
p for p in params
if p.kind in (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD)
]
if has_var_positional or positional:
return handler(dct)
return handler()
except (TypeError, ValueError):
try:
return handler(dct)
except TypeError:
return handler()
[docs]
def wait(self, 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.
"""
start = time.time()
delay = abs(delay)
intervall = abs(intervall)
while time.time() - start < delay:
self._invoke_wait_handler(uitester, dct)
time.sleep(intervall)
[docs]
def out(self, item):
"""Print a nested object recursively on one line.
Parameters
----------
item : object
Mapping, sequence, or scalar value to print.
"""
if hasattr(item, 'keys'): # a dict like object
print("{", end=' ')
for k in list(item.keys()):
print(str(k) + ":", end=' ')
self.out(item[k])
print("}", end=' ')
elif hasattr(item, 'append'): # a list like object
print("[", end=' ')
for i in item:
self.out(i)
print("]", end=' ')
elif issequence(item): # other sequence
print("(", end=' ')
for i in item:
self.out(i)
print(")", end=' ')
else:
print(item, end=' ')
[docs]
def set_autosave_interval(self, interval):
"""Set the minimum interval between automatic saves.
Parameters
----------
interval : float
Minimum interval in seconds.
"""
self.autosave_interval = interval
[docs]
def stdlogger(self, block, *args):
"""The standard method to write messages to log file.
Print *block* to `self.logfile` or to `stdout` (if `self.logfile` is `None`).
If *block* has attribute `keys` (i.e. is a :class:`dict`), the elements are
processed with the local function :meth:`out_block`. Else, the block is printed
directly.
Parameters
----------
block : object
Mapping or scalar message to log.
*args : object
Additional positional values retained for callback compatibility;
they are ignored by the standard logger.
"""
def out_block(b):
"""Helper function to log something.
"""
assert hasattr(b, 'keys'), "Argument b has to be a dict."
try:
print(repr(b['comment']), end=' ')
except KeyError:
pass
try:
par = b['parameter']
for des, p in par.items():
print(des, end=' ')
out_block(p)
try:
item = b['value']
except KeyError:
item = None
self.out(item)
except KeyError:
pass
stdout = sys.stdout # save stdout
if self.logfile is not None:
sys.stdout = self.logfile
try:
try:
for des, bd in block.items():
print(tstamp(), des, end=' ')
out_block(bd)
print() # New Line
except AttributeError:
print(block)
finally:
try:
sys.stdout.flush()
finally:
sys.stdout = stdout # restore stdout
[docs]
def stdUserMessenger(self,
msg: str = "Are you ready?",
but: list[str] | None = None,
level: str = '',
dct: dict[Any, Any] | None = None) -> int:
"""Present a message and optionally wait for a button selection.
Parameters
----------
msg : str, optional
Message shown to the operator.
but : list of str, optional
Button labels selectable by their initial character.
level : str, optional
Message category; ``"email"`` enables the legacy email path.
dct : mapping, optional
Supplemental message data, including legacy email fields.
Returns
-------
int
Selected button index, or ``-1`` when no buttons are supplied.
"""
if but is None:
but = ["Ok", "Quit"]
if dct is None:
dct = {}
print(msg)
self.ui.emit_log(msg, but, level, dct)
if level in ('email',):
try:
util.send_email(to=dct['to'], fr=dct['from'], subj=dct['subject'], msg=msg)
except (NameError, KeyError):
LogError(self.messenger)
if len(but): # button(s) are given -> wait
if _tts:
_tts.say(msg)
_tts.runAndWait()
self.PreUserEvent()
try:
while True:
try:
key = chr(keypress())
except KeyboardUnavailableError:
return self._std_user_messenger_line_prompt(but)
key = key.lower()
for s in but:
if s.lower().startswith(key):
if _tts:
_tts.say(s) # , pyTTS.tts_purge_before_speak)
_tts.runAndWait()
return but.index(s)
finally:
self.PostUserEvent()
else:
return -1
@staticmethod
def _std_user_messenger_line_prompt(buttons: list[str]) -> int:
"""Ask for a button choice using line input.
This is a fallback for debuggers and IDE consoles where stdin is not a
terminal and single-key POSIX input is unavailable.
"""
choices = ", ".join(f"{index}: {button}" for index, button in enumerate(buttons))
prompt = f"Select [{choices}]: "
while True:
try:
answer = input(prompt)
except EOFError as exc:
raise KeyboardUnavailableError(
"Cannot read user choice because stdin is not interactive"
) from exc
answer = answer.strip().lower()
if not answer:
continue
if answer.isdigit():
index = int(answer)
if 0 <= index < len(buttons):
return index
for index, button in enumerate(buttons):
if button.lower().startswith(answer[0]):
return index
print(f"Please enter one of: {choices}")
[docs]
@staticmethod
def stdUserInterruptTester() -> int | None:
"""Poll the standard keyboard source for a user interrupt.
Returns
-------
int or None
Key code from :func:`mpylab.tools.keyboard.anykeyevent`, or
``None`` when no key is available.
"""
return anykeyevent()
[docs]
def set_logfile(self, name):
"""Open or replace the append-only measurement log file.
Parameters
----------
name : path-like
Requested log-file path. The filename component is sanitized.
"""
import pathvalidate
# import unicodedata
# import string
# import re
# validFilenameChars = "-_.() %s%s" % (string.ascii_letters, string.digits)
#
# def slugify(value, allow_unicode=False):
# """
# Taken from https://github.com/django/django/blob/master/django/utils/text.py
# Convert to ASCII if 'allow_unicode' is False. Convert spaces or repeated
# dashes to single dashes. Remove characters that aren't alphanumerics,
# underscores, or hyphens. Convert to lowercase. Also strip leading and
# trailing whitespace, dashes, and underscores.
# """
# value = str(value)
# if allow_unicode:
# value = unicodedata.normalize('NFKC', value)
# else:
# value = unicodedata.normalize('NFKD', value).encode('ascii', 'ignore').decode('ascii')
# value = re.sub(r'[^\w\s-]', '', value.lower())
# return re.sub(r'[-\s]+', '-', value).strip('-_')
#
# def removeDisallowedFilenameChars(filename):
# try:
# cleanedFilename = unicodedata.normalize('NFKD', filename).encode('ASCII', 'ignore')
# except TypeError:
# cleanedFilename = unicodedata.normalize('NFKD', str(filename)).encode('ASCII', 'ignore')
# return ''.join(c for c in cleanedFilename if c in validFilenameChars)
log = None
# name = removeDisallowedFilenameChars(name)
# name = slugify(name)
log_path = Path(os.fspath(name))
sanitized_name = pathvalidate.sanitize_filename(log_path.name)
log_path = log_path.with_name(sanitized_name)
try:
if log_path.parent != Path("."):
log_path.parent.mkdir(parents=True, exist_ok=True)
log = open(log_path, "a+")
except IOError:
LogError(self.messenger)
else:
if self.logfile is not None:
try:
self.logfile.close()
except IOError:
LogError(self.messenger)
self.logfilename = str(log_path)
self.logfile = log
[docs]
def set_logger(self, logger=None):
"""Configure the callbacks receiving measurement log messages.
Parameters
----------
logger : callable or iterable of callable, optional
Logger callbacks. :meth:`stdlogger` is used by default.
"""
if logger is None:
logger = [self.stdlogger]
logger = flatten(logger) # ensure flat list
self.logger = [l for l in logger if callable(l)]
self.ui.set_logger(self.logger)
[docs]
def set_messenger(self, messenger):
"""Configure the callback used for user-facing messages.
Parameters
----------
messenger : callable
Messenger compatible with :meth:`stdUserMessenger`.
"""
if callable(messenger):
self.ui.set_messenger(messenger)
self.messenger = self.ui.ask
[docs]
def set_ui_adapter(self, adapter):
"""Replace the complete measurement UI adapter.
Parameters
----------
adapter : UIAdapter
Adapter providing messaging, logging, and interaction hooks.
"""
self.ui = adapter
self.messenger = self.ui.ask
self.UserInterruptTester = self.ui.check_interrupt
self.PollKey = self.ui.poll_key
self.PreUserEvent = self.ui.pre_user_event
self.PostUserEvent = self.ui.post_user_event
[docs]
def set_user_interrupt_tester(self, tester):
"""Configure the non-blocking user-interrupt callback.
Parameters
----------
tester : callable
Callback compatible with :meth:`stdUserInterruptTester`.
"""
if callable(tester):
self.ui.set_interrupt_tester(tester)
self.UserInterruptTester = self.ui.check_interrupt
self.PollKey = self.ui.poll_key
[docs]
def set_user_interrupt_Tester(self, tester):
"""Call :meth:`set_user_interrupt_tester` for legacy clients.
Parameters
----------
tester : callable
Non-blocking user-interrupt callback.
"""
self.set_user_interrupt_tester(tester)
[docs]
def set_pre_user_event(self, event_cb):
"""Configure the callback run before user-facing interactions.
Parameters
----------
event_cb : callable
Callback invoked before an interaction begins.
"""
if callable(event_cb):
self.ui.set_pre_user_event(event_cb)
self.PreUserEvent = self.ui.pre_user_event
[docs]
def set_post_user_event(self, event_cb):
"""Configure the callback run after user-facing interactions.
Parameters
----------
event_cb : callable
Callback invoked after an interaction finishes.
"""
if callable(event_cb):
self.ui.set_post_user_event(event_cb)
self.PostUserEvent = self.ui.post_user_event
[docs]
def set_interactive_runner(self, runner):
"""Configure the callback used for interactive sessions.
Parameters
----------
runner : callable
Callback starting an interactive session.
"""
if callable(runner):
self.ui.set_interactive_runner(runner)
[docs]
def set_autosave(self, name):
"""Set the autosave filename used by :meth:`do_autosave`.
Parameters
----------
name : path-like or None
Autosave destination, or ``None`` to disable file output.
"""
self.asname = name
def _init_measurement_devices(self, mg, do_zero=False, do_rfoff=False):
"""Initialize measurement devices with optional safe defaults.
Returns:
int: status from init path (`0` means success).
"""
self.messenger(tstamp() + " Init devices...", [])
err = mg.Init_Devices()
if err:
self.messenger(tstamp() + " ...faild with err %d" % (err), [])
return err
self.messenger(tstamp() + " ...done", [])
if do_rfoff:
mg.RFOff_Devices()
if do_zero:
self.messenger(tstamp() + " Zero devices...", [])
mg.Zero_Devices()
self.messenger(tstamp() + " ...done", [])
return 0
def _finalize_measurement_devices(self, mg, do_rfoff=True, do_quit=True):
"""Finalize measurement devices in a fail-safe way.
Returns:
int: last device status seen (`0` by default).
"""
stat = 0
if mg is None:
return stat
if do_rfoff:
self.messenger(tstamp() + " RF Off...", [])
try:
stat = mg.RFOff_Devices()
except Exception:
LogError(self.messenger)
if do_quit:
self.messenger(tstamp() + " Quit...", [])
try:
stat = mg.Quit_Devices()
except Exception:
LogError(self.messenger)
return stat
def _poll_device_batteries(
self,
mg,
interval=mgraph.DEFAULT_BATTERY_POLL_INTERVAL,
):
"""Poll graph batteries and report low states or communication errors."""
poll = getattr(mg, 'pollBatteryState_Devices', None)
if callable(poll):
result = poll(interval=interval)
if result is None:
return None
else:
# Compatibility for graph-like test doubles and external adapters.
result = {
'checked_at_monotonic': None,
'low_devices': mg.getBatteryLow_Devices(),
'device_errors': {},
}
if result['low_devices']:
self.messenger(
util.tstamp()
+ " WARNING: Low battery status detected for: %s"
% str(result['low_devices']),
[],
)
if result['device_errors']:
self.messenger(
util.tstamp()
+ " WARNING: Battery status query failed for: %s"
% str(result['device_errors']),
[],
)
return result
def _handle_user_interrupt_common(
self,
dct,
ignorelist='',
set_level_cb=None,
do_leveling_cb=None,
wait_handler=None,
on_resume_cb=None,
):
"""Shared interrupt/suspend/resume flow used by measurement classes.
Returns:
bool: `True` if an interrupt was handled, else `False`.
"""
key = self.UserInterruptTester()
if not key or chr(key) in ignorelist:
return False
# Empty key buffer.
_k = self.UserInterruptTester()
while _k is not None:
_k = self.UserInterruptTester()
mg = dct['mg']
names = dct.get('names', {})
f = dct.get('f')
SGLevel = dct.get('SGLevel')
leveling = dct.get('leveling')
hassg = (SGLevel is not None and leveling is not None)
delay = dct.get('delay', 0)
nblist = dct.get('nblist', [])
self.messenger(tstamp() + " RF Off...", [])
mg.RFOff_Devices()
msg1 = (
"The measurement has been interrupted by the user.\n"
"How do you want to proceed?\n\n"
"Continue: go ahead...\n"
"Suspend: Quit devices, go ahead later after reinit...\n"
"Interactive: Go to interactive mode...\n"
"Quit: Quit measurement..."
)
but1 = ['Continue', 'Suspend', 'Interactive', 'Quit']
answer = self.messenger(msg1, but1)
if answer == but1.index('Quit'):
self.messenger(tstamp() + " measurment terminated by user.", [])
raise UserWarning
if answer == but1.index('Interactive'):
self.ui.run_interactive(
self, "Press CTRL-D (Linux,MacOS) or CTRL-Z (Windows) plus Return to exit"
)
elif answer == but1.index('Suspend'):
self.messenger(tstamp() + " measurment suspended by user.", [])
mg.Quit_Devices()
msg2 = "Measurement is suspended.\n\nResume: Reinit and continue\nQuit: Quit measurement..."
but2 = ['Resume', 'Quit']
answer = self.messenger(msg2, but2)
if answer == but2.index('Resume'):
self._init_measurement_devices(mg, do_zero=True, do_rfoff=True)
if hassg and callable(set_level_cb):
try:
set_level_cb(mg, names, SGLevel)
except AmplifierProtectionError as _e:
self.messenger(
tstamp() + " Can not set signal generator level. Amplifier protection raised with message: %s"
% _e.message,
[],
)
if f is not None:
mg.SetFreq_Devices(f)
mg.EvaluateConditions(context={"frequency": f})
if callable(on_resume_cb):
on_resume_cb(mg, names, dct)
else:
self.messenger(tstamp() + " measurment terminated by user.", [])
raise UserWarning
self.messenger(tstamp() + " RF On...", [])
mg.RFOn_Devices()
if hassg and callable(do_leveling_cb):
do_leveling_cb(leveling, mg, names, dct)
if wait_handler is None:
wait_handler = self._handle_user_interrupt_common
try:
self.messenger(tstamp() + " Going to sleep for %d seconds ..." % (delay), [])
self.wait(delay, dct, wait_handler)
self.messenger(tstamp() + " ... back.", [])
except Exception:
pass
mg.NBTrigger(nblist)
return True
[docs]
def do_autosave(self, name_or_obj=None, depth=None, prefixes=None):
"""Serialize the measurement state using :mod:`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.
"""
if depth is None:
depth = 1
if name_or_obj is None:
name_or_obj = getattr(self, 'asname', None)
# we want to save the cmd that has been used
# (in order to get all the calling parameters)
try:
self.autosave = True # mark the state
calling_sequence = calling.get_calling_sequence(prefixes=prefixes) or []
calling_sequence = [cs for cs in calling_sequence if cs != '<string>']
# print calling_sequence
if calling_sequence:
try:
ascmd = calling_sequence[depth]
except IndexError:
ascmd = calling_sequence[-1]
else:
ascmd = None
if isinstance(ascmd, str) and ascmd.startswith('exec'):
# print self.ascmd
ascmd = ascmd[ascmd.index('(') + 1: ascmd.rindex(')')].strip() # part between brackets
var = get_var_from_nearest_outerframe(ascmd)
if var:
ascmd = var
self.ascmd = ascmd
# print "Measure.py; 363:", self.ascmd
# now, we can serialize 'self'
destination = None
external_stream = None
if isinstance(name_or_obj, (str, os.PathLike)):
destination = Path(name_or_obj)
elif hasattr(name_or_obj, 'write'):
external_stream = name_or_obj
elif name_or_obj is not None:
raise TypeError(
"autosave destination must be path-like or a writable "
"binary stream"
)
saved = False
if destination is not None:
self.lastautosave_filename = str(destination)
try:
_atomic_dump_pickle_path(self, destination)
saved = True
except OSError:
LogError(self.messenger)
if external_stream is not None:
self.lastautosave_filename = None
try:
dump_pickle(self, external_stream)
_flush_pickle_file(external_stream)
saved = True
except OSError:
LogError(self.messenger)
if not saved:
fd, fname = tempfile.mkstemp(suffix='.p', prefix='autosave', dir='.', text=False)
pfile = os.fdopen(fd, 'wb')
self.lastautosave_filename = str(Path(fname).resolve())
try:
dump_pickle(self, pfile)
_flush_pickle_file(pfile)
saved = True
finally:
try:
pfile.close()
except OSError:
LogError(self.messenger)
if saved:
self.lastautosave = time.time()
finally:
self.autosave = False
# print self.ascmd
def _run_with_output_target(self, fname, fn, *args, **kwargs):
"""Run a printer function with optional stdout redirection to file."""
stdout = sys.stdout
fp = None
if fname:
fp = open(fname, "w")
sys.stdout = fp
try:
return fn(*args, **kwargs)
finally:
try:
if fp is not None:
fp.close()
except IOError:
LogError(self.messenger)
sys.stdout = stdout
[docs]
@staticmethod
def stdPreUserEvent():
#"""Just calls :meth:`mpylab.tools.unixcrt.unbuffer_stdin()`.
# See there...
#"""
#crt.unbuffer_stdin()
"""stdPreUserEvent method."""
pass
[docs]
@staticmethod
def stdPostUserEvent():
#"""Just calls :meth:`mpylab.tools.unixcrt.restore_stdin()`
# See there...
#"""
#crt.restore_stdin()
"""stdPostUserEvent method."""
pass
[docs]
@staticmethod
def stdInteractiveSession(obj, banner):
"""Start the default terminal-based interactive session.
Parameters
----------
obj : object
Object exposed to the interactive session.
banner : str
Introductory text shown by the interactive console.
"""
interactive(obj=obj, banner=banner)
# def do_leveling(self, leveling, mg, names, dct):
# """Perform leveling on the measurement graph.
# - *leveling*: sequence of dicts with leveling records. Each record is a dict with keys
# 'conditions', 'actor', 'watch', 'nominal', 'reader', 'path', 'actor_min', and 'actor_max'.
# The meaning is:
# - condition: has to be True in order that this lewveling takes place. The condition is evaluated in the global namespace and in C{dct}.
# - actor: at the moment, this can only be a signalgenerator 'sg'
# - watch: the point in the graph to be monitored (e.g. antena input)
# - nominal: the desired value at watch
# - reader: the device reading the value for watch (e.g. forward poer meter)
# - path: Path between reader and watch
# - actor_min, actor_max: valid range for actor values
# - *mg*: the measurement graph
# - *names*: mapping between symbolic names and real names in the dot file
# - *dct*: namespace used for the evaluation of *condition*
# Return: the level set at the actor
# """
# for l in leveling:
# if eval(l['condition'], globals(), dct):
# actor = l['actor']
# watch = l['watch']
# nominal = l['nominal']
# reader = l['reader']
# path = l['path']
# ac_min = l['actor_min']
# ac_max = l['actor_max']
# if actor not in ['sg']:
# self.messenger(util.tstamp()+" Only signal generator can be used as leveling actor.", [])
# break
# for dev in [watch, reader]:
# if dev not in names:
# self.messenger(util.tstamp()+" Device '%s' not found"%dev, [])
# break
# c_level = device.UMDCMResult(complex(0.0,mg.zero(umddevice.UMD_dB)),umddevice.UMD_dB)
# for cpath in path:
# if mg.find_shortest_path(names[cpath[0]],names[cpath[-1]]):
# c_level *= mg.get_path_correction(names[cpath[0]],names[cpath[-1]], umddevice.UMD_dB)['total']
# elif mg.find_shortest_path(names[cpath[-1]],names[cpath[0]]):
# c_level /= mg.get_path_correction(names[cpath[-1]],names[cpath[0]], umddevice.UMD_dB)['total']
# else:
# self.messenger(util.tstamp()+" can't find path from %s tp %s (looked for both directions)."%(cpath[0],cpath[-1]), [])
# break
# if ac_min == ac_max:
# return self.set_level(mg, names, ac_min)
# def __objective (x, mg=mg):
# self.set_level(mg, names, x)
# actual = mg.Read([names[reader]])[names[reader]]
# actual = device.UMDCMResult(actual)
# cond, a, n = self.__test_leveling_condition(actual, nominal, c_level)
# return a-n
# l = util.secant_solve(__objective, ac_min, ac_max, nominal.get_u()-nominal.get_v(), 0.1)
# return self.set_level(mg, names, l)
# #break # only first true condition ie evaluated
# return None
@staticmethod
def _sg_level_quantity(level):
if isinstance(level, Quantity):
level = level.reduce_to(WATT)
return Quantity(WATT, level.get_expectation_value_as_float())
# numeric signal-generator levels are dBm
return UConv.to_quantity('dBm', level)
#return Quantity(WATT, 10 ** (0.1 * level) * 0.001)
@staticmethod
def _numeric_quantity_like(quantity):
quantity = quantity.reduce_to(quantity._unit)
return Quantity(quantity._unit, quantity.get_expectation_value_as_float())
[docs]
def set_level(self, mg, l, leveler=None):
"""Set the signal-generator level, optionally respecting a leveler.
Parameters
----------
mg : MGraph
Active measurement graph containing the signal generator.
l : Quantity or float
Requested power. Numeric values are interpreted as dBm.
leveler : Leveler, optional
Leveler whose ``MaxSafe`` value limits the request.
Returns
-------
Quantity
Level reported as applied by the signal generator.
"""
sg = mg.instrumentation[mg.name.sg]
l = self._sg_level_quantity(l)
if leveler is None: # try to use instance leveler
try:
leveler = self.leveler_inst # (**self.leveler_par)
except AttributeError:
pass # stay with None
if leveler: # use MaxSafe
l = min(l, leveler.MaxSafe)
err, lv = sg.SetLevel(l)
# is_save, message = mg.AmplifierProtect (names['sg'], names['a2'], l, sg_unit, typ='lasy')
# if not is_save:
# raise AmplifierProtectionError, message
self.messenger(tstamp() + " Signal Generator set to %s" % (lv), [])
return lv
[docs]
def set_level_protected(self, mg, level, output=None, actor=None, leveler=None, reason=None):
"""Set a signal-generator level without exceeding graph safety limits.
If ``output`` is given, :meth:`MGraph.AmplifierProtect` is used as the
hard safety check. Unsafe requests are clipped to
:meth:`MGraph.MaxSafeLevel`; the level is still applied and metadata is
returned so measurement code can record that the requested target was
unreachable due to amplifier protection.
Parameters
----------
mg : MGraph
Active measurement graph.
level : Quantity or float
Requested power. Numeric values are interpreted as dBm.
output : str, optional
End node of the path checked for amplifier protection.
actor : str, optional
Signal-generator node. The graph's ``sg`` name is used by default.
leveler : Leveler, optional
Leveler providing an additional ``MaxSafe`` constraint.
reason : str, optional
Operation recorded when protection limits the request.
Returns
-------
tuple of (Quantity, dict)
Applied level and protection metadata. ``max_safe_level`` contains
the strictest effective limit when multiple protection sources are
active.
"""
actor_key = actor if actor is not None else mg.name.sg
sg = mg.instrumentation[actor_key]
requested = self._sg_level_quantity(level)
applied = requested
metadata = {
'requested_level': requested,
'applied_level': None,
'limited_by_amplifier_protection': False,
'amplifier_protection_output': output,
'amplifier_protection_actor': actor_key,
'amplifier_protection_message': '',
'amplifier_protection_reason': reason,
'max_safe_level': None,
}
def record_max_safe(candidate):
candidate = self._numeric_quantity_like(candidate).reduce_to(WATT)
current = metadata['max_safe_level']
metadata['max_safe_level'] = (
candidate if current is None else min(current, candidate)
)
return candidate
if leveler is None: # try to use instance leveler
try:
leveler = self.leveler_inst # (**self.leveler_par)
except AttributeError:
pass # stay with None
if leveler: # use MaxSafe
max_safe = record_max_safe(leveler.MaxSafe)
applied = min(applied, max_safe)
if applied != requested:
metadata['limited_by_amplifier_protection'] = True
if output is not None and hasattr(mg, 'AmplifierProtect'):
if actor is None:
actor = actor_key
is_safe, message = mg.AmplifierProtect(actor, output, requested)
if not is_safe:
max_safe = None
if hasattr(mg, 'MaxSafeLevel'):
max_safe = mg.MaxSafeLevel(actor, output)
if max_safe is None:
raise AmplifierProtectionError(
"Amplifier protection rejected %s, but no maximum safe level "
"could be determined. %s" % (requested, message)
)
max_safe = record_max_safe(max_safe)
applied = min(applied, max_safe)
metadata.update({
'limited_by_amplifier_protection': True,
'amplifier_protection_message': message,
})
err, lv = sg.SetLevel(applied)
if err:
raise ValueError("signal generator SetLevel failed with error %r" % err)
metadata['applied_level'] = self._sg_level_quantity(lv if lv is not None else applied)
if metadata['limited_by_amplifier_protection']:
self.messenger(
tstamp()
+ " Amplifier protection limited signal generator level from %s to %s%s"
% (
metadata['requested_level'],
metadata['applied_level'],
"" if reason is None else " before %s" % reason,
),
[],
)
else:
self.messenger(tstamp() + " Signal Generator set to %s" % (lv), [])
return metadata['applied_level'], metadata
# --- Backward compatible legacy aliases ---------------------------------
[docs]
def setLevel(self, mg, level_or_names, level_or_leveler=None):
"""Backward-compatible wrapper for legacy callers.
Supported call shapes:
- ``setLevel(mg, level_dBm)``
- ``setLevel(mg, level_dBm, leveler)``
- ``setLevel(mg, names_dict, level_dBm)`` (legacy TEM/Univers code)
Parameters
----------
mg : MGraph
Active measurement graph.
level_or_names : Quantity, float, or mapping
Requested level, or a legacy names mapping.
level_or_leveler : Quantity, float, or Leveler, optional
Legacy level argument or leveler instance.
Returns
-------
Quantity
Applied signal-generator level returned by :meth:`set_level`.
"""
if isinstance(level_or_names, dict):
# Legacy signature: (mg, names, level_dBm)
return self.set_level(mg, level_or_leveler, leveler=None)
return self.set_level(mg, level_or_names, leveler=level_or_leveler)
[docs]
def doLeveling(self, leveling, mg, names, dct):
"""Backward-compatible no-op stub for removed legacy leveling API.
The legacy callers expect this method to exist and to return either a
new level or ``None``. Current code path keeps behavior by returning
``None``.
Parameters
----------
leveling : object
Unused legacy leveling configuration.
mg : MGraph
Unused measurement graph.
names : mapping
Unused instrumentation names.
dct : mapping
Unused legacy evaluation context.
"""
_ = (leveling, mg, names, dct)
return None
[docs]
def do_leveling(self, leveling, mg, names, dct):
"""Call the compatibility :meth:`doLeveling` implementation.
Parameters
----------
leveling : object
Legacy leveling configuration.
mg : MGraph
Active measurement graph.
names : mapping
Instrumentation names.
dct : mapping
Legacy evaluation context.
Returns
-------
None
The removed legacy implementation is intentionally a no-op.
"""
return self.doLeveling(leveling, mg, names, dct)
# def __test_leveling_condition(self, actual, nominal, c_level):
# cond = True
# actual = util.flatten(actual) # ensure lists
# nominal= util.flatten(nominal)
# for ac,nom in zip(actual,nominal):
# ac *= c_level
# if hasattr(nom.get_v(), 'mag'): # a complex
# nom = nom.mag()
# ac = ac.mag()
# ac = ac.convert(nominal.unit)
# cond &= (nom.get_l() <= ac.get_v() <= nom.get_u())
# return cond, actual.get_v(), nominal.get_v()
[docs]
def make_deslist(self, thedata, description):
"""Select available data-set descriptions.
Parameters
----------
thedata : mapping
Data indexed by description.
description : str, iterable of str, or None
Requested descriptions, or ``None`` for all available entries.
Returns
-------
list
Requested descriptions that exist in ``thedata``.
"""
if description is None:
description = list(thedata.keys())
if issequence(description): # a sequence
deslist = [des for des in description if des in thedata]
else:
if description in thedata:
deslist = [description]
else:
deslist = []
return deslist
[docs]
def MakeDeslist(self, thedata, description):
"""Call :meth:`make_deslist` for legacy clients.
Parameters
----------
thedata : mapping
Data indexed by description.
description : str, iterable of str, or None
Requested descriptions.
Returns
-------
list
Available requested descriptions.
"""
return self.make_deslist(thedata, description)
[docs]
def make_whatlist(self, thedata, what):
"""Select available result-channel names.
Parameters
----------
thedata : mapping
Data sets containing channel mappings.
what : str, iterable of str, or None
Requested channels, or ``None`` for all available channels.
Returns
-------
list
Requested channels present in the data.
"""
allwhat_withdupes = flatten([list(v.keys()) for v in thedata.values()])
allwhat = list(dict.fromkeys(allwhat_withdupes))
if what is None:
whatlist = allwhat
else:
whatlist = []
what = flatten(what)
whatlist = [w for w in what if w in allwhat]
return whatlist
[docs]
def MakeWhatlist(self, thedata, what):
"""Call :meth:`make_whatlist` for legacy clients.
Parameters
----------
thedata : mapping
Data sets containing channel mappings.
what : str, iterable of str, or None
Requested channels.
Returns
-------
list
Available requested channels.
"""
return self.make_whatlist(thedata, what)
[docs]
@staticmethod
def stdEutStatusChecker(status):
"""Return whether a legacy EUT status denotes normal operation.
Parameters
----------
status : object
Legacy EUT status value.
Returns
-------
bool
``True`` only for ``"ok"`` or ``"OK"``.
"""
return status in ['ok', 'OK']
[docs]
@staticmethod
def std_eut_status_checker(status):
"""Call :meth:`stdEutStatusChecker` using the snake-case name.
Parameters
----------
status : object
Legacy EUT status value.
Returns
-------
bool
Whether the status denotes normal operation.
"""
return Measure.stdEutStatusChecker(status)
class Error(Exception):
"""Base class for all exceptions of this module
"""
pass
class AmplifierProtectionError(Error):
"""Report that a requested RF level violates amplifier protection.
Parameters
----------
message : str
Human-readable protection failure details.
"""
def __init__(self, message):
self.message = message