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 
 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
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
class InstrumentDriver(ABC):
    """Abstract Base Class for all instrument drivers following the 'Abstract Hardware' spec."""

    # Explicit simulated-driver marker (GH #160). Simulated/digital-twin
    # driver classes set this True; real drivers leave it False. Factory
    # filtering MUST use this flag instead of string-matching class names.
    is_simulated = False

    def __init__(self, resource: str) -> None:
        self.resource = resource
        self.connected = False
        # Instance flag mirrors the class-level simulated marker so both
        # class- and instance-level checks agree (GH #160).
        self.is_simulated = type(self).is_simulated

        # 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 sync_name and hasattr(self, sync_name):
                sync_method = getattr(self, sync_name)
                if not callable(sync_method):
                    raise AttributeError(
                        f"'{self.__class__.__name__}' has no async counterpart "
                        f"for non-callable attribute '{sync_name}'"
                    )
                async def wrapper(*args: Any, **kwargs: Any) -> Any:
                    return await asyncio.to_thread(sync_method, *args, **kwargs)
                return wrapper
            # Likely typo'd async_ name: the sync attribute doesn't exist.
            raise AttributeError(
                f"'{self.__class__.__name__}' has no attribute '{sync_name}' "
                f"(resolved from async_{sync_name}); did you mean "
                f"'async_{name[6:]}' on the sync driver?"
            )
        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

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

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

    @abstractmethod
    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

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

    @_unsupported
    def load_state(self, index: Union[int, str]) -> None:
        """Recalls state from memory."""

    # --- 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:
        model = self.identity.get("model") or "Instrument"
        logger.warning(
            "Feature '%s' is not supported by %s", feature_name, model
        )

    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")

    def supports(self, feature: str) -> bool:
        """True if this driver implements ``feature`` (a method name).

        A method counts as unsupported when the driver inherits the
        library's generic ``_unsupported_feature`` default for it.
        """
        method = getattr(type(self), feature, None)
        return callable(method) and not getattr(method, "_unsupported", False)

    # --- Measurements (optional; Oscilloscope / FrequencyCounter make them required) ---
    @_unsupported
    def measure_frequency(self) -> MeasurementResult:
        return MeasurementResult(0.0, "Hz")

    @_unsupported
    def measure_duty_cycle(self) -> MeasurementResult:
        return MeasurementResult(0.0, "%")

    @_unsupported
    def measure_v_peak_to_peak(self) -> MeasurementResult:
        return MeasurementResult(0.0, "V")

    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 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def __getattr__(self, name: str) -> Any:
    """Dynamic async wrapper for all driver methods."""
    if name.startswith("async_"):
        sync_name = name[6:]
        if sync_name and hasattr(self, sync_name):
            sync_method = getattr(self, sync_name)
            if not callable(sync_method):
                raise AttributeError(
                    f"'{self.__class__.__name__}' has no async counterpart "
                    f"for non-callable attribute '{sync_name}'"
                )
            async def wrapper(*args: Any, **kwargs: Any) -> Any:
                return await asyncio.to_thread(sync_method, *args, **kwargs)
            return wrapper
        # Likely typo'd async_ name: the sync attribute doesn't exist.
        raise AttributeError(
            f"'{self.__class__.__name__}' has no attribute '{sync_name}' "
            f"(resolved from async_{sync_name}); did you mean "
            f"'async_{name[6:]}' on the sync driver?"
        )
    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 
141
142
143
144
@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 
121
122
123
124
@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 
80
81
82
83
@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 
85
86
87
88
@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 
155
156
157
158
159
160
161
162
163
164
165
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 
150
151
152
@_unsupported
def load_state(self, index: Union[int, str]) -> None:
    """Recalls state from memory."""

query_ascii(command) abstractmethod

Sends command, reads response, and checks for errors.

Source code in src/instrumation/drivers/base.py 
104
105
106
107
@abstractmethod
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) abstractmethod

High-speed binary data transfer.

Source code in src/instrumation/drivers/base.py 
109
110
111
112
@abstractmethod
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) abstractmethod

Sends command and immediately checks SYST:ERR?.

Source code in src/instrumation/drivers/base.py 
 99
100
101
102
@abstractmethod
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 
146
147
148
@_unsupported
def save_state(self, index: Union[int, str]) -> None:
    """Saves current state to memory."""

shutdown_safety() abstractmethod

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

Source code in src/instrumation/drivers/base.py 
136
137
138
139
@abstractmethod
def shutdown_safety(self) -> None:
    """Emergency shutdown protocol (Outputs OFF, Power/Volt 0)."""
    pass

supports(feature)

True if this driver implements feature (a method name).

A method counts as unsupported when the driver inherits the library's generic _unsupported_feature default for it.

Source code in src/instrumation/drivers/base.py 
185
186
187
188
189
190
191
192
def supports(self, feature: str) -> bool:
    """True if this driver implements ``feature`` (a method name).

    A method counts as unsupported when the driver inherits the
    library's generic ``_unsupported_feature`` default for it.
    """
    method = getattr(type(self), feature, None)
    return callable(method) and not getattr(method, "_unsupported", False)

sync_config() abstractmethod

Executes CLS and WAI for a clean slate.

Source code in src/instrumation/drivers/base.py 
126
127
128
129
@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 
131
132
133
134
@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 
459
460
461
462
463
464
465
466
467
468
469
470
471
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

DC power supply.

Every per-output method takes channel: None means the default (first) output, so single-output drivers may ignore it. CHANNELS lists the addressable outputs.

Source code in src/instrumation/drivers/base.py 
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
class PowerSupply(InstrumentDriver):
    """DC power supply.

    Every per-output method takes ``channel``: ``None`` means the default
    (first) output, so single-output drivers may ignore it. ``CHANNELS``
    lists the addressable outputs.
    """

    CHANNELS: Tuple[int, ...] = (1,)

    @staticmethod
    def _ch_kw(channel: Optional[int]) -> Dict[str, int]:
        # Only pass channel= through when set, so aliases also work on
        # single-output drivers whose methods take no channel argument.
        return {} if channel is None else {"channel": channel}

    @abstractmethod
    def set_voltage(self, voltage: float, channel: Optional[int] = None) -> None: pass
    @abstractmethod
    def get_voltage(self, channel: Optional[int] = None) -> float: pass
    @abstractmethod
    def set_current_limit(self, current: float, channel: Optional[int] = None) -> None: pass
    def set_current(self, current: float, channel: Optional[int] = None) -> None:
        """Generalized alias for set_current_limit."""
        self.set_current_limit(current, **self._ch_kw(channel))
    @abstractmethod
    def get_current(self, channel: Optional[int] = None) -> MeasurementResult: pass
    @abstractmethod
    def set_output(self, state: bool, channel: Optional[int] = None) -> None: pass
    @abstractmethod
    def get_output(self, channel: Optional[int] = None) -> bool: pass
    @_unsupported
    def set_ovp(self, voltage: float, channel: Optional[int] = None) -> None: pass
    @_unsupported
    def set_ocp(self, current: float, channel: Optional[int] = None) -> None: pass
    @abstractmethod
    def measure_voltage_actual(self, channel: Optional[int] = None) -> MeasurementResult: pass
    @abstractmethod
    def measure_current(self, channel: Optional[int] = None) -> MeasurementResult: pass

    def set_voltage_limit(self, voltage: float, channel: Optional[int] = None) -> None:
        """Generalized alias for Over-Voltage Protection (OVP)."""
        self.set_ovp(voltage, **self._ch_kw(channel))

    def measure_voltage(self, channel: Optional[int] = None) -> MeasurementResult:
        """Generalized alias for measure_voltage_actual."""
        return self.measure_voltage_actual(**self._ch_kw(channel))

    @_unsupported
    def clear_protection(self) -> None: pass

    @_unsupported
    def measure_power(self, channel: Optional[int] = None) -> MeasurementResult:
        """Queries the actual measured output power (Watts)."""
        return MeasurementResult(0.0, "W")

    @_unsupported
    def set_foldback_mode(self, mode: str) -> None:
        """Sets the foldback protection mode (OFF, CC, or CV)."""

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

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

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

get_mode()

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

Source code in src/instrumation/drivers/base.py 
541
542
543
544
@_unsupported
def get_mode(self) -> str:
    """Returns the current operation mode (CV, CC, or OFF)."""
    return "OFF"

measure_power(channel=None)

Queries the actual measured output power (Watts).

Source code in src/instrumation/drivers/base.py 
524
525
526
527
@_unsupported
def measure_power(self, channel: Optional[int] = None) -> MeasurementResult:
    """Queries the actual measured output power (Watts)."""
    return MeasurementResult(0.0, "W")

measure_voltage(channel=None)

Generalized alias for measure_voltage_actual.

Source code in src/instrumation/drivers/base.py 
517
518
519
def measure_voltage(self, channel: Optional[int] = None) -> MeasurementResult:
    """Generalized alias for measure_voltage_actual."""
    return self.measure_voltage_actual(**self._ch_kw(channel))

set_autostart(state)

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

Source code in src/instrumation/drivers/base.py 
537
538
539
@_unsupported
def set_autostart(self, state: bool) -> None:
    """Sets the Power-ON state (SAFE/OFF or AUTO/ON)."""

set_current(current, channel=None)

Generalized alias for set_current_limit.

Source code in src/instrumation/drivers/base.py 
495
496
497
def set_current(self, current: float, channel: Optional[int] = None) -> None:
    """Generalized alias for set_current_limit."""
    self.set_current_limit(current, **self._ch_kw(channel))

set_foldback_delay(seconds)

Sets the delay for foldback protection.

Source code in src/instrumation/drivers/base.py 
533
534
535
@_unsupported
def set_foldback_delay(self, seconds: float) -> None:
    """Sets the delay for foldback protection."""

set_foldback_mode(mode)

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

Source code in src/instrumation/drivers/base.py 
529
530
531
@_unsupported
def set_foldback_mode(self, mode: str) -> None:
    """Sets the foldback protection mode (OFF, CC, or CV)."""

set_voltage_limit(voltage, channel=None)

Generalized alias for Over-Voltage Protection (OVP).

Source code in src/instrumation/drivers/base.py 
513
514
515
def set_voltage_limit(self, voltage: float, channel: Optional[int] = None) -> None:
    """Generalized alias for Over-Voltage Protection (OVP)."""
    self.set_ovp(voltage, **self._ch_kw(channel))

Bases: InstrumentDriver

Source code in src/instrumation/drivers/base.py 
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
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 
566
567
568
569
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 
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
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

    @_unsupported
    def set_center_freq(self, freq_hz: float) -> None: 
        pass

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

    @_unsupported
    def set_span(self, span_hz: float) -> None: 
        pass

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

    @_unsupported
    def set_if_bandwidth(self, hz: float) -> None: 
        pass

    @_unsupported
    def set_power_level(self, dbm: float) -> None: 
        pass

    @_unsupported
    def set_sweep_type(self, sweep_type: str) -> None: 
        pass

    @_unsupported
    def set_averaging(self, state: bool, count: int = 10) -> None: 
        pass

    @_unsupported
    def set_continuous(self, state: bool) -> None: 
        pass

    @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

    @_unsupported
    def get_smith_data(self, measurement_name: str = "CH1_S11_1") -> MeasurementResult:
        return MeasurementResult([], "Z")

    @_unsupported
    def peak_search(self, marker: int = 1) -> None: 
        pass

    @_unsupported
    def get_marker_x(self, marker: int = 1) -> float: 
        return 0.0

    @_unsupported
    def get_marker_y(self, marker: int = 1) -> float: 
        return 0.0

    @_unsupported
    def save_state(self, filename: str) -> None: 
        pass

    @_unsupported
    def load_state(self, filename: str) -> None: 
        pass

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

set_center_frequency(freq_hz)

Alias for set_center_freq.

Source code in src/instrumation/drivers/base.py 
581
582
583
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 
645
646
647
@_unsupported
def wait_for_sweep(self) -> None:
    """Wait for the current sweep to complete."""

Bases: InstrumentDriver

Source code in src/instrumation/drivers/base.py 
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
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 
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
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
    @_unsupported
    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 
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
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

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

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

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

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

clear_protection()

Clears any tripped protection status.

Source code in src/instrumation/drivers/base.py 
309
310
311
312
@_unsupported
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 
234
235
236
237
@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 
274
275
276
277
@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 
224
225
226
227
@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 
264
265
266
267
@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 
254
255
256
257
@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 
244
245
246
247
@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 
284
285
286
287
@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 
289
290
291
292
@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 
279
280
281
282
@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 
229
230
231
232
@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 
269
270
271
272
@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 
219
220
221
222
@abstractmethod
def set_mode(self, mode: str) -> None:
    """Sets the operating mode, typically CC, CV, CR, or CP."""
    pass

set_ocp(current)

Sets the over-current protection limit.

Source code in src/instrumation/drivers/base.py 
299
300
301
302
@_unsupported
def set_ocp(self, current: float) -> None:
    """Sets the over-current protection limit."""
    pass

set_opp(power)

Sets the over-power protection limit.

Source code in src/instrumation/drivers/base.py 
304
305
306
307
@_unsupported
def set_opp(self, power: float) -> None:
    """Sets the over-power protection limit."""
    pass

set_ovp(voltage)

Sets the over-voltage protection limit.

Source code in src/instrumation/drivers/base.py 
294
295
296
297
@_unsupported
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 
259
260
261
262
@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 
249
250
251
252
@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 
239
240
241
242
@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 
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
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 
317
318
319
320
@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 
322
323
324
325
@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 
327
328
329
330
@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 
347
348
349
350
@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 
342
343
344
345
@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 
332
333
334
335
@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 
337
338
339
340
@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 
 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
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
@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 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 
91
92
93
94
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 
76
77
78
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 
96
97
98
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 
112
113
114
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 
129
130
131
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 
254
255
256
257
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 
191
192
193
194
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 
173
174
175
176
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 
269
270
271
272
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 
221
222
223
224
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 
236
237
238
239
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 
206
207
208
209
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 
354
355
356
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 
346
347
348
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 
331
332
333
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 
85
86
87
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 
153
154
155
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 
553
554
555
556
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 
289
290
291
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 
313
314
315
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 
301
302
303
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 
319
320
321
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 
373
374
375
376
377
378
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 
369
370
371
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 
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
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 
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
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 
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
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 
404
405
406
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 
453
454
455
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 
457
458
459
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 
461
462
463
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 
465
466
467
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 
469
470
471
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 
473
474
475
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 
543
544
545
546
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 
538
539
540
541
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 
548
549
550
551
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)

run()

:RUN — Start acquisition.

Source code in src/instrumation/drivers/rigol.py 
100
101
102
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 
133
134
135
136
137
138
139
140
141
142
143
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 
145
146
147
148
149
150
151
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 
118
119
120
121
122
123
124
125
126
127
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 
241
242
243
244
245
246
247
248
249
250
251
252
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 
178
179
180
181
182
183
184
185
186
187
188
189
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 
163
164
165
166
167
168
169
170
171
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 
259
260
261
262
263
264
265
266
267
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 
211
212
213
214
215
216
217
218
219
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 
226
227
228
229
230
231
232
233
234
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 
196
197
198
199
200
201
202
203
204
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 
274
275
276
277
278
279
280
281
282
283
284
285
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 
350
351
352
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 
335
336
337
338
339
340
341
342
343
344
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 
323
324
325
326
327
328
329
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 
305
306
307
308
309
310
311
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 
293
294
295
296
297
298
299
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 
380
381
382
383
384
385
386
387
388
389
390
391
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 
358
359
360
361
362
363
364
365
366
367
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 
419
420
421
422
423
424
425
426
427
428
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 
408
409
410
411
412
413
414
415
416
417
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 
395
396
397
398
399
400
401
402
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 
560
561
562
563
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 
108
109
110
def single(self) -> None:
    """:SINGle — Single-shot acquisition."""
    self.write(":SINGle")

stop()

:STOP — Stop acquisition.

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

sync_config()

CLS + WAI — Synchronize.

Source code in src/instrumation/drivers/rigol.py 
80
81
82
83
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", "ACPSU", "DAQ", "PEAKPM", "TEMP" or "GENERIC". Defaults to "GENERIC", which accepts any instrument. probe_asrl : bool, optional When True (default), an explicit ASRL connection gets a best-effort TDK-Lambda Z+ handshake (INST:NSEL 6) before *IDN? so the unit can identify itself. During "AUTO" discovery this is disabled: vendor-specific commands are never sent to arbitrary serial devices (gh #150).

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 
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
def get_instrument(resource_address: str, driver_type: str = "GENERIC", probe_asrl: bool = True) -> 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"``, ``"ACPSU"``,
        ``"DAQ"``, ``"PEAKPM"``, ``"TEMP"`` or ``"GENERIC"``. Defaults to
        ``"GENERIC"``, which accepts any instrument.
    probe_asrl : bool, optional
        When True (default), an explicit ASRL connection gets a best-effort
        TDK-Lambda Z+ handshake (``INST:NSEL 6``) before ``*IDN?`` so the
        unit can identify itself. During ``"AUTO"`` discovery this is
        disabled: vendor-specific commands are never sent to arbitrary
        serial devices (gh #150).

    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 drv_cls.is_simulated:
                # 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 by transport priority tier (HiSLIP/TCPIP > USB > GPIB >
            # serial), preserving input (recency) order within the same tier
            # via the stable sort.
            candidates.sort(key=_discovery_priority, 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 _skip_serial_probe(res):
                    return None
                dev = get_instrument(res, driver_type, probe_asrl=False)
                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). Runs only on
            # explicit connections -- during AUTO discovery vendor-specific
            # commands must never be sent to arbitrary serial devices (#150).
            if "ASRL" in resource_address and probe_asrl:
                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 (pyvisa.Error, TimeoutError, ConnectionLost) as e:
        # Expected transport failures (device unreachable, VISA timeout) are
        # logged and degrade to "unidentified". Genuine programming errors are
        # NOT swallowed -- they propagate so regressions surface loudly (#149).
        logger.warning(f"Identification failed for {resource_address}: {e}")
        idn = ""

    # Smart Routing based on IDN
    drv_cls = route_idn(idn)
    final_drv = drv_cls(resource_address) if drv_cls else None

    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 not d.is_simulated]
        if len(candidates) == 1:
            final_drv = candidates[0](resource_address)
        elif not candidates and driver_type not in KNOWN_DRIVER_TYPES:
            # Issue #146: mirror SIM mode's loud failure. A type that is
            # neither registered nor a canonical category has no driver to
            # fall back to — do NOT silently return GENERIC.
            raise ValueError(
                f"No real driver found for type: {driver_type}"
            )
        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 
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
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

Generic wrapper for VISA instruments.

Opens a VISA resource on construction and clears its status register.

Parameters

address : str VISA resource string, for example "TCPIP::192.168.1.5::INSTR". timeout : int, optional Read timeout in milliseconds, applied to the opened resource. Defaults to 5000.

Notes

Construction never raises. If the resource cannot be opened, the error is printed and :attr:inst is set to None, leaving an object whose methods are all no-ops. Check driver.inst is not None before relying on it.

Attributes

rm : pyvisa.ResourceManager The shared resource manager from :func:~instrumation.factory.get_rm. address : str The resource address this driver was constructed with. inst : pyvisa.Resource or None The open resource, or None if the connection failed.

Source code in src/instrumation/transport.py 
 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
class VisaDriver:
    """Generic wrapper for VISA instruments.

    Opens a VISA resource on construction and clears its status register.

    Parameters
    ----------
    address : str
        VISA resource string, for example ``"TCPIP::192.168.1.5::INSTR"``.
    timeout : int, optional
        Read timeout in milliseconds, applied to the opened resource.
        Defaults to ``5000``.

    Notes
    -----
    Construction never raises. If the resource cannot be opened, the error is
    printed and :attr:`inst` is set to ``None``, leaving an object whose methods
    are all no-ops. Check ``driver.inst is not None`` before relying on it.

    Attributes
    ----------
    rm : pyvisa.ResourceManager
        The shared resource manager from :func:`~instrumation.factory.get_rm`.
    address : str
        The resource address this driver was constructed with.
    inst : pyvisa.Resource or None
        The open resource, or ``None`` if the connection failed.
    """
    def __init__(self, address: str, timeout: int = 5000) -> None:
        from .factory import get_rm
        self.rm = get_rm()
        self.address = address
        try:
            self.inst = self.rm.open_resource(address)
            self.inst.timeout = timeout
            self.inst.write("*CLS")
        except Exception as e:
            print(f"Error connecting to {address}: {e}")
            self.inst = None

    def query_value(self, command: str) -> Optional[str]:
        """Send a query and return the stripped response.

        Parameters
        ----------
        command : str
            SCPI query to send, for example ``"MEAS:VOLT:DC?"``.

        Returns
        -------
        str or None
            The stripped response string on success. Returns ``None`` if
            the query raised, or if the driver never connected -- so a
            failed read is never mistaken for a genuine ``0.0`` reading.

        Notes
        -----
        The previous behaviour of returning ``0.0`` on failure was ambiguous
        with a real zero reading; callers should treat ``None`` as "no data".
        Callers that need to know why the read failed should check
        ``self.inst`` first and use the underlying resource directly.
        """
        if self.inst:
            try:
                return self.inst.query(command).strip()
            except Exception as e:
                print(f"VISA Query Error: {e}")
                return None
        return None

    def write(self, command: str) -> None:
        """Send a command without reading a response.

        Parameters
        ----------
        command : str
            SCPI command to send, for example ``"OUTP ON"``.

        Notes
        -----
        Silently does nothing if the driver never connected. Write errors are
        caught and printed rather than raised -- so a successful return does
        not guarantee the command reached the device. This matches the
        forgiving contract of :meth:`query_value` (and
        :class:`SerialDriver`): a failed connection leaves the object usable
        but inert, and callers that must know a write landed should check
        ``self.inst`` first and use the underlying resource directly.
        """
        if self.inst:
            try:
                self.inst.write(command)
            except Exception as e:
                print(f"VISA Write Error: {e}")

    def close(self) -> None:
        """Close the VISA resource, leaving the shared resource manager open.

        Safe to call when the driver never connected. The resource manager is
        a process-wide singleton shared with every other driver, so it is
        deliberately not closed here.
        """
        if self.inst:
            self.inst.close()

close()

Close the VISA resource, leaving the shared resource manager open.

Safe to call when the driver never connected. The resource manager is a process-wide singleton shared with every other driver, so it is deliberately not closed here.

Source code in src/instrumation/transport.py 
113
114
115
116
117
118
119
120
121
def close(self) -> None:
    """Close the VISA resource, leaving the shared resource manager open.

    Safe to call when the driver never connected. The resource manager is
    a process-wide singleton shared with every other driver, so it is
    deliberately not closed here.
    """
    if self.inst:
        self.inst.close()

query_value(command)

Send a query and return the stripped response.

Parameters

command : str SCPI query to send, for example "MEAS:VOLT:DC?".

Returns

str or None The stripped response string on success. Returns None if the query raised, or if the driver never connected -- so a failed read is never mistaken for a genuine 0.0 reading.

Notes

The previous behaviour of returning 0.0 on failure was ambiguous with a real zero reading; callers should treat None as "no data". Callers that need to know why the read failed should check self.inst first and use the underlying resource directly.

Source code in src/instrumation/transport.py 
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
def query_value(self, command: str) -> Optional[str]:
    """Send a query and return the stripped response.

    Parameters
    ----------
    command : str
        SCPI query to send, for example ``"MEAS:VOLT:DC?"``.

    Returns
    -------
    str or None
        The stripped response string on success. Returns ``None`` if
        the query raised, or if the driver never connected -- so a
        failed read is never mistaken for a genuine ``0.0`` reading.

    Notes
    -----
    The previous behaviour of returning ``0.0`` on failure was ambiguous
    with a real zero reading; callers should treat ``None`` as "no data".
    Callers that need to know why the read failed should check
    ``self.inst`` first and use the underlying resource directly.
    """
    if self.inst:
        try:
            return self.inst.query(command).strip()
        except Exception as e:
            print(f"VISA Query Error: {e}")
            return None
    return None

write(command)

Send a command without reading a response.

Parameters

command : str SCPI command to send, for example "OUTP ON".

Notes

Silently does nothing if the driver never connected. Write errors are caught and printed rather than raised -- so a successful return does not guarantee the command reached the device. This matches the forgiving contract of :meth:query_value (and :class:SerialDriver): a failed connection leaves the object usable but inert, and callers that must know a write landed should check self.inst first and use the underlying resource directly.

Source code in src/instrumation/transport.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def write(self, command: str) -> None:
    """Send a command without reading a response.

    Parameters
    ----------
    command : str
        SCPI command to send, for example ``"OUTP ON"``.

    Notes
    -----
    Silently does nothing if the driver never connected. Write errors are
    caught and printed rather than raised -- so a successful return does
    not guarantee the command reached the device. This matches the
    forgiving contract of :meth:`query_value` (and
    :class:`SerialDriver`): a failed connection leaves the object usable
    but inert, and callers that must know a write landed should check
    ``self.inst`` first and use the underlying resource directly.
    """
    if self.inst:
        try:
            self.inst.write(command)
        except Exception as e:
            print(f"VISA Write Error: {e}")

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 
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
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 
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
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 
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
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 
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
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 
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
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

Configuration

Returns the configuration dictionary for the test station.

Resolution precedence (highest wins): 1. Environment variables (VISA_ADDRESS, SERIAL_PORT, INSTRUMENT_TYPE, LOG_FILE) 2. A config file, located via the INSTRUMATION_CONFIG env var or instrumation.{json,yaml,yml} in the current directory or the user's home directory 3. Built-in defaults

Example::

config = get_config()
visa = config["visa_address"]
Source code in src/instrumation/config.py 
 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
def get_config() -> dict:
    """
    Returns the configuration dictionary for the test station.

    Resolution precedence (highest wins):
        1. Environment variables (``VISA_ADDRESS``, ``SERIAL_PORT``,
           ``INSTRUMENT_TYPE``, ``LOG_FILE``)
        2. A config file, located via the ``INSTRUMATION_CONFIG`` env var or
           ``instrumation.{json,yaml,yml}`` in the current directory or the
           user's home directory
        3. Built-in defaults

    Example::

        config = get_config()
        visa = config["visa_address"]
    """
    cfg = _default_config()

    config_file = _find_config_file()
    if config_file is not None:
        cfg.update(_load_config_file(config_file))

    # Environment variables always win over defaults and the file.
    for key, env_name in _ENV_MAP.items():
        if os.environ.get(env_name):
            cfg[key] = os.environ[env_name]
    return cfg

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
124
125
126
127
128
129
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"])

    addr_to_identities = defaultdict(set)
    for device in devices:
        # Identity = (type, desc), so two genuinely different physical devices
        # that happen to share a description string are still detected.
        addr_to_identities[device["id"]].add((device.get("type", ""), device["desc"]))

    conflicts = []
    for addr, identities in addr_to_identities.items():
        if len(identities) > 1:
            unique_descs = sorted({desc for _, desc in identities})
            conflicts.append({
                "address": addr,
                "identities": unique_descs,
                "count": len(addr_to_descs[addr]),
            })

    return conflicts