Thomas Ales [M S E]
1679 lines
53 KiB
Python
1679 lines
53 KiB
Python
"""
|
|
Base class for Tektronix oscilloscope control with raw SCPI command support.
|
|
"""
|
|
|
|
import socket
|
|
|
|
|
|
class TektronixOscilloscopeBase:
|
|
"""
|
|
Base class for Tektronix oscilloscope instruments.
|
|
|
|
Provides low-level SCPI command interface for fast data transfer.
|
|
"""
|
|
|
|
# Valid acquisition modes (short and long forms)
|
|
# SCPI uses capital letters for short form, scope responds with full uppercase
|
|
ACQUIRE_MODES = {
|
|
'SAMPLE': ['SAM', 'SAMPLE', 'SAMple'],
|
|
'PEAKDETECT': ['PEAK', 'PEAKDETECT', 'PEAKdetect'],
|
|
'HIRES': ['HIR', 'HIRES', 'HIRes'],
|
|
'AVERAGE': ['AVE', 'AVERAGE', 'AVErage'],
|
|
'ENVELOPE': ['ENV', 'ENVELOPE', 'ENVelope']
|
|
}
|
|
|
|
# Valid trigger coupling modes
|
|
TRIGGER_COUPLING = {
|
|
'DC': ['DC'],
|
|
'HFREJ': ['HFRej', 'HFREJ'],
|
|
'LFREJ': ['LFRej', 'LFREJ'],
|
|
'NOISEREJ': ['NOISErej', 'NOISEREJ']
|
|
}
|
|
|
|
# Valid trigger slope options
|
|
TRIGGER_SLOPE = {
|
|
'RISE': ['RISe', 'RISE'],
|
|
'FALL': ['FALL'],
|
|
'EITHER': ['EITher', 'EITHER']
|
|
}
|
|
|
|
# Valid trigger modes
|
|
TRIGGER_MODE = {
|
|
'AUTO': ['AUTO'],
|
|
'NORMAL': ['NORMal', 'NORMAL']
|
|
}
|
|
|
|
# Valid channel coupling modes
|
|
CHANNEL_COUPLING = {
|
|
'AC': ['AC'],
|
|
'DC': ['DC']
|
|
}
|
|
|
|
# Valid channel termination values (ohms)
|
|
CHANNEL_TERMINATION = [50, 1000000]
|
|
|
|
# Valid data encoding formats
|
|
DATA_ENCODING = {
|
|
'ASCII': ['ASCIi', 'ASCII'],
|
|
'RIBINARY': ['RIBinary', 'RIBINARY'], # Signed integer, MSB first (recommended)
|
|
'RPBINARY': ['RPBinary', 'RPBINARY'], # Positive integer, MSB first
|
|
'FPBINARY': ['FPBinary', 'FPBINARY'], # Floating point binary
|
|
'SRIBINARY': ['SRIbinary', 'SRIBINARY'], # Signed integer, byte-swapped
|
|
'SRPBINARY': ['SRPbinary', 'SRPBINARY'], # Positive integer, byte-swapped
|
|
'SFPBINARY': ['SFPbinary', 'SFPBINARY'] # Floating point binary, byte-swapped
|
|
}
|
|
|
|
# Valid waveform output preamble encoding
|
|
WFMOUTPRE_ENCODING = {
|
|
'BINARY': ['BINary', 'BINARY'],
|
|
'ASCII': ['ASCii', 'ASCII']
|
|
}
|
|
|
|
# Valid byte order options
|
|
BYTE_ORDER = {
|
|
'LSB': ['LSB'], # Least significant byte first
|
|
'MSB': ['MSB'] # Most significant byte first
|
|
}
|
|
|
|
def __init__(self, resource_name=None, port=4000, timeout=5.0, terminator='\n'):
|
|
"""
|
|
Initialize the oscilloscope base class.
|
|
|
|
Args:
|
|
resource_name: IP address or hostname of the instrument
|
|
port: TCP port for SCPI communication (default: 4000)
|
|
timeout: Socket timeout in seconds (default: 5.0)
|
|
terminator: Line terminator for SCPI commands (default: '\n')
|
|
"""
|
|
self.resource_name = resource_name
|
|
self.port = port
|
|
self.timeout = timeout
|
|
self.terminator = terminator
|
|
self.socket = None
|
|
self._connected = False
|
|
|
|
def connect(self):
|
|
"""
|
|
Establish TCP socket connection to the oscilloscope.
|
|
|
|
Raises:
|
|
ValueError: If resource_name is not provided
|
|
ConnectionError: If connection fails
|
|
"""
|
|
if not self.resource_name:
|
|
raise ValueError("resource_name (IP address/hostname) must be provided")
|
|
|
|
if self._connected:
|
|
return
|
|
|
|
try:
|
|
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
self.socket.settimeout(self.timeout)
|
|
self.socket.connect((self.resource_name, self.port))
|
|
self._connected = True
|
|
except socket.error as e:
|
|
self.socket = None
|
|
self._connected = False
|
|
raise ConnectionError(f"Failed to connect to {self.resource_name}:{self.port} - {e}")
|
|
|
|
def disconnect(self):
|
|
"""
|
|
Close TCP socket connection to the oscilloscope.
|
|
"""
|
|
if self.socket:
|
|
try:
|
|
self.socket.close()
|
|
except Exception:
|
|
pass
|
|
finally:
|
|
self.socket = None
|
|
self._connected = False
|
|
|
|
def _normalize_channel(self, channel):
|
|
"""
|
|
Normalize channel input to integer (1-4).
|
|
|
|
Args:
|
|
channel: Channel as int (1-4) or string ('CH1'-'CH4')
|
|
|
|
Returns:
|
|
int: Channel number (1-4)
|
|
|
|
Raises:
|
|
ValueError: If channel is invalid
|
|
"""
|
|
if isinstance(channel, str):
|
|
channel_upper = channel.upper()
|
|
if not (channel_upper.startswith('CH') and len(channel_upper) == 3 and channel_upper[2].isdigit()):
|
|
raise ValueError(f"Invalid channel: {channel}. Must be CH1, CH2, CH3, or CH4")
|
|
channel = int(channel_upper[2])
|
|
|
|
if not isinstance(channel, int) or channel < 1 or channel > 4:
|
|
raise ValueError(f"Invalid channel: {channel}. Must be 1-4")
|
|
|
|
return channel
|
|
|
|
def identify(self):
|
|
"""
|
|
Query instrument identification string (*IDN?).
|
|
|
|
Returns:
|
|
str: Instrument identification string
|
|
|
|
Raises:
|
|
RuntimeError: If not connected
|
|
"""
|
|
return self.query("*IDN?")
|
|
|
|
def get_acquire_mode(self):
|
|
"""
|
|
Query the current acquisition mode.
|
|
|
|
Returns:
|
|
str: Current acquisition mode (SAMple, PEAKdetect, HIRes, AVErage, or ENVelope)
|
|
|
|
Raises:
|
|
RuntimeError: If not connected
|
|
"""
|
|
return self.query("ACQuire:MODe?")
|
|
|
|
def set_acquire_mode(self, mode):
|
|
"""
|
|
Set the acquisition mode.
|
|
|
|
Args:
|
|
mode: Acquisition mode to set (case-insensitive). Valid options:
|
|
'SAMple' or 'SAM' or 'SAMPLE' - Sample mode (default)
|
|
'PEAKdetect' or 'PEAK' or 'PEAKDETECT' - Peak detect mode
|
|
'HIRes' or 'HIR' or 'HIRES' - High resolution mode
|
|
'AVErage' or 'AVE' or 'AVERAGE' - Average mode
|
|
'ENVelope' or 'ENV' or 'ENVELOPE' - Envelope mode
|
|
|
|
Raises:
|
|
ValueError: If mode is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
# Normalize mode to uppercase for comparison
|
|
mode_upper = mode.upper()
|
|
|
|
# Check if mode is valid (any short or long form)
|
|
valid = False
|
|
for long_form, variants in self.ACQUIRE_MODES.items():
|
|
if mode_upper in [v.upper() for v in variants]:
|
|
valid = True
|
|
break
|
|
|
|
if not valid:
|
|
valid_options = []
|
|
for variants in self.ACQUIRE_MODES.values():
|
|
valid_options.extend(variants)
|
|
raise ValueError(f"Invalid acquisition mode: {mode}. "
|
|
f"Valid options: {', '.join(valid_options)}")
|
|
|
|
self.write(f"ACQuire:MODe {mode}")
|
|
|
|
def get_fastframe_state(self):
|
|
"""
|
|
Query the current FastFrame state.
|
|
|
|
Returns:
|
|
int: 0 if FastFrame is off, 1 if active
|
|
|
|
Raises:
|
|
RuntimeError: If not connected
|
|
"""
|
|
response = self.query("HORizontal:FASTframe:STATE?")
|
|
return int(response)
|
|
|
|
def set_fastframe_state(self, state):
|
|
"""
|
|
Enable or disable FastFrame mode.
|
|
|
|
Args:
|
|
state: FastFrame state (0 = off, 1 = active)
|
|
Can be int (0/1) or bool (False/True)
|
|
|
|
Raises:
|
|
ValueError: If state is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
# Convert boolean to int if needed
|
|
if isinstance(state, bool):
|
|
state = 1 if state else 0
|
|
|
|
# Validate state
|
|
if state not in (0, 1):
|
|
raise ValueError(f"Invalid FastFrame state: {state}. Must be 0 (off) or 1 (active)")
|
|
|
|
self.write(f"HORizontal:FASTframe:STATE {state}")
|
|
|
|
def get_fastframe_count(self):
|
|
"""
|
|
Query the current FastFrame frame count.
|
|
|
|
Returns:
|
|
int: Number of frames configured
|
|
|
|
Raises:
|
|
RuntimeError: If not connected
|
|
"""
|
|
response = self.query("HORizontal:FASTframe:COUNt?")
|
|
return int(response)
|
|
|
|
def set_fastframe_count(self, count):
|
|
"""
|
|
Set the number of FastFrame frames.
|
|
|
|
Args:
|
|
count: Number of frames to capture (must be positive integer)
|
|
|
|
Raises:
|
|
ValueError: If count is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
# Validate count
|
|
if not isinstance(count, int) or count <= 0:
|
|
raise ValueError(f"Invalid frame count: {count}. Must be a positive integer")
|
|
|
|
self.write(f"HORizontal:FASTframe:COUNt {count}")
|
|
|
|
def get_record_length(self):
|
|
"""
|
|
Query the current horizontal record length.
|
|
|
|
Returns:
|
|
int: Current record length in samples
|
|
|
|
Raises:
|
|
RuntimeError: If not connected
|
|
"""
|
|
response = self.query("HORizontal:MODe:RECOrdlength?")
|
|
return int(response)
|
|
|
|
def set_record_length(self, length):
|
|
"""
|
|
Set the horizontal record length.
|
|
|
|
Args:
|
|
length: Record length in samples (must be positive integer)
|
|
|
|
Raises:
|
|
ValueError: If length is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
# Validate length
|
|
if not isinstance(length, int) or length <= 0:
|
|
raise ValueError(f"Invalid record length: {length}. Must be a positive integer")
|
|
|
|
self.write(f"HORizontal:MODe:RECOrdlength {length}")
|
|
|
|
def get_sample_rate(self):
|
|
"""
|
|
Query the current horizontal sample rate.
|
|
|
|
Returns:
|
|
float: Current sample rate in samples per second
|
|
|
|
Raises:
|
|
RuntimeError: If not connected
|
|
"""
|
|
response = self.query("HORizontal:MODe:SAMPLERate?")
|
|
return float(response)
|
|
|
|
def set_sample_rate(self, rate):
|
|
"""
|
|
Set the horizontal sample rate.
|
|
|
|
Args:
|
|
rate: Sample rate in samples per second (must be positive number)
|
|
|
|
Raises:
|
|
ValueError: If rate is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
# Validate rate
|
|
if not isinstance(rate, (int, float)) or rate <= 0:
|
|
raise ValueError(f"Invalid sample rate: {rate}. Must be a positive number")
|
|
|
|
self.write(f"HORizontal:MODe:SAMPLERate {rate}")
|
|
|
|
def get_time_scale(self):
|
|
"""
|
|
Query the current horizontal time scale.
|
|
|
|
Returns:
|
|
float: Current time scale in seconds per division
|
|
|
|
Raises:
|
|
RuntimeError: If not connected
|
|
"""
|
|
response = self.query("HORizontal:MODe:SCAle?")
|
|
return float(response)
|
|
|
|
def set_time_scale(self, scale):
|
|
"""
|
|
Set the horizontal time scale.
|
|
|
|
Args:
|
|
scale: Time scale in seconds per division (must be positive number)
|
|
|
|
Raises:
|
|
ValueError: If scale is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
# Validate scale
|
|
if not isinstance(scale, (int, float)) or scale <= 0:
|
|
raise ValueError(f"Invalid time scale: {scale}. Must be a positive number")
|
|
|
|
self.write(f"HORizontal:MODe:SCAle {scale}")
|
|
|
|
def get_trigger_coupling(self):
|
|
"""
|
|
Query the current trigger coupling mode.
|
|
|
|
Returns:
|
|
str: Current trigger coupling (DC, HFRej, LFRej, or NOISErej)
|
|
|
|
Raises:
|
|
RuntimeError: If not connected
|
|
"""
|
|
return self.query("TRIGger:A:EDGE:COUPling?")
|
|
|
|
def set_trigger_coupling(self, coupling):
|
|
"""
|
|
Set the trigger coupling mode.
|
|
|
|
Args:
|
|
coupling: Trigger coupling mode (case-insensitive). Valid options:
|
|
'DC' - DC coupling
|
|
'HFRej' or 'HFREJ' - High frequency reject
|
|
'LFRej' or 'LFREJ' - Low frequency reject
|
|
'NOISErej' or 'NOISEREJ' - Noise reject
|
|
|
|
Raises:
|
|
ValueError: If coupling is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
coupling_upper = coupling.upper()
|
|
|
|
# Check if coupling is valid
|
|
valid = False
|
|
for long_form, variants in self.TRIGGER_COUPLING.items():
|
|
if coupling_upper in [v.upper() for v in variants]:
|
|
valid = True
|
|
break
|
|
|
|
if not valid:
|
|
valid_options = []
|
|
for variants in self.TRIGGER_COUPLING.values():
|
|
valid_options.extend(variants)
|
|
raise ValueError(f"Invalid trigger coupling: {coupling}. "
|
|
f"Valid options: {', '.join(valid_options)}")
|
|
|
|
self.write(f"TRIGger:A:EDGE:COUPling {coupling}")
|
|
|
|
def get_trigger_slope(self):
|
|
"""
|
|
Query the current trigger slope.
|
|
|
|
Returns:
|
|
str: Current trigger slope (RISe, FALL, or EITher)
|
|
|
|
Raises:
|
|
RuntimeError: If not connected
|
|
"""
|
|
return self.query("TRIGger:A:EDGE:SLOpe?")
|
|
|
|
def set_trigger_slope(self, slope):
|
|
"""
|
|
Set the trigger slope.
|
|
|
|
Args:
|
|
slope: Trigger slope (case-insensitive). Valid options:
|
|
'RISe' or 'RISE' - Rising edge
|
|
'FALL' - Falling edge
|
|
'EITher' or 'EITHER' - Either edge
|
|
|
|
Raises:
|
|
ValueError: If slope is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
slope_upper = slope.upper()
|
|
|
|
# Check if slope is valid
|
|
valid = False
|
|
for long_form, variants in self.TRIGGER_SLOPE.items():
|
|
if slope_upper in [v.upper() for v in variants]:
|
|
valid = True
|
|
break
|
|
|
|
if not valid:
|
|
valid_options = []
|
|
for variants in self.TRIGGER_SLOPE.values():
|
|
valid_options.extend(variants)
|
|
raise ValueError(f"Invalid trigger slope: {slope}. "
|
|
f"Valid options: {', '.join(valid_options)}")
|
|
|
|
self.write(f"TRIGger:A:EDGE:SLOpe {slope}")
|
|
|
|
def get_trigger_source(self):
|
|
"""
|
|
Query the current trigger source.
|
|
|
|
Returns:
|
|
str: Current trigger source (e.g., CH1, CH2, CH3, CH4)
|
|
|
|
Raises:
|
|
RuntimeError: If not connected
|
|
"""
|
|
return self.query("TRIGger:A:EDGE:SOUrce?")
|
|
|
|
def set_trigger_source(self, source):
|
|
"""
|
|
Set the trigger source.
|
|
|
|
Args:
|
|
source: Trigger source channel. Valid options:
|
|
'CH1', 'CH2', 'CH3', 'CH4' (case-insensitive)
|
|
Can also pass as integer 1-4
|
|
|
|
Raises:
|
|
ValueError: If source is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
# Convert integer to channel string if needed
|
|
if isinstance(source, int):
|
|
if source < 1 or source > 4:
|
|
raise ValueError(f"Invalid trigger source: {source}. Must be 1-4")
|
|
source = f"CH{source}"
|
|
|
|
# Validate channel format
|
|
source_upper = source.upper()
|
|
if not (source_upper.startswith('CH') and len(source_upper) == 3 and source_upper[2].isdigit()):
|
|
raise ValueError(f"Invalid trigger source: {source}. Must be CH1, CH2, CH3, or CH4")
|
|
|
|
channel_num = int(source_upper[2])
|
|
if channel_num < 1 or channel_num > 4:
|
|
raise ValueError(f"Invalid trigger source: {source}. Channel must be 1-4")
|
|
|
|
self.write(f"TRIGger:A:EDGE:SOUrce {source}")
|
|
|
|
def get_trigger_level(self, channel):
|
|
"""
|
|
Query the trigger level for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
|
|
Returns:
|
|
float: Trigger level in volts
|
|
|
|
Raises:
|
|
ValueError: If channel is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
# Convert to channel number if string
|
|
if isinstance(channel, str):
|
|
channel_upper = channel.upper()
|
|
if not (channel_upper.startswith('CH') and len(channel_upper) == 3 and channel_upper[2].isdigit()):
|
|
raise ValueError(f"Invalid channel: {channel}. Must be CH1, CH2, CH3, or CH4")
|
|
channel = int(channel_upper[2])
|
|
|
|
# Validate channel number
|
|
if not isinstance(channel, int) or channel < 1 or channel > 4:
|
|
raise ValueError(f"Invalid channel: {channel}. Must be 1-4")
|
|
|
|
response = self.query(f"TRIGger:A:LEVel:CH{channel}?")
|
|
return float(response)
|
|
|
|
def set_trigger_level(self, channel, level):
|
|
"""
|
|
Set the trigger level for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
level: Trigger level in volts
|
|
|
|
Raises:
|
|
ValueError: If channel or level is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
# Convert to channel number if string
|
|
if isinstance(channel, str):
|
|
channel_upper = channel.upper()
|
|
if not (channel_upper.startswith('CH') and len(channel_upper) == 3 and channel_upper[2].isdigit()):
|
|
raise ValueError(f"Invalid channel: {channel}. Must be CH1, CH2, CH3, or CH4")
|
|
channel = int(channel_upper[2])
|
|
|
|
# Validate channel number
|
|
if not isinstance(channel, int) or channel < 1 or channel > 4:
|
|
raise ValueError(f"Invalid channel: {channel}. Must be 1-4")
|
|
|
|
# Validate level
|
|
if not isinstance(level, (int, float)):
|
|
raise ValueError(f"Invalid trigger level: {level}. Must be a number")
|
|
|
|
self.write(f"TRIGger:A:LEVel:CH{channel} {level}")
|
|
|
|
def get_trigger_mode(self):
|
|
"""
|
|
Query the current trigger mode.
|
|
|
|
Returns:
|
|
str: Current trigger mode (AUTO or NORMal)
|
|
|
|
Raises:
|
|
RuntimeError: If not connected
|
|
"""
|
|
return self.query("TRIGger:A:MODe?")
|
|
|
|
def set_trigger_mode(self, mode):
|
|
"""
|
|
Set the trigger mode.
|
|
|
|
Args:
|
|
mode: Trigger mode (case-insensitive). Valid options:
|
|
'AUTO' - Auto trigger mode
|
|
'NORMal' or 'NORMAL' - Normal trigger mode
|
|
|
|
Raises:
|
|
ValueError: If mode is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
mode_upper = mode.upper()
|
|
|
|
# Check if mode is valid
|
|
valid = False
|
|
for long_form, variants in self.TRIGGER_MODE.items():
|
|
if mode_upper in [v.upper() for v in variants]:
|
|
valid = True
|
|
break
|
|
|
|
if not valid:
|
|
valid_options = []
|
|
for variants in self.TRIGGER_MODE.values():
|
|
valid_options.extend(variants)
|
|
raise ValueError(f"Invalid trigger mode: {mode}. "
|
|
f"Valid options: {', '.join(valid_options)}")
|
|
|
|
self.write(f"TRIGger:A:MODe {mode}")
|
|
|
|
# ========== Channel Control Methods ==========
|
|
|
|
def query_channel(self, channel):
|
|
"""
|
|
Query all parameters for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
|
|
Returns:
|
|
str: All channel parameters
|
|
|
|
Raises:
|
|
ValueError: If channel is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
return self.query(f"CH{channel}?")
|
|
|
|
def get_channel_bandwidth(self, channel):
|
|
"""
|
|
Query the bandwidth setting for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
|
|
Returns:
|
|
str: Channel bandwidth setting
|
|
|
|
Raises:
|
|
ValueError: If channel is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
return self.query(f"CH{channel}:BANdwidth?")
|
|
|
|
def set_channel_bandwidth(self, channel, bandwidth):
|
|
"""
|
|
Set the bandwidth for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
bandwidth: Bandwidth setting (depends on scope model)
|
|
|
|
Raises:
|
|
ValueError: If channel is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
self.write(f"CH{channel}:BANdwidth {bandwidth}")
|
|
|
|
def get_channel_coupling(self, channel):
|
|
"""
|
|
Query the coupling mode for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
|
|
Returns:
|
|
str: Coupling mode (AC or DC)
|
|
|
|
Raises:
|
|
ValueError: If channel is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
return self.query(f"CH{channel}:COUPling?")
|
|
|
|
def set_channel_coupling(self, channel, coupling):
|
|
"""
|
|
Set the coupling mode for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
coupling: Coupling mode (AC or DC, case-insensitive)
|
|
|
|
Raises:
|
|
ValueError: If channel or coupling is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
coupling_upper = coupling.upper()
|
|
|
|
# Validate coupling
|
|
valid = False
|
|
for long_form, variants in self.CHANNEL_COUPLING.items():
|
|
if coupling_upper in [v.upper() for v in variants]:
|
|
valid = True
|
|
break
|
|
|
|
if not valid:
|
|
valid_options = []
|
|
for variants in self.CHANNEL_COUPLING.values():
|
|
valid_options.extend(variants)
|
|
raise ValueError(f"Invalid coupling mode: {coupling}. "
|
|
f"Valid options: {', '.join(valid_options)}")
|
|
|
|
self.write(f"CH{channel}:COUPling {coupling}")
|
|
|
|
def get_channel_label_color(self, channel):
|
|
"""
|
|
Query the label color for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
|
|
Returns:
|
|
str: Color as hex string (e.g., '#FF0000')
|
|
|
|
Raises:
|
|
ValueError: If channel is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
return self.query(f"CH{channel}:LABel:COLor?")
|
|
|
|
def set_channel_label_color(self, channel, color):
|
|
"""
|
|
Set the label color for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
color: Color as hex string (e.g., '#FF0000' or '#AABBCC'), or empty string
|
|
|
|
Raises:
|
|
ValueError: If channel or color is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
|
|
# Allow empty strings (including SCPI quoted empty strings like '""')
|
|
if color == '' or color == '""':
|
|
self.write(f'CH{channel}:LABel:COLor ""')
|
|
return
|
|
|
|
# Basic validation of hex color format
|
|
if not isinstance(color, str) or not color.startswith('#') or len(color) != 7:
|
|
raise ValueError(f"Invalid color format: {color}. Must be '#AABBCC' format or empty string")
|
|
|
|
# Validate hex digits
|
|
try:
|
|
int(color[1:], 16)
|
|
except ValueError:
|
|
raise ValueError(f"Invalid color format: {color}. Must contain valid hex digits")
|
|
|
|
self.write(f"CH{channel}:LABel:COLor {color}")
|
|
|
|
def get_channel_label_font_size(self, channel):
|
|
"""
|
|
Query the label font size for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
|
|
Returns:
|
|
int: Font size in points
|
|
|
|
Raises:
|
|
ValueError: If channel is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
response = self.query(f"CH{channel}:LABel:FONT:SIZE?")
|
|
return int(float(response))
|
|
|
|
def set_channel_label_font_size(self, channel, size):
|
|
"""
|
|
Set the label font size for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
size: Font size in points (positive integer)
|
|
|
|
Raises:
|
|
ValueError: If channel or size is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
|
|
if not isinstance(size, int) or size <= 0:
|
|
raise ValueError(f"Invalid font size: {size}. Must be a positive integer")
|
|
|
|
self.write(f"CH{channel}:LABel:FONT:SIZE {size}")
|
|
|
|
def get_channel_label_font_type(self, channel):
|
|
"""
|
|
Query the label font type for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
|
|
Returns:
|
|
str: Font name
|
|
|
|
Raises:
|
|
ValueError: If channel is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
return self.query(f"CH{channel}:LABel:FONT:TYPE?")
|
|
|
|
def set_channel_label_font_type(self, channel, font_name):
|
|
"""
|
|
Set the label font type for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
font_name: Font name as string
|
|
|
|
Raises:
|
|
ValueError: If channel is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
|
|
if not isinstance(font_name, str) or not font_name:
|
|
raise ValueError(f"Invalid font name: {font_name}. Must be a non-empty string")
|
|
|
|
self.write(f"CH{channel}:LABel:FONT:TYPE {font_name}")
|
|
|
|
def get_channel_label_name(self, channel):
|
|
"""
|
|
Query the label name for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
|
|
Returns:
|
|
str: Channel label name
|
|
|
|
Raises:
|
|
ValueError: If channel is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
return self.query(f"CH{channel}:LABel:NAMe?")
|
|
|
|
def set_channel_label_name(self, channel, name):
|
|
"""
|
|
Set the label name for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
name: Label name as string
|
|
|
|
Raises:
|
|
ValueError: If channel is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
|
|
if not isinstance(name, str):
|
|
raise ValueError(f"Invalid label name: {name}. Must be a string")
|
|
|
|
# Need to quote the string for SCPI
|
|
self.write(f'CH{channel}:LABel:NAMe "{name}"')
|
|
|
|
def get_channel_label_xpos(self, channel):
|
|
"""
|
|
Query the label x-position for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
|
|
Returns:
|
|
int: X-position in pixels
|
|
|
|
Raises:
|
|
ValueError: If channel is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
response = self.query(f"CH{channel}:LABel:XPOS?")
|
|
return int(float(response))
|
|
|
|
def set_channel_label_xpos(self, channel, xpos):
|
|
"""
|
|
Set the label x-position for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
xpos: X-position in pixels (non-negative integer)
|
|
|
|
Raises:
|
|
ValueError: If channel or xpos is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
|
|
if not isinstance(xpos, int) or xpos < 0:
|
|
raise ValueError(f"Invalid x-position: {xpos}. Must be a non-negative integer")
|
|
|
|
self.write(f"CH{channel}:LABel:XPOS {xpos}")
|
|
|
|
def get_channel_label_ypos(self, channel):
|
|
"""
|
|
Query the label y-position for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
|
|
Returns:
|
|
int: Y-position in pixels
|
|
|
|
Raises:
|
|
ValueError: If channel is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
response = self.query(f"CH{channel}:LABel:YPOS?")
|
|
return int(float(response))
|
|
|
|
def set_channel_label_ypos(self, channel, ypos):
|
|
"""
|
|
Set the label y-position for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
ypos: Y-position in pixels (non-negative integer)
|
|
|
|
Raises:
|
|
ValueError: If channel or ypos is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
|
|
if not isinstance(ypos, int) or ypos < 0:
|
|
raise ValueError(f"Invalid y-position: {ypos}. Must be a non-negative integer")
|
|
|
|
self.write(f"CH{channel}:LABel:YPOS {ypos}")
|
|
|
|
def get_channel_offset(self, channel):
|
|
"""
|
|
Query the vertical offset for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
|
|
Returns:
|
|
float: Vertical offset in volts
|
|
|
|
Raises:
|
|
ValueError: If channel is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
response = self.query(f"CH{channel}:OFFset?")
|
|
return float(response)
|
|
|
|
def set_channel_offset(self, channel, offset):
|
|
"""
|
|
Set the vertical offset for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
offset: Vertical offset in volts
|
|
|
|
Raises:
|
|
ValueError: If channel or offset is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
|
|
if not isinstance(offset, (int, float)):
|
|
raise ValueError(f"Invalid offset: {offset}. Must be a number")
|
|
|
|
self.write(f"CH{channel}:OFFset {offset}")
|
|
|
|
def get_channel_position(self, channel):
|
|
"""
|
|
Query the vertical position for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
|
|
Returns:
|
|
float: Vertical position in divisions
|
|
|
|
Raises:
|
|
ValueError: If channel is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
response = self.query(f"CH{channel}:POSition?")
|
|
return float(response)
|
|
|
|
def set_channel_position(self, channel, position):
|
|
"""
|
|
Set the vertical position for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
position: Vertical position in divisions (can be negative or positive)
|
|
|
|
Raises:
|
|
ValueError: If channel or position is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
|
|
if not isinstance(position, (int, float)):
|
|
raise ValueError(f"Invalid position: {position}. Must be a number")
|
|
|
|
self.write(f"CH{channel}:POSition {position}")
|
|
|
|
def get_channel_scale(self, channel):
|
|
"""
|
|
Query the vertical scale for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
|
|
Returns:
|
|
float: Vertical scale in volts per division
|
|
|
|
Raises:
|
|
ValueError: If channel is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
response = self.query(f"CH{channel}:SCAle?")
|
|
return float(response)
|
|
|
|
def set_channel_scale(self, channel, scale):
|
|
"""
|
|
Set the vertical scale for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
scale: Vertical scale in volts per division (must be positive)
|
|
|
|
Raises:
|
|
ValueError: If channel or scale is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
|
|
if not isinstance(scale, (int, float)) or scale <= 0:
|
|
raise ValueError(f"Invalid scale: {scale}. Must be a positive number")
|
|
|
|
self.write(f"CH{channel}:SCAle {scale}")
|
|
|
|
def get_channel_termination(self, channel):
|
|
"""
|
|
Query the termination setting for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
|
|
Returns:
|
|
int: Termination in ohms (50 or 1000000)
|
|
|
|
Raises:
|
|
ValueError: If channel is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
response = self.query(f"CH{channel}:TERmination?")
|
|
return int(float(response))
|
|
|
|
def set_channel_termination(self, channel, termination):
|
|
"""
|
|
Set the termination for a specific channel.
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
termination: Termination in ohms (50 or 1000000)
|
|
|
|
Raises:
|
|
ValueError: If channel or termination is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
channel = self._normalize_channel(channel)
|
|
|
|
if termination not in self.CHANNEL_TERMINATION:
|
|
raise ValueError(f"Invalid termination: {termination}. "
|
|
f"Valid options: {', '.join(map(str, self.CHANNEL_TERMINATION))}")
|
|
|
|
self.write(f"CH{channel}:TERmination {termination}")
|
|
|
|
# ========== Waveform Transfer Methods ==========
|
|
|
|
def get_data_encoding(self):
|
|
"""
|
|
Query the current data encoding format.
|
|
|
|
Returns:
|
|
str: Current data encoding (e.g., RIBinary, ASCII, etc.)
|
|
|
|
Raises:
|
|
RuntimeError: If not connected
|
|
"""
|
|
return self.query("DATa:ENCdg?")
|
|
|
|
def set_data_encoding(self, encoding):
|
|
"""
|
|
Set the data encoding format for waveform transfer.
|
|
|
|
Args:
|
|
encoding: Encoding format (case-insensitive). Valid options:
|
|
'ASCIi' or 'ASCII' - ASCII format
|
|
'RIBinary' or 'RIBINARY' - Signed integer, MSB first (recommended)
|
|
'RPBinary' or 'RPBINARY' - Positive integer, MSB first
|
|
'FPBinary' or 'FPBINARY' - Floating point binary
|
|
'SRIbinary' or 'SRIBINARY' - Signed integer, byte-swapped
|
|
'SRPbinary' or 'SRPBINARY' - Positive integer, byte-swapped
|
|
'SFPbinary' or 'SFPBINARY' - Floating point binary, byte-swapped
|
|
|
|
Raises:
|
|
ValueError: If encoding is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
encoding_upper = encoding.upper()
|
|
|
|
# Check if encoding is valid
|
|
valid = False
|
|
for long_form, variants in self.DATA_ENCODING.items():
|
|
if encoding_upper in [v.upper() for v in variants]:
|
|
valid = True
|
|
break
|
|
|
|
if not valid:
|
|
valid_options = []
|
|
for variants in self.DATA_ENCODING.values():
|
|
valid_options.extend(variants)
|
|
raise ValueError(f"Invalid data encoding: {encoding}. "
|
|
f"Valid options: {', '.join(valid_options)}")
|
|
|
|
self.write(f"DATa:ENCdg {encoding}")
|
|
|
|
def get_data_source(self):
|
|
"""
|
|
Query the current data source for waveform transfer.
|
|
|
|
Returns:
|
|
str: Current data source (e.g., CH1, CH2, CH3, CH4)
|
|
|
|
Raises:
|
|
RuntimeError: If not connected
|
|
"""
|
|
return self.query("DATa:SOUrce?")
|
|
|
|
def set_data_source(self, source):
|
|
"""
|
|
Set the data source for waveform transfer.
|
|
|
|
Args:
|
|
source: Data source channel. Valid options:
|
|
'CH1', 'CH2', 'CH3', 'CH4' (case-insensitive)
|
|
Can also pass as integer 1-4
|
|
|
|
Raises:
|
|
ValueError: If source is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
# Convert integer to channel string if needed
|
|
if isinstance(source, int):
|
|
if source < 1 or source > 4:
|
|
raise ValueError(f"Invalid data source: {source}. Must be 1-4")
|
|
source = f"CH{source}"
|
|
|
|
# Validate channel format
|
|
source_upper = source.upper()
|
|
if not (source_upper.startswith('CH') and len(source_upper) == 3 and source_upper[2].isdigit()):
|
|
raise ValueError(f"Invalid data source: {source}. Must be CH1, CH2, CH3, or CH4")
|
|
|
|
channel_num = int(source_upper[2])
|
|
if channel_num < 1 or channel_num > 4:
|
|
raise ValueError(f"Invalid data source: {source}. Channel must be 1-4")
|
|
|
|
self.write(f"DATa:SOUrce {source}")
|
|
|
|
def get_wfmoutpre_encoding(self):
|
|
"""
|
|
Query the waveform output preamble encoding.
|
|
|
|
Returns:
|
|
str: Current encoding (BINary or ASCii)
|
|
|
|
Raises:
|
|
RuntimeError: If not connected
|
|
"""
|
|
return self.query("WFMOutpre:ENCdg?")
|
|
|
|
def set_wfmoutpre_encoding(self, encoding):
|
|
"""
|
|
Set the waveform output preamble encoding.
|
|
|
|
Args:
|
|
encoding: Encoding format (case-insensitive). Valid options:
|
|
'BINary' or 'BINARY' - Binary format (recommended)
|
|
'ASCii' or 'ASCII' - ASCII format
|
|
|
|
Raises:
|
|
ValueError: If encoding is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
encoding_upper = encoding.upper()
|
|
|
|
# Check if encoding is valid
|
|
valid = False
|
|
for long_form, variants in self.WFMOUTPRE_ENCODING.items():
|
|
if encoding_upper in [v.upper() for v in variants]:
|
|
valid = True
|
|
break
|
|
|
|
if not valid:
|
|
valid_options = []
|
|
for variants in self.WFMOUTPRE_ENCODING.values():
|
|
valid_options.extend(variants)
|
|
raise ValueError(f"Invalid waveform output encoding: {encoding}. "
|
|
f"Valid options: {', '.join(valid_options)}")
|
|
|
|
self.write(f"WFMOutpre:ENCdg {encoding}")
|
|
|
|
def get_wfmoutpre_byte_count(self):
|
|
"""
|
|
Query the waveform output preamble byte count (BYT_Nr).
|
|
|
|
Returns:
|
|
int: Number of bytes per data point (1 or 2)
|
|
|
|
Raises:
|
|
RuntimeError: If not connected
|
|
"""
|
|
response = self.query("WFMOutpre:BYT_Nr?")
|
|
return int(response)
|
|
|
|
def set_wfmoutpre_byte_count(self, byte_count):
|
|
"""
|
|
Set the waveform output preamble byte count (BYT_Nr).
|
|
|
|
Args:
|
|
byte_count: Number of bytes per data point
|
|
1 = 8-bit data (0-255 for unsigned, -128 to 127 for signed)
|
|
2 = 16-bit data (0-65535 for unsigned, -32768 to 32767 for signed)
|
|
|
|
Raises:
|
|
ValueError: If byte_count is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
if byte_count not in (1, 2):
|
|
raise ValueError(f"Invalid byte count: {byte_count}. Must be 1 or 2")
|
|
|
|
self.write(f"WFMOutpre:BYT_Nr {byte_count}")
|
|
|
|
def get_wfmoutpre_byte_order(self):
|
|
"""
|
|
Query the waveform output preamble byte order.
|
|
|
|
Returns:
|
|
str: Current byte order (LSB or MSB)
|
|
|
|
Raises:
|
|
RuntimeError: If not connected
|
|
"""
|
|
return self.query("WFMOutpre:BYT_Or?")
|
|
|
|
def set_wfmoutpre_byte_order(self, byte_order):
|
|
"""
|
|
Set the waveform output preamble byte order.
|
|
|
|
Args:
|
|
byte_order: Byte order (case-insensitive). Valid options:
|
|
'LSB' - Least significant byte first
|
|
'MSB' - Most significant byte first
|
|
|
|
Raises:
|
|
ValueError: If byte_order is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
byte_order_upper = byte_order.upper()
|
|
|
|
# Check if byte order is valid
|
|
valid = False
|
|
for long_form, variants in self.BYTE_ORDER.items():
|
|
if byte_order_upper in [v.upper() for v in variants]:
|
|
valid = True
|
|
break
|
|
|
|
if not valid:
|
|
valid_options = []
|
|
for variants in self.BYTE_ORDER.values():
|
|
valid_options.extend(variants)
|
|
raise ValueError(f"Invalid byte order: {byte_order}. "
|
|
f"Valid options: {', '.join(valid_options)}")
|
|
|
|
self.write(f"WFMOutpre:BYT_Or {byte_order}")
|
|
|
|
def query_wfmoutpre(self):
|
|
"""
|
|
Query all waveform output preamble parameters.
|
|
|
|
Returns:
|
|
str: Complete waveform preamble string with all parameters
|
|
|
|
Raises:
|
|
RuntimeError: If not connected
|
|
"""
|
|
return self.query("WFMOutpre?")
|
|
|
|
def get_fastframe_selected(self):
|
|
"""
|
|
Query the currently selected FastFrame frame number.
|
|
|
|
Returns:
|
|
int: Currently selected frame number
|
|
|
|
Raises:
|
|
RuntimeError: If not connected
|
|
"""
|
|
response = self.query("HORizontal:FASTframe:SELECTED?")
|
|
return int(response)
|
|
|
|
def set_fastframe_selected(self, frame_number):
|
|
"""
|
|
Select a specific FastFrame frame for data transfer.
|
|
|
|
Args:
|
|
frame_number: Frame number to select (must be positive integer)
|
|
|
|
Raises:
|
|
ValueError: If frame_number is not valid
|
|
RuntimeError: If not connected
|
|
"""
|
|
if not isinstance(frame_number, int) or frame_number <= 0:
|
|
raise ValueError(f"Invalid frame number: {frame_number}. Must be a positive integer")
|
|
|
|
self.write(f"HORizontal:FASTframe:SELECTED {frame_number}")
|
|
|
|
def transfer_curve(self):
|
|
"""
|
|
Transfer waveform curve data from the oscilloscope.
|
|
|
|
This method reads binary curve data according to the current DATa:SOUrce,
|
|
DATa:ENCdg, and WFMOutpre settings.
|
|
|
|
Returns:
|
|
bytes: Raw binary curve data
|
|
|
|
Raises:
|
|
RuntimeError: If not connected or transfer fails
|
|
"""
|
|
# Send the CURVe? query
|
|
self.write("CURVe?")
|
|
|
|
|
|
# Read the binary data using IEEE 488.2 format
|
|
return self.read_raw()
|
|
|
|
def transfer_waveform(self):
|
|
"""
|
|
Transfer complete waveform data including preamble and curve.
|
|
|
|
This method queries WFMOutpre? for the preamble and CURVe? for the data
|
|
as separate operations, which is the standard approach for Tektronix scopes.
|
|
|
|
Returns:
|
|
tuple: (preamble_str, curve_bytes)
|
|
preamble_str: Waveform preamble as string
|
|
curve_bytes: Raw binary curve data
|
|
|
|
Raises:
|
|
RuntimeError: If not connected or transfer fails
|
|
"""
|
|
# First query the waveform preamble
|
|
preamble_str = self.query_wfmoutpre()
|
|
|
|
# Then transfer the curve data
|
|
curve_bytes = self.transfer_curve()
|
|
|
|
return (preamble_str, curve_bytes)
|
|
|
|
def transfer_fastframe(self, parse=True, byte_count=1, signed=True, byte_order='MSB'):
|
|
"""
|
|
Transfer all FastFrame waveform data from the oscilloscope.
|
|
|
|
When FastFrame is enabled, a single CURVe? query returns all frames
|
|
as sequential IEEE 488.2 binary blocks. This method reads all frames
|
|
efficiently in a single operation.
|
|
|
|
Args:
|
|
parse: If True, parse the raw bytes into integer arrays (default: True)
|
|
byte_count: Number of bytes per sample for parsing (1 or 2)
|
|
signed: True for signed integer parsing, False for unsigned
|
|
byte_order: 'MSB' or 'LSB' for byte order when parsing
|
|
|
|
Returns:
|
|
list: List of waveforms. If parse=True, each waveform is a list of
|
|
integer ADC values. If parse=False, each waveform is raw bytes.
|
|
|
|
Raises:
|
|
RuntimeError: If not connected, FastFrame not enabled, or transfer fails
|
|
ValueError: If parsing parameters are invalid
|
|
|
|
Example:
|
|
# Enable FastFrame with 1000 frames
|
|
scope.set_fastframe_state(True)
|
|
scope.set_fastframe_count(1000)
|
|
|
|
# Acquire data
|
|
scope.write("ACQuire:STATE RUN")
|
|
time.sleep(2)
|
|
scope.write("ACQuire:STATE STOP")
|
|
|
|
# Transfer all frames at once
|
|
waveforms = scope.transfer_fastframe()
|
|
print(f"Got {len(waveforms)} frames, {len(waveforms[0])} points each")
|
|
"""
|
|
# Verify FastFrame is enabled
|
|
if not self.get_fastframe_state():
|
|
raise RuntimeError("FastFrame is not enabled. Enable it with set_fastframe_state(True)")
|
|
|
|
# Use the number of frames actually acquired, not the configured maximum.
|
|
# If the stage stops early, fewer triggers arrive and the scope captures
|
|
# fewer frames than configured — reading the configured count would block.
|
|
acquired = int(self.query("ACQuire:NUMFRAMESACQuired?"))
|
|
if acquired <= 0:
|
|
raise RuntimeError(f"Scope acquired 0 FastFrame frames — no data to read")
|
|
frame_count = acquired
|
|
|
|
# Send a single CURVe? query - scope will return all frames
|
|
self.write("CURVe?")
|
|
|
|
# Read all frames as sequential IEEE 488.2 binary blocks
|
|
waveforms = []
|
|
for _ in range(frame_count):
|
|
curve_bytes = self.read_raw()
|
|
|
|
if parse:
|
|
waveform = self.parse_curve_data(
|
|
curve_bytes,
|
|
byte_count=byte_count,
|
|
signed=signed,
|
|
byte_order=byte_order
|
|
)
|
|
else:
|
|
waveform = curve_bytes
|
|
|
|
waveforms.append(waveform)
|
|
|
|
return waveforms
|
|
|
|
def parse_curve_data(self, curve_bytes, byte_count=1, signed=True, byte_order='MSB'):
|
|
"""
|
|
Parse raw curve data into integer array.
|
|
|
|
Args:
|
|
curve_bytes: Raw binary curve data
|
|
byte_count: Number of bytes per sample (1 or 2)
|
|
signed: True for signed integer, False for unsigned
|
|
byte_order: 'MSB' or 'LSB' for byte order
|
|
|
|
Returns:
|
|
list: List of integer values from the curve data
|
|
|
|
Raises:
|
|
ValueError: If parameters are invalid
|
|
"""
|
|
if byte_count not in (1, 2):
|
|
raise ValueError(f"Invalid byte count: {byte_count}. Must be 1 or 2")
|
|
|
|
if byte_order not in ('MSB', 'LSB'):
|
|
raise ValueError(f"Invalid byte order: {byte_order}. Must be 'MSB' or 'LSB'")
|
|
|
|
values = []
|
|
|
|
if byte_count == 1:
|
|
# 8-bit data
|
|
for byte in curve_bytes:
|
|
if signed:
|
|
# Convert unsigned byte to signed
|
|
value = byte if byte < 128 else byte - 256
|
|
else:
|
|
value = byte
|
|
values.append(value)
|
|
else:
|
|
# 16-bit data
|
|
num_samples = len(curve_bytes) // 2
|
|
for i in range(num_samples):
|
|
byte1 = curve_bytes[i * 2]
|
|
byte2 = curve_bytes[i * 2 + 1]
|
|
|
|
if byte_order == 'MSB':
|
|
# Most significant byte first
|
|
value = (byte1 << 8) | byte2
|
|
else:
|
|
# Least significant byte first
|
|
value = (byte2 << 8) | byte1
|
|
|
|
if signed:
|
|
# Convert unsigned to signed
|
|
if value >= 32768:
|
|
value = value - 65536
|
|
|
|
values.append(value)
|
|
|
|
return values
|
|
|
|
def acquire_waveform(self, channel, frame_number=None):
|
|
"""
|
|
High-level method to acquire waveform data from a channel.
|
|
|
|
This method:
|
|
1. Sets up data encoding (RIBinary, BINary, 1 byte, MSB)
|
|
2. Sets the data source to the specified channel
|
|
3. Optionally selects a FastFrame frame
|
|
4. Transfers the curve data
|
|
5. Parses it into a list of integer values
|
|
|
|
Args:
|
|
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
|
|
frame_number: Optional FastFrame frame number to acquire (default: None)
|
|
|
|
Returns:
|
|
list: List of integer ADC values from the waveform
|
|
|
|
Raises:
|
|
ValueError: If channel or frame_number is invalid
|
|
RuntimeError: If not connected or transfer fails
|
|
"""
|
|
# Normalize channel input
|
|
if isinstance(channel, str):
|
|
channel_upper = channel.upper()
|
|
if not (channel_upper.startswith('CH') and len(channel_upper) == 3 and channel_upper[2].isdigit()):
|
|
raise ValueError(f"Invalid channel: {channel}. Must be CH1, CH2, CH3, or CH4")
|
|
source = channel_upper
|
|
elif isinstance(channel, int):
|
|
if channel < 1 or channel > 4:
|
|
raise ValueError(f"Invalid channel: {channel}. Must be 1-4")
|
|
source = f"CH{channel}"
|
|
else:
|
|
raise ValueError(f"Invalid channel type: {type(channel)}. Must be int or str")
|
|
|
|
# Configure data transfer format
|
|
self.set_data_encoding('RIBinary')
|
|
self.set_wfmoutpre_encoding('BINary')
|
|
self.set_wfmoutpre_byte_count(1)
|
|
self.set_wfmoutpre_byte_order('MSB')
|
|
|
|
# Set data source
|
|
self.set_data_source(source)
|
|
|
|
# Select FastFrame frame if specified
|
|
if frame_number is not None:
|
|
self.set_fastframe_selected(frame_number)
|
|
|
|
# Transfer curve data
|
|
curve_bytes = self.transfer_curve()
|
|
|
|
# Parse the data
|
|
values = self.parse_curve_data(curve_bytes, byte_count=1, signed=True, byte_order='MSB')
|
|
|
|
return values
|
|
|
|
def write(self, command):
|
|
"""
|
|
Send a raw SCPI command to the instrument.
|
|
|
|
Args:
|
|
command: SCPI command string
|
|
|
|
Raises:
|
|
RuntimeError: If not connected
|
|
socket.error: If communication fails
|
|
"""
|
|
if not self._connected or not self.socket:
|
|
raise RuntimeError("Not connected to instrument")
|
|
|
|
if not command.endswith(self.terminator):
|
|
command += self.terminator
|
|
|
|
self.socket.sendall(command.encode('ascii'))
|
|
|
|
def query(self, command):
|
|
"""
|
|
Send a SCPI query and return the response.
|
|
|
|
Args:
|
|
command: SCPI query command string
|
|
|
|
Returns:
|
|
str: Response from the instrument (stripped of trailing newline)
|
|
|
|
Raises:
|
|
RuntimeError: If not connected
|
|
socket.error: If communication fails
|
|
"""
|
|
if not self._connected or not self.socket:
|
|
raise RuntimeError("Not connected to instrument")
|
|
|
|
self.write(command)
|
|
|
|
response = b''
|
|
terminator_bytes = self.terminator.encode('ascii')
|
|
while True:
|
|
chunk = self.socket.recv(4096)
|
|
if not chunk:
|
|
break
|
|
response += chunk
|
|
if terminator_bytes in chunk:
|
|
break
|
|
|
|
return response.decode('ascii').strip()
|
|
|
|
def read_raw(self):
|
|
"""
|
|
Read raw binary data from the instrument.
|
|
|
|
Useful for fast waveform data transfer. Handles IEEE 488.2 binary block format.
|
|
Format: #<N><digits><data><newline>
|
|
where N is a digit indicating how many digits follow,
|
|
and those digits specify the length of the data block.
|
|
|
|
Returns:
|
|
bytes: Raw binary data (without IEEE 488.2 header)
|
|
|
|
Raises:
|
|
RuntimeError: If not connected or invalid format
|
|
socket.error: If communication fails
|
|
"""
|
|
if not self._connected or not self.socket:
|
|
raise RuntimeError("Not connected to instrument")
|
|
|
|
# Read the '#' character
|
|
header = self.socket.recv(1)
|
|
if header != b'#':
|
|
# Debug: read more to see what's actually there
|
|
self.socket.setblocking(False)
|
|
try:
|
|
more = self.socket.recv(100)
|
|
except:
|
|
more = b''
|
|
self.socket.setblocking(True)
|
|
raise RuntimeError(f"Invalid IEEE 488.2 header: expected '#', got {header}, followed by: {more[:50]}")
|
|
|
|
# Read the digit indicating length of length field
|
|
length_of_length = self.socket.recv(1)
|
|
if not length_of_length.isdigit():
|
|
raise RuntimeError(f"Invalid length specifier: {length_of_length}")
|
|
|
|
num_digits = int(length_of_length)
|
|
|
|
# Read the data length
|
|
length_bytes = self.socket.recv(num_digits)
|
|
if len(length_bytes) != num_digits:
|
|
raise RuntimeError("Failed to read data length")
|
|
|
|
data_length = int(length_bytes)
|
|
|
|
# Read the actual binary data
|
|
data = b''
|
|
remaining = data_length
|
|
while remaining > 0:
|
|
chunk = self.socket.recv(min(remaining, 65536))
|
|
if not chunk:
|
|
raise RuntimeError("Connection closed while reading data")
|
|
data += chunk
|
|
remaining -= len(chunk)
|
|
|
|
# Read the trailing newline
|
|
self.socket.recv(1)
|
|
|
|
return data
|
|
|
|
@property
|
|
def is_connected(self):
|
|
"""Check if instrument is connected."""
|
|
return self._connected
|
|
|
|
def __enter__(self):
|
|
"""Context manager entry."""
|
|
self.connect()
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
"""Context manager exit."""
|
|
self.disconnect()
|