Skip to content

API Reference

This page provides the automatically generated documentation for the core classes and drivers.

Base Classes

Bases: ABC

Abstract Base Class for all instrument drivers following the 'Abstract Hardware' spec.

Source code in src/instrumation/drivers/base.py 
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
class InstrumentDriver(ABC):
    """Abstract Base Class for all instrument drivers following the 'Abstract Hardware' spec."""
    def __init__(self, resource: str):
        self.resource = resource
        self.connected = False
        self.is_simulated = False

        # Identity & Capabilities
        self.identity: Dict[str, str] = {"manufacturer": "", "model": "", "serial": "", "version": ""}
        self.options: List[str] = []
        self.error_stack: List[str] = []

        # Software Safety Guardrails
        self.min_frequency = 0.0
        self.max_frequency = 1e12
        self.max_power_dbm = 0.0
        self.max_voltage = 0.0

    def __getattr__(self, name: str):
        """Dynamic async wrapper for all driver methods."""
        if name.startswith("async_"):
            sync_name = name[6:]
            if hasattr(self, sync_name):
                sync_method = getattr(self, sync_name)
                async def wrapper(*args, **kwargs):
                    return await asyncio.to_thread(sync_method, *args, **kwargs)
                return wrapper
        raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{name}'")


    @property
    def resource_address(self) -> str:
        return self.resource

    @abstractmethod
    def connect(self):
        """Establishes connection and performs identity/option discovery."""
        pass

    @abstractmethod
    def disconnect(self):
        """Safely tears down connection."""
        pass

    def close(self):
        self.disconnect()

    @abstractmethod
    def write(self, command: str): pass

    @abstractmethod
    def query(self, command: str) -> str: pass

    def safe_send(self, command: str):
        """Sends command and immediately checks SYST:ERR?."""
        raise NotImplementedError()

    def query_ascii(self, command: str) -> str:
        """Sends command, reads response, and checks for errors."""
        raise NotImplementedError()

    def query_binary_values(self, command: str, datatype: str = 'f', is_big_endian: bool = False) -> List[float]:
        """High-speed binary data transfer."""
        raise NotImplementedError()

    @abstractmethod
    def get_id(self) -> str: pass

    # --- Global Logic & Synchronization ---
    @abstractmethod
    def preset(self, automation_optimized: bool = True): pass

    @abstractmethod
    def clear_status(self):
        """Executes *CLS."""
        pass

    @abstractmethod
    def sync_config(self):
        """Executes *CLS and *WAI for a clean slate."""
        pass

    @abstractmethod
    def wait_ready(self, timeout: float = 30.0):
        """Standard polling loop for *OPC?."""
        pass

    @abstractmethod
    def shutdown_safety(self):
        """Emergency shutdown protocol (Outputs OFF, Power/Volt 0)."""
        pass

    @abstractmethod
    def check_errors(self):
        """Queries SYST:ERR? and updates local error_stack."""
        pass

    def save_state(self, index: Union[int, str]):
        """Saves current state to memory."""
        self._unsupported_feature("save_state")

    def load_state(self, index: Union[int, str]):
        """Recalls state from memory."""
        self._unsupported_feature("load_state")

    # --- Unit Guards & Formatting ---
    def format_frequency(self, val: Union[float, str]) -> str:
        """Ensures input is Hz and formats for SCPI (e.g. 1.5e9 -> '1.5 GHz')."""
        hz = float(val)
        self._validate_frequency(hz)
        if hz >= 1e9:
            return f"{hz/1e9:.6f} GHz"
        if hz >= 1e6:
            return f"{hz/1e6:.6f} MHz"
        if hz >= 1e3:
            return f"{hz/1e3:.6f} kHz"
        return f"{hz:.0f} Hz"

    def format_power(self, dbm: float) -> str:
        self._validate_power(dbm)
        return f"{dbm:.2f} DBM"

    def _unsupported_feature(self, feature_name: str):
        print(f"Warning: Feature '{feature_name}' is not supported by {self.identity.get('model', 'Instrument')}")

    def _validate_frequency(self, hz: float):
        if hz < self.min_frequency or hz > self.max_frequency:
            raise ConfigurationError(f"Frequency {hz} Hz out of safety range")

    def _validate_power(self, dbm: float):
        if dbm > self.max_power_dbm:
            raise OverloadError(f"Power {dbm} dBm exceeds safety limit")

    # --- Measurements ---
    @abstractmethod
    def measure_frequency(self) -> MeasurementResult: pass
    @abstractmethod
    def measure_duty_cycle(self) -> MeasurementResult: pass
    @abstractmethod
    def measure_v_peak_to_peak(self) -> MeasurementResult: pass

    def __enter__(self):
        self.connect()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        try:
            self.shutdown_safety()
        except Exception:
            pass
        self.disconnect()

__getattr__(name)

Dynamic async wrapper for all driver methods.

Source code in src/instrumation/drivers/base.py 
25
26
27
28
29
30
31
32
33
34
def __getattr__(self, name: str):
    """Dynamic async wrapper for all driver methods."""
    if name.startswith("async_"):
        sync_name = name[6:]
        if hasattr(self, sync_name):
            sync_method = getattr(self, sync_name)
            async def wrapper(*args, **kwargs):
                return await asyncio.to_thread(sync_method, *args, **kwargs)
            return wrapper
    raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{name}'")

check_errors() abstractmethod

Queries SYST:ERR? and updates local error_stack.

Source code in src/instrumation/drivers/base.py 
 99
100
101
102
@abstractmethod
def check_errors(self):
    """Queries SYST:ERR? and updates local error_stack."""
    pass

clear_status() abstractmethod

Executes *CLS.

Source code in src/instrumation/drivers/base.py 
79
80
81
82
@abstractmethod
def clear_status(self):
    """Executes *CLS."""
    pass

connect() abstractmethod

Establishes connection and performs identity/option discovery.

Source code in src/instrumation/drivers/base.py 
41
42
43
44
@abstractmethod
def connect(self):
    """Establishes connection and performs identity/option discovery."""
    pass

disconnect() abstractmethod

Safely tears down connection.

Source code in src/instrumation/drivers/base.py 
46
47
48
49
@abstractmethod
def disconnect(self):
    """Safely tears down connection."""
    pass

format_frequency(val)

Ensures input is Hz and formats for SCPI (e.g. 1.5e9 -> '1.5 GHz').

Source code in src/instrumation/drivers/base.py 
113
114
115
116
117
118
119
120
121
122
123
def format_frequency(self, val: Union[float, str]) -> str:
    """Ensures input is Hz and formats for SCPI (e.g. 1.5e9 -> '1.5 GHz')."""
    hz = float(val)
    self._validate_frequency(hz)
    if hz >= 1e9:
        return f"{hz/1e9:.6f} GHz"
    if hz >= 1e6:
        return f"{hz/1e6:.6f} MHz"
    if hz >= 1e3:
        return f"{hz/1e3:.6f} kHz"
    return f"{hz:.0f} Hz"

load_state(index)

Recalls state from memory.

Source code in src/instrumation/drivers/base.py 
108
109
110
def load_state(self, index: Union[int, str]):
    """Recalls state from memory."""
    self._unsupported_feature("load_state")

query_ascii(command)

Sends command, reads response, and checks for errors.

Source code in src/instrumation/drivers/base.py 
64
65
66
def query_ascii(self, command: str) -> str:
    """Sends command, reads response, and checks for errors."""
    raise NotImplementedError()

query_binary_values(command, datatype='f', is_big_endian=False)

High-speed binary data transfer.

Source code in src/instrumation/drivers/base.py 
68
69
70
def query_binary_values(self, command: str, datatype: str = 'f', is_big_endian: bool = False) -> List[float]:
    """High-speed binary data transfer."""
    raise NotImplementedError()

safe_send(command)

Sends command and immediately checks SYST:ERR?.

Source code in src/instrumation/drivers/base.py 
60
61
62
def safe_send(self, command: str):
    """Sends command and immediately checks SYST:ERR?."""
    raise NotImplementedError()

save_state(index)

Saves current state to memory.

Source code in src/instrumation/drivers/base.py 
104
105
106
def save_state(self, index: Union[int, str]):
    """Saves current state to memory."""
    self._unsupported_feature("save_state")

shutdown_safety() abstractmethod

Emergency shutdown protocol (Outputs OFF, Power/Volt 0).

Source code in src/instrumation/drivers/base.py 
94
95
96
97
@abstractmethod
def shutdown_safety(self):
    """Emergency shutdown protocol (Outputs OFF, Power/Volt 0)."""
    pass

sync_config() abstractmethod

Executes CLS and WAI for a clean slate.

Source code in src/instrumation/drivers/base.py 
84
85
86
87
@abstractmethod
def sync_config(self):
    """Executes *CLS and *WAI for a clean slate."""
    pass

wait_ready(timeout=30.0) abstractmethod

Standard polling loop for *OPC?.

Source code in src/instrumation/drivers/base.py 
89
90
91
92
@abstractmethod
def wait_ready(self, timeout: float = 30.0):
    """Standard polling loop for *OPC?."""
    pass

Bases: InstrumentDriver

Source code in src/instrumation/drivers/base.py 
293
294
295
296
297
298
299
300
301
302
303
304
305
class Multimeter(InstrumentDriver):
    @abstractmethod
    def configure_voltage_dc(self): pass
    @abstractmethod
    def configure_voltage_ac(self): pass
    @abstractmethod
    def measure_voltage(self, ac: bool = False) -> MeasurementResult: pass
    @abstractmethod
    def measure_resistance(self, four_wire: bool = False) -> MeasurementResult: pass
    @abstractmethod
    def measure_current(self, ac: bool = False) -> MeasurementResult: pass
    @abstractmethod
    def set_auto_range(self, state: bool): pass

Bases: InstrumentDriver

Source code in src/instrumation/drivers/base.py 
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
class PowerSupply(InstrumentDriver):
    @abstractmethod
    def set_voltage(self, voltage: float): pass
    @abstractmethod
    def get_voltage(self) -> float: pass
    @abstractmethod
    def set_current_limit(self, current: float): pass
    def set_current(self, current: float):
        """Generalized alias for set_current_limit."""
        self.set_current_limit(current)
    @abstractmethod
    def get_current(self) -> MeasurementResult: pass
    @abstractmethod
    def set_output(self, state: bool): pass
    @abstractmethod
    def get_output(self) -> bool: pass
    @abstractmethod
    def set_ovp(self, voltage: float): pass
    @abstractmethod
    def set_ocp(self, current: float): pass
    @abstractmethod
    def measure_voltage_actual(self) -> MeasurementResult: pass
    @abstractmethod
    def measure_current(self) -> MeasurementResult: pass

    def set_voltage_limit(self, voltage: float):
        """Generalized alias for Over-Voltage Protection (OVP)."""
        self.set_ovp(voltage)

    def measure_voltage(self) -> MeasurementResult:
        """Generalized alias for measure_voltage_actual."""
        return self.measure_voltage_actual()

    @abstractmethod
    def clear_protection(self): pass

    def measure_power(self) -> MeasurementResult:
        """Queries the actual measured output power (Watts)."""
        self._unsupported_feature("measure_power")
        return MeasurementResult(0.0, "W")

    def set_foldback_mode(self, mode: str):
        """Sets the foldback protection mode (OFF, CC, or CV)."""
        self._unsupported_feature("set_foldback_mode")

    def set_foldback_delay(self, seconds: float):
        """Sets the delay for foldback protection."""
        self._unsupported_feature("set_foldback_delay")

    def set_autostart(self, state: bool):
        """Sets the Power-ON state (SAFE/OFF or AUTO/ON)."""
        self._unsupported_feature("set_autostart")

    def get_mode(self) -> str:
        """Returns the current operation mode (CV, CC, or OFF)."""
        self._unsupported_feature("get_mode")
        return "OFF"

get_mode()

Returns the current operation mode (CV, CC, or OFF).

Source code in src/instrumation/drivers/base.py 
360
361
362
363
def get_mode(self) -> str:
    """Returns the current operation mode (CV, CC, or OFF)."""
    self._unsupported_feature("get_mode")
    return "OFF"

measure_power()

Queries the actual measured output power (Watts).

Source code in src/instrumation/drivers/base.py 
343
344
345
346
def measure_power(self) -> MeasurementResult:
    """Queries the actual measured output power (Watts)."""
    self._unsupported_feature("measure_power")
    return MeasurementResult(0.0, "W")

measure_voltage()

Generalized alias for measure_voltage_actual.

Source code in src/instrumation/drivers/base.py 
336
337
338
def measure_voltage(self) -> MeasurementResult:
    """Generalized alias for measure_voltage_actual."""
    return self.measure_voltage_actual()

set_autostart(state)

Sets the Power-ON state (SAFE/OFF or AUTO/ON).

Source code in src/instrumation/drivers/base.py 
356
357
358
def set_autostart(self, state: bool):
    """Sets the Power-ON state (SAFE/OFF or AUTO/ON)."""
    self._unsupported_feature("set_autostart")

set_current(current)

Generalized alias for set_current_limit.

Source code in src/instrumation/drivers/base.py 
314
315
316
def set_current(self, current: float):
    """Generalized alias for set_current_limit."""
    self.set_current_limit(current)

set_foldback_delay(seconds)

Sets the delay for foldback protection.

Source code in src/instrumation/drivers/base.py 
352
353
354
def set_foldback_delay(self, seconds: float):
    """Sets the delay for foldback protection."""
    self._unsupported_feature("set_foldback_delay")

set_foldback_mode(mode)

Sets the foldback protection mode (OFF, CC, or CV).

Source code in src/instrumation/drivers/base.py 
348
349
350
def set_foldback_mode(self, mode: str):
    """Sets the foldback protection mode (OFF, CC, or CV)."""
    self._unsupported_feature("set_foldback_mode")

set_voltage_limit(voltage)

Generalized alias for Over-Voltage Protection (OVP).

Source code in src/instrumation/drivers/base.py 
332
333
334
def set_voltage_limit(self, voltage: float):
    """Generalized alias for Over-Voltage Protection (OVP)."""
    self.set_ovp(voltage)

Bases: InstrumentDriver

Source code in src/instrumation/drivers/base.py 
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
class SpectrumAnalyzer(InstrumentDriver):
    @abstractmethod
    def peak_search(self): pass
    @abstractmethod
    def get_marker_amplitude(self) -> MeasurementResult: pass
    @abstractmethod
    def set_center_freq(self, hz: float): pass
    @abstractmethod
    def get_center_freq(self) -> float: pass
    @abstractmethod
    def set_span(self, hz: float): pass
    @abstractmethod
    def get_span(self) -> float: pass
    @abstractmethod
    def set_rbw(self, hz: float): pass
    @abstractmethod
    def set_vbw(self, hz: float): pass
    @abstractmethod
    def get_trace_data(self) -> MeasurementResult: pass

    def get_peak_value(self) -> MeasurementResult:
        """Helper: Performs peak search and returns marker amplitude."""
        self.peak_search()
        return self.get_marker_amplitude()

get_peak_value()

Helper: Performs peak search and returns marker amplitude.

Source code in src/instrumation/drivers/base.py 
385
386
387
388
def get_peak_value(self) -> MeasurementResult:
    """Helper: Performs peak search and returns marker amplitude."""
    self.peak_search()
    return self.get_marker_amplitude()

Bases: InstrumentDriver

Source code in src/instrumation/drivers/base.py 
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
class SignalGenerator(InstrumentDriver):
    @abstractmethod
    def set_frequency(self, hz: float): pass
    @abstractmethod
    def set_amplitude(self, dbm: float): pass
    @abstractmethod
    def set_output(self, state: bool): pass
    @abstractmethod
    def set_mod_state(self, mod_type: str, state: bool): pass
    @abstractmethod
    def start_sweep(self, start: float, stop: float, points: int, dwell: float): pass
    @abstractmethod
    def configure_list_sweep(self, freq_list: List[float], power_list: List[float]): pass
    @abstractmethod
    def set_reference_clock(self, source: str): pass

Bases: InstrumentDriver

Abstract Base for Frequency Counters / Timer/Counter instruments.

Source code in src/instrumation/drivers/base.py 
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
class FrequencyCounter(InstrumentDriver):
    """Abstract Base for Frequency Counters / Timer/Counter instruments."""

    @abstractmethod
    def measure_frequency(self, range: str = "AUTO") -> MeasurementResult:
        """Measures frequency. Range can be 'AUTO' or a specific range in Hz."""
        pass

    @abstractmethod
    def measure_period(self, range: str = "AUTO") -> MeasurementResult:
        """Measures period. Range can be 'AUTO' or a specific range in seconds."""
        pass

    @abstractmethod
    def measure_time_interval(self, start_trigger: str, stop_trigger: str) -> MeasurementResult:
        """Measures time interval between two events (e.g. 'CH1', 'CH2')."""
        pass

    @abstractmethod
    def set_impedance(self, ohms: float):
        """Sets input impedance (50 or 1e6)."""
        pass

    @abstractmethod
    def set_trigger_level(self, volts: float):
        """Sets the trigger level voltage."""
        pass

    @abstractmethod
    def set_coupling(self, dc_ac: str):
        """Sets input coupling — 'DC' or 'AC'."""
        pass

    @abstractmethod
    def set_auto_range(self, state: bool):
        """Enables or disables auto-ranging."""
        pass

measure_frequency(range='AUTO') abstractmethod

Measures frequency. Range can be 'AUTO' or a specific range in Hz.

Source code in src/instrumation/drivers/base.py 
258
259
260
261
@abstractmethod
def measure_frequency(self, range: str = "AUTO") -> MeasurementResult:
    """Measures frequency. Range can be 'AUTO' or a specific range in Hz."""
    pass

measure_period(range='AUTO') abstractmethod

Measures period. Range can be 'AUTO' or a specific range in seconds.

Source code in src/instrumation/drivers/base.py 
263
264
265
266
@abstractmethod
def measure_period(self, range: str = "AUTO") -> MeasurementResult:
    """Measures period. Range can be 'AUTO' or a specific range in seconds."""
    pass

measure_time_interval(start_trigger, stop_trigger) abstractmethod

Measures time interval between two events (e.g. 'CH1', 'CH2').

Source code in src/instrumation/drivers/base.py 
268
269
270
271
@abstractmethod
def measure_time_interval(self, start_trigger: str, stop_trigger: str) -> MeasurementResult:
    """Measures time interval between two events (e.g. 'CH1', 'CH2')."""
    pass

set_auto_range(state) abstractmethod

Enables or disables auto-ranging.

Source code in src/instrumation/drivers/base.py 
288
289
290
291
@abstractmethod
def set_auto_range(self, state: bool):
    """Enables or disables auto-ranging."""
    pass

set_coupling(dc_ac) abstractmethod

Sets input coupling — 'DC' or 'AC'.

Source code in src/instrumation/drivers/base.py 
283
284
285
286
@abstractmethod
def set_coupling(self, dc_ac: str):
    """Sets input coupling — 'DC' or 'AC'."""
    pass

set_impedance(ohms) abstractmethod

Sets input impedance (50 or 1e6).

Source code in src/instrumation/drivers/base.py 
273
274
275
276
@abstractmethod
def set_impedance(self, ohms: float):
    """Sets input impedance (50 or 1e6)."""
    pass

set_trigger_level(volts) abstractmethod

Sets the trigger level voltage.

Source code in src/instrumation/drivers/base.py 
278
279
280
281
@abstractmethod
def set_trigger_level(self, volts: float):
    """Sets the trigger level voltage."""
    pass

Results

Standardized object for measurement results.

Attributes:

Name Type Description
value Any

The measured value (float, complex, list, or numpy array).
unit str

The physical unit of the measurement (e.g., 'V', 'Hz', 'dBm').
timestamp datetime

The time the measurement was taken.
status str

Status of the measurement ('OK', 'ERROR', 'OVERLOAD', etc.).
channel Optional[Union[int, str]]

Optional channel index for multi-channel instruments.
metadata Optional[Dict[str, Any]]

Optional additional information from the driver.
Source code in src/instrumation/results.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@dataclass
class MeasurementResult:
    """Standardized object for measurement results.

    Attributes:
        value: The measured value (float, complex, list, or numpy array).
        unit: The physical unit of the measurement (e.g., 'V', 'Hz', 'dBm').
        timestamp: The time the measurement was taken.
        status: Status of the measurement ('OK', 'ERROR', 'OVERLOAD', etc.).
        channel: Optional channel index for multi-channel instruments.
        metadata: Optional additional information from the driver.
    """
    value: Any
    unit: str
    timestamp: datetime = field(default_factory=datetime.now)
    status: str = "OK"
    channel: Optional[Union[int, str]] = None
    metadata: Optional[Dict[str, Any]] = field(default_factory=dict)

    def __str__(self):
        chan_str = f" [CH {self.channel}]" if self.channel is not None else ""
        return f"{self.value} {self.unit}{chan_str} ({self.status}) @ {self.timestamp.isoformat()}"

    def __format__(self, format_spec):
        """Allows MeasurementResult to be used in f-strings with float formatting."""
        if isinstance(self.value, (float, int)):
            return format(float(self.value), format_spec)
        return str(self.value)

    def __float__(self):
        """Allows direct conversion to float if the value is a scalar."""
        return float(self.value)

    def __len__(self):
        """Allows MeasurementResult to be used with len() if the value is a collection."""
        return len(self.value)

    def __getitem__(self, key):
        """Allows indexing into the MeasurementResult value."""
        return self.value[key]

    def __iter__(self):
        """Allows iterating over the MeasurementResult value."""
        return iter(self.value)

    def to_dict(self) -> Dict[str, Any]:
        """Converts the result to a JSON-serializable dictionary."""
        val = self.value

        # Handle numpy arrays
        if np and isinstance(val, np.ndarray):
            val = val.tolist()

        # Handle complex numbers (common in VNA/IQ data)
        if isinstance(val, complex):
            val = {"real": val.real, "imag": val.imag}
        elif isinstance(val, list):
            # Recursively handle complex numbers in lists
            val = [
                {"real": v.real, "imag": v.imag} if isinstance(v, complex) else v 
                for v in val
            ]

        return {
            "value": val,
            "unit": self.unit,
            "timestamp": self.timestamp.isoformat(),
            "status": self.status,
            "channel": self.channel,
            "metadata": self.metadata
        }

    def to_json(self) -> str:
        """Returns the JSON string representation of the result."""
        return json.dumps(self.to_dict())

__float__()

Allows direct conversion to float if the value is a scalar.

Source code in src/instrumation/results.py 
40
41
42
def __float__(self):
    """Allows direct conversion to float if the value is a scalar."""
    return float(self.value)

__format__(format_spec)

Allows MeasurementResult to be used in f-strings with float formatting.

Source code in src/instrumation/results.py 
34
35
36
37
38
def __format__(self, format_spec):
    """Allows MeasurementResult to be used in f-strings with float formatting."""
    if isinstance(self.value, (float, int)):
        return format(float(self.value), format_spec)
    return str(self.value)

__getitem__(key)

Allows indexing into the MeasurementResult value.

Source code in src/instrumation/results.py 
48
49
50
def __getitem__(self, key):
    """Allows indexing into the MeasurementResult value."""
    return self.value[key]

__iter__()

Allows iterating over the MeasurementResult value.

Source code in src/instrumation/results.py 
52
53
54
def __iter__(self):
    """Allows iterating over the MeasurementResult value."""
    return iter(self.value)

__len__()

Allows MeasurementResult to be used with len() if the value is a collection.

Source code in src/instrumation/results.py 
44
45
46
def __len__(self):
    """Allows MeasurementResult to be used with len() if the value is a collection."""
    return len(self.value)

to_dict()

Converts the result to a JSON-serializable dictionary.

Source code in src/instrumation/results.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def to_dict(self) -> Dict[str, Any]:
    """Converts the result to a JSON-serializable dictionary."""
    val = self.value

    # Handle numpy arrays
    if np and isinstance(val, np.ndarray):
        val = val.tolist()

    # Handle complex numbers (common in VNA/IQ data)
    if isinstance(val, complex):
        val = {"real": val.real, "imag": val.imag}
    elif isinstance(val, list):
        # Recursively handle complex numbers in lists
        val = [
            {"real": v.real, "imag": v.imag} if isinstance(v, complex) else v 
            for v in val
        ]

    return {
        "value": val,
        "unit": self.unit,
        "timestamp": self.timestamp.isoformat(),
        "status": self.status,
        "channel": self.channel,
        "metadata": self.metadata
    }

to_json()

Returns the JSON string representation of the result.

Source code in src/instrumation/results.py 
83
84
85
def to_json(self) -> str:
    """Returns the JSON string representation of the result."""
    return json.dumps(self.to_dict())

Factory

Source code in src/instrumation/factory.py 
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
def get_instrument(resource_address: str, driver_type: str = "GENERIC") -> any:
    # 0. Check for replay mode (Highest Priority)
    if resource_address.startswith("replay://"):
        file_path = resource_address.replace("replay://", "")
        from .drivers.replay import ReplayDriver
        return ReplayDriver(resource_address, master_file=file_path)

    # 1. Handle Simulation Mode (The Digital Twin Path)
    if is_sim_mode():
        from .drivers.simulated import SimulatedMultimeter
        drivers = DriverRegistry.get_drivers_by_type(driver_type)
        for drv_cls in drivers:
            if "Simulated" in drv_cls.__name__:
                # Use the requested address or a mock one
                addr = resource_address if resource_address != "AUTO" else "USB0::SIM::INSTR"
                drv = drv_cls(addr)
                drv.connect()
                return drv

        # If explicitly requested a type and not found, raise error (don't fallback to DMM silently)
        if driver_type != "GENERIC":
            raise ValueError(f"No simulated driver found for type: {driver_type}")

        # Fallback for GENERIC only
        drv = SimulatedMultimeter(resource_address if resource_address != "AUTO" else "USB0::SIM::INSTR")
        drv.connect()
        return drv

    # 2. Handle AUTO discovery
    if resource_address == "AUTO":
        import json
        from concurrent.futures import ThreadPoolExecutor, as_completed
        cache_file = ".visa_cache.json"

        # 1. Load Cache & LAN (The Fast Resources)
        cached_resources = []
        if os.path.exists(cache_file):
            try:
                with open(cache_file, "r") as f:
                    cached_resources = json.load(f)
            except (IOError, OSError, json.JSONDecodeError):
                pass

        lan_resources = _discover_lan_resources()
        tried = set()

        def run_probe(resources, desc):
            # Sort by priority and recency
            candidates = []
            for r in resources:
                if r not in tried:
                    candidates.append(r)

            if not candidates:
                return None

            # Sort: Priority first, then preserve order (recency)
            candidates.sort(key=lambda x: "ASRL5" in x or "TCPIP" in x or "USB0" in x, reverse=True)

            logger.info(f"AUTO-Discovery checking {desc}: {candidates}")

            if len(candidates) <= 2:
                for res in candidates:
                    tried.add(res)
                    result = probe_resource(res)
                    if result:
                        update_cache(result.resource_address)
                        return result
                return None

            with ThreadPoolExecutor(max_workers=4) as executor:
                future_to_res = {executor.submit(probe_resource, res): res for res in candidates}
                for future in as_completed(future_to_res):
                    tried.add(future_to_res[future])
                    result = future.result()
                    if result:
                        update_cache(result.resource_address)
                        return result
            return None

        def update_cache(res):
            try:
                # Move successful resource to the front of the cache
                new_cache = [res] + [r for r in cached_resources if r != res]
                with open(cache_file, "w") as f:
                    json.dump(new_cache[:10], f) # Keep top 10 for speed
            except (IOError, OSError):
                pass

        def probe_resource(res):
            try:
                if "ASRL" in res and any(p in res for p in ["1", "2", "3", "4"]):
                    return None
                dev = get_instrument(res, driver_type)
                type_map = {"SCOPE": Oscilloscope, "SA": SpectrumAnalyzer, "SG": (SignalGenerator, FunctionGenerator), "PSU": PowerSupply, "DMM": Multimeter, "VNA": NetworkAnalyzer, "NA": NetworkAnalyzer, "LOAD": ElectronicLoad, "ELOAD": ElectronicLoad, "COUNTER": FrequencyCounter}
                if driver_type == "GENERIC" or (type_map.get(driver_type) and isinstance(dev, type_map.get(driver_type))):
                    return dev
                dev.disconnect()
            except Exception:
                pass
            return None

        # --- Phase 0: Try ONLY Cache (Super Fast) ---
        if cached_resources:
            result = run_probe(cached_resources, "Super Fast (Cache Only)")
            if result:
                return result

        # --- Phase 0.5: Try mDNS (Bonjour) ---
        mdns_resources = _discover_mdns_resources()
        result = run_probe(mdns_resources, "mDNS/Bonjour")
        if result:
            return result

        # --- Phase 1: Try LAN (Quick Search) ---
        result = run_probe(lan_resources, "Fast Track (LAN)")
        if result:
            return result

        # --- Phase 2: Try Full VISA Scan (The 10s Tax) ---
        rm = get_rm()
        visa_resources = list(rm.list_resources())
        result = run_probe(visa_resources, "Slow Track (Full Scan)")
        if result:
            return result

        raise ValueError(f"AUTO-Discovery could not find a suitable {driver_type} instrument.")

    # 4. Real Hardware Logic
    idn = ""
    try:
        if "SIM" in resource_address or "MOCK" in resource_address:
            idn = ""
        else:
            base_dev = RealDriver(resource_address, rm=get_rm())

            # Smart Probe for Serial Ports (like TDK-Lambda)
            if "ASRL" in resource_address:
                try:
                    base_dev.inst = base_dev.rm.open_resource(resource_address)
                    base_dev.inst.baud_rate = 9600
                    base_dev.inst.read_termination = '\r\n'
                    base_dev.inst.write_termination = '\r\n'
                    base_dev.inst.timeout = 500 # 500ms is enough for local Serial
                    base_dev.inst.write('INST:NSEL 6')
                    time.sleep(0.2)
                    idn = base_dev.inst.query("*IDN?").upper()
                    base_dev.inst.close()
                except Exception:
                    pass

            if not idn:
                base_dev.connect()
                # Set a safer timeout for the ID query during discovery
                base_dev.inst.timeout = 2000
                idn = base_dev.get_id().upper()
                base_dev.disconnect()
    except Exception as e:
        logger.warning(f"Identification failed for {resource_address}: {e}")
        idn = ""

    # Smart Routing based on IDN
    final_drv = None
    if "TEKTRONIX" in idn:
        if "AFG" in idn:
            from .drivers.tektronix import TektronixAFG
            final_drv = TektronixAFG(resource_address)
        else:
            from .drivers.tektronix import TektronixTDS
            final_drv = TektronixTDS(resource_address)
    elif "KEYSIGHT" in idn or "AGILENT" in idn or "HEWLETT-PACKARD" in idn or "HP" in idn:
        if any(m in idn for m in ["DSO-X", "MSO-X", "DSOX", "MSOX"]):
            from .drivers.keysight import KeysightInfiniiVision
            final_drv = KeysightInfiniiVision(resource_address)
        elif any(m in idn for m in ["N9030", "N9020", "N9010", "PXA", "MXA", "EXA"]):
            from .drivers.keysight import KeysightPXA
            final_drv = KeysightPXA(resource_address)
        elif any(m in idn for m in ["E8257", "N5181", "N5182", "N5183", "PSG", "MXG", "EXG"]):
            from .drivers.keysight import KeysightSG
            final_drv = KeysightSG(resource_address)
        elif "N99" in idn or "FIELD FOX" in idn:
            from .drivers.keysight import KeysightFieldFox
            final_drv = KeysightFieldFox(resource_address)
        elif "34461" in idn or "34460" in idn:
            from .drivers.keysight import Keysight34461A
            final_drv = Keysight34461A(resource_address)
        elif "E83" in idn or "N52" in idn or "PNA" in idn:
            from .drivers.keysight import KeysightPNA
            final_drv = KeysightPNA(resource_address)
        elif any(m in idn for m in ["34401", "34410", "34411", "34420"]):
            from .drivers.keysight import Keysight34461A
            final_drv = Keysight34461A(resource_address)
    elif "SIGLENT" in idn:
        from .drivers.siglent import SiglentSDS
        final_drv = SiglentSDS(resource_address)
    elif "KEITHLEY" in idn:
        if "2400" in idn:
            from .drivers.keithley import Keithley2400
            final_drv = Keithley2400(resource_address)
        elif "2000" in idn:
            from .drivers.keithley import Keithley2000
            final_drv = Keithley2000(resource_address)
    elif "TDK-LAMBDA" in idn or "Z+" in idn:
        from .drivers.tdk import TDKLambdaZPlus
        final_drv = TDKLambdaZPlus(resource_address)

    if not final_drv:
        drivers = DriverRegistry.get_drivers_by_type(driver_type)
        for drv_cls in drivers:
            if "Simulated" not in drv_cls.__name__:
                final_drv = drv_cls(resource_address)
                break
    if not final_drv:
        final_drv = RealDriver(resource_address, rm=get_rm())


    final_drv.connect()

    # Update cache with successful manual connection to enable future AUTO discovery
    if resource_address != "AUTO":
        try:
            cache_file = ".visa_cache.json"
            cached_resources = []
            if os.path.exists(cache_file):
                with open(cache_file, "r") as f:
                    cached_resources = json.load(f)
            new_cache = [resource_address] + [r for r in cached_resources if r != resource_address]
            with open(cache_file, "w") as f:
                json.dump(new_cache[:10], f)
        except Exception:
            pass

    return final_drv

Dynamically loads all available instrument drivers.

Source code in src/instrumation/factory.py 
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
def load_plugins(plugin_path: str = None):
    """Dynamically loads all available instrument drivers."""
    import importlib
    import pkgutil
    import sys

    # 1. Load built-in drivers
    import instrumation.drivers as drivers_pkg
    for _, name, _ in pkgutil.iter_modules(drivers_pkg.__path__):
        importlib.import_module(f"instrumation.drivers.{name}")

    # 2. Load from external path if provided
    if plugin_path:
        if plugin_path not in sys.path:
            sys.path.insert(0, plugin_path)
        for _, name, _ in pkgutil.iter_modules([plugin_path]):
            importlib.import_module(name)