Keysight Instruments

import pyvisa

# Connect to Keysight instrument via USB
rm = pyvisa.ResourceManager()

# Find Keysight instruments (Vendor ID: 0x2A8D)
resources = rm.list_resources()
keysight_instruments = [r for r in resources if "0x2A8D" in r]

print("Found Keysight instruments:")
for instrument in keysight_instruments:
    try:
        inst = rm.open_resource(instrument)
        idn = inst.query("*IDN?")
        print(f"  {instrument}: {idn.strip()}")
        inst.close()
    except:
        print(f"  {instrument}: Connection failed")

# Direct IP connection (fastest)
scope_ip = "192.168.1.100"
scope = rm.open_resource(f"TCPIP0::{scope_ip}::5025::SOCKET")

# VXI-11 connection (more compatible)
scope_vxi11 = rm.open_resource(f"TCPIP0::{scope_ip}::inst0::INSTR")

# Auto-discovery using Keysight Connection Expert
# (Install Keysight IO Libraries Suite first)

class Keysight34461A:
    """Keysight 34461A/34470A Truevolt DMM Controller"""
    
    def __init__(self, resource_string):
        self.rm = pyvisa.ResourceManager()
        self.dmm = self.rm.open_resource(resource_string)
        
        # Configure for optimal performance
        self.dmm.timeout = 10000
        self.dmm.write_termination = '\n'
        self.dmm.read_termination = '\n'
        
        # Verify connection
        idn = self.dmm.query("*IDN?")
        if "34461A" not in idn and "34470A" not in idn:
            raise Exception(f"Expected Keysight 34461A/34470A, got: {idn}")
        
        print(f"Connected to: {idn.strip()}")
        
        # Initialize to known state
        self.dmm.write("*RST")
        self.dmm.write("*CLS")
    
    def configure_high_accuracy_dc_voltage(self, range_val=10):
        """Configure for highest accuracy DC voltage measurements"""
        
        self.dmm.write("CONF:VOLT:DC")
        self.dmm.write(f"VOLT:DC:RANGE {range_val}")
        self.dmm.write("VOLT:DC:NPLC 100")          # Maximum integration time
        self.dmm.write("VOLT:DC:ZERO:AUTO ON")      # Enable autozero
        self.dmm.write("VOLT:DC:IMP:AUTO OFF")      # Fixed high impedance
        self.dmm.write("VOLT:DC:IMP 10e9")          # 10 GOhm
        
        print(f"Configured for high-accuracy DC voltage, {range_val}V range")
    
    def fast_dc_voltage_measurement(self, samples=100):
        """Fast DC voltage measurements for monitoring applications"""
        
        self.dmm.write("CONF:VOLT:DC")
        self.dmm.write("VOLT:DC:NPLC 0.02")         # Minimum integration time
        self.dmm.write("VOLT:DC:ZERO:AUTO OFF")     # Disable autozero for speed
        self.dmm.write(f"SAMP:COUN {samples}")
        
        # Trigger and read all samples
        start_time = time.time()
        self.dmm.write("READ?")
        response = self.dmm.read()
        measurement_time = time.time() - start_time
        
        # Parse results
        values = [float(x) for x in response.split(',')]
        rate = len(values) / measurement_time
        
        print(f"Fast measurement: {rate:.0f} samples/second")
        return values
    
    def measure_with_statistics(self, measurement_type="VOLT:DC", samples=20):
        """Perform measurement with comprehensive statistics"""
        
        import statistics
        
        # Configure measurement
        self.dmm.write(f"CONF:{measurement_type}")
        if measurement_type == "VOLT:DC":
            self.dmm.write("VOLT:DC:NPLC 10")  # Good balance of speed/accuracy
        
        measurements = []
        print(f"Taking {samples} {measurement_type} measurements...")
        
        for i in range(samples):
            self.dmm.write("READ?")
            value = float(self.dmm.read())
            measurements.append(value)
            print(f"  #{i+1}: {value:.8f}")
            time.sleep(0.1)
        
        # Calculate statistics
        stats = {
            'mean': statistics.mean(measurements),
            'stdev': statistics.stdev(measurements) if len(measurements) > 1 else 0,
            'min': min(measurements),
            'max': max(measurements),
            'range': max(measurements) - min(measurements),
            'samples': measurements
        }
        
        return stats
    
    def close(self):
        """Clean up resources"""
        self.dmm.close()
        self.rm.close()

# Example usage
dmm = Keysight34461A("USB0::0x2A8D::0x0101::MY53220001::INSTR")

# High accuracy measurement
dmm.configure_high_accuracy_dc_voltage(10)
voltage_stats = dmm.measure_with_statistics("VOLT:DC", 10)
print(f"Voltage: {voltage_stats['mean']:.6f} ± {voltage_stats['stdev']:.6f} V")

# Fast monitoring
fast_data = dmm.fast_dc_voltage_measurement(100)
print(f"Fast measurements: {len(fast_data)} samples")

dmm.close()

def keysight_dmm_advanced_features(dmm):
    """Demonstrate advanced Keysight DMM features"""
    
    # Math functions
    dmm.write("CALC:FUNC NULL")  # Enable null function
    dmm.write("CALC:NULL:OFFS 5.000")  # Set null offset to 5V
    dmm.write("CALC:STAT ON")    # Enable math
    
    # Limits testing
    dmm.write("CALC2:FUNC LIM")  # Limit testing
    dmm.write("CALC2:LIM:LOW 4.5")   # Lower limit
    dmm.write("CALC2:LIM:UPP 5.5")   # Upper limit
    dmm.write("CALC2:STAT ON")
    
    # Perform measurement with math
    dmm.write("MEAS:VOLT:DC?")
    raw_value = float(dmm.read())
    
    # Get calculated (null) value
    dmm.write("CALC:DATA?")
    null_value = float(dmm.read())
    
    # Get limit test result
    dmm.write("CALC2:DATA?")
    limit_result = dmm.read().strip()  # "PASS" or "FAIL"
    
    print(f"Raw measurement: {raw_value:.6f} V")
    print(f"Null value: {null_value:.6f} V")
    print(f"Limit test: {limit_result}")
    
    return {
        'raw': raw_value,
        'null': null_value,
        'limit_test': limit_result
    }

class KeysightInfiniiVision:
    """Keysight InfiniiVision Oscilloscope Controller"""
    
    def __init__(self, resource_string):
        self.rm = pyvisa.ResourceManager()
        self.scope = self.rm.open_resource(resource_string)
        
        # Configure for large data transfers
        self.scope.timeout = 30000
        self.scope.chunk_size = 1024 * 1024  # 1MB chunks
        
        # Verify connection
        idn = self.scope.query("*IDN?")
        print(f"Connected to: {idn.strip()}")
        
        # Initialize
        self.scope.write("*RST")
        self.scope.write(":WAV:FORM BYTE")  # 8-bit data format
        self.scope.write(":WAV:MODE NORM")  # Normal mode
    
    def setup_acquisition(self, channel=1, scale=1.0, time_scale=1e-3):
        """Setup basic acquisition parameters"""
        
        # Channel setup
        self.scope.write(f":CHAN{channel}:DISP ON")
        self.scope.write(f":CHAN{channel}:SCAL {scale}")
        self.scope.write(f":CHAN{channel}:OFFS 0")
        
        # Time base setup
        self.scope.write(f":TIM:SCAL {time_scale}")
        self.scope.write(":TIM:POS 0")
        
        # Trigger setup
        self.scope.write(f":TRIG:EDGE:SOUR CHAN{channel}")
        self.scope.write(":TRIG:EDGE:LEV 0")
        self.scope.write(":TRIG:EDGE:SLOP POS")
        
        print(f"Setup complete: CH{channel}, {scale}V/div, {time_scale*1000}ms/div")
    
    def acquire_waveform(self, channel=1, points=1000):
        """Acquire waveform data efficiently"""
        
        # Set data source
        self.scope.write(f":WAV:SOUR CHAN{channel}")
        self.scope.write(f":WAV:POIN {points}")
        
        # Single acquisition
        self.scope.write(":SING")
        
        # Wait for trigger
        self.scope.query("*OPC?")
        
        # Get waveform preamble (scaling info)
        preamble = self.scope.query(":WAV:PRE?").split(',')
        y_increment = float(preamble[7])
        y_origin = float(preamble[8])
        y_reference = float(preamble[9])
        x_increment = float(preamble[4])
        x_origin = float(preamble[5])
        
        # Get waveform data
        self.scope.write(":WAV:DATA?")
        raw_data = self.scope.read_raw()
        
        # Remove IEEE header (first few bytes)
        header_len = 2 + int(chr(raw_data[1]))
        waveform_data = raw_data[header_len:-1]  # Remove header and terminator
        
        # Convert to numpy array
        import numpy as np
        data_array = np.frombuffer(waveform_data, dtype=np.uint8)
        
        # Scale to actual voltage values
        voltage_data = (data_array - y_reference) * y_increment + y_origin
        
        # Create time array
        time_data = np.arange(len(voltage_data)) * x_increment + x_origin
        
        return time_data, voltage_data, {
            'points': len(voltage_data),
            'time_scale': x_increment,
            'voltage_scale': y_increment,
            'sample_rate': 1/x_increment
        }
    
    def measure_parameters(self, channel=1):
        """Get automated measurements"""
        
        measurements = {}
        
        # Voltage measurements
        measurements['vpp'] = float(self.scope.query(f":MEAS:VPP? CHAN{channel}"))
        measurements['vmax'] = float(self.scope.query(f":MEAS:VMAX? CHAN{channel}"))
        measurements['vmin'] = float(self.scope.query(f":MEAS:VMIN? CHAN{channel}"))
        measurements['vavg'] = float(self.scope.query(f":MEAS:VAV? CHAN{channel}"))
        measurements['vrms'] = float(self.scope.query(f":MEAS:VRMS? CHAN{channel}"))
        
        # Time measurements
        try:
            measurements['frequency'] = float(self.scope.query(f":MEAS:FREQ? CHAN{channel}"))
            measurements['period'] = float(self.scope.query(f":MEAS:PER? CHAN{channel}"))
            measurements['rise_time'] = float(self.scope.query(f":MEAS:RIS? CHAN{channel}"))
            measurements['fall_time'] = float(self.scope.query(f":MEAS:FALL? CHAN{channel}"))
        except:
            # Measurements might fail if signal is not periodic
            pass
        
        return measurements
    
    def screenshot(self, filename="screenshot.png"):
        """Capture oscilloscope screenshot"""
        
        # Set image format
        self.scope.write(":DISP:DATA? PNG, COL")
        image_data = self.scope.read_raw()
        
        # Remove SCPI header
        header_len = 2 + int(chr(image_data[1]))
        png_data = image_data[header_len:]
        
        # Save to file
        with open(filename, 'wb') as f:
            f.write(png_data)
        
        print(f"Screenshot saved as {filename}")
        return filename
    
    def close(self):
        """Clean up resources"""
        self.scope.close()
        self.rm.close()

# Example usage
scope = KeysightInfiniiVision("USB0::0x2A8D::0x0001::MY52345678::INSTR")

# Setup and acquire
scope.setup_acquisition(channel=1, scale=2.0, time_scale=1e-3)  # 2V/div, 1ms/div
time_data, voltage_data, info = scope.acquire_waveform(channel=1, points=2000)

print(f"Acquired {info['points']} points at {info['sample_rate']/1e6:.1f} MSa/s")

# Get measurements
measurements = scope.measure_parameters(channel=1)
print("Automated measurements:")
for param, value in measurements.items():
    print(f"  {param}: {value}")

# Take screenshot
scope.screenshot("waveform_capture.png")

scope.close()

class Keysight33500B:
    """Keysight 33500B Series Function/Arbitrary Waveform Generator"""
    
    def __init__(self, resource_string):
        self.rm = pyvisa.ResourceManager()
        self.gen = self.rm.open_resource(resource_string)
        
        # Configure connection
        self.gen.timeout = 10000
        
        # Verify and initialize
        idn = self.gen.query("*IDN?")
        print(f"Connected to: {idn.strip()}")
        
        self.gen.write("*RST")
        self.gen.write("*CLS")
    
    def generate_sine_wave(self, channel=1, frequency=1000, amplitude=1.0, offset=0):
        """Generate basic sine wave"""
        
        # Configure waveform
        self.gen.write(f"SOUR{channel}:FUNC SIN")
        self.gen.write(f"SOUR{channel}:FREQ {frequency}")
        self.gen.write(f"SOUR{channel}:VOLT {amplitude}")
        self.gen.write(f"SOUR{channel}:VOLT:OFFS {offset}")
        
        # Enable output
        self.gen.write(f"OUTP{channel} ON")
        
        print(f"CH{channel}: {frequency}Hz sine wave, {amplitude}Vpp, {offset}V offset")
    
    def generate_square_wave(self, channel=1, frequency=1000, amplitude=5.0, duty_cycle=50):
        """Generate square wave with adjustable duty cycle"""
        
        self.gen.write(f"SOUR{channel}:FUNC SQU")
        self.gen.write(f"SOUR{channel}:FREQ {frequency}")
        self.gen.write(f"SOUR{channel}:VOLT {amplitude}")
        self.gen.write(f"SOUR{channel}:FUNC:SQU:DCYC {duty_cycle}")
        
        self.gen.write(f"OUTP{channel} ON")
        
        print(f"CH{channel}: {frequency}Hz square wave, {duty_cycle}% duty cycle")
    
    def sweep_frequency(self, channel=1, start_freq=100, stop_freq=10000, 
                       sweep_time=10, amplitude=1.0):
        """Generate frequency sweep"""
        
        # Configure basic waveform
        self.gen.write(f"SOUR{channel}:FUNC SIN")
        self.gen.write(f"SOUR{channel}:VOLT {amplitude}")
        
        # Configure sweep
        self.gen.write(f"SOUR{channel}:FREQ:START {start_freq}")
        self.gen.write(f"SOUR{channel}:FREQ:STOP {stop_freq}")
        self.gen.write(f"SOUR{channel}:SWE:TIME {sweep_time}")
        self.gen.write(f"SOUR{channel}:SWE:STAT ON")
        
        self.gen.write(f"OUTP{channel} ON")
        
        print(f"CH{channel}: Frequency sweep {start_freq}-{stop_freq}Hz in {sweep_time}s")
    
    def arbitrary_waveform(self, channel=1, waveform_data, sample_rate=1e6):
        """Upload and generate arbitrary waveform"""
        
        import numpy as np
        
        # Normalize waveform data to ±1
        if isinstance(waveform_data, list):
            waveform_data = np.array(waveform_data)
        
        waveform_normalized = waveform_data / np.max(np.abs(waveform_data))
        
        # Convert to comma-separated string
        waveform_str = ','.join([f"{val:.6f}" for val in waveform_normalized])
        
        # Upload waveform
        self.gen.write(f"SOUR{channel}:DATA:VOL:CLE")  # Clear memory
        self.gen.write(f"SOUR{channel}:DATA VOLATILE,{waveform_str}")
        
        # Configure arbitrary function
        self.gen.write(f"SOUR{channel}:FUNC ARB")
        self.gen.write(f"SOUR{channel}:FUNC:ARB VOLATILE")
        
        # Set sample rate (determines frequency)
        points = len(waveform_data)
        frequency = sample_rate / points
        self.gen.write(f"SOUR{channel}:FREQ {frequency}")
        
        self.gen.write(f"OUTP{channel} ON")
        
        print(f"CH{channel}: Arbitrary waveform, {points} points, {frequency:.0f}Hz")
    
    def burst_mode(self, channel=1, burst_count=10, frequency=1000):
        """Configure burst mode operation"""
        
        # Configure basic waveform
        self.gen.write(f"SOUR{channel}:FUNC SIN")
        self.gen.write(f"SOUR{channel}:FREQ {frequency}")
        
        # Configure burst
        self.gen.write(f"SOUR{channel}:BURS:STAT ON")
        self.gen.write(f"SOUR{channel}:BURS:MODE TRIG")
        self.gen.write(f"SOUR{channel}:BURS:NCYC {burst_count}")
        
        # Configure trigger
        self.gen.write("TRIG:SOUR BUS")
        
        self.gen.write(f"OUTP{channel} ON")
        
        print(f"CH{channel}: Burst mode, {burst_count} cycles per trigger")
    
    def trigger_burst(self):
        """Trigger a burst"""
        self.gen.write("*TRG")
        print("Burst triggered")
    
    def close(self):
        """Clean up resources"""
        # Turn off all outputs
        self.gen.write("OUTP1 OFF")
        self.gen.write("OUTP2 OFF")
        self.gen.close()
        self.rm.close()

# Example usage
gen = Keysight33500B("USB0::0x2A8D::0x0001::MY52345678::INSTR")

# Basic sine wave
gen.generate_sine_wave(channel=1, frequency=1000, amplitude=2.0)

# Square wave on channel 2
gen.generate_square_wave(channel=2, frequency=500, amplitude=3.3, duty_cycle=25)

# Create and upload arbitrary waveform
import numpy as np
t = np.linspace(0, 2*np.pi, 1000)
custom_wave = np.sin(t) + 0.3*np.sin(3*t)  # Fundamental + 3rd harmonic
gen.arbitrary_waveform(channel=1, waveform_data=custom_wave, sample_rate=1e6)

gen.close()

def optimize_keysight_connection(instrument):
    """Apply Keysight-specific optimizations"""
    
    # Use largest possible buffer sizes
    instrument.chunk_size = 2 * 1024 * 1024  # 2MB
    instrument.timeout = 60000  # 60 seconds
    
    # Keysight instruments support fast binary transfers
    try:
        # Set binary data format when available
        instrument.write("FORM:DATA REAL,64")  # Double precision
        # or
        instrument.write("FORM:DATA INT,32")   # 32-bit integers
    except:
        pass
    
    # Disable unnecessary features for speed
    try:
        instrument.write("SYST:DISP:UPD OFF")  # Disable display updates
        instrument.write("SYST:BEEP:STAT OFF") # Disable beep
    except:
        pass

def keysight_error_checking(instrument):
    """Comprehensive error checking for Keysight instruments"""
    
    # Check system errors
    error_count = 0
    while True:
        error = instrument.query("SYST:ERR?")
        if error.startswith("+0,"):  # No error
            break
        print(f"System error: {error}")
        error_count += 1
        if error_count > 10:  # Prevent infinite loop
            break
    
    # Check specific instrument status
    try:
        # For oscilloscopes
        acq_status = instrument.query(":OPER:COND?")
        print(f"Acquisition status: {acq_status}")
        
        # For signal generators
        output_status = instrument.query("OUTP:PROT:TRIP?")
        if output_status.strip() == "1":
            print("Warning: Output protection tripped!")
            
    except:
        pass
    
    return error_count == 0

def keysight_advanced_features():
    """Demonstrate advanced Keysight-specific features"""
    
    # Segmented memory (oscilloscopes)
    def setup_segmented_memory(scope, segments=100):
        scope.write(f":ACQ:SEGM:COUN {segments}")
        scope.write(":ACQ:SEGM:STAT ON")
        print(f"Segmented memory: {segments} segments")
    
    # High-speed digitizing mode
    def setup_high_speed_digitizing(scope):
        scope.write(":ACQ:MODE HRES")  # High resolution mode
        scope.write(":ACQ:SRAT:AUTO OFF")
        scope.write(":ACQ:SRAT MAX")  # Maximum sample rate
        print("High-speed digitizing mode enabled")
    
    # Waveform math (oscilloscopes)
    def setup_waveform_math(scope):
        scope.write(":FUNC:DISP ON")
        scope.write(":FUNC:OPER ADD")  # Add channels
        scope.write(":FUNC:SOUR1 CHAN1")
        scope.write(":FUNC:SOUR2 CHAN2")
        print("Math function: CH1 + CH2")
    
    # Sequence mode (signal generators)
    def setup_sequence_mode(generator):
        # Create sequence with multiple waveforms
        generator.write("SOUR:FUNC:ARB:SEQ:LENG 3")
        generator.write("SOUR:FUNC:ARB:SEQ:TRAC:WFOR SINE")
        generator.write("SOUR:FUNC:ARB:SEQ:TRAC:FREQ 1000")
        # Add more sequence steps...
        print("Sequence mode configured")

def troubleshoot_keysight_connection():
    """Troubleshoot common Keysight connection issues"""
    
    print("=== Keysight Instrument Troubleshooting ===")
    
    # Check for Keysight IO Libraries
    try:
        import win32api
        io_libs_path = r"C:\Program Files (x86)\Keysight\IO Libraries Suite"
        if os.path.exists(io_libs_path):
            print("Keysight IO Libraries Suite found")
        else:
            print("Keysight IO Libraries Suite not found")
            print("   Download from: https://www.keysight.com/find/iosuite")
    except:
        pass
    
    # Test USB connection
    rm = pyvisa.ResourceManager()
    keysight_usb = [r for r in rm.list_resources() if "0x2A8D" in r]
    
    if keysight_usb:
        print(f"Found {len(keysight_usb)} Keysight USB instruments")
        for resource in keysight_usb:
            try:
                inst = rm.open_resource(resource)
                idn = inst.query("*IDN?")
                print(f"  {resource}: {idn.strip()}")
                inst.close()
            except Exception as e:
                print(f"  {resource}: Error - {e}")
    else:
        print("No Keysight USB instruments found")
        print("   Check USB connection and drivers")
    
    # Test common network addresses
    common_ips = ["192.168.1.100", "10.0.0.100", "169.254.1.1"]
    for ip in common_ips:
        try:
            inst = rm.open_resource(f"TCPIP0::{ip}::5025::SOCKET", timeout=2000)
            idn = inst.query("*IDN?")
            if "Keysight" in idn or "Agilent" in idn:
                print(f"Found Keysight instrument at {ip}: {idn.strip()}")
            inst.close()
        except:
            pass
    
    rm.close()