diff --git a/docs/source/instruments/RSFSV.rst b/docs/source/instruments/RSFSV.rst new file mode 100644 index 0000000..1198a47 --- /dev/null +++ b/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 `_. + +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. diff --git a/docs/source/instruments/modules.rst b/docs/source/instruments/modules.rst index 5300501..fec3a89 100644 --- a/docs/source/instruments/modules.rst +++ b/docs/source/instruments/modules.rst @@ -13,6 +13,7 @@ Supported Instruments SMA100B VALON519 RSZNB + RSFSV N9916A triton proteox diff --git a/src/qtics/__init__.py b/src/qtics/__init__.py index 5e26762..55e49b1 100644 --- a/src/qtics/__init__.py +++ b/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 diff --git a/src/qtics/instruments/network/RS_FSV3030.py b/src/qtics/instruments/network/RS_FSV3030.py new file mode 100644 index 0000000..dea70a5 --- /dev/null +++ b/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 diff --git a/src/qtics/instruments/network/RS_ZNB.py b/src/qtics/instruments/network/RS_ZNB.py index 0d8f912..59d615d 100644 --- a/src/qtics/instruments/network/RS_ZNB.py +++ b/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.""" diff --git a/src/qtics/instruments/network/utils.py b/src/qtics/instruments/network/utils.py new file mode 100644 index 0000000..60fe658 --- /dev/null +++ b/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)