RS Spectrum driver

docs/source/instruments/RSFSV.rst

@@ -0,0 +1,44 @@
+R&S®FSV3030 Spectrum Analyzer
+=============================
+
+Spectrum analyzer drivers for the FSV3030 series by Rohde & Schwarz.
+This driver has been tested with the `FSV3030` spectrum analyzer. Documentation is available at `RS website <https://www.rohde-schwarz.com/it/prodotti/misura-e-collaudo/analizzatori-da-banco/rs-fsv3000-signal-and-spectrum-analyzer_63493-601503.html>`_.
+
+Base class: :class:`qtics.instruments.network_inst.NetworkInst`.
+
+Commands
+""""""""
+Other than featuring all the methods of :class:`qtics.instruments.network_inst.NetworkInst`, the base :class:`qtics.instruments.network.RSFsv3030.FSV3030` class contains the following methods and properties:
+
+Functions
+---------
+
+- clear(): Clear error queue and status registers.
+- wait(): Wait until all commands are processed.
+- single_sweep(): Trigger a single sweep and wait until completion.
+- continuous(state: bool = True): Enable or disable continuous sweep mode.
+- autoscale(): Autoscale the display (reference level and range).
+- read_trace_data(trace: int = 1) -> np.ndarray: Read the trace data in dBm.
+- read_freqs() -> np.ndarray: Return frequency axis for current span.
+- snapshot(trace: int = 1) -> tuple[np.ndarray, np.ndarray]: Perform single sweep and return (freqs, trace).
+- marker_to_peak(marker: int = 1): Move marker to maximum peak.
+- marker_freq(marker: int = 1) -> float: Query marker frequency in Hz.
+- marker_power(marker: int = 1) -> float: Query marker power in dBm.
+- set_max_hold(trace: int = 1): Set a trace to max-hold mode.
+- clear_max_hold(trace: int = 1): Clear the max-hold trace (reset).
+- read_max_hold(trace: int = 1) -> np.ndarray: Acquire current max-hold trace data.
+
+Properties
+----------
+
+- f_center: Center frequency in Hz.
+- f_span: Frequency span in Hz.
+- f_start: Start frequency in Hz.
+- f_stop: Stop frequency in Hz.
+- rbw: Resolution bandwidth in Hz.
+- vbw: Video bandwidth in Hz.
+- detector: Detector type (`POS`, `NEG`, `AVER`, `SAMP`, `RMS`, etc.).
+- sweep_points: Number of sweep points.
+- sweep_time: Sweep time in seconds.
+- ref_level: Reference level in dBm.
+- is_completed: Boolean indicating if the last operation is complete.

docs/source/instruments/modules.rst

@@ -13,6 +13,7 @@ Supported Instruments
    SMA100B
    VALON519
    RSZNB
+   RSFSV
    N9916A
    triton
    proteox

src/qtics/__init__.py

@@ -9,6 +9,7 @@
 from qtics.experiment import BaseExperiment, Experiment, MonitorExperiment
 from qtics.instruments.network.NA_N9916A import SAN9916A, VNAN9916A
 from qtics.instruments.network.proteox.proteox import Proteox
+from qtics.instruments.network.RS_FSV3030 import FSV3030
 from qtics.instruments.network.RS_SMA100B import SMA100B
 from qtics.instruments.network.RS_ZNB import RSZNB
 from qtics.instruments.network.triton_ctrl import Triton

src/qtics/instruments/network/RS_FSV3030.py

@@ -0,0 +1,211 @@
+"""Controller of the R&S FSV3030 Spectrum Analyzer."""
+
+import numpy as np
+
+from qtics.instruments import NetworkInst
+
+from .utils import query_data
+
+
+class FSV3030(NetworkInst):
+    """R&S FSV3030 Spectrum Analyzer by Rohde & Schwarz."""
+
+    def clear(self):
+        """Clear the error queue and status registers."""
+        self.write("*CLS")
+
+    def wait(self):
+        """Wait until all commands are processed."""
+        self.write("*WAI")
+
+    @property
+    def is_completed(self) -> bool:
+        """Query if last operation is complete."""
+        return self.query("*OPC?") == "1"
+
+    # ============================================================
+    # Frequency control
+    # ============================================================
+
+    @property
+    def f_center(self) -> float:
+        """Center frequency in Hz."""
+        return float(self.query("FREQ:CENT?"))
+
+    @f_center.setter
+    def f_center(self, f: float):
+        f = self.validate_range(f, 9e3, 30e9)  # FSV3030: 9 kHz – 30 GHz
+        self.write(f"FREQ:CENT {f}")
+
+    @property
+    def f_span(self) -> float:
+        """Frequency span in Hz."""
+        return float(self.query("FREQ:SPAN?"))
+
+    @f_span.setter
+    def f_span(self, f: float):
+        f = self.validate_range(f, 0, 30e9)
+        self.write(f"FREQ:SPAN {f}")
+
+    @property
+    def f_start(self) -> float:
+        """Start frequency in Hz."""
+        return float(self.query("FREQ:STAR?"))
+
+    @f_start.setter
+    def f_start(self, f: float):
+        f = self.validate_range(f, 9e3, 30e9)
+        self.write(f"FREQ:STAR {f}")
+
+    @property
+    def f_stop(self) -> float:
+        """Stop frequency in Hz."""
+        return float(self.query("FREQ:STOP?"))
+
+    @f_stop.setter
+    def f_stop(self, f: float):
+        f = self.validate_range(f, 9e3, 30e9)
+        self.write(f"FREQ:STOP {f}")
+
+    # ============================================================
+    # Bandwidth and detector
+    # ============================================================
+
+    @property
+    def rbw(self) -> float:
+        """Resolution bandwidth (Hz)."""
+        return float(self.query("BAND:RES?"))
+
+    @rbw.setter
+    def rbw(self, bw: float):
+        bw = self.validate_range(bw, 1, 10e6)
+        self.write(f"BAND:RES {bw}")
+
+    @property
+    def vbw(self) -> float:
+        """Video bandwidth (Hz)."""
+        return float(self.query("BAND:VID?"))
+
+    @vbw.setter
+    def vbw(self, bw: float):
+        bw = self.validate_range(bw, 1, 10e6)
+        self.write(f"BAND:VID {bw}")
+
+    @property
+    def detector(self) -> str:
+        """Detector type."""
+        return self.query("DET:FUNC?")
+
+    @detector.setter
+    def detector(self, mode: str = "POS"):
+        """Set detector mode (POS, NEG, AVER, SAMP, RMS, etc.)."""
+        self.validate_opt(mode, ("POS", "NEG", "AVER", "SAMP", "RMS"))
+        self.write(f"DET:FUNC {mode}")
+
+    # ============================================================
+    # Sweep and trace settings
+    # ============================================================
+
+    @property
+    def sweep_points(self) -> int:
+        """Number of sweep points."""
+        return int(self.query("SWE:POIN?"))
+
+    @sweep_points.setter
+    def sweep_points(self, n: int):
+        n = self.validate_range(n, 101, 10001)
+        self.write(f"SWE:POIN {n}")
+
+    @property
+    def sweep_time(self) -> float:
+        """Sweep time in seconds."""
+        return float(self.query("SWE:TIME?"))
+
+    @sweep_time.setter
+    def sweep_time(self, t: float):
+        t = self.validate_range(t, 1e-3, 1000)
+        self.write(f"SWE:TIME {t}")
+
+    def single_sweep(self):
+        """Trigger a single sweep and wait until completion."""
+        self.write("INIT:CONT OFF")
+        self.write("INIT:IMM")
+        self.wait()
+
+    def continuous(self, state: bool = True):
+        """Set continuous sweep mode."""
+        self.write(f"INIT:CONT {int(state)}")
+
+    # ============================================================
+    # Trace and measurement
+    # ============================================================
+
+    def autoscale(self):
+        """Autoscale display (adjust reference level and range)."""
+        self.write("DISP:WIND:TRAC:Y:AUTO")
+
+    @property
+    def ref_level(self) -> float:
+        """Reference level in dBm."""
+        return float(self.query("DISP:WIND:TRAC:Y:SCAL:RLEV?"))
+
+    @ref_level.setter
+    def ref_level(self, level: float):
+        self.write(f"DISP:WIND:TRAC:Y:SCAL:RLEV {level}")
+
+    def read_trace_data(self, trace: int = 1) -> np.ndarray:
+        """Read trace data (in dBm) as a numpy array."""
+        self.single_sweep()
+        data = query_data(self, f"TRAC? TRACE{trace}")
+        return data
+
+    def read_freqs(self) -> np.ndarray:
+        """Return frequency axis corresponding to current span."""
+        start = self.f_start
+        stop = self.f_stop
+        points = self.sweep_points
+        return np.linspace(start, stop, points)
+
+    def snapshot(self, trace: int = 1) -> tuple[np.ndarray, np.ndarray]:
+        """Perform single sweep and return (freqs, trace)."""
+        self.single_sweep()
+        freqs = self.read_freqs()
+        tot_trace = self.read_trace_data(trace)
+        return freqs, tot_trace
+
+    # ============================================================
+    # Marker control
+    # ============================================================
+
+    def marker_to_peak(self, marker: int = 1):
+        """Move marker to maximum peak."""
+        self.write(f"CALC:MARK{marker}:MAX")
+
+    def marker_freq(self, marker: int = 1) -> float:
+        """Query marker frequency."""
+        return float(self.query(f"CALC:MARK{marker}:X?"))
+
+    def marker_power(self, marker: int = 1) -> float:
+        """Query marker power in dBm."""
+        return float(self.query(f"CALC:MARK{marker}:Y?"))
+
+    # ============================================================
+    # Max-hold acquisition
+    # ============================================================
+
+    def set_max_hold(self, trace: int = 1):
+        """Set a trace to max-hold mode."""
+        self.write(f"TRAC{trace}:MODE MAXH")
+        self.single_sweep()  # Start sweep in max-hold mode
+
+    def clear_max_hold(self, trace: int = 1):
+        """Clear the max-hold trace (reset)."""
+        self.write(f"TRAC{trace}:MODE WRIT")
+        self.single_sweep()
+
+    def read_max_hold(self, trace: int = 1) -> np.ndarray:
+        """Acquire the current max-hold trace data."""
+        self.set_max_hold(trace)
+        # optional: wait for a few sweeps if needed to build the max-hold
+        data = query_data(self, f"TRAC? TRACE{trace}")
+        return data

src/qtics/instruments/network/RS_ZNB.py

@@ -13,6 +13,8 @@
 from qtics import log
 from qtics.instruments import NetworkInst
 
+from .utils import query_data
+
 MEAS_TIME_FACTOR = 1.02
 
 
@@ -178,52 +180,6 @@ def data_format(self, form: str):
         self.validate_opt(form, ("REAL,32", "REAL,64", "ASC,0"))
         self.write("FORMat:DATA " + form)
 
-    def query_data(self, cmd, datatype="REAL,64") -> np.ndarray:
-        """
-        Send a command and parses response in IEEE 488.2 binary block format.
-
-        Similar to N9916A implementation but adapted for R&S ZNB.
-        """
-        self.data_format = datatype
-
-        if datatype == "ASC,0":
-            return np.array(self.query(cmd).split(",")).astype(float)
-
-        map_types = {"REAL,32": np.float32, "REAL,64": np.float64}
-        if datatype not in map_types:
-            raise ValueError("Invalid data type selected.")
-
-        self.write(cmd)
-
-        assert self.socket is not None
-
-        # Read # character, raise exception if not present.
-        if self.socket.recv(1) != b"#":
-            raise ValueError("Data in buffer is not in binblock format.")
-
-        # Extract header length and number of bytes in binblock.
-        header_length = int(self.socket.recv(1).decode("utf-8"), 16)
-        n_bytes = int(self.socket.recv(header_length).decode("utf-8"))
-
-        # Create a buffer and expose a memoryview for efficient socket reading
-        raw_data = bytearray(n_bytes)
-        buf = memoryview(raw_data)
-
-        while n_bytes:
-            # Read data from instrument into buffer.
-            bytes_recv = self.socket.recv_into(buf, n_bytes)
-            # Slice buffer to preserve data already written to it.
-            buf = buf[bytes_recv:]
-            # Subtract bytes received from total bytes.
-            n_bytes -= bytes_recv
-
-        # Receive termination character.
-        term = self.socket.recv(1)
-        if term != b"\n":
-            raise ValueError("Data not terminated correctly.")
-
-        return np.frombuffer(raw_data, dtype=map_types[datatype]).astype(float)
-
     @property
     def s_par(self) -> str:
         """The current scattering matrix parameter."""
@@ -355,7 +311,7 @@ def read_trace_data(self, yformat=None) -> np.ndarray:
             # Read complex S-parameter data
             self.sweep()
             self.activate_trace()
-            IQ = self.query_data(f"CALCulate{self._channel}:DATA? SDATa")
+            IQ = query_data(self, f"CALCulate{self._channel}:DATA? SDATa")
             len_2 = int(len(IQ) / 2)
             z = np.empty(len_2, dtype=np.complex128)
             z.real = IQ[0::2]
@@ -366,7 +322,7 @@ def read_trace_data(self, yformat=None) -> np.ndarray:
             self.yformat = yformat
             self.sweep()
             self.activate_trace()
-            return self.query_data(f"CALCulate{self._channel}:DATA? FDATa")
+            return query_data(self, f"CALCulate{self._channel}:DATA? FDATa")
 
     def snapshot(self, yformat=None, **kwargs) -> Tuple[np.ndarray, np.ndarray]:
         """Get frequency and trace values for a single sweep."""

src/qtics/instruments/network/utils.py

@@ -0,0 +1,46 @@
+"""Common utils for network instruments."""
+
+import numpy as np
+
+
+def query_data(inst, cmd, datatype="REAL,64") -> np.ndarray:
+    """Send a command and parses response in IEEE 488.2 binary block format."""
+    inst.data_format = datatype
+
+    if datatype == "ASC,0":
+        return np.array(inst.query(cmd).split(",")).astype(float)
+
+    map_types = {"REAL,32": np.float32, "REAL,64": np.float64}
+    if datatype not in map_types:
+        raise ValueError("Invalid data type selected.")
+
+    inst.write(cmd)
+
+    assert inst.socket is not None
+
+    # Read # character, raise exception if not present.
+    if inst.socket.recv(1) != b"#":
+        raise ValueError("Data in buffer is not in binblock format.")
+
+    # Extract header length and number of bytes in binblock.
+    header_length = int(inst.socket.recv(1).decode("utf-8"), 16)
+    n_bytes = int(inst.socket.recv(header_length).decode("utf-8"))
+
+    # Create a buffer and expose a memoryview for efficient socket reading
+    raw_data = bytearray(n_bytes)
+    buf = memoryview(raw_data)
+
+    while n_bytes:
+        # Read data from instrument into buffer.
+        bytes_recv = inst.socket.recv_into(buf, n_bytes)
+        # Slice buffer to preserve data already written to it.
+        buf = buf[bytes_recv:]
+        # Subtract bytes received from total bytes.
+        n_bytes -= bytes_recv
+
+    # Receive termination character.
+    term = inst.socket.recv(1)
+    if term != b"\n":
+        raise ValueError("Data not terminated correctly.")
+
+    return np.frombuffer(raw_data, dtype=map_types[datatype]).astype(float)