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) -> None:
        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) -> Any:
        """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: Any, **kwargs: Any) -> Any:
                    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) -> None:
        """Establishes connection and performs identity/option discovery."""
        pass

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

    def close(self) -> None:
        self.disconnect()

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

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

    def safe_send(self, command: str) -> None:
        """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) -> None: pass

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

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

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

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

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

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

    def load_state(self, index: Union[int, str]) -> None:
        """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) -> None:
        print(f"Warning: Feature '{feature_name}' is not supported by {self.identity.get('model', 'Instrument')}")

    def _validate_frequency(self, hz: float) -> None:
        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) -> None:
        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) -> "InstrumentDriver":
        self.connect()
        return self

    def __exit__(self, exc_type: Optional[type], exc_val: Optional[BaseException], exc_tb: Any) -> None:
        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) -> Any:
    """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: Any, **kwargs: Any) -> Any:
                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) -> None:
    """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) -> None:
    """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) -> None:
    """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) -> None:
    """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]) -> None:
    """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) -> None:
    """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]) -> None:
    """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) -> None:
    """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) -> None:
    """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) -> None:
    """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) -> None: pass
    @abstractmethod
    def configure_voltage_ac(self) -> None: 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) -> None: 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) -> None: pass
    @abstractmethod
    def get_voltage(self) -> float: pass
    @abstractmethod
    def set_current_limit(self, current: float) -> None: pass
    def set_current(self, current: float) -> None:
        """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) -> None: pass
    @abstractmethod
    def get_output(self) -> bool: pass
    @abstractmethod
    def set_ovp(self, voltage: float) -> None: pass
    @abstractmethod
    def set_ocp(self, current: float) -> None: pass
    @abstractmethod
    def measure_voltage_actual(self) -> MeasurementResult: pass
    @abstractmethod
    def measure_current(self) -> MeasurementResult: pass

    def set_voltage_limit(self, voltage: float) -> None:
        """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) -> None: 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) -> None:
        """Sets the foldback protection mode (OFF, CC, or CV)."""
        self._unsupported_feature("set_foldback_mode")

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

    def set_autostart(self, state: bool) -> None:
        """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) -> None:
    """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) -> None:
    """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) -> None:
    """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) -> None:
    """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) -> None:
    """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) -> None: pass
    @abstractmethod
    def get_marker_amplitude(self) -> MeasurementResult: pass
    @abstractmethod
    def set_center_freq(self, hz: float) -> None: pass
    @abstractmethod
    def get_center_freq(self) -> float: pass
    @abstractmethod
    def set_span(self, hz: float) -> None: pass
    @abstractmethod
    def get_span(self) -> float: pass
    @abstractmethod
    def set_rbw(self, hz: float) -> None: pass
    @abstractmethod
    def set_vbw(self, hz: float) -> None: 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 
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
class NetworkAnalyzer(InstrumentDriver):
    @abstractmethod
    def set_start_frequency(self, freq_hz: float) -> None: pass
    @abstractmethod
    def set_stop_frequency(self, freq_hz: float) -> None: pass

    def set_center_freq(self, freq_hz: float) -> None: 
        self._unsupported_feature("set_center_freq")

    def set_center_frequency(self, freq_hz: float) -> None:
        """Alias for set_center_freq."""
        self.set_center_freq(freq_hz)

    def set_span(self, span_hz: float) -> None: 
        self._unsupported_feature("set_span")

    @abstractmethod
    def set_points(self, num_points: int) -> None: pass

    def set_if_bandwidth(self, hz: float) -> None: 
        self._unsupported_feature("set_if_bandwidth")

    def set_power_level(self, dbm: float) -> None: 
        self._unsupported_feature("set_power_level")

    def set_sweep_type(self, sweep_type: str) -> None: 
        self._unsupported_feature("set_sweep_type")

    def set_averaging(self, state: bool, count: int = 10) -> None: 
        self._unsupported_feature("set_averaging")

    def set_continuous(self, state: bool) -> None: 
        self._unsupported_feature("set_continuous")

    @abstractmethod
    def set_parameter(self, parameter: str) -> None: pass  # e.g., "S11", "S21"

    @abstractmethod
    def get_trace_data(self, measurement_name: str = "CH1_S11_1") -> MeasurementResult: pass

    @abstractmethod
    def get_complex_trace(self, measurement_name: str = "CH1_S11_1") -> MeasurementResult: pass

    @abstractmethod
    def get_smith_data(self, measurement_name: str = "CH1_S11_1") -> MeasurementResult: pass

    def peak_search(self, marker: int = 1) -> None: 
        self._unsupported_feature("peak_search")

    def get_marker_x(self, marker: int = 1) -> float: 
        self._unsupported_feature("get_marker_x")
        return 0.0

    def get_marker_y(self, marker: int = 1) -> float: 
        self._unsupported_feature("get_marker_y")
        return 0.0

    def save_state(self, filename: str) -> None: 
        self._unsupported_feature("save_state")

    def load_state(self, filename: str) -> None: 
        self._unsupported_feature("load_state")

    def wait_for_sweep(self) -> None:
        """Wait for the current sweep to complete."""
        self._unsupported_feature("wait_for_sweep")

set_center_frequency(freq_hz)

Alias for set_center_freq.

Source code in src/instrumation/drivers/base.py 
399
400
401
def set_center_frequency(self, freq_hz: float) -> None:
    """Alias for set_center_freq."""
    self.set_center_freq(freq_hz)

wait_for_sweep()

Wait for the current sweep to complete.

Source code in src/instrumation/drivers/base.py 
453
454
455
def wait_for_sweep(self) -> None:
    """Wait for the current sweep to complete."""
    self._unsupported_feature("wait_for_sweep")

Bases: InstrumentDriver

Source code in src/instrumation/drivers/base.py 
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
class Oscilloscope(InstrumentDriver):
    @abstractmethod
    def run(self) -> None: pass
    @abstractmethod
    def stop(self) -> None: pass
    @abstractmethod
    def single(self) -> None: pass
    @abstractmethod
    def get_waveform(self, channel: int) -> MeasurementResult: pass
    @abstractmethod
    def auto_scale(self) -> None: pass
    @abstractmethod
    def set_trigger(self, source: str, level: float, slope: str) -> None: pass
    @abstractmethod
    def get_screenshot(self) -> bytes: pass
    @abstractmethod
    def measure_frequency(self, channel: int = 1) -> MeasurementResult: pass
    @abstractmethod
    def measure_duty_cycle(self, channel: int = 1) -> MeasurementResult: pass
    @abstractmethod
    def measure_v_peak_to_peak(self, channel: int = 1) -> MeasurementResult: pass

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) -> None: pass
    @abstractmethod
    def set_amplitude(self, dbm: float) -> None: pass
    @abstractmethod
    def set_output(self, state: bool) -> None: pass
    @abstractmethod
    def set_mod_state(self, mod_type: str, state: bool) -> None: pass
    @abstractmethod
    def start_sweep(self, start: float, stop: float, points: int, dwell: float) -> None: pass
    @abstractmethod
    def configure_list_sweep(self, freq_list: List[float], power_list: List[float]) -> None: pass
    @abstractmethod
    def set_reference_clock(self, source: str) -> None: pass

Bases: InstrumentDriver

Source code in src/instrumation/drivers/base.py 
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
class ElectronicLoad(InstrumentDriver):
    @abstractmethod
    def set_mode(self, mode: str) -> None:
        """Sets the operating mode, typically CC, CV, CR, or CP."""
        pass

    @abstractmethod
    def get_mode(self) -> str:
        """Returns the active operating mode."""
        pass

    @abstractmethod
    def set_current(self, amps: float) -> None:
        """Sets the constant current value in CC mode."""
        pass

    @abstractmethod
    def get_current(self) -> float:
        """Returns the set current value in CC mode."""
        pass

    @abstractmethod
    def set_voltage(self, volts: float) -> None:
        """Sets the constant voltage value in CV mode."""
        pass

    @abstractmethod
    def get_voltage(self) -> float:
        """Returns the set voltage value in CV mode."""
        pass

    @abstractmethod
    def set_resistance(self, ohms: float) -> None:
        """Sets the constant resistance value in CR mode."""
        pass

    @abstractmethod
    def get_resistance(self) -> float:
        """Returns the set resistance value in CR mode."""
        pass

    @abstractmethod
    def set_power(self, watts: float) -> None:
        """Sets the constant power value in CP mode."""
        pass

    @abstractmethod
    def get_power(self) -> float:
        """Returns the set power value in CP mode."""
        pass

    @abstractmethod
    def set_input(self, state: bool) -> None:
        """Turns the load input ON (True) or OFF (False)."""
        pass

    @abstractmethod
    def get_input(self) -> bool:
        """Returns the input state (ON/OFF)."""
        pass

    @abstractmethod
    def measure_voltage(self) -> MeasurementResult:
        """Measures the actual input voltage at the load terminals."""
        pass

    @abstractmethod
    def measure_current(self) -> MeasurementResult:
        """Measures the actual current being drawn by the load."""
        pass

    @abstractmethod
    def measure_power(self) -> MeasurementResult:
        """Measures the actual power being consumed by the load."""
        pass

    @abstractmethod
    def set_ovp(self, voltage: float) -> None:
        """Sets the over-voltage protection limit."""
        pass

    @abstractmethod
    def set_ocp(self, current: float) -> None:
        """Sets the over-current protection limit."""
        pass

    @abstractmethod
    def set_opp(self, power: float) -> None:
        """Sets the over-power protection limit."""
        pass

    @abstractmethod
    def clear_protection(self) -> None:
        """Clears any tripped protection status."""
        pass

clear_protection() abstractmethod

Clears any tripped protection status.

Source code in src/instrumation/drivers/base.py 
250
251
252
253
@abstractmethod
def clear_protection(self) -> None:
    """Clears any tripped protection status."""
    pass

get_current() abstractmethod

Returns the set current value in CC mode.

Source code in src/instrumation/drivers/base.py 
175
176
177
178
@abstractmethod
def get_current(self) -> float:
    """Returns the set current value in CC mode."""
    pass

get_input() abstractmethod

Returns the input state (ON/OFF).

Source code in src/instrumation/drivers/base.py 
215
216
217
218
@abstractmethod
def get_input(self) -> bool:
    """Returns the input state (ON/OFF)."""
    pass

get_mode() abstractmethod

Returns the active operating mode.

Source code in src/instrumation/drivers/base.py 
165
166
167
168
@abstractmethod
def get_mode(self) -> str:
    """Returns the active operating mode."""
    pass

get_power() abstractmethod

Returns the set power value in CP mode.

Source code in src/instrumation/drivers/base.py 
205
206
207
208
@abstractmethod
def get_power(self) -> float:
    """Returns the set power value in CP mode."""
    pass

get_resistance() abstractmethod

Returns the set resistance value in CR mode.

Source code in src/instrumation/drivers/base.py 
195
196
197
198
@abstractmethod
def get_resistance(self) -> float:
    """Returns the set resistance value in CR mode."""
    pass

get_voltage() abstractmethod

Returns the set voltage value in CV mode.

Source code in src/instrumation/drivers/base.py 
185
186
187
188
@abstractmethod
def get_voltage(self) -> float:
    """Returns the set voltage value in CV mode."""
    pass

measure_current() abstractmethod

Measures the actual current being drawn by the load.

Source code in src/instrumation/drivers/base.py 
225
226
227
228
@abstractmethod
def measure_current(self) -> MeasurementResult:
    """Measures the actual current being drawn by the load."""
    pass

measure_power() abstractmethod

Measures the actual power being consumed by the load.

Source code in src/instrumation/drivers/base.py 
230
231
232
233
@abstractmethod
def measure_power(self) -> MeasurementResult:
    """Measures the actual power being consumed by the load."""
    pass

measure_voltage() abstractmethod

Measures the actual input voltage at the load terminals.

Source code in src/instrumation/drivers/base.py 
220
221
222
223
@abstractmethod
def measure_voltage(self) -> MeasurementResult:
    """Measures the actual input voltage at the load terminals."""
    pass

set_current(amps) abstractmethod

Sets the constant current value in CC mode.

Source code in src/instrumation/drivers/base.py 
170
171
172
173
@abstractmethod
def set_current(self, amps: float) -> None:
    """Sets the constant current value in CC mode."""
    pass

set_input(state) abstractmethod

Turns the load input ON (True) or OFF (False).

Source code in src/instrumation/drivers/base.py 
210
211
212
213
@abstractmethod
def set_input(self, state: bool) -> None:
    """Turns the load input ON (True) or OFF (False)."""
    pass

set_mode(mode) abstractmethod

Sets the operating mode, typically CC, CV, CR, or CP.

Source code in src/instrumation/drivers/base.py 
160
161
162
163
@abstractmethod
def set_mode(self, mode: str) -> None:
    """Sets the operating mode, typically CC, CV, CR, or CP."""
    pass

set_ocp(current) abstractmethod

Sets the over-current protection limit.

Source code in src/instrumation/drivers/base.py 
240
241
242
243
@abstractmethod
def set_ocp(self, current: float) -> None:
    """Sets the over-current protection limit."""
    pass

set_opp(power) abstractmethod

Sets the over-power protection limit.

Source code in src/instrumation/drivers/base.py 
245
246
247
248
@abstractmethod
def set_opp(self, power: float) -> None:
    """Sets the over-power protection limit."""
    pass

set_ovp(voltage) abstractmethod

Sets the over-voltage protection limit.

Source code in src/instrumation/drivers/base.py 
235
236
237
238
@abstractmethod
def set_ovp(self, voltage: float) -> None:
    """Sets the over-voltage protection limit."""
    pass

set_power(watts) abstractmethod

Sets the constant power value in CP mode.

Source code in src/instrumation/drivers/base.py 
200
201
202
203
@abstractmethod
def set_power(self, watts: float) -> None:
    """Sets the constant power value in CP mode."""
    pass

set_resistance(ohms) abstractmethod

Sets the constant resistance value in CR mode.

Source code in src/instrumation/drivers/base.py 
190
191
192
193
@abstractmethod
def set_resistance(self, ohms: float) -> None:
    """Sets the constant resistance value in CR mode."""
    pass

set_voltage(volts) abstractmethod

Sets the constant voltage value in CV mode.

Source code in src/instrumation/drivers/base.py 
180
181
182
183
@abstractmethod
def set_voltage(self, volts: float) -> None:
    """Sets the constant voltage value in CV mode."""
    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) -> None:
        """Sets input impedance (50 or 1e6)."""
        pass

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

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

    @abstractmethod
    def set_auto_range(self, state: bool) -> None:
        """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) -> None:
    """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) -> None:
    """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) -> None:
    """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) -> None:
    """Sets the trigger level voltage."""
    pass

Oscilloscopes

Bases: RealDriver, Oscilloscope

Driver for Rigol DS1054Z Digital Oscilloscope.

Supports the MSO1000Z/DS1000Z Series (DS1054Z, DS1104Z, MSO1054Z, etc.). Implements edge trigger only; LA, :SOURce, :DECoder, :MASK, :FUNCtion commands are excluded (option-gated or -S variant features).

Source code in src/instrumation/drivers/rigol.py 
 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
296
297
298
299
300
301
302
303
304
305
306
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
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
@register_driver("SCOPE")
class RigolDS1054Z(RealDriver, Oscilloscope):
    """Driver for Rigol DS1054Z Digital Oscilloscope.

    Supports the MSO1000Z/DS1000Z Series (DS1054Z, DS1104Z, MSO1054Z, etc.).
    Implements edge trigger only; LA, :SOURce, :DECoder, :MASK, :FUNCtion
    commands are excluded (option-gated or -S variant features).
    """

    def __init__(self, resource: str, rm=None) -> None:
        super().__init__(resource, rm)
        self.max_voltage = 40.0
        self._channel_count = 4

    def connect(self) -> None:
        super().connect()
        self.inst.timeout = 10000
        self.write("*CLS")
        self.write("*WAI")

    # ── IEEE-488.2 Common Commands ─────────────────────────────

    def preset(self, automation_optimized: bool = True) -> None:
        """*RST — Factory default reset."""
        self.write("*RST")
        self.wait_ready()

    def clear_status(self) -> None:
        """*CLS — Clear status registers."""
        self.write("*CLS")

    def sync_config(self) -> None:
        """*CLS + *WAI — Synchronize."""
        self.write("*CLS")
        self.write("*WAI")

    def get_id(self) -> str:
        """*IDN? — Query instrument identification."""
        return self.query("*IDN?")

    # ── Basic Control ──────────────────────────────────────────

    def auto_scale(self) -> None:
        """:AUToscale — Automatically adjust scale for all channels."""
        self.write(":AUToscale")
        self.wait_ready()

    def clear_waveform(self) -> None:
        """:CLEar — Clear waveform data."""
        self.write(":CLEar")

    def run(self) -> None:
        """:RUN — Start acquisition."""
        self.write(":RUN")

    def stop(self) -> None:
        """:STOP — Stop acquisition."""
        self.write(":STOP")

    def single(self) -> None:
        """:SINGle — Single-shot acquisition."""
        self.write(":SINGle")

    def force_trigger(self) -> None:
        """:TFORce — Force a trigger event."""
        self.write(":TFORce")

    # ── Acquisition Configuration ──────────────────────────────

    def set_acquire_type(self, acquire_type: str) -> None:
        """:ACQuire:TYPE — Set acquisition mode.

        Args:
            acquire_type: NORMal, AVERages, PEAK, or HRESolution.
        """
        valid = {"NORMAL", "AVERAGES", "PEAK", "HRESOLUTION"}
        if acquire_type.upper() not in valid:
            raise ValueError(f"Invalid acquire type: {acquire_type}")
        self.safe_send(f":ACQuire:TYPE {acquire_type}")

    def get_acquire_type(self) -> str:
        """:ACQuire:TYPE? — Query acquisition mode."""
        return self.query(":ACQuire:TYPE?")

    def set_acquire_averages(self, count: int) -> None:
        """:ACQuire:AVERages — Set averaging count (2^n, n=1..10).

        Args:
            count: Must be a power of 2 between 2 and 1024.
        """
        if count < 2 or count > 1024:
            raise ValueError(f"Average count must be 2^n (2..1024), got {count}")
        if count & (count - 1) != 0:
            raise ValueError(f"Average count must be a power of 2, got {count}")
        self.safe_send(f":ACQuire:AVERages {count}")

    def set_acquire_memory_depth(self, mdep: int) -> None:
        """:ACQuire:MDEPth — Set memory depth.

        Args:
            mdep: Memory depth in points (14k, 140k, or 14M for DS1054Z).
        """
        self.safe_send(f":ACQuire:MDEPth {mdep}")

    def get_sample_rate(self) -> float:
        """:ACQuire:SRATe? — Query current sample rate (samples/sec)."""
        return float(self.query(":ACQuire:SRATe?"))

    # ── Channel Configuration (per channel n=1..4) ─────────────

    def _validate_channel(self, channel: int) -> None:
        if channel < 1 or channel > self._channel_count:
            raise ValueError(f"Channel must be 1..{self._channel_count}, got {channel}")

    def set_channel_display(self, channel: int, state: bool) -> None:
        """:CHANnel<n>:DISPlay — Enable/disable channel display.

        Args:
            channel: 1-4
            state: True=ON, False=OFF
        """
        self._validate_channel(channel)
        self.safe_send(f":CHANnel{channel}:DISPlay {'ON' if state else 'OFF'}")

    def get_channel_display(self, channel: int) -> bool:
        """:CHANnel<n>:DISPlay? — Query channel display state."""
        self._validate_channel(channel)
        return self.query(f":CHANnel{channel}:DISPlay?") == "1"

    def set_channel_coupling(self, channel: int, coupling: str) -> None:
        """:CHANnel<n>:COUPling — Set input coupling.

        Args:
            channel: 1-4
            coupling: AC, DC, or GND
        """
        self._validate_channel(channel)
        valid = {"AC", "DC", "GND"}
        if coupling.upper() not in valid:
            raise ValueError(f"Invalid coupling: {coupling}")
        self.safe_send(f":CHANnel{channel}:COUPling {coupling.upper()}")

    def get_channel_coupling(self, channel: int) -> str:
        """:CHANnel<n>:COUPling? — Query input coupling."""
        self._validate_channel(channel)
        return self.query(f":CHANnel{channel}:COUPling?")

    def set_channel_scale(self, channel: int, scale: float) -> None:
        """:CHANnel<n>:SCALe — Set vertical scale (volts/div).

        Args:
            channel: 1-4
            scale: Volts per division (e.g. 0.01 to 10.0)
        """
        self._validate_channel(channel)
        self.safe_send(f":CHANnel{channel}:SCALe {scale}")

    def get_channel_scale(self, channel: int) -> float:
        """:CHANnel<n>:SCALe? — Query vertical scale."""
        self._validate_channel(channel)
        return float(self.query(f":CHANnel{channel}:SCALe?"))

    def set_channel_offset(self, channel: int, offset: float) -> None:
        """:CHANnel<n>:OFFSet — Set vertical offset (volts).

        Args:
            channel: 1-4
            offset: DC offset voltage
        """
        self._validate_channel(channel)
        self.safe_send(f":CHANnel{channel}:OFFSet {offset}")

    def get_channel_offset(self, channel: int) -> float:
        """:CHANnel<n>:OFFSet? — Query vertical offset."""
        self._validate_channel(channel)
        return float(self.query(f":CHANnel{channel}:OFFSet?"))

    def set_channel_probe(self, channel: int, attenuation: float) -> None:
        """:CHANnel<n>:PROBe — Set probe attenuation factor.

        Args:
            channel: 1-4
            attenuation: Probe ratio (e.g. 1.0, 10.0)
        """
        self._validate_channel(channel)
        self.safe_send(f":CHANnel{channel}:PROBe {attenuation}")

    def get_channel_probe(self, channel: int) -> float:
        """:CHANnel<n>:PROBe? — Query probe attenuation."""
        self._validate_channel(channel)
        return float(self.query(f":CHANnel{channel}:PROBe?"))

    def set_channel_bw_limit(self, channel: int, bw: str) -> None:
        """:CHANnel<n>:BWLimit — Set bandwidth limit.

        Args:
            channel: 1-4
            bw: 20M (20 MHz) or OFF
        """
        self._validate_channel(channel)
        valid = {"20M", "OFF"}
        if bw.upper() not in valid:
            raise ValueError(f"Invalid bandwidth limit: {bw}")
        self.safe_send(f":CHANnel{channel}:BWLimit {bw.upper()}")

    def get_channel_bw_limit(self, channel: int) -> str:
        """:CHANnel<n>:BWLimit? — Query bandwidth limit."""
        self._validate_channel(channel)
        return self.query(f":CHANnel{channel}:BWLimit?")

    def set_channel_invert(self, channel: int, state: bool) -> None:
        """:CHANnel<n>:INVert — Enable/disable channel inversion.

        Args:
            channel: 1-4
            state: True=ON, False=OFF
        """
        self._validate_channel(channel)
        self.safe_send(f":CHANnel{channel}:INVert {'ON' if state else 'OFF'}")

    def get_channel_invert(self, channel: int) -> bool:
        """:CHANnel<n>:INVert? — Query channel inversion state."""
        self._validate_channel(channel)
        return self.query(f":CHANnel{channel}:INVert?") == "1"

    def set_channel_units(self, channel: int, units: str) -> None:
        """:CHANnel<n>:UNITs — Set channel display units.

        Args:
            channel: 1-4
            units: VOLTage, WATT, AMPere, or UNKNown
        """
        self._validate_channel(channel)
        valid = {"VOLTAGE", "WATT", "AMPERE", "UNKNOWN"}
        if units.upper() not in valid:
            raise ValueError(f"Invalid units: {units}")
        self.safe_send(f":CHANnel{channel}:UNITs {units.upper()}")

    # ── Timebase Configuration ─────────────────────────────────

    def get_timebase_mode(self) -> str:
        """:TIMebase:MODE — Query timebase mode (MAIN, XY, ROLL)."""
        return self.query(":TIMebase:MODE?")

    def set_timebase_scale(self, scale: float) -> None:
        """:TIMebase[:MAIN]:SCALe — Set horizontal scale (seconds/div).

        Args:
            scale: Seconds per division (e.g. 1e-9 to 50.0)
        """
        self.safe_send(f":TIMebase:SCALe {scale}")

    def get_timebase_scale(self) -> float:
        """:TIMebase[:MAIN]:SCALe? — Query horizontal scale."""
        return float(self.query(":TIMebase:SCALe?"))

    def set_timebase_offset(self, offset: float) -> None:
        """:TIMebase[:MAIN]:OFFSet — Set horizontal offset (seconds).

        Args:
            offset: Time offset from trigger point
        """
        self.safe_send(f":TIMebase:OFFSet {offset}")

    def get_timebase_offset(self) -> float:
        """:TIMebase[:MAIN]:OFFSet? — Query horizontal offset."""
        return float(self.query(":TIMebase:OFFSet?"))

    # ── Trigger Configuration (Edge Only) ──────────────────────

    def get_trigger_mode(self) -> str:
        """:TRIGger:MODE — Query trigger mode (EDGE, PULSE, etc.)."""
        return self.query(":TRIGger:MODE?")

    def set_edge_trigger_source(self, source: str) -> None:
        """:TRIGger:EDGe:SOURce — Set edge trigger source.

        Args:
            source: CHANnel1, CHANnel2, CHANnel3, CHANnel4, EXT, or DEMath
        """
        self.safe_send(f":TRIGger:EDGe:SOURce {source}")

    def get_edge_trigger_source(self) -> str:
        """:TRIGger:EDGe:SOURce? — Query edge trigger source."""
        return self.query(":TRIGger:EDGe:SOURce?")

    def set_edge_trigger_slope(self, slope: str) -> None:
        """:TRIGger:EDGe:SLOPe — Set edge trigger slope.

        Args:
            slope: POSitive, NEGative, or RFALl
        """
        valid = {"POSITIVE", "NEGATIVE", "RFALL"}
        if slope.upper() not in valid:
            raise ValueError(f"Invalid slope: {slope}")
        self.safe_send(f":TRIGger:EDGe:SLOPe {slope}")

    def get_edge_trigger_slope(self) -> str:
        """:TRIGger:EDGe:SLOPe? — Query edge trigger slope."""
        return self.query(":TRIGger:EDGe:SLOPe?")

    def set_edge_trigger_level(self, level: float) -> None:
        """:TRIGger:EDGe:LEVel — Set trigger level voltage."""
        self.safe_send(f":TRIGger:EDGe:LEVel {level}")

    def get_edge_trigger_level(self) -> float:
        """:TRIGger:EDGe:LEVel? — Query trigger level voltage."""
        return float(self.query(":TRIGger:EDGe:LEVel?"))

    def set_trigger_sweep(self, sweep: str) -> None:
        """:TRIGger:SWEep — Set trigger sweep mode.

        Args:
            sweep: AUTO, NORMal, or SINGle
        """
        valid = {"AUTO", "NORMAL", "SINGLE"}
        if sweep.upper() not in valid:
            raise ValueError(f"Invalid sweep mode: {sweep}")
        self.safe_send(f":TRIGger:SWEep {sweep.upper()}")

    def get_trigger_sweep(self) -> str:
        """:TRIGger:SWEep? — Query trigger sweep mode."""
        return self.query(":TRIGger:SWEep?")

    def get_trigger_status(self) -> str:
        """:TRIGger:STATus? — Query trigger status.

        Returns: TD (triggered), WAIT, AUTO, STOP, T'D, etc.
        """
        return self.query(":TRIGger:STATus?")

    def set_trigger(self, source: str, level: float, slope: str) -> None:
        """Configure edge trigger (Oscilloscope interface convenience method).

        Args:
            source: CHANnel1, CHANnel2, etc.
            level: Trigger level in volts
            slope: POSITIVE, NEGATIVE, or RFALL
        """
        self.safe_send(":TRIGger:MODE EDGE")
        self.set_edge_trigger_source(source)
        self.set_edge_trigger_level(level)
        self.set_edge_trigger_slope(slope)

    # ── Waveform Readout ───────────────────────────────────────

    def set_waveform_source(self, channel: int) -> None:
        """:WAVeform:SOURce — Select waveform data source.

        Args:
            channel: 1-4 (maps to CHANnel1..CHANnel4)
        """
        self._validate_channel(channel)
        self.safe_send(f":WAVeform:SOURce CHANnel{channel}")

    def get_waveform_source(self) -> str:
        """:WAVeform:SOURce? — Query waveform data source."""
        return self.query(":WAVeform:SOURce?")

    def set_waveform_mode(self, mode: str) -> None:
        """:WAVeform:MODE — Set waveform readout mode.

        Args:
            mode: NORMal, MAXimum, or RAW
        """
        valid = {"NORMAL", "MAXIMUM", "RAW"}
        if mode.upper() not in valid:
            raise ValueError(f"Invalid waveform mode: {mode}")
        self.safe_send(f":WAVeform:MODE {mode.upper()}")

    def set_waveform_format(self, fmt: str) -> None:
        """:WAVeform:FORMat — Set waveform data format.

        Args:
            fmt: WORD, BYTE, or ASCii
        """
        valid = {"WORD", "BYTE", "ASCII"}
        if fmt.upper() not in valid:
            raise ValueError(f"Invalid waveform format: {fmt}")
        self.safe_send(f":WAVeform:FORMat {fmt.upper()}")

    def get_waveform_preamble(self) -> dict:
        """:WAVeform:PREamble? — Query and parse the waveform preamble.

        Returns a dict with keys: format, type, points, count,
        x_increment, x_origin, x_reference, y_increment, y_origin, y_reference.
        """
        resp = self.query(":WAVeform:PREamble?")
        parts = resp.split(",")
        if len(parts) < 10:
            raise ValueError(f"Unexpected preamble format: {resp}")
        return {
            "format": int(parts[0]),
            "type": int(parts[1]),
            "points": int(parts[2]),
            "count": int(parts[3]),
            "x_increment": float(parts[4]),
            "x_origin": float(parts[5]),
            "x_reference": int(parts[6]),
            "y_increment": float(parts[7]),
            "y_origin": float(parts[8]),
            "y_reference": int(parts[9]),
        }

    def get_waveform_x_increment(self) -> float:
        """:WAVeform:XINCrement? — Query X-axis increment (seconds/sample)."""
        return float(self.query(":WAVeform:XINCrement?"))

    def get_waveform_x_origin(self) -> float:
        """:WAVeform:XORigin? — Query X-axis origin (seconds)."""
        return float(self.query(":WAVeform:XORigin?"))

    def get_waveform_x_reference(self) -> int:
        """:WAVeform:XREFerence? — Query X-axis reference point."""
        return int(self.query(":WAVeform:XREFerence?"))

    def get_waveform_y_increment(self) -> float:
        """:WAVeform:YINCrement? — Query Y-axis increment (volts/code)."""
        return float(self.query(":WAVeform:YINCrement?"))

    def get_waveform_y_origin(self) -> float:
        """:WAVeform:YORigin? — Query Y-axis origin (volts)."""
        return float(self.query(":WAVeform:YORigin?"))

    def get_waveform_y_reference(self) -> int:
        """:WAVeform:YREFerence? — Query Y-axis reference code."""
        return int(self.query(":WAVeform:YREFerence?"))

    def get_waveform_raw(self, channel: int) -> List[int]:
        """:WAVeform:DATA? — Fetch raw waveform data as unsigned integers.

        Args:
            channel: 1-4

        Returns:
            List of raw ADC codes (unsigned 8-bit or 16-bit).
        """
        self.set_waveform_source(channel)
        self.set_waveform_format("WORD")
        self.write(":WAVeform:BYTEorder LSBFirst")
        raw = self.query_binary_values(
            ":WAVeform:DATA?", datatype="H", is_big_endian=False
        )
        return [int(v) for v in raw]

    def get_waveform(self, channel: int) -> MeasurementResult:
        """Fetch calibrated waveform data for a channel.

        Queries :WAVeform:PREamble? and :WAVeform:DATA?, converts raw ADC
        codes to real voltage values using: V = (raw - yref) * yinc + yor
        and generates a time axis using: t = (n - xref) * xinc + xor

        Args:
            channel: 1-4

        Returns:
            MeasurementResult with value=(time_array, voltage_array), unit="V".
        """
        preamble = self.get_waveform_preamble()
        raw = self.get_waveform_raw(channel)

        y_inc = preamble["y_increment"]
        y_origin = preamble["y_origin"]
        y_ref = preamble["y_reference"]
        x_inc = preamble["x_increment"]
        x_origin = preamble["x_origin"]
        x_ref = preamble["x_reference"]

        voltage = [((v - y_ref) * y_inc) + y_origin for v in raw]
        time_axis = [
            ((n - x_ref) * x_inc) + x_origin for n in range(len(raw))
        ]

        if np is not None:
            time_arr = np.array(time_axis)
            voltage_arr = np.array(voltage)
        else:
            time_arr = time_axis
            voltage_arr = voltage

        return MeasurementResult(
            value=(time_arr, voltage_arr),
            unit="V",
            channel=channel,
            metadata={"preamble": preamble},
        )

    # ── Measurement Helpers (Oscilloscope interface) ───────────

    def measure_frequency(self, channel: int = 1) -> MeasurementResult:
        """:MEASure:FREQuency? — Measure frequency on a channel."""
        val = self.query(f":MEASure:FREQuency? CHANnel{channel}")
        return MeasurementResult(float(val), "Hz", channel=channel)

    def measure_duty_cycle(self, channel: int = 1) -> MeasurementResult:
        """:MEASure:DUTYcycle? — Measure duty cycle on a channel."""
        val = self.query(f":MEASure:DUTYcycle? CHANnel{channel}")
        return MeasurementResult(float(val), "%", channel=channel)

    def measure_v_peak_to_peak(self, channel: int = 1) -> MeasurementResult:
        """:MEASure:VPP? — Measure Vpp on a channel."""
        val = self.query(f":MEASure:VPP? CHANnel{channel}")
        return MeasurementResult(float(val), "V", channel=channel)

    def get_screenshot(self) -> bytes:
        """:DISPlay:DATA? — Capture display screenshot as PNG."""
        self.write(":DISPlay:DATA? PNG, COLor")
        return self.inst.read_raw()

    # ── Safety & Shutdown ──────────────────────────────────────

    def shutdown_safety(self) -> None:
        """Stop acquisition and sync."""
        self.stop()
        self.sync_config()

auto_scale()

:AUToscale — Automatically adjust scale for all channels.

Source code in src/instrumation/drivers/rigol.py 
110
111
112
113
def auto_scale(self) -> None:
    """:AUToscale — Automatically adjust scale for all channels."""
    self.write(":AUToscale")
    self.wait_ready()

clear_status()

*CLS — Clear status registers.

Source code in src/instrumation/drivers/rigol.py 
95
96
97
def clear_status(self) -> None:
    """*CLS — Clear status registers."""
    self.write("*CLS")

clear_waveform()

:CLEar — Clear waveform data.

Source code in src/instrumation/drivers/rigol.py 
115
116
117
def clear_waveform(self) -> None:
    """:CLEar — Clear waveform data."""
    self.write(":CLEar")

force_trigger()

:TFORce — Force a trigger event.

Source code in src/instrumation/drivers/rigol.py 
131
132
133
def force_trigger(self) -> None:
    """:TFORce — Force a trigger event."""
    self.write(":TFORce")

get_acquire_type()

:ACQuire:TYPE? — Query acquisition mode.

Source code in src/instrumation/drivers/rigol.py 
148
149
150
def get_acquire_type(self) -> str:
    """:ACQuire:TYPE? — Query acquisition mode."""
    return self.query(":ACQuire:TYPE?")

get_channel_bw_limit(channel)

:CHANnel:BWLimit? — Query bandwidth limit.

Source code in src/instrumation/drivers/rigol.py 
273
274
275
276
def get_channel_bw_limit(self, channel: int) -> str:
    """:CHANnel<n>:BWLimit? — Query bandwidth limit."""
    self._validate_channel(channel)
    return self.query(f":CHANnel{channel}:BWLimit?")

get_channel_coupling(channel)

:CHANnel:COUPling? — Query input coupling.

Source code in src/instrumation/drivers/rigol.py 
210
211
212
213
def get_channel_coupling(self, channel: int) -> str:
    """:CHANnel<n>:COUPling? — Query input coupling."""
    self._validate_channel(channel)
    return self.query(f":CHANnel{channel}:COUPling?")

get_channel_display(channel)

:CHANnel:DISPlay? — Query channel display state.

Source code in src/instrumation/drivers/rigol.py 
192
193
194
195
def get_channel_display(self, channel: int) -> bool:
    """:CHANnel<n>:DISPlay? — Query channel display state."""
    self._validate_channel(channel)
    return self.query(f":CHANnel{channel}:DISPlay?") == "1"

get_channel_invert(channel)

:CHANnel:INVert? — Query channel inversion state.

Source code in src/instrumation/drivers/rigol.py 
288
289
290
291
def get_channel_invert(self, channel: int) -> bool:
    """:CHANnel<n>:INVert? — Query channel inversion state."""
    self._validate_channel(channel)
    return self.query(f":CHANnel{channel}:INVert?") == "1"

get_channel_offset(channel)

:CHANnel:OFFSet? — Query vertical offset.

Source code in src/instrumation/drivers/rigol.py 
240
241
242
243
def get_channel_offset(self, channel: int) -> float:
    """:CHANnel<n>:OFFSet? — Query vertical offset."""
    self._validate_channel(channel)
    return float(self.query(f":CHANnel{channel}:OFFSet?"))

get_channel_probe(channel)

:CHANnel:PROBe? — Query probe attenuation.

Source code in src/instrumation/drivers/rigol.py 
255
256
257
258
def get_channel_probe(self, channel: int) -> float:
    """:CHANnel<n>:PROBe? — Query probe attenuation."""
    self._validate_channel(channel)
    return float(self.query(f":CHANnel{channel}:PROBe?"))

get_channel_scale(channel)

:CHANnel:SCALe? — Query vertical scale.

Source code in src/instrumation/drivers/rigol.py 
225
226
227
228
def get_channel_scale(self, channel: int) -> float:
    """:CHANnel<n>:SCALe? — Query vertical scale."""
    self._validate_channel(channel)
    return float(self.query(f":CHANnel{channel}:SCALe?"))

get_edge_trigger_level()

:TRIGger:EDGe:LEVel? — Query trigger level voltage.

Source code in src/instrumation/drivers/rigol.py 
373
374
375
def get_edge_trigger_level(self) -> float:
    """:TRIGger:EDGe:LEVel? — Query trigger level voltage."""
    return float(self.query(":TRIGger:EDGe:LEVel?"))

get_edge_trigger_slope()

:TRIGger:EDGe:SLOPe? — Query edge trigger slope.

Source code in src/instrumation/drivers/rigol.py 
365
366
367
def get_edge_trigger_slope(self) -> str:
    """:TRIGger:EDGe:SLOPe? — Query edge trigger slope."""
    return self.query(":TRIGger:EDGe:SLOPe?")

get_edge_trigger_source()

:TRIGger:EDGe:SOURce? — Query edge trigger source.

Source code in src/instrumation/drivers/rigol.py 
350
351
352
def get_edge_trigger_source(self) -> str:
    """:TRIGger:EDGe:SOURce? — Query edge trigger source."""
    return self.query(":TRIGger:EDGe:SOURce?")

get_id()

*IDN? — Query instrument identification.

Source code in src/instrumation/drivers/rigol.py 
104
105
106
def get_id(self) -> str:
    """*IDN? — Query instrument identification."""
    return self.query("*IDN?")

get_sample_rate()

:ACQuire:SRATe? — Query current sample rate (samples/sec).

Source code in src/instrumation/drivers/rigol.py 
172
173
174
def get_sample_rate(self) -> float:
    """:ACQuire:SRATe? — Query current sample rate (samples/sec)."""
    return float(self.query(":ACQuire:SRATe?"))

get_screenshot()

:DISPlay:DATA? — Capture display screenshot as PNG.

Source code in src/instrumation/drivers/rigol.py 
572
573
574
575
def get_screenshot(self) -> bytes:
    """:DISPlay:DATA? — Capture display screenshot as PNG."""
    self.write(":DISPlay:DATA? PNG, COLor")
    return self.inst.read_raw()

get_timebase_mode()

:TIMebase:MODE — Query timebase mode (MAIN, XY, ROLL).

Source code in src/instrumation/drivers/rigol.py 
308
309
310
def get_timebase_mode(self) -> str:
    """:TIMebase:MODE — Query timebase mode (MAIN, XY, ROLL)."""
    return self.query(":TIMebase:MODE?")

get_timebase_offset()

:TIMebase[:MAIN]:OFFSet? — Query horizontal offset.

Source code in src/instrumation/drivers/rigol.py 
332
333
334
def get_timebase_offset(self) -> float:
    """:TIMebase[:MAIN]:OFFSet? — Query horizontal offset."""
    return float(self.query(":TIMebase:OFFSet?"))

get_timebase_scale()

:TIMebase[:MAIN]:SCALe? — Query horizontal scale.

Source code in src/instrumation/drivers/rigol.py 
320
321
322
def get_timebase_scale(self) -> float:
    """:TIMebase[:MAIN]:SCALe? — Query horizontal scale."""
    return float(self.query(":TIMebase:SCALe?"))

get_trigger_mode()

:TRIGger:MODE — Query trigger mode (EDGE, PULSE, etc.).

Source code in src/instrumation/drivers/rigol.py 
338
339
340
def get_trigger_mode(self) -> str:
    """:TRIGger:MODE — Query trigger mode (EDGE, PULSE, etc.)."""
    return self.query(":TRIGger:MODE?")

get_trigger_status()

:TRIGger:STATus? — Query trigger status.

Returns: TD (triggered), WAIT, AUTO, STOP, T'D, etc.

Source code in src/instrumation/drivers/rigol.py 
392
393
394
395
396
397
def get_trigger_status(self) -> str:
    """:TRIGger:STATus? — Query trigger status.

    Returns: TD (triggered), WAIT, AUTO, STOP, T'D, etc.
    """
    return self.query(":TRIGger:STATus?")

get_trigger_sweep()

:TRIGger:SWEep? — Query trigger sweep mode.

Source code in src/instrumation/drivers/rigol.py 
388
389
390
def get_trigger_sweep(self) -> str:
    """:TRIGger:SWEep? — Query trigger sweep mode."""
    return self.query(":TRIGger:SWEep?")

get_waveform(channel)

Fetch calibrated waveform data for a channel.

Queries :WAVeform:PREamble? and :WAVeform:DATA?, converts raw ADC codes to real voltage values using: V = (raw - yref) * yinc + yor and generates a time axis using: t = (n - xref) * xinc + xor

Parameters:

Name Type Description Default
channel int

1-4

 required

Returns:

Type Description
MeasurementResult

MeasurementResult with value=(time_array, voltage_array), unit="V".
Source code in src/instrumation/drivers/rigol.py 
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
def get_waveform(self, channel: int) -> MeasurementResult:
    """Fetch calibrated waveform data for a channel.

    Queries :WAVeform:PREamble? and :WAVeform:DATA?, converts raw ADC
    codes to real voltage values using: V = (raw - yref) * yinc + yor
    and generates a time axis using: t = (n - xref) * xinc + xor

    Args:
        channel: 1-4

    Returns:
        MeasurementResult with value=(time_array, voltage_array), unit="V".
    """
    preamble = self.get_waveform_preamble()
    raw = self.get_waveform_raw(channel)

    y_inc = preamble["y_increment"]
    y_origin = preamble["y_origin"]
    y_ref = preamble["y_reference"]
    x_inc = preamble["x_increment"]
    x_origin = preamble["x_origin"]
    x_ref = preamble["x_reference"]

    voltage = [((v - y_ref) * y_inc) + y_origin for v in raw]
    time_axis = [
        ((n - x_ref) * x_inc) + x_origin for n in range(len(raw))
    ]

    if np is not None:
        time_arr = np.array(time_axis)
        voltage_arr = np.array(voltage)
    else:
        time_arr = time_axis
        voltage_arr = voltage

    return MeasurementResult(
        value=(time_arr, voltage_arr),
        unit="V",
        channel=channel,
        metadata={"preamble": preamble},
    )

get_waveform_preamble()

:WAVeform:PREamble? — Query and parse the waveform preamble.

Returns a dict with keys: format, type, points, count, x_increment, x_origin, x_reference, y_increment, y_origin, y_reference.

Source code in src/instrumation/drivers/rigol.py 
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
def get_waveform_preamble(self) -> dict:
    """:WAVeform:PREamble? — Query and parse the waveform preamble.

    Returns a dict with keys: format, type, points, count,
    x_increment, x_origin, x_reference, y_increment, y_origin, y_reference.
    """
    resp = self.query(":WAVeform:PREamble?")
    parts = resp.split(",")
    if len(parts) < 10:
        raise ValueError(f"Unexpected preamble format: {resp}")
    return {
        "format": int(parts[0]),
        "type": int(parts[1]),
        "points": int(parts[2]),
        "count": int(parts[3]),
        "x_increment": float(parts[4]),
        "x_origin": float(parts[5]),
        "x_reference": int(parts[6]),
        "y_increment": float(parts[7]),
        "y_origin": float(parts[8]),
        "y_reference": int(parts[9]),
    }

get_waveform_raw(channel)

:WAVeform:DATA? — Fetch raw waveform data as unsigned integers.

Parameters:

Name Type Description Default
channel int

1-4

 required

Returns:

Type Description
List[int]

List of raw ADC codes (unsigned 8-bit or 16-bit).
Source code in src/instrumation/drivers/rigol.py 
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
def get_waveform_raw(self, channel: int) -> List[int]:
    """:WAVeform:DATA? — Fetch raw waveform data as unsigned integers.

    Args:
        channel: 1-4

    Returns:
        List of raw ADC codes (unsigned 8-bit or 16-bit).
    """
    self.set_waveform_source(channel)
    self.set_waveform_format("WORD")
    self.write(":WAVeform:BYTEorder LSBFirst")
    raw = self.query_binary_values(
        ":WAVeform:DATA?", datatype="H", is_big_endian=False
    )
    return [int(v) for v in raw]

get_waveform_source()

:WAVeform:SOURce? — Query waveform data source.

Source code in src/instrumation/drivers/rigol.py 
423
424
425
def get_waveform_source(self) -> str:
    """:WAVeform:SOURce? — Query waveform data source."""
    return self.query(":WAVeform:SOURce?")

get_waveform_x_increment()

:WAVeform:XINCrement? — Query X-axis increment (seconds/sample).

Source code in src/instrumation/drivers/rigol.py 
472
473
474
def get_waveform_x_increment(self) -> float:
    """:WAVeform:XINCrement? — Query X-axis increment (seconds/sample)."""
    return float(self.query(":WAVeform:XINCrement?"))

get_waveform_x_origin()

:WAVeform:XORigin? — Query X-axis origin (seconds).

Source code in src/instrumation/drivers/rigol.py 
476
477
478
def get_waveform_x_origin(self) -> float:
    """:WAVeform:XORigin? — Query X-axis origin (seconds)."""
    return float(self.query(":WAVeform:XORigin?"))

get_waveform_x_reference()

:WAVeform:XREFerence? — Query X-axis reference point.

Source code in src/instrumation/drivers/rigol.py 
480
481
482
def get_waveform_x_reference(self) -> int:
    """:WAVeform:XREFerence? — Query X-axis reference point."""
    return int(self.query(":WAVeform:XREFerence?"))

get_waveform_y_increment()

:WAVeform:YINCrement? — Query Y-axis increment (volts/code).

Source code in src/instrumation/drivers/rigol.py 
484
485
486
def get_waveform_y_increment(self) -> float:
    """:WAVeform:YINCrement? — Query Y-axis increment (volts/code)."""
    return float(self.query(":WAVeform:YINCrement?"))

get_waveform_y_origin()

:WAVeform:YORigin? — Query Y-axis origin (volts).

Source code in src/instrumation/drivers/rigol.py 
488
489
490
def get_waveform_y_origin(self) -> float:
    """:WAVeform:YORigin? — Query Y-axis origin (volts)."""
    return float(self.query(":WAVeform:YORigin?"))

get_waveform_y_reference()

:WAVeform:YREFerence? — Query Y-axis reference code.

Source code in src/instrumation/drivers/rigol.py 
492
493
494
def get_waveform_y_reference(self) -> int:
    """:WAVeform:YREFerence? — Query Y-axis reference code."""
    return int(self.query(":WAVeform:YREFerence?"))

measure_duty_cycle(channel=1)

:MEASure:DUTYcycle? — Measure duty cycle on a channel.

Source code in src/instrumation/drivers/rigol.py 
562
563
564
565
def measure_duty_cycle(self, channel: int = 1) -> MeasurementResult:
    """:MEASure:DUTYcycle? — Measure duty cycle on a channel."""
    val = self.query(f":MEASure:DUTYcycle? CHANnel{channel}")
    return MeasurementResult(float(val), "%", channel=channel)

measure_frequency(channel=1)

:MEASure:FREQuency? — Measure frequency on a channel.

Source code in src/instrumation/drivers/rigol.py 
557
558
559
560
def measure_frequency(self, channel: int = 1) -> MeasurementResult:
    """:MEASure:FREQuency? — Measure frequency on a channel."""
    val = self.query(f":MEASure:FREQuency? CHANnel{channel}")
    return MeasurementResult(float(val), "Hz", channel=channel)

measure_v_peak_to_peak(channel=1)

:MEASure:VPP? — Measure Vpp on a channel.

Source code in src/instrumation/drivers/rigol.py 
567
568
569
570
def measure_v_peak_to_peak(self, channel: int = 1) -> MeasurementResult:
    """:MEASure:VPP? — Measure Vpp on a channel."""
    val = self.query(f":MEASure:VPP? CHANnel{channel}")
    return MeasurementResult(float(val), "V", channel=channel)

preset(automation_optimized=True)

*RST — Factory default reset.

Source code in src/instrumation/drivers/rigol.py 
90
91
92
93
def preset(self, automation_optimized: bool = True) -> None:
    """*RST — Factory default reset."""
    self.write("*RST")
    self.wait_ready()

run()

:RUN — Start acquisition.

Source code in src/instrumation/drivers/rigol.py 
119
120
121
def run(self) -> None:
    """:RUN — Start acquisition."""
    self.write(":RUN")

set_acquire_averages(count)

:ACQuire:AVERages — Set averaging count (2^n, n=1..10).

Parameters:

Name Type Description Default
count int

Must be a power of 2 between 2 and 1024.

 required
Source code in src/instrumation/drivers/rigol.py 
152
153
154
155
156
157
158
159
160
161
162
def set_acquire_averages(self, count: int) -> None:
    """:ACQuire:AVERages — Set averaging count (2^n, n=1..10).

    Args:
        count: Must be a power of 2 between 2 and 1024.
    """
    if count < 2 or count > 1024:
        raise ValueError(f"Average count must be 2^n (2..1024), got {count}")
    if count & (count - 1) != 0:
        raise ValueError(f"Average count must be a power of 2, got {count}")
    self.safe_send(f":ACQuire:AVERages {count}")

set_acquire_memory_depth(mdep)

:ACQuire:MDEPth — Set memory depth.

Parameters:

Name Type Description Default
mdep int

Memory depth in points (14k, 140k, or 14M for DS1054Z).

 required
Source code in src/instrumation/drivers/rigol.py 
164
165
166
167
168
169
170
def set_acquire_memory_depth(self, mdep: int) -> None:
    """:ACQuire:MDEPth — Set memory depth.

    Args:
        mdep: Memory depth in points (14k, 140k, or 14M for DS1054Z).
    """
    self.safe_send(f":ACQuire:MDEPth {mdep}")

set_acquire_type(acquire_type)

:ACQuire:TYPE — Set acquisition mode.

Parameters:

Name Type Description Default
acquire_type str

NORMal, AVERages, PEAK, or HRESolution.

 required
Source code in src/instrumation/drivers/rigol.py 
137
138
139
140
141
142
143
144
145
146
def set_acquire_type(self, acquire_type: str) -> None:
    """:ACQuire:TYPE — Set acquisition mode.

    Args:
        acquire_type: NORMal, AVERages, PEAK, or HRESolution.
    """
    valid = {"NORMAL", "AVERAGES", "PEAK", "HRESOLUTION"}
    if acquire_type.upper() not in valid:
        raise ValueError(f"Invalid acquire type: {acquire_type}")
    self.safe_send(f":ACQuire:TYPE {acquire_type}")

set_channel_bw_limit(channel, bw)

:CHANnel:BWLimit — Set bandwidth limit.

Parameters:

Name Type Description Default
channel int

1-4

 required
bw str

20M (20 MHz) or OFF

 required
Source code in src/instrumation/drivers/rigol.py 
260
261
262
263
264
265
266
267
268
269
270
271
def set_channel_bw_limit(self, channel: int, bw: str) -> None:
    """:CHANnel<n>:BWLimit — Set bandwidth limit.

    Args:
        channel: 1-4
        bw: 20M (20 MHz) or OFF
    """
    self._validate_channel(channel)
    valid = {"20M", "OFF"}
    if bw.upper() not in valid:
        raise ValueError(f"Invalid bandwidth limit: {bw}")
    self.safe_send(f":CHANnel{channel}:BWLimit {bw.upper()}")

set_channel_coupling(channel, coupling)

:CHANnel:COUPling — Set input coupling.

Parameters:

Name Type Description Default
channel int

1-4

 required
coupling str

AC, DC, or GND

 required
Source code in src/instrumation/drivers/rigol.py 
197
198
199
200
201
202
203
204
205
206
207
208
def set_channel_coupling(self, channel: int, coupling: str) -> None:
    """:CHANnel<n>:COUPling — Set input coupling.

    Args:
        channel: 1-4
        coupling: AC, DC, or GND
    """
    self._validate_channel(channel)
    valid = {"AC", "DC", "GND"}
    if coupling.upper() not in valid:
        raise ValueError(f"Invalid coupling: {coupling}")
    self.safe_send(f":CHANnel{channel}:COUPling {coupling.upper()}")

set_channel_display(channel, state)

:CHANnel:DISPlay — Enable/disable channel display.

Parameters:

Name Type Description Default
channel int

1-4

 required
state bool

True=ON, False=OFF

 required
Source code in src/instrumation/drivers/rigol.py 
182
183
184
185
186
187
188
189
190
def set_channel_display(self, channel: int, state: bool) -> None:
    """:CHANnel<n>:DISPlay — Enable/disable channel display.

    Args:
        channel: 1-4
        state: True=ON, False=OFF
    """
    self._validate_channel(channel)
    self.safe_send(f":CHANnel{channel}:DISPlay {'ON' if state else 'OFF'}")

set_channel_invert(channel, state)

:CHANnel:INVert — Enable/disable channel inversion.

Parameters:

Name Type Description Default
channel int

1-4

 required
state bool

True=ON, False=OFF

 required
Source code in src/instrumation/drivers/rigol.py 
278
279
280
281
282
283
284
285
286
def set_channel_invert(self, channel: int, state: bool) -> None:
    """:CHANnel<n>:INVert — Enable/disable channel inversion.

    Args:
        channel: 1-4
        state: True=ON, False=OFF
    """
    self._validate_channel(channel)
    self.safe_send(f":CHANnel{channel}:INVert {'ON' if state else 'OFF'}")

set_channel_offset(channel, offset)

:CHANnel:OFFSet — Set vertical offset (volts).

Parameters:

Name Type Description Default
channel int

1-4

 required
offset float

DC offset voltage

 required
Source code in src/instrumation/drivers/rigol.py 
230
231
232
233
234
235
236
237
238
def set_channel_offset(self, channel: int, offset: float) -> None:
    """:CHANnel<n>:OFFSet — Set vertical offset (volts).

    Args:
        channel: 1-4
        offset: DC offset voltage
    """
    self._validate_channel(channel)
    self.safe_send(f":CHANnel{channel}:OFFSet {offset}")

set_channel_probe(channel, attenuation)

:CHANnel:PROBe — Set probe attenuation factor.

Parameters:

Name Type Description Default
channel int

1-4

 required
attenuation float

Probe ratio (e.g. 1.0, 10.0)

 required
Source code in src/instrumation/drivers/rigol.py 
245
246
247
248
249
250
251
252
253
def set_channel_probe(self, channel: int, attenuation: float) -> None:
    """:CHANnel<n>:PROBe — Set probe attenuation factor.

    Args:
        channel: 1-4
        attenuation: Probe ratio (e.g. 1.0, 10.0)
    """
    self._validate_channel(channel)
    self.safe_send(f":CHANnel{channel}:PROBe {attenuation}")

set_channel_scale(channel, scale)

:CHANnel:SCALe — Set vertical scale (volts/div).

Parameters:

Name Type Description Default
channel int

1-4

 required
scale float

Volts per division (e.g. 0.01 to 10.0)

 required
Source code in src/instrumation/drivers/rigol.py 
215
216
217
218
219
220
221
222
223
def set_channel_scale(self, channel: int, scale: float) -> None:
    """:CHANnel<n>:SCALe — Set vertical scale (volts/div).

    Args:
        channel: 1-4
        scale: Volts per division (e.g. 0.01 to 10.0)
    """
    self._validate_channel(channel)
    self.safe_send(f":CHANnel{channel}:SCALe {scale}")

set_channel_units(channel, units)

:CHANnel:UNITs — Set channel display units.

Parameters:

Name Type Description Default
channel int

1-4

 required
units str

VOLTage, WATT, AMPere, or UNKNown

 required
Source code in src/instrumation/drivers/rigol.py 
293
294
295
296
297
298
299
300
301
302
303
304
def set_channel_units(self, channel: int, units: str) -> None:
    """:CHANnel<n>:UNITs — Set channel display units.

    Args:
        channel: 1-4
        units: VOLTage, WATT, AMPere, or UNKNown
    """
    self._validate_channel(channel)
    valid = {"VOLTAGE", "WATT", "AMPERE", "UNKNOWN"}
    if units.upper() not in valid:
        raise ValueError(f"Invalid units: {units}")
    self.safe_send(f":CHANnel{channel}:UNITs {units.upper()}")

set_edge_trigger_level(level)

:TRIGger:EDGe:LEVel — Set trigger level voltage.

Source code in src/instrumation/drivers/rigol.py 
369
370
371
def set_edge_trigger_level(self, level: float) -> None:
    """:TRIGger:EDGe:LEVel — Set trigger level voltage."""
    self.safe_send(f":TRIGger:EDGe:LEVel {level}")

set_edge_trigger_slope(slope)

:TRIGger:EDGe:SLOPe — Set edge trigger slope.

Parameters:

Name Type Description Default
slope str

POSitive, NEGative, or RFALl

 required
Source code in src/instrumation/drivers/rigol.py 
354
355
356
357
358
359
360
361
362
363
def set_edge_trigger_slope(self, slope: str) -> None:
    """:TRIGger:EDGe:SLOPe — Set edge trigger slope.

    Args:
        slope: POSitive, NEGative, or RFALl
    """
    valid = {"POSITIVE", "NEGATIVE", "RFALL"}
    if slope.upper() not in valid:
        raise ValueError(f"Invalid slope: {slope}")
    self.safe_send(f":TRIGger:EDGe:SLOPe {slope}")

set_edge_trigger_source(source)

:TRIGger:EDGe:SOURce — Set edge trigger source.

Parameters:

Name Type Description Default
source str

CHANnel1, CHANnel2, CHANnel3, CHANnel4, EXT, or DEMath

 required
Source code in src/instrumation/drivers/rigol.py 
342
343
344
345
346
347
348
def set_edge_trigger_source(self, source: str) -> None:
    """:TRIGger:EDGe:SOURce — Set edge trigger source.

    Args:
        source: CHANnel1, CHANnel2, CHANnel3, CHANnel4, EXT, or DEMath
    """
    self.safe_send(f":TRIGger:EDGe:SOURce {source}")

set_timebase_offset(offset)

:TIMebase[:MAIN]:OFFSet — Set horizontal offset (seconds).

Parameters:

Name Type Description Default
offset float

Time offset from trigger point

 required
Source code in src/instrumation/drivers/rigol.py 
324
325
326
327
328
329
330
def set_timebase_offset(self, offset: float) -> None:
    """:TIMebase[:MAIN]:OFFSet — Set horizontal offset (seconds).

    Args:
        offset: Time offset from trigger point
    """
    self.safe_send(f":TIMebase:OFFSet {offset}")

set_timebase_scale(scale)

:TIMebase[:MAIN]:SCALe — Set horizontal scale (seconds/div).

Parameters:

Name Type Description Default
scale float

Seconds per division (e.g. 1e-9 to 50.0)

 required
Source code in src/instrumation/drivers/rigol.py 
312
313
314
315
316
317
318
def set_timebase_scale(self, scale: float) -> None:
    """:TIMebase[:MAIN]:SCALe — Set horizontal scale (seconds/div).

    Args:
        scale: Seconds per division (e.g. 1e-9 to 50.0)
    """
    self.safe_send(f":TIMebase:SCALe {scale}")

set_trigger(source, level, slope)

Configure edge trigger (Oscilloscope interface convenience method).

Parameters:

Name Type Description Default
source str

CHANnel1, CHANnel2, etc.

 required
level float

Trigger level in volts

 required
slope str

POSITIVE, NEGATIVE, or RFALL

 required
Source code in src/instrumation/drivers/rigol.py 
399
400
401
402
403
404
405
406
407
408
409
410
def set_trigger(self, source: str, level: float, slope: str) -> None:
    """Configure edge trigger (Oscilloscope interface convenience method).

    Args:
        source: CHANnel1, CHANnel2, etc.
        level: Trigger level in volts
        slope: POSITIVE, NEGATIVE, or RFALL
    """
    self.safe_send(":TRIGger:MODE EDGE")
    self.set_edge_trigger_source(source)
    self.set_edge_trigger_level(level)
    self.set_edge_trigger_slope(slope)

set_trigger_sweep(sweep)

:TRIGger:SWEep — Set trigger sweep mode.

Parameters:

Name Type Description Default
sweep str

AUTO, NORMal, or SINGle

 required
Source code in src/instrumation/drivers/rigol.py 
377
378
379
380
381
382
383
384
385
386
def set_trigger_sweep(self, sweep: str) -> None:
    """:TRIGger:SWEep — Set trigger sweep mode.

    Args:
        sweep: AUTO, NORMal, or SINGle
    """
    valid = {"AUTO", "NORMAL", "SINGLE"}
    if sweep.upper() not in valid:
        raise ValueError(f"Invalid sweep mode: {sweep}")
    self.safe_send(f":TRIGger:SWEep {sweep.upper()}")

set_waveform_format(fmt)

:WAVeform:FORMat — Set waveform data format.

Parameters:

Name Type Description Default
fmt str

WORD, BYTE, or ASCii

 required
Source code in src/instrumation/drivers/rigol.py 
438
439
440
441
442
443
444
445
446
447
def set_waveform_format(self, fmt: str) -> None:
    """:WAVeform:FORMat — Set waveform data format.

    Args:
        fmt: WORD, BYTE, or ASCii
    """
    valid = {"WORD", "BYTE", "ASCII"}
    if fmt.upper() not in valid:
        raise ValueError(f"Invalid waveform format: {fmt}")
    self.safe_send(f":WAVeform:FORMat {fmt.upper()}")

set_waveform_mode(mode)

:WAVeform:MODE — Set waveform readout mode.

Parameters:

Name Type Description Default
mode str

NORMal, MAXimum, or RAW

 required
Source code in src/instrumation/drivers/rigol.py 
427
428
429
430
431
432
433
434
435
436
def set_waveform_mode(self, mode: str) -> None:
    """:WAVeform:MODE — Set waveform readout mode.

    Args:
        mode: NORMal, MAXimum, or RAW
    """
    valid = {"NORMAL", "MAXIMUM", "RAW"}
    if mode.upper() not in valid:
        raise ValueError(f"Invalid waveform mode: {mode}")
    self.safe_send(f":WAVeform:MODE {mode.upper()}")

set_waveform_source(channel)

:WAVeform:SOURce — Select waveform data source.

Parameters:

Name Type Description Default
channel int

1-4 (maps to CHANnel1..CHANnel4)

 required
Source code in src/instrumation/drivers/rigol.py 
414
415
416
417
418
419
420
421
def set_waveform_source(self, channel: int) -> None:
    """:WAVeform:SOURce — Select waveform data source.

    Args:
        channel: 1-4 (maps to CHANnel1..CHANnel4)
    """
    self._validate_channel(channel)
    self.safe_send(f":WAVeform:SOURce CHANnel{channel}")

shutdown_safety()

Stop acquisition and sync.

Source code in src/instrumation/drivers/rigol.py 
579
580
581
582
def shutdown_safety(self) -> None:
    """Stop acquisition and sync."""
    self.stop()
    self.sync_config()

single()

:SINGle — Single-shot acquisition.

Source code in src/instrumation/drivers/rigol.py 
127
128
129
def single(self) -> None:
    """:SINGle — Single-shot acquisition."""
    self.write(":SINGle")

stop()

:STOP — Stop acquisition.

Source code in src/instrumation/drivers/rigol.py 
123
124
125
def stop(self) -> None:
    """:STOP — Stop acquisition."""
    self.write(":STOP")

sync_config()

CLS + WAI — Synchronize.

Source code in src/instrumation/drivers/rigol.py 
 99
100
101
102
def sync_config(self) -> None:
    """*CLS + *WAI — Synchronize."""
    self.write("*CLS")
    self.write("*WAI")

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

Connect to an instrument and return a driver instance for it.

The address is resolved in priority order:

  1. Replay -- an address beginning with replay:// returns a ReplayDriver that reads from the file named after the prefix.
  2. Simulation -- when :func:is_sim_mode is true, a simulated driver registered for driver_type is returned instead of touching hardware.
  3. Auto-discovery -- the literal address "AUTO" searches for a matching instrument, trying the on-disk cache first, then mDNS, then the LAN ARP table, then a full VISA scan.
  4. Real hardware -- any other address is opened directly, identified via *IDN?, and routed to the matching vendor driver.

Parameters

resource_address : str A VISA resource string such as "TCPIP::192.168.1.5::INSTR", a replay:// file path, or the literal "AUTO" to auto-discover. driver_type : str, optional Instrument category used to select and validate the driver. One of "SCOPE", "SA", "SG", "PSU", "DMM", "VNA", "NA", "LOAD", "ELOAD", "COUNTER" or "GENERIC". Defaults to "GENERIC", which accepts any instrument.

Returns

object A connected driver instance. The concrete class depends on which instrument was identified.

Raises

ValueError If simulation mode is active, driver_type is not "GENERIC" and no simulated driver is registered for that type; or if "AUTO" discovery finds no matching instrument.

Notes

Resources found through "AUTO" discovery are written to .visa_cache.json in the working directory, most recent first, so later lookups try them before falling back to a full scan.

Examples

dmm = get_instrument("TCPIP::192.168.1.5::INSTR", "DMM") scope = get_instrument("AUTO", "SCOPE")

Source code in src/instrumation/factory.py 
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
296
297
298
299
300
301
302
303
304
305
306
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
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
def get_instrument(resource_address: str, driver_type: str = "GENERIC") -> any:
    """Connect to an instrument and return a driver instance for it.

    The address is resolved in priority order:

    1. **Replay** -- an address beginning with ``replay://`` returns a
       ``ReplayDriver`` that reads from the file named after the prefix.
    2. **Simulation** -- when :func:`is_sim_mode` is true, a simulated driver
       registered for ``driver_type`` is returned instead of touching hardware.
    3. **Auto-discovery** -- the literal address ``"AUTO"`` searches for a
       matching instrument, trying the on-disk cache first, then mDNS, then the
       LAN ARP table, then a full VISA scan.
    4. **Real hardware** -- any other address is opened directly, identified via
       ``*IDN?``, and routed to the matching vendor driver.

    Parameters
    ----------
    resource_address : str
        A VISA resource string such as ``"TCPIP::192.168.1.5::INSTR"``, a
        ``replay://`` file path, or the literal ``"AUTO"`` to auto-discover.
    driver_type : str, optional
        Instrument category used to select and validate the driver. One of
        ``"SCOPE"``, ``"SA"``, ``"SG"``, ``"PSU"``, ``"DMM"``, ``"VNA"``,
        ``"NA"``, ``"LOAD"``, ``"ELOAD"``, ``"COUNTER"`` or ``"GENERIC"``.
        Defaults to ``"GENERIC"``, which accepts any instrument.

    Returns
    -------
    object
        A connected driver instance. The concrete class depends on which
        instrument was identified.

    Raises
    ------
    ValueError
        If simulation mode is active, ``driver_type`` is not ``"GENERIC"`` and
        no simulated driver is registered for that type; or if ``"AUTO"``
        discovery finds no matching instrument.

    Notes
    -----
    Resources found through ``"AUTO"`` discovery are written to
    ``.visa_cache.json`` in the working directory, most recent first, so later
    lookups try them before falling back to a full scan.

    Examples
    --------
    >>> dmm = get_instrument("TCPIP::192.168.1.5::INSTR", "DMM")
    >>> scope = get_instrument("AUTO", "SCOPE")
    """
    # 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 SimulatedGeneric
        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 = SimulatedGeneric(resource_address if resource_address != "AUTO" else "USB0::SIM::INSTR")
        drv.connect()
        return drv

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

        # 1. Load Cache & LAN (The Fast Resources)
        cached_resources = []
        if cache_file.exists():
            try:
                cached_resources = json.loads(cache_file.read_text())
            except (IOError, OSError, json.JSONDecodeError):
                pass
        else:
            # Create an empty cache file on first run so AUTO doesn't
            # always fall through to the slow full VISA scan.
            try:
                cache_file.write_text("[]")
            except (IOError, OSError):
                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]
                cache_file.write_text(json.dumps(new_cache[:10]))  # 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 "RIGOL" in idn:
        if any(m in idn for m in ["DS1054Z", "DS1104Z", "DS1074Z", "DS1102Z",
                                   "MSO1054Z", "MSO1104Z", "MSO1074Z",
                                   "DS1000Z", "MSO1000Z"]):
            from .drivers.rigol import RigolDS1054Z
            final_drv = RigolDS1054Z(resource_address)
        else:
            from .drivers.rigol import RigolDSA
            final_drv = RigolDSA(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)
    elif "ANRITSU" in idn:
        if "MS2035" in idn:
            from .drivers.anritsu import AnritsuMS2035B
            final_drv = AnritsuMS2035B(resource_address)
        elif "SHOCKLINE" in idn or "MS4" in idn:
            from .drivers.anritsu import AnritsuShockLineVNA
            final_drv = AnritsuShockLineVNA(resource_address)
        elif "VNA" in idn or "MS20" in idn:
            from .drivers.anritsu import AnritsuVNA
            final_drv = AnritsuVNA(resource_address)
        else:
            from .drivers.anritsu import AnritsuSA
            final_drv = AnritsuSA(resource_address)
    elif "PROLOGIX" in idn:
        from .drivers.prologix import PrologixDriver
        final_drv = PrologixDriver(resource_address)

    if not final_drv:
        # No brand matched the IDN. If exactly one driver is registered for the
        # requested type (e.g. a plugin driver with no brand siblings), there's
        # no ambiguity and it's safe to use it. If multiple candidates exist,
        # picking one would be guessing a brand's SCPI dialect for an
        # unidentified instrument, so fall back to the explicit GENERIC driver.
        candidates = [d for d in DriverRegistry.get_drivers_by_type(driver_type) if "Simulated" not in d.__name__]
        if len(candidates) == 1:
            final_drv = candidates[0](resource_address)
        else:
            if idn:
                logger.warning(
                    f"Unrecognized instrument IDN '{idn.strip()}' at {resource_address}; "
                    f"using GENERIC driver instead of guessing a {driver_type} brand driver."
                )
            final_drv = GenericDriver(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_path = Path(".visa_cache.json")
            cached_resources = []
            if cache_path.exists():
                try:
                    cached_resources = json.loads(cache_path.read_text())
                except (IOError, OSError, json.JSONDecodeError):
                    pass
            new_cache = [resource_address] + [r for r in cached_resources if r != resource_address]
            cache_path.write_text(json.dumps(new_cache[:10]))
        except (IOError, OSError, json.JSONDecodeError):
            pass

    return final_drv

Dynamically loads all available instrument drivers.

Source code in src/instrumation/factory.py 
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
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)

Transport Utilities

Detect which line-termination character an instrument responds to.

Tries each common terminator (LF, CR, CRLF) against a safe SCPI query and returns the first one that produces a valid response.

Steps
  1. Save the instrument's current read_termination setting.
  2. For each candidate terminator in ["\n", "\r", "\r\n"]: a. Set instrument.read_termination to the candidate. b. Send the query (default "IDN?"). c. If the response is non-empty and looks like a valid IDN? reply (contains at least one comma), return the candidate terminator.
  3. If none work, restore the original termination and raise RuntimeError.
  4. Restore the original termination before returning.

Parameters:

Name Type Description Default
instrument Any

A connected VisaDriver or pyvisa Resource object with .write(), .query(), and .read_termination attributes.

 required
query str

The SCPI query to test with. Defaults to "*IDN?".

 '*IDN?'

Returns:

Type Description
str

The working terminator string: "\n", "\r", or "\r\n".

Raises:

Type Description
RuntimeError

If no candidate terminator produces a valid response.
Source code in src/instrumation/transport.py 
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
def detect_line_termination(instrument: Any, query: str = "*IDN?") -> str:
    """Detect which line-termination character an instrument responds to.

    Tries each common terminator (LF, CR, CRLF) against a safe SCPI query
    and returns the first one that produces a valid response.

    Steps:
        1. Save the instrument's current read_termination setting.
        2. For each candidate terminator in ["\\n", "\\r", "\\r\\n"]:
           a. Set instrument.read_termination to the candidate.
           b. Send the query (default "*IDN?").
           c. If the response is non-empty and looks like a valid *IDN? reply
              (contains at least one comma), return the candidate terminator.
        3. If none work, restore the original termination and raise RuntimeError.
        4. Restore the original termination before returning.

    Args:
        instrument: A connected VisaDriver or pyvisa Resource object with
            .write(), .query(), and .read_termination attributes.
        query: The SCPI query to test with. Defaults to "*IDN?".

    Returns:
        The working terminator string: "\\n", "\\r", or "\\r\\n".

    Raises:
        RuntimeError: If no candidate terminator produces a valid response.
    """
    original_termination = instrument.read_termination

    for candidate in ["\n", "\r", "\r\n"]:
        instrument.read_termination = candidate
        try:
            response = instrument.query(query)
            if response and "," in response:
                instrument.read_termination = original_termination
                return candidate
        except Exception:
            continue

    instrument.read_termination = original_termination
    raise RuntimeError(f"No terminator produced a valid response for query: {query}")

Find the smallest safe timeout value for an instrument.

Tries a list of candidate timeout values (in milliseconds) against a safe SCPI query and returns the first one that completes without timing out.

Steps
  1. If candidates is None, use [100, 250, 500, 1000, 2500, 5000].
  2. Save the instrument's current timeout setting.
  3. Sort candidates ascending (smallest first).
  4. For each candidate timeout: a. Set instrument.timeout to the candidate. b. Send the query. c. If the response is non-empty, restore the original timeout and return the candidate. d. If a VisaIOError or timeout exception occurs, continue to the next.
  5. If no candidate works, restore the original timeout and raise RuntimeError.

Parameters:

Name Type Description Default
instrument Any

A connected VisaDriver or pyvisa Resource object with .write(), .query(), and .timeout attributes.

 required
query str

The SCPI query to test with. Defaults to "*IDN?".

 '*IDN?'
candidates Optional[List[int]]

List of timeout values in ms to try. Defaults to [100, 250, 500, 1000, 2500, 5000].

 None

Returns:

Type Description
int

The smallest timeout value (in ms) that worked.

Raises:

Type Description
RuntimeError

If no candidate timeout produces a valid response.
Source code in src/instrumation/transport.py 
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
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
def find_minimum_timeout(
    instrument: Any,
    query: str = "*IDN?",
    candidates: Optional[List[int]] = None,
) -> int:
    """Find the smallest safe timeout value for an instrument.

    Tries a list of candidate timeout values (in milliseconds) against a safe
    SCPI query and returns the first one that completes without timing out.

    Steps:
        1. If candidates is None, use [100, 250, 500, 1000, 2500, 5000].
        2. Save the instrument's current timeout setting.
        3. Sort candidates ascending (smallest first).
        4. For each candidate timeout:
           a. Set instrument.timeout to the candidate.
           b. Send the query.
           c. If the response is non-empty, restore the original timeout and
              return the candidate.
           d. If a VisaIOError or timeout exception occurs, continue to the next.
        5. If no candidate works, restore the original timeout and raise RuntimeError.

    Args:
        instrument: A connected VisaDriver or pyvisa Resource object with
            .write(), .query(), and .timeout attributes.
        query: The SCPI query to test with. Defaults to "*IDN?".
        candidates: List of timeout values in ms to try. Defaults to
            [100, 250, 500, 1000, 2500, 5000].

    Returns:
        The smallest timeout value (in ms) that worked.

    Raises:
        RuntimeError: If no candidate timeout produces a valid response.
    """
    if candidates is None:
        candidates = [100, 250, 500, 1000, 2500, 5000]

    original_timeout = instrument.timeout
    candidates_sorted = sorted(candidates)

    for candidate in candidates_sorted:
        instrument.timeout = candidate
        try:
            response = instrument.query(query)
            if response:
                instrument.timeout = original_timeout
                return candidate
        except (TimeoutError, Exception):
            continue

    instrument.timeout = original_timeout
    raise RuntimeError(f"No timeout candidate worked for query: {query}")

Poll an instrument's status byte for the MAV (Message Available) bit.

Instead of reading immediately or sleeping blindly, this polls the Status Byte Register (STB) until bit 4 (MAV) is set, indicating the instrument has data ready to read.

Steps
  1. Record the start time.
  2. Loop until timeout is exceeded: a. Send "*STB?" or "STB?" to read the status byte. b. Parse the response as an integer. c. If bit 4 (value & 0x10) is set, return — data is ready. d. Sleep for poll_interval seconds.
  3. If the loop exits without MAV being set, raise InstrumentTimeout.

Parameters:

Name Type Description Default
instrument Any

A connected VisaDriver or pyvisa Resource object.

 required
timeout float

Maximum seconds to wait for MAV. Defaults to 10.0.

 10.0
poll_interval float

Seconds between polls. Defaults to 0.1.

 0.1

Raises:

Type Description
InstrumentTimeout

If MAV is not set within the timeout period.
Source code in src/instrumation/transport.py 
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
def poll_for_mav(
    instrument: Any,
    timeout: float = 10.0,
    poll_interval: float = 0.1,
) -> None:
    """Poll an instrument's status byte for the MAV (Message Available) bit.

    Instead of reading immediately or sleeping blindly, this polls the Status
    Byte Register (STB) until bit 4 (MAV) is set, indicating the instrument
    has data ready to read.

    Steps:
        1. Record the start time.
        2. Loop until timeout is exceeded:
           a. Send "*STB?" or "STB?" to read the status byte.
           b. Parse the response as an integer.
           c. If bit 4 (value & 0x10) is set, return — data is ready.
           d. Sleep for poll_interval seconds.
        3. If the loop exits without MAV being set, raise InstrumentTimeout.

    Args:
        instrument: A connected VisaDriver or pyvisa Resource object.
        timeout: Maximum seconds to wait for MAV. Defaults to 10.0.
        poll_interval: Seconds between polls. Defaults to 0.1.

    Raises:
        InstrumentTimeout: If MAV is not set within the timeout period.
    """
    from .exceptions import InstrumentTimeout

    start = time.time()
    while time.time() - start < timeout:
        try:
            response = instrument.query("*STB?")
            value = int(response)
            if value & 0x10:
                return
        except (ValueError, Exception):
            pass
        time.sleep(poll_interval)

    raise InstrumentTimeout(f"MAV bit not set within {timeout}s timeout")

Poll for operation-complete (*OPC?) with exponential backoff delay.

Sends *OPC? and waits for "1" as the response, but instead of polling at a fixed interval, uses exponential backoff to reduce bus traffic during long operations while still responding quickly to fast completions.

Steps
  1. Record the start time. Set current_delay = initial_delay.
  2. Loop until timeout is exceeded: a. Send "*OPC?" and read the response. b. If the response stripped is "1", return — operation complete. c. Sleep for current_delay seconds. d. Multiply current_delay by 2, capped at max_delay.
  3. If the loop exits without completion, raise InstrumentTimeout.

Parameters:

Name Type Description Default
instrument Any

A connected VisaDriver or pyvisa Resource object.

 required
timeout float

Maximum seconds to wait. Defaults to 30.0.

 30.0
initial_delay float

Starting poll delay in seconds. Defaults to 0.1.

 0.1
max_delay float

Maximum delay between polls in seconds. Defaults to 1.0.

 1.0

Raises:

Type Description
InstrumentTimeout

If *OPC? does not return "1" within the timeout.
Source code in src/instrumation/transport.py 
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
def poll_opc_with_backoff(
    instrument: Any,
    timeout: float = 30.0,
    initial_delay: float = 0.1,
    max_delay: float = 1.0,
) -> None:
    """Poll for operation-complete (*OPC?) with exponential backoff delay.

    Sends *OPC? and waits for "1" as the response, but instead of polling at
    a fixed interval, uses exponential backoff to reduce bus traffic during
    long operations while still responding quickly to fast completions.

    Steps:
        1. Record the start time. Set current_delay = initial_delay.
        2. Loop until timeout is exceeded:
           a. Send "*OPC?" and read the response.
           b. If the response stripped is "1", return — operation complete.
           c. Sleep for current_delay seconds.
           d. Multiply current_delay by 2, capped at max_delay.
        3. If the loop exits without completion, raise InstrumentTimeout.

    Args:
        instrument: A connected VisaDriver or pyvisa Resource object.
        timeout: Maximum seconds to wait. Defaults to 30.0.
        initial_delay: Starting poll delay in seconds. Defaults to 0.1.
        max_delay: Maximum delay between polls in seconds. Defaults to 1.0.

    Raises:
        InstrumentTimeout: If *OPC? does not return "1" within the timeout.
    """
    from .exceptions import InstrumentTimeout

    start = time.time()
    current_delay = initial_delay

    while time.time() - start < timeout:
        try:
            response = instrument.query("*OPC?")
            if response.strip() == "1":
                return
        except Exception:
            pass
        time.sleep(current_delay)
        current_delay = min(current_delay * 2, max_delay)

    raise InstrumentTimeout(f"Operation did not complete within {timeout}s timeout")

Send multiple SCPI queries and return a dictionary of results.

This is useful for fetching multiple instrument settings or readings in a single call, reducing round-trips and improving efficiency on slow connections. Failed queries are logged with their error messages rather than raising immediately (unless stop_on_error=True).

Steps
  1. Initialize an empty results dictionary.
  2. For each query in the list: a. Send the query via instrument.query(). b. Store the stripped response in results[query]. c. If the query fails and stop_on_error is True, raise the exception. d. If stop_on_error is False, store the error message as the value.
  3. For each (write_cmd, read_cmd) pair in write_then_read: a. Send write_cmd via instrument.write(), then read_cmd via instrument.query(). b. Store the stripped response in results[write_cmd]. c. Same stop_on_error/error-message behavior as queries.
  4. Return the results dictionary.

Parameters:

Name Type Description Default
instrument Any

A connected VisaDriver or pyvisa Resource object.

 required
queries List[str]

List of SCPI query strings to send.

 required
stop_on_error bool

If True, raise on first error. If False (default), store error messages and continue with remaining queries.

 False
write_then_read Optional[List[Tuple[str, str]]]

Optional list of (write_cmd, read_cmd) pairs for instruments that need a separate write before the read (e.g. writing a register address, then reading its value). Results are keyed by write_cmd.

 None

Returns:

Type Description
dict

A dictionary mapping each query string (or write_then_read write_cmd)
dict

to its response (or error message).
Example

results = batch_query(dmm, ["IDN?", "MEAS:VOLT:DC?", "STB?"]) for cmd, resp in results.items(): ... print(f"{cmd} -> {resp}")

results = batch_query(inst, [], write_then_read=[("REG 0", "REG?")]) results["REG 0"]

Source code in src/instrumation/transport.py 
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
def batch_query(
    instrument: Any,
    queries: List[str],
    stop_on_error: bool = False,
    write_then_read: Optional[List[Tuple[str, str]]] = None,
) -> dict:
    """Send multiple SCPI queries and return a dictionary of results.

    This is useful for fetching multiple instrument settings or readings in
    a single call, reducing round-trips and improving efficiency on slow
    connections. Failed queries are logged with their error messages rather
    than raising immediately (unless stop_on_error=True).

    Steps:
        1. Initialize an empty results dictionary.
        2. For each query in the list:
           a. Send the query via instrument.query().
           b. Store the stripped response in results[query].
           c. If the query fails and stop_on_error is True, raise the exception.
           d. If stop_on_error is False, store the error message as the value.
        3. For each (write_cmd, read_cmd) pair in write_then_read:
           a. Send write_cmd via instrument.write(), then read_cmd via instrument.query().
           b. Store the stripped response in results[write_cmd].
           c. Same stop_on_error/error-message behavior as queries.
        4. Return the results dictionary.

    Args:
        instrument: A connected VisaDriver or pyvisa Resource object.
        queries: List of SCPI query strings to send.
        stop_on_error: If True, raise on first error. If False (default),
            store error messages and continue with remaining queries.
        write_then_read: Optional list of (write_cmd, read_cmd) pairs for
            instruments that need a separate write before the read (e.g.
            writing a register address, then reading its value). Results
            are keyed by write_cmd.

    Returns:
        A dictionary mapping each query string (or write_then_read write_cmd)
        to its response (or error message).

    Example:
        >>> results = batch_query(dmm, ["*IDN?", "MEAS:VOLT:DC?", "*STB?"])
        >>> for cmd, resp in results.items():
        ...     print(f"{cmd} -> {resp}")

        >>> results = batch_query(inst, [], write_then_read=[("REG 0", "REG?")])
        >>> results["REG 0"]
    """
    results = {}
    for query in queries:
        try:
            response = instrument.query(query)
            results[query] = response.strip() if isinstance(response, str) else response
        except Exception as e:
            if stop_on_error:
                raise
            results[query] = f"ERROR: {e}"

    for write_cmd, read_cmd in write_then_read or []:
        try:
            instrument.write(write_cmd)
            response = instrument.query(read_cmd)
            results[write_cmd] = response.strip() if isinstance(response, str) else response
        except Exception as e:
            if stop_on_error:
                raise
            results[write_cmd] = f"ERROR: {e}"

    return results

Scanner Utilities

Find addresses that appear more than once with different identities.

On shared buses (GPIB, RS-485), two instruments configured with the same address will corrupt each other's responses. This function flags addresses that show up multiple times in a scan result with differing descriptions, which is the telltale sign of a bus conflict.

Steps
  1. Build a dict mapping each device "id" (address) to a list of its "desc" values across all scan results.
  2. For each address that has more than one distinct description: a. Create a conflict entry with:
    • "address": the duplicated address string
    • "identities": the list of distinct descriptions seen
    • "count": how many times it appeared b. Append it to the results list.
  3. Return the list of conflict entries. Empty list means no conflicts.

Parameters:

Name Type Description Default
devices List[Dict[str, str]]

The list of device dicts returned by scan(), each having keys "type", "id", and "desc".

 required

Returns:

Type Description
List[Dict[str, Any]]

A list of dicts, each with keys "address" (str), "identities" (list
List[Dict[str, Any]]

of str), and "count" (int). Empty list if no duplicates found, and
List[Dict[str, Any]]

also for empty or None input.
Example

devices = scan() conflicts = find_duplicate_addresses(devices) for c in conflicts: ... print(f"CONFLICT on {c['address']}: {c['identities']}")

Source code in src/instrumation/scanner.py 
 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
def find_duplicate_addresses(devices: List[Dict[str, str]]) -> List[Dict[str, Any]]:
    """Find addresses that appear more than once with different identities.

    On shared buses (GPIB, RS-485), two instruments configured with the same
    address will corrupt each other's responses.  This function flags addresses
    that show up multiple times in a scan result with differing descriptions,
    which is the telltale sign of a bus conflict.

    Steps:
        1. Build a dict mapping each device "id" (address) to a list of its
           "desc" values across all scan results.
        2. For each address that has more than one distinct description:
           a. Create a conflict entry with:
              - "address": the duplicated address string
              - "identities": the list of distinct descriptions seen
              - "count": how many times it appeared
           b. Append it to the results list.
        3. Return the list of conflict entries. Empty list means no conflicts.

    Args:
        devices: The list of device dicts returned by scan(), each having
            keys "type", "id", and "desc".

    Returns:
        A list of dicts, each with keys "address" (str), "identities" (list
        of str), and "count" (int).  Empty list if no duplicates found, and
        also for empty or ``None`` input.

    Example:
        >>> devices = scan()
        >>> conflicts = find_duplicate_addresses(devices)
        >>> for c in conflicts:
        ...     print(f"CONFLICT on {c['address']}: {c['identities']}")
    """
    if not devices:
        return []

    from collections import defaultdict

    addr_to_descs = defaultdict(list)
    for device in devices:
        addr_to_descs[device["id"]].append(device["desc"])

    conflicts = []
    for addr, descs in addr_to_descs.items():
        unique_descs = list(set(descs))
        if len(unique_descs) > 1:
            conflicts.append({
                "address": addr,
                "identities": unique_descs,
                "count": len(descs),
            })

    return conflicts