Instrument Control Guide

import pyvisa
import time

# Connect to DMM
rm = pyvisa.ResourceManager()
dmm = rm.open_resource("USB0::0x2A8D::0x0101::MY53220001::INSTR")

# High-precision voltage measurement
dmm.write("CONF:VOLT:DC")
dmm.write("VOLT:DC:NPLC 10")  # High accuracy mode
voltage = float(dmm.query("READ?"))

print(f"Measured voltage: {voltage:.6f} V")
dmm.close()

# Connect to oscilloscope
scope = rm.open_resource("USB0::0x0699::0x0363::C065089::INSTR")

# Configure for signal capture
scope.write("CH1:SCALE 1.0")      # 1V/div
scope.write("HOR:SCALE 1E-3")     # 1ms/div
scope.write("TRIG:MAIN:LEV 0.5")  # 500mV trigger

# Capture waveform
scope.write("SING")  # Single acquisition
scope.query("*OPC?")  # Wait for completion

# Get measurement
frequency = float(scope.query("MEASU:FREQ?"))
amplitude = float(scope.query("MEASU:PK2PK?"))

print(f"Signal: {frequency:.0f} Hz, {amplitude:.3f} Vpp")

# Connect to power supply  
psu = rm.open_resource("ASRL/dev/ttyUSB0::INSTR", baud_rate=115200)

# Configure output
psu.write("VOLT 12.0")    # Set 12V
psu.write("CURR 2.0")     # Limit to 2A  
psu.write("OUTP ON")      # Enable output

# Monitor load
voltage = float(psu.query("MEAS:VOLT?"))
current = float(psu.query("MEAS:CURR?"))
power = voltage * current

print(f"Load: {voltage:.2f}V, {current:.3f}A, {power:.2f}W")

# Connect to signal generator
sig_gen = rm.open_resource("USB0::0x2A8D::0x0001::MY52345678::INSTR")

# Generate sine wave
sig_gen.write("SOUR:FUNC SIN")
sig_gen.write("SOUR:FREQ 1000")     # 1 kHz
sig_gen.write("SOUR:VOLT 2.0")      # 2 Vpp
sig_gen.write("OUTP ON")            # Enable output

print("Generating 1 kHz sine wave, 2 Vpp")

# Connect to network analyzer
na = rm.open_resource("TCPIP0::192.168.1.100::5025::SOCKET")

# Configure S-parameter measurement
na.write("CALC:PAR:DEF 'S11',S11")
na.write("SENS:FREQ:START 1E6")    # 1 MHz start  
na.write("SENS:FREQ:STOP 1E9")     # 1 GHz stop
na.write("SENS:SWE:POIN 201")      # 201 points

# Perform measurement
na.write("INIT:IMM")
na.query("*OPC?")

# Get data
data = na.query("CALC:DATA:FDATA?")
s11_data = [float(x) for x in data.split(',')]

print(f"S11 measurement complete: {len(s11_data)} points")

import pandas as pd
import matplotlib.pyplot as plt
from datetime import datetime

class InstrumentLogger:
    """Professional instrument data logging"""
    
    def __init__(self, instrument, log_file="data.csv"):
        self.instrument = instrument
        self.log_file = log_file
        self.data = []
    
    def log_measurement(self, measurement_cmd="MEAS:VOLT:DC?"):
        """Log single measurement with timestamp"""
        timestamp = datetime.now()
        value = float(self.instrument.query(measurement_cmd))
        
        self.data.append({
            'timestamp': timestamp,
            'value': value
        })
        
        return timestamp, value
    
    def continuous_logging(self, duration_minutes=60, interval_seconds=10):
        """Log data continuously"""
        import time
        
        end_time = time.time() + (duration_minutes * 60)
        
        print(f"Starting {duration_minutes} minute data logging...")
        
        while time.time() < end_time:
            timestamp, value = self.log_measurement()
            print(f"{timestamp.strftime('%H:%M:%S')}: {value:.6f}")
            time.sleep(interval_seconds)
        
        self.save_data()
    
    def save_data(self):
        """Save logged data to CSV"""
        df = pd.DataFrame(self.data)
        df.to_csv(self.log_file, index=False)
        print(f"Data saved to {self.log_file}")
    
    def plot_data(self):
        """Create time series plot"""
        df = pd.DataFrame(self.data)
        plt.figure(figsize=(12, 6))
        plt.plot(df['timestamp'], df['value'])
        plt.title('Instrument Measurements Over Time')
        plt.xlabel('Time')
        plt.ylabel('Measurement')
        plt.grid(True)
        plt.xticks(rotation=45)
        plt.tight_layout()
        plt.show()

# Example usage
dmm = rm.open_resource("USB0::0x2A8D::0x0101::MY53220001::INSTR")
logger = InstrumentLogger(dmm, "battery_test.csv")

# Log for 10 minutes, one reading per 30 seconds
logger.continuous_logging(duration_minutes=10, interval_seconds=30)
logger.plot_data()

class TestSequenceController:
    """Coordinate multiple instruments for complex tests"""
    
    def __init__(self):
        self.instruments = {}
        self.rm = pyvisa.ResourceManager()
    
    def add_instrument(self, name, resource_string, **config):
        """Add instrument to test setup"""
        inst = self.rm.open_resource(resource_string)
        
        # Apply configuration
        if 'timeout' in config:
            inst.timeout = config['timeout']
        if 'termination' in config:
            inst.write_termination = config['termination']
            inst.read_termination = config['termination']
        
        self.instruments[name] = inst
        print(f"Added {name}: {inst.query('*IDN?').strip()}")
    
    def device_under_test_sequence(self):
        """Example: Device characterization sequence"""
        
        # Step 1: Apply power
        psu = self.instruments['power_supply']
        psu.write("VOLT 5.0")
        psu.write("CURR 1.0") 
        psu.write("OUTP ON")
        time.sleep(1)  # Settling time
        
        # Step 2: Generate test signal
        sig_gen = self.instruments['signal_generator']
        sig_gen.write("SOUR:FUNC SIN")
        sig_gen.write("SOUR:FREQ 1000")
        sig_gen.write("SOUR:VOLT 1.0")
        sig_gen.write("OUTP ON")
        time.sleep(0.5)
        
        # Step 3: Measure response
        scope = self.instruments['oscilloscope']
        scope.write("SING")
        scope.query("*OPC?")
        
        frequency = float(scope.query("MEASU:FREQ?"))
        amplitude = float(scope.query("MEASU:PK2PK?"))
        
        # Step 4: Measure power consumption
        supply_voltage = float(psu.query("MEAS:VOLT?"))
        supply_current = float(psu.query("MEAS:CURR?"))
        power_consumption = supply_voltage * supply_current
        
        # Step 5: Cleanup
        sig_gen.write("OUTP OFF")
        psu.write("OUTP OFF")
        
        results = {
            'input_frequency': 1000,
            'output_frequency': frequency,
            'output_amplitude': amplitude,
            'power_consumption': power_consumption,
            'test_passed': abs(frequency - 1000) < 10  # 10 Hz tolerance
        }
        
        return results
    
    def close_all(self):
        """Clean up all instruments"""
        for name, inst in self.instruments.items():
            try:
                inst.close()
            except:
                pass
        self.rm.close()

# Example usage
controller = TestSequenceController()
controller.add_instrument('power_supply', 'ASRL/dev/ttyUSB0::INSTR', 
                         timeout=5000, termination='\n')
controller.add_instrument('signal_generator', 'USB0::0x2A8D::0x0001::MY52345678::INSTR')
controller.add_instrument('oscilloscope', 'USB0::0x0699::0x0363::C065089::INSTR')

# Run test sequence
test_results = controller.device_under_test_sequence()
print(f"Test results: {test_results}")

controller.close_all()

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy import signal

def analyze_waveform_data(scope_resource):
    """Complete waveform analysis example"""
    
    scope = rm.open_resource(scope_resource)
    
    # Configure scope for data acquisition
    scope.write("DATA:ENCDG RIBINARY")  # Binary for speed
    scope.write("DATA:WIDTH 2")         # 16-bit data
    scope.write("WFMO:YMULT?")
    y_mult = float(scope.read())
    
    # Acquire waveform
    scope.write("SING")
    scope.query("*OPC?")
    scope.write("CURVE?")
    
    # Get binary data
    raw_data = scope.read_raw()
    
    # Convert to numpy array (skip IEEE header)
    waveform = np.frombuffer(raw_data[2:], dtype=np.int16) * y_mult
    
    # Create time array
    sample_rate = 1e6  # 1 MSa/s (get from instrument)
    time_array = np.arange(len(waveform)) / sample_rate
    
    # Analyze waveform
    analysis = {
        'mean': np.mean(waveform),
        'std': np.std(waveform),
        'pk_pk': np.max(waveform) - np.min(waveform),
        'rms': np.sqrt(np.mean(waveform**2))
    }
    
    # Frequency domain analysis
    freqs, power_spectrum = signal.welch(waveform, sample_rate)
    fundamental_freq = freqs[np.argmax(power_spectrum[1:])]  # Skip DC
    
    analysis['fundamental_frequency'] = fundamental_freq
    
    # Create comprehensive plot
    fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8))
    
    # Time domain
    ax1.plot(time_array * 1000, waveform)  # Convert to ms
    ax1.set_xlabel('Time (ms)')
    ax1.set_ylabel('Voltage (V)')
    ax1.set_title('Waveform - Time Domain')
    ax1.grid(True)
    
    # Frequency domain
    ax2.loglog(freqs, power_spectrum)
    ax2.set_xlabel('Frequency (Hz)')
    ax2.set_ylabel('Power Spectral Density')
    ax2.set_title('Waveform - Frequency Domain')
    ax2.grid(True)
    
    plt.tight_layout()
    plt.show()
    
    scope.close()
    return analysis, waveform, time_array

# Example usage
# analysis = analyze_waveform_data("USB0::0x0699::0x0363::C065089::INSTR")
# print(f"Waveform analysis: {analysis}")

from sklearn.linear_model import LinearRegression
from sklearn.preprocessing import PolynomialFeatures
import joblib

class InstrumentCalibration:
    """ML-powered instrument calibration"""
    
    def __init__(self, reference_instrument, dut_instrument):
        self.reference = reference_instrument
        self.dut = dut_instrument
        self.calibration_model = None
    
    def collect_calibration_data(self, test_points=50):
        """Collect calibration data points"""
        
        reference_readings = []
        dut_readings = []
        
        # Generate test voltages (example for DMM calibration)
        test_voltages = np.linspace(0, 10, test_points)
        
        for voltage in test_voltages:
            # Set reference voltage (assuming programmable reference)
            # In practice, you'd use a precision calibrator
            
            # Read both instruments
            ref_reading = float(self.reference.query("MEAS:VOLT:DC?"))
            dut_reading = float(self.dut.query("MEAS:VOLT:DC?"))
            
            reference_readings.append(ref_reading)
            dut_readings.append(dut_reading)
            
            time.sleep(0.1)  # Settling time
        
        return np.array(reference_readings), np.array(dut_readings)
    
    def train_calibration_model(self, degree=2):
        """Train polynomial calibration model"""
        
        ref_data, dut_data = self.collect_calibration_data()
        
        # Use polynomial features for non-linear calibration
        poly_features = PolynomialFeatures(degree=degree)
        dut_poly = poly_features.fit_transform(dut_data.reshape(-1, 1))
        
        # Train linear regression on polynomial features
        model = LinearRegression()
        model.fit(dut_poly, ref_data)
        
        # Store model and preprocessing
        self.calibration_model = {
            'model': model,
            'poly_features': poly_features,
            'degree': degree
        }
        
        # Calculate calibration accuracy
        predictions = self.apply_calibration(dut_data)
        accuracy = np.mean(np.abs(predictions - ref_data))
        
        print(f"Calibration model trained (degree {degree})")
        print(f"Mean absolute error: {accuracy:.6f}")
        
        return accuracy
    
    def apply_calibration(self, raw_readings):
        """Apply calibration to raw readings"""
        
        if self.calibration_model is None:
            raise Exception("Calibration model not trained")
        
        model = self.calibration_model['model']
        poly_features = self.calibration_model['poly_features']
        
        # Ensure raw_readings is numpy array
        if not isinstance(raw_readings, np.ndarray):
            raw_readings = np.array([raw_readings])
        
        # Apply polynomial transformation
        raw_poly = poly_features.transform(raw_readings.reshape(-1, 1))
        
        # Get calibrated values
        calibrated = model.predict(raw_poly)
        
        return calibrated
    
    def save_calibration(self, filename):
        """Save calibration model"""
        joblib.dump(self.calibration_model, filename)
        print(f"Calibration saved to {filename}")
    
    def load_calibration(self, filename):
        """Load calibration model"""
        self.calibration_model = joblib.load(filename)
        print(f"Calibration loaded from {filename}")

# Example usage
# reference_dmm = rm.open_resource("USB0::0x2A8D::0x0101::REF123456::INSTR")
# test_dmm = rm.open_resource("USB0::0x2A8D::0x0101::DUT789012::INSTR")
# 
# calibrator = InstrumentCalibration(reference_dmm, test_dmm)
# calibrator.train_calibration_model(degree=3)
# calibrator.save_calibration("dmm_calibration.pkl")

# LabVIEW-style error handling in Python
def labview_style_function():
    """Python function with LabVIEW-style error handling"""
    error_code = 0
    error_message = ""
    result = None
    
    try:
        # Your instrument code here
        rm = pyvisa.ResourceManager()
        inst = rm.open_resource("USB0::0x2A8D::0x0101::MY53220001::INSTR")
        result = float(inst.query("MEAS:VOLT:DC?"))
        inst.close()
        
    except Exception as e:
        error_code = -1
        error_message = str(e)
    
    return result, error_code, error_message

# Usage
voltage, err_code, err_msg = labview_style_function()
if err_code == 0:
    print(f"Measurement: {voltage} V")
else:
    print(f"Error {err_code}: {err_msg}")

# Always use try/finally or context managers
try:
    rm = pyvisa.ResourceManager()
    inst = rm.open_resource("USB0::0x1234::0x5678::SN::INSTR")
    # Your code here
finally:
    inst.close()
    rm.close()

# Or better yet, use context managers
with pyvisa.ResourceManager() as rm:
    with rm.open_resource("USB0::0x1234::0x5678::SN::INSTR") as inst:
        # Your code here
        pass  # Resources automatically closed

from pyvisa import VisaIOError

try:
    inst.write("*IDN?")
    response = inst.read()
except VisaIOError as e:
    if e.error_code == -1073807339:  # Timeout
        print("Instrument not responding")
    elif e.error_code == -1073807343:  # Invalid resource
        print("Device not found")
    else:
        print(f"VISA error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

# Optimize for speed
instrument.chunk_size = 1024 * 1024  # Large buffer for data transfers
instrument.timeout = 30000  # Appropriate timeout

# Use binary data format for large datasets
instrument.write("DATA:ENCDG RIBINARY")

# Batch operations when possible
commands = ["CONF:VOLT:DC", "SAMP:COUN 100", "READ?"]
for cmd in commands[:-1]:
    instrument.write(cmd)
data = instrument.query(commands[-1])