mpylab.tools.mgraph module

This is mpylab.tools.mgraph.

Provides the MGraph class (mainly)

author:

Hans Georg Krauthäuser (main author)

license:

GPL-3 or higher

exception mpylab.tools.mgraph.ConditionContextError

Bases: GraphValidationError

Raised when a condition variable cannot be resolved explicitly.

exception mpylab.tools.mgraph.DeviceConfigurationError

Bases: ValueError

Raised when a graph node cannot be mapped to a configured driver.

class mpylab.tools.mgraph.DictObj

Bases: dict

A dict with object-like attributes.

Instead of dct[‘name’] you can do dct.name

class mpylab.tools.mgraph.GName(mginst: MGraph)

Bases: object

Provide attribute access to logical-to-physical device names.

Parameters:

mginst (MGraph) – Measurement graph whose name mapping is exposed.

Notes

For an MGraph instance mg, mg.name.logical_name returns the corresponding physical node name from the DOT graph.

class mpylab.tools.mgraph.Graph(fname_or_data=None, SearchPaths=None)

Bases: object

Represent a directed graph based on pydot.

Parameters:

  • fname_or_data (path-like, file-like, str, pydot.Dot, or object) – DOT file, readable stream, DOT source text, existing pydot graph, or structured input accepted by a pydot graph constructor.

  • SearchPaths (sequence of path-like or path-like, optional) – Directories searched for a relative DOT filename. The current working directory is used by default.

Notes

Structured input is passed to pydot’s edge-, adjacency-matrix-, and incidence-matrix constructors in that order for compatibility with the historical API.

find_all_paths(start, end, path=None, edge=None)

Find every acyclic active path between two nodes.

Parameters:

  • start (str) – Name of the start node.

  • end (str) – Name of the destination node.

  • path (list, optional) – Recursive path state. Applications should leave this as None.

  • edge (pydot.Edge, optional) – Edge entering the current recursion level. Applications should leave this as None.

Returns:

All active, acyclic paths. The list is empty when none exists.

Return type:

list[list[pydot.Edge]]

find_path(start, end, path=None)

Return the first active path between two nodes.

Parameters:

  • start (str) – Name of the start node.

  • end (str) – Name of the destination node.

  • path (list, optional) – Initial recursive path state. Applications should leave this as None.

Returns:

Edges forming the first active path, or None if no path exists.

Return type:

list or None

find_shortest_path(start, end, path=None)

Return the active path containing the fewest edges.

Parameters:

  • start (str) – Name of the start node.

  • end (str) – Name of the destination node.

  • path (list, optional) – Initial recursive path state. Applications should leave this as None.

Returns:

Shortest active path, or None if no path exists.

Return type:

list[pydot.Edge] or None

get_common_parent(n1, n2)

Find an active common ancestor of two nodes.

Parameters:

  • n1 (str) – First node name.

  • n2 (str) – Second node name.

Returns:

Common ancestor node, or None if none is found.

Return type:

str or None

exception mpylab.tools.mgraph.GraphContextError

Bases: RuntimeError

Raised when a graph context cannot be applied to its controllers.

class mpylab.tools.mgraph.GraphContextResult(context: dict[str, object], frequency_result: object = None, expected_controller_states: dict[str, object] = <factory>, actual_controller_states: dict[str, object] = <factory>)

Bases: object

Result of applying a logical context to a measurement graph.

actual_controller_states: dict[str, object]
context: dict[str, object]
expected_controller_states: dict[str, object]
frequency_result: object = None
exception mpylab.tools.mgraph.GraphInputError

Bases: ValueError

Raised when a measurement graph input cannot be loaded or parsed.

class mpylab.tools.mgraph.GraphPathCheckPoint(value: float, paths: tuple[tuple[str, ...], ...], active_branches: tuple[tuple[str, tuple[str, ...]], ...] = (), condition_errors: tuple[str, ...] = ())

Bases: object

Path-selection result for one parameter value.

active_branches: tuple[tuple[str, tuple[str, ...]], ...] = ()
condition_errors: tuple[str, ...] = ()
paths: tuple[tuple[str, ...], ...]
property status

Return the most severe path-selection outcome at this value.

value: float
class mpylab.tools.mgraph.GraphPathCheckReport(start: str, end: str, parameter: str, mode: str, results: list[GraphPathCheckPoint] = <factory>, critical_values: tuple[float, ...]=(), boundary_analysis_complete: bool = True)

Bases: object

Structured result returned by MGraph.check_paths().

boundary_analysis_complete: bool = True
critical_values: tuple[float, ...] = ()
end: str
property errors

Return path-check points classified as errors in this mode.

format_text()

Return a compact human-readable validation report.

Returns:

Multi-line graph path validation summary.

Return type:

str

mode: str
property ok

Whether the report contains no path-selection errors.

parameter: str
results: list[GraphPathCheckPoint]
start: str
property warnings

Return path-check points classified as warnings in this mode.

exception mpylab.tools.mgraph.GraphStateMismatchError(mismatches)

Bases: GraphContextError

Raised when physical controller state differs from graph context.

Parameters:

mismatches (Mapping[str, Mapping[str, object]]) – Expected and observed states indexed by controller node name.

exception mpylab.tools.mgraph.GraphValidationError

Bases: ValueError

Raised when measurement-graph references are semantically invalid.

exception mpylab.tools.mgraph.LegacyConditionContextWarning

Bases: FutureWarning

Warn that condition variables were resolved from the caller frame.

class mpylab.tools.mgraph.Leveler(mg, actor, output, lpoint, observer, pin=None, datafunc=None, min_actor=None)

Bases: object

Iterative level control based on measurement-graph corrections.

Parameters:

  • mg (MGraph) – Active measurement graph.

  • actor (str) – Device node whose SetLevel method controls the stimulus.

  • output (str) – End node used to determine the maximum safe actor level.

  • lpoint (str) – Graph node at which the target value is defined.

  • observer (str) – Device node measuring the controlled quantity.

  • pin (Quantity or iterable of Quantity, optional) – Initial actor-level samples. Three fractions of the safe maximum are used by default.

  • datafunc (callable, optional) – Transformation applied to observer values before path correction.

  • min_actor (Quantity or float, optional) – Minimum permitted actor level. Numeric values use watts.

add_samples(pin)

Add actor-level samples and their measured observation values.

Parameters:

pin (Quantity or iterable of Quantity) – Actor levels to apply and measure.

Returns:

Accepted actor levels after minimum and maximum constraints.

Return type:

list of Quantity

adjust_level(soll, maxiter=10, relerr=0.01)

Iteratively adjust the actor and retain a structured outcome.

The return value remains the historical (applied, observed) tuple. Detailed completion information is available through last_result.

Parameters:

  • soll (Quantity) – Positive target value at the leveling point.

  • maxiter (int, optional) – Maximum number of additional measurement iterations.

  • relerr (float, optional) – Required maximum relative error.

Returns:

Applied actor level and observed value at the leveling point.

Return type:

tuple of (Quantity, Quantity)

update_interpol()

Rebuild forward and inverse interpolators from current sample set.

exception mpylab.tools.mgraph.LevelingDataError

Bases: ValueError

Raised when graph-based leveling receives unusable numeric data.

exception mpylab.tools.mgraph.LevelingDeviceError

Bases: RuntimeError

Raised when a device operation fails during graph-based leveling.

class mpylab.tools.mgraph.LevelingResult(status: str, applied_level: Quantity, observed_value: Quantity, target_value: Quantity, relative_error: float, iterations: int, limited_by_amplifier_protection: bool = False, reason: str = '')

Bases: object

Machine-readable outcome of Leveler.adjust_level().

applied_level: Quantity
as_dict()

Return a serialization-friendly representation.

Returns:

Leveling outcome and all diagnostic fields.

Return type:

dict

property converged

Whether the leveler reached the requested target tolerance.

iterations: int
limited_by_amplifier_protection: bool = False
observed_value: Quantity
reason: str = ''
relative_error: float
status: str
target_value: Quantity
class mpylab.tools.mgraph.MGraph(fname_or_data=None, themap=None, SearchPaths=None, condition_map=None, allow_legacy_condition_context=True)

Bases: Graph

Represent devices, paths, and conditions in a measurement graph.

Parameters:

  • fname_or_data (path-like, file-like, str, pydot.Dot, or object) – Graph input accepted by Graph.

  • themap (Mapping[str, str], optional) – Mapping from logical application names to physical DOT node names.

  • SearchPaths (sequence of path-like or path-like, optional) – Directories searched for graph, INI, driver, and data files.

  • condition_map (Mapping[str, str], optional) – Mapping from variable names used in DOT conditions to application-side condition names.

  • allow_legacy_condition_context (bool, optional) – Permit condition variables to be resolved from legacy outer-frame context when no explicit value was supplied.

AmplifierProtect(start, end, startlevel, sg_unit=<scuq.units.AlternateUnit object>, typ='save')

Check amplifier input limits along all active paths.

Parameters:

  • start (str) – Node at which startlevel is applied.

  • end (str) – End node delimiting paths that must be inspected.

  • startlevel (Quantity or float) – Applied level at start.

  • sg_unit (scuq.units.Unit, optional) – Unit assigned to numeric start levels.

  • typ (str, optional) – Retained for API compatibility; it does not alter the check.

Returns:

Safety result and details for every exceeded MAXIN limit.

Return type:

tuple of (bool, str)

ApplyContext(context, doAction=True, set_frequency=True, force_controllers=False)

Apply logical conditions and declared hardware controllers.

Context-controller nodes opt in through the DOT attribute context_controller. Their drivers plan, apply, and read back the concrete hardware state while callers provide only logical values.

Parameters:

  • context (mapping) – Application-level condition values.

  • doAction (bool, optional) – Execute compatible DOT actions while conditions are evaluated.

  • set_frequency (bool, optional) – Call SetFreq_Devices() when the context contains the mapped frequency value.

  • force_controllers (bool, optional) – Reapply controllers even when their declared context keys did not change. Intended for explicit recovery after user confirmation.

Returns:

Applied context, frequency result, and controller read-back.

Return type:

GraphContextResult

CalcLevelFrom(sg, limiter, what)

Validate a level-calculation path and compute its correction.

Parameters:

  • sg (str) – Source node name.

  • limiter (str) – Limiting destination node name.

  • what (object) – Retained for compatibility with the historical API.

Returns:

Zero after successful validation and path-correction lookup.

Return type:

int

CheckContextState()

Verify that physical context controllers still match the graph.

Returns:

Current controller states indexed by physical graph node.

Return type:

dict

Raises:

GraphStateMismatchError – If read-back fails or a controller differs from its expected state. RF is switched off before the exception is raised.

CmdDevices(IgnoreInactive, cmd, *args)

Invoke one command on selected graph devices.

Parameters:

  • IgnoreInactive (bool) – If True, invoke the command only on active devices. If False, include inactive devices as well.

  • cmd (str) – Driver method name.

  • *args (object) – Positional arguments passed to the driver method.

Returns:

Sum of returned status codes; zero normally indicates success.

Return type:

int

Notes

Individual status and error values are stored in nodes[name]['ret'] and nodes[name]['err'].

ConfReceivers(conf, IgnoreInactive=True)

Configure spectrum analyzers and receivers in the graph.

Parameters:

  • conf (Mapping[str, object]) – Requested values for supported receiver parameters such as rbw, vbw, min_attenuation, att, preamp, reflevel, detector, tracemode, sweeptime, sweepcount, and span. Missing or None values are read from the device when a corresponding getter exists.

  • IgnoreInactive (bool, optional) – If True, configure only active devices. If False, include inactive devices as well.

Returns:

Applied or read-back values as result[node][parameter].

Return type:

dict

CreateDevices()

Create and configure device instances represented by graph nodes.

This method initializes node activity, reads referenced INI files, and stores each instantiated driver in nodes[name]['inst']. It should normally be called once after constructing the graph.

Returns:

Mapping from physical graph node names to device instances.

Return type:

DictObj

EvaluateConditions(doAction=True, context=None)

Evaluate node and edge conditions and update their active state.

context contains application-level names. condition_map maps names used in DOT expressions to those application names. If context is omitted, caller-frame lookup remains available as a deprecated compatibility fallback.

Parameters:

  • doAction (bool, optional) – Execute actions associated with condition state changes.

  • context (mapping, optional) – Application-level values used to resolve DOT conditions.

GetAntennaEfficiency(node)

Read efficiency from the first active antenna connected to a node.

Parameters:

node (str) – Graph node connected to the antenna, often a virtual node such as "ant".

Returns:

Antenna efficiency returned by the driver, or None if no suitable antenna is found.

Return type:

object or None

Init_Devices(IgnoreInactive=True)

Initialize selected graph devices.

Parameters:

IgnoreInactive (bool, optional) – If True, initialize only active devices. If False, include inactive devices as well.

Returns:

Sum of device status codes.

Return type:

int

Raises:

UserWarning – If a device fails to initialize.

MaxSafeLevel(start, end, typ='save')

Return the minimum safe start level across all active paths.

Parameters:

  • start (str) – Node at which the returned level would be applied.

  • end (str) – End node delimiting paths that must be inspected.

  • typ (str, optional) – Retained for API compatibility; it does not alter the calculation.

Returns:

Smallest source level allowed by downstream MAXIN limits, or None when no applicable limit exists.

Return type:

scuq.quantities.Quantity or None

NBRead(lst, result)

Continue a non-blocking read from active devices.

Parameters:

  • lst (str or iterable of str) – Physical graph node names to read.

  • result (dict) – Results already collected by an earlier call.

Returns:

Updated result mapping. Devices without ready data remain absent.

Return type:

dict

NBTrigger(lst)

Trigger every capable active device in a node list.

Parameters:

lst (iterable of str) – Physical graph node names to trigger.

Returns:

Trigger return values indexed by nodes that could be triggered.

Return type:

dict

PollContextState(interval=0.5, force=False)

Poll declared context controllers at a bounded rate.

Parameters:

  • interval (float, optional) – Minimum seconds between hardware queries.

  • force (bool, optional) – Query immediately regardless of the previous poll time.

Returns:

Last or newly read controller states.

Return type:

dict

Quit_Devices(IgnoreInactive=True)

Quit selected devices using CmdDevices().

Parameters:

IgnoreInactive (bool, optional) – If True, quit only active devices. If False, include inactive devices as well.

Returns:

Sum of device status codes.

Return type:

int

RFOff_Devices(IgnoreInactive=True)

Disable RF output on selected devices.

Parameters:

IgnoreInactive (bool, optional) – If True, disable only active devices. If False, include inactive devices as well.

Returns:

Sum of device status codes.

Return type:

int

RFOn_Devices(IgnoreInactive=True)

Enable RF output on selected devices.

Parameters:

IgnoreInactive (bool, optional) – If True, enable only active devices. If False, include inactive devices as well.

Returns:

Sum of device status codes.

Return type:

int

Read(lst)

Read active devices using their blocking data method.

Parameters:

lst (str or iterable of str) – Physical graph node names to read.

Returns:

Device values indexed by physical node name.

Return type:

dict

ReapplyContext()

Reapply the stored context after explicit recovery approval.

Returns:

Reapplied context, frequency result, and controller read-back.

Return type:

GraphContextResult

Raises:

  • GraphContextError – If the measurement graph has no stored context or a context controller cannot apply it.

  • GraphStateMismatchError – If a controller’s physical state differs from the reapplied context. RF is switched off before the exception is raised.

RunReceiverScans(conf, receivers=None, IgnoreInactive=True, cancel_callback=None, progress_callback=None)

Run receiver scans for graph receiver nodes.

conf is forwarded to each receiver RunScan call. If conf contains a default entry and/or entries named like receiver nodes, those dictionaries are merged for the corresponding receiver. Progress callbacks receive a copy of each point dictionary with an added receiver key.

Parameters:

  • conf (mapping) – Default and optional receiver-specific RunScan arguments.

  • receivers (str or iterable of str, optional) – Receiver nodes to scan. All eligible nodes are used by default.

  • IgnoreInactive (bool, optional) – Skip receiver nodes that are currently inactive.

  • cancel_callback (callable, optional) – Callback used by receiver scans to request cancellation.

  • progress_callback (callable, optional) – Callback receiving each scan point and its receiver name.

Returns:

Per-receiver error status and scan result.

Return type:

dict

SetCisprRbwScanFrequencies_Devices(frequencies=None, terminal_boundary_policy='previous', IgnoreInactive=True)

Set optional CISPR RBW scan context on capable receivers.

Parameters:

  • frequencies (iterable of float, optional) – Frequencies in the upcoming scan, in hertz.

  • terminal_boundary_policy (str, optional) – Policy used when the final scan frequency lies on a CISPR band boundary.

  • IgnoreInactive (bool, optional) – If True, configure only active devices. If False, include inactive devices as well.

Returns:

Applied scan contexts indexed by physical receiver node name.

Return type:

dict

SetFreq_Devices(freq, IgnoreInactive=True)

Set the frequency on selected devices.

Parameters:

  • freq (Quantity or float) – Requested frequency in the representation accepted by each driver.

  • IgnoreInactive (bool, optional) – If True, configure only active devices. If False, include inactive devices as well.

Returns:

Minimum and maximum frequencies reported by capable devices.

Return type:

tuple

Trigger_Devices(IgnoreInactive=True)

Trigger selected devices using CmdDevices().

Parameters:

IgnoreInactive (bool, optional) – If True, trigger only active devices. If False, include inactive devices as well.

Returns:

Sum of device status codes.

Return type:

int

Zero_Devices(IgnoreInactive=True)

Zero selected devices using CmdDevices().

Parameters:

IgnoreInactive (bool, optional) – If True, zero only active devices. If False, include inactive devices as well.

Returns:

Sum of device status codes.

Return type:

int

active_paths(start, end, names=None)

Return active node paths for an explicit condition-name context.

Unlike EvaluateConditions(), this method neither changes graph attributes nor executes actions. Condition failures are raised as GraphValidationError.

Parameters:

  • start (str) – Start node.

  • end (str) – End node.

  • names (mapping, optional) – Application-level condition context.

Returns:

Active node paths from start to end.

Return type:

list of tuple of str

static apply_path_correction(value, correction, *, operation='divide', magnitude=True, output_unit=None)

Apply a path correction while preserving SCUQ semantics.

Exactly known ratio corrections use a constant-scaling fast path. Uncertain or otherwise unsupported inputs automatically use the full SCUQ expression implementation.

Parameters:

  • value (scuq.quantities.Quantity) – Measured value to which the path correction is applied.

  • correction (scuq.quantities.Quantity or mpylab.tools.quantity_uncertainty.PreparedRatioCorrection) – Ratio correction, or a correction prepared by prepare_path_correction(). When a prepared correction is supplied, its stored operation, magnitude setting, and output unit are used.

  • operation (str) – Apply an unprepared correction by "divide" (default) or "multiply". This argument is ignored when correction is already prepared.

  • magnitude (bool) – If true (default), return the magnitude after applying an unprepared correction. If false, preserve phase. This argument is ignored when correction is already prepared.

  • output_unit (scuq.units.Unit or None) – Optional unit to which the corrected value is reduced. This argument is ignored when correction is already prepared.

Returns:

Corrected measurement value.

Return type:

scuq.quantities.Quantity

Raises:

  • TypeError – If value or an unprepared correction is not a SCUQ quantity.

  • ValueError – If operation is neither "divide" nor "multiply", or the requested correction or unit conversion is invalid.

check_paths(start, end, parameter, *, values=None, start_value=None, stop_value=None, context=None, mode='exactly_one_path')

Check condition-controlled paths over a parameter range.

Simple comparison boundaries involving parameter are checked at the boundary itself and at the immediately adjacent floating-point values. Explicit values are checked as well. For more complex parameter expressions, GraphPathCheckReport.boundary_analysis_complete is false because only the supplied values can be sampled.

Parameters:

  • start (str) – Start node of the checked path.

  • end (str) – End node of the checked path.

  • parameter (str) – Application-level parameter varied by the check.

  • values (iterable of float, optional) – Explicit parameter values to check.

  • start_value (float, optional) – Lower boundary of the checked range.

  • stop_value (float, optional) – Upper boundary of the checked range.

  • context (mapping, optional) – Fixed application-level condition values.

  • mode ({"exactly_one_path", "warn_parallel"}, optional) – Classification policy for multiple active paths.

Returns:

Structured results at explicit and detected critical values.

Return type:

GraphPathCheckReport

condition_variables()

Return variable names referenced by node and edge conditions.

Returns:

Names as written in DOT condition expressions.

Return type:

set of str

getBatteryLow_Devices(IgnoreInactive=True)

Return selected devices reporting a low battery state.

Parameters:

IgnoreInactive (bool, optional) – If True, query only active devices. If False, include inactive devices as well.

Returns:

Physical node names whose battery state is low.

Return type:

list[str]

get_gname(name: str) → str | None

Resolve a logical or physical device name to a DOT node name.

Parameters:

name (str) – Logical application name or physical graph node name.

Returns:

Physical DOT node name, or None when the name is unknown.

Return type:

str or None

get_path_correction(start, end, unit=None)

Return the total active path correction from start to end.

Parameters:

  • start (str) – Start node or mapped instrumentation name.

  • end (str) – End node or mapped instrumentation name.

  • unit (scuq.units.Unit, optional) – AMPLITUDERATIO or POWERRATIO. Amplitude ratio is the default.

Returns:

Total S21-based correction reduced to the requested ratio unit.

Return type:

scuq.quantities.Quantity

get_path_corrections(start, end, unit=None)

Return individual and total corrections from start to end.

Parameters:

  • start (str) – Start node or mapped instrumentation name.

  • end (str) – End node or mapped instrumentation name.

  • unit (scuq.units.Unit, optional) – AMPLITUDERATIO or POWERRATIO. Amplitude ratio is the default.

Returns:

SCUQ corrections for path elements; "total" contains the combined correction.

Return type:

dict

pollBatteryState_Devices(interval=60.0, IgnoreInactive=True, force=False)

Poll battery devices when the configured interval has elapsed.

Return None when polling is disabled or not yet due. Otherwise return a dictionary containing low_devices, device_errors, and checked_at_monotonic. An interval of zero polls on every call. force=True performs an immediate poll regardless of the interval.

Parameters:

  • interval (float or None, optional) – Minimum seconds between checks, or None to disable polling.

  • IgnoreInactive (bool, optional) – Skip devices that are currently inactive.

  • force (bool, optional) – Poll immediately regardless of the previous check time.

Returns:

Battery status and device errors when a poll occurs, otherwise None.

Return type:

dict or None

static prepare_path_correction(correction, *, operation, magnitude, output_unit=None)

Prepare an invariant path correction for repeated application.

Preparation analyzes the correction once. An exact ratio correction can subsequently use constant scaling when magnitude output and an explicit output unit are requested. All other cases retain the full SCUQ calculation.

Parameters:

  • correction (scuq.quantities.Quantity) – Amplitude- or power-ratio correction to prepare.

  • operation (str) – Apply the correction by "divide" or "multiply". This keyword argument is required so that the correction direction is selected explicitly.

  • magnitude (bool) – If true, return magnitudes when the prepared correction is applied. If false, preserve phase. This keyword argument is required so that discarding phase is explicit.

  • output_unit (scuq.units.Unit or None) – Optional unit to which corrected values are reduced.

Returns:

Reusable correction containing the analyzed fast-path properties and all application options.

Return type:

mpylab.tools.quantity_uncertainty.PreparedRatioCorrection

Raises:

  • TypeError – If correction is not a SCUQ quantity.

  • ValueError – If operation is neither "divide" nor "multiply".

Reuse the returned object only while the graph path, frequency, active conditions, and relevant calibration data remain unchanged.

validate_created_devices()

Ensure every edge dev reference has a created device instance.

validate_graph_structure()

Validate edge device references and logical-to-physical name mappings.

mpylab.tools.mgraph.normalize_battery_poll_interval(interval)

Return a validated battery polling interval in seconds.

Parameters:

interval (float or None) – Requested interval in seconds. None disables polling.

Returns:

The finite, non-negative interval, or None when disabled.

Return type:

float or None

mpylab.tools.mgraph.safe_action_exec(expr, names)

Safely execute one method-call action expression with literal arguments.

Parameters:

  • expr (str) – Expression containing one method call on a named object.

  • names (Mapping[str, object]) – Objects available to the expression, indexed by their permitted names.

Returns:

Return value of the invoked method.

Return type:

object