diff --git a/CHANGELOG.md b/CHANGELOG.md index ac6a221..55e0606 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,34 @@ API changes can land even when the minor version is unchanged. ## [Unreleased] +### Added +- `SR830Device` (and `DebugSR830Device`): Stanford Research SR830 DSP lock-in + amplifier over a Prologix GPIB-USB controller. It combines several capabilities: + `AnalogInputStreamDevice` (the four rear-panel Aux A/D inputs via `OAUX?`, plus + hardware-timed buffered acquisition of the demodulated outputs from the internal + data buffer), `AnalogOutputDevice` (the four rear-panel Aux D/A outputs via + `AUXV`), `PhaseLockedDetectionDevice` (X/Y/R/theta, reference frequency, signal + input source, sensitivity, and time constant), and `TriggerableDevice` (the + rear-panel TRIG IN). Enums: `AuxInput`, `AuxOutput`, `StreamChannel`, + `InputSource`. `doInitializeDevice` self-discovers the Prologix among the + connected FTDI adaptors by confirming `*IDN?`, and pins the adaptor's serial. +- `PrologixGPIBPort` (`hardwarelibrary/communication/`): a `SerialPort` subclass + that speaks the Prologix GPIB-USB controller `++` protocol. GPIB instruments + talk to it with the ordinary `readString`/`writeString` primitives; the `++read + eoi` handshake is encapsulated in its `readString`. +- New DAQ capability contracts in `daq/daqdevice.py`: `PhaseLockedDetectionDevice` + (lock-in / phase-locked detection), `TriggerableDevice` with the `TriggerSource` + enum, and the `SampleClock` enum for stream sample clocking. + +### Changed +- `AnalogInputStreamDevice`: the sample-rate parameter of + `configureStream`/`acquireWaveform` is renamed `scanRate` -> `sampleRate`. + `LabjackDevice.configureStream` keeps `scanRate` as a temporary deprecated + synonym for existing callers. +- `LabjackDevice` now imports `u3` (LabJackPython) lazily at point of use, so + `import hardwarelibrary.daq` (and the new SR830 driver) works on hosts that do + not have LabJackPython installed. + ## [1.3.3] - 2026-07-08 ### Changed diff --git a/hardwarelibrary/communication/__init__.py b/hardwarelibrary/communication/__init__.py index f16dc4a..d9d11b9 100644 --- a/hardwarelibrary/communication/__init__.py +++ b/hardwarelibrary/communication/__init__.py @@ -1,5 +1,6 @@ from .communicationport import * from .serialport import SerialPort +from .prologixgpibport import PrologixGPIBPort from .tcpport import TCPPort, UnableToOpenTCPPort from .labviewtcpport import LabviewTCPPort from .usbport import USBPort diff --git a/hardwarelibrary/communication/prologixgpibport.py b/hardwarelibrary/communication/prologixgpibport.py new file mode 100644 index 0000000..a4302bb --- /dev/null +++ b/hardwarelibrary/communication/prologixgpibport.py @@ -0,0 +1,75 @@ +from .serialport import SerialPort + + +class PrologixGPIBPort(SerialPort): + """A GPIB instrument reached through a Prologix GPIB-USB controller. + + The Prologix controller enumerates as a generic FTDI serial port + (VID 0x0403 / PID 0x6001), so this IS a SerialPort: it inherits the FTDI + discovery and every read/write primitive unchanged, and adds only the + controller handshake in open(). GPIB payloads are ordinary ASCII, so callers + use the inherited readString/writeString (and the writeStringReadMatch + family); there is no GPIB-specific read/write method. + + The controller runs in manual-read mode (++auto 0): a query is read back by + an explicit '++read eoi', which this class issues inside readString(). So the + driver still talks to the port with plain writeString/readString (and the + writeStringReadMatch family) and never touches a GPIB-specific method. Manual + read is used rather than ++auto 1 because auto mode addresses the instrument + to talk after every command, including commands with no reply (e.g. SENS), + which leaves the bus in an error state. + """ + + def __init__(self, gpibAddress, portPath=None, idVendor=0x0403, + idProduct=0x6001, serialNumber=None): + """Create a port for the instrument at GPIB address gpibAddress. + + gpibAddress is the addressed instrument's GPIB (IEEE-488) primary + address. The remaining arguments locate the FTDI-based Prologix adaptor + itself and are forwarded to SerialPort: an explicit portPath, or the + adaptor's idVendor/idProduct/serialNumber for discovery. + """ + super().__init__(idVendor=idVendor, idProduct=idProduct, + serialNumber=serialNumber, portPath=portPath) + self.gpibAddress = gpibAddress + + def open(self, baudRate=115200, timeout=1.0): + """Open the serial line, then apply the Prologix controller handshake. + + Configuring the controller in open() (rather than a separate step the + caller must remember) mirrors SerialPort.open()/USBPort.open(): once + open() returns, the port is ready for readString/writeString. + """ + super().open(baudRate=baudRate, timeout=timeout) + self.configureController() + + def configureController(self): + """Send the one-time Prologix controller handshake. + + ++eos 2 appends a line-feed to commands forwarded to the instrument (the + SR830 and most GPIB instruments accept LF or EOI as a terminator). + ++eot_enable 0 keeps the controller from appending its own terminator to + replies: the instrument already ends its response with CR-LF, so a second + appended LF would be left stranded in the buffer and corrupt the next + read. + """ + for command in ("++mode 1", + "++addr {0}".format(self.gpibAddress), + "++auto 0", + "++eoi 1", + "++eos 2", + "++eot_enable 0", + "++read_tmo_ms 1500"): + self.writeString(command + "\n") + + def readString(self, endPoint=None) -> str: + """Read one reply from the addressed instrument. + + In manual-read mode (++auto 0) the controller does not forward a reply + until told to, so this issues '++read eoi' (read until the instrument + asserts EOI) and then reads the forwarded bytes with the inherited serial + readString. Encapsulating the handshake here lets the driver use the + ordinary readString/writeStringReadMatch primitives unchanged. + """ + self.writeString("++read eoi\n") + return super().readString(endPoint) diff --git a/hardwarelibrary/daq/__init__.py b/hardwarelibrary/daq/__init__.py index c7d896f..a2cefe9 100644 --- a/hardwarelibrary/daq/__init__.py +++ b/hardwarelibrary/daq/__init__.py @@ -3,5 +3,11 @@ from .daqdevice import ( AnalogInputDevice, AnalogOutputDevice, AnalogIODevice, AnalogInputStreamDevice, DigitalInputDevice, DigitalOutputDevice, DigitalIODevice, + PhaseLockedDetectionDevice, InputSource, + TriggerableDevice, TriggerSource, SampleClock, ) -from .labjackdevice import LabjackDevice, DebugLabjackDevice \ No newline at end of file +from .labjackdevice import LabjackDevice, DebugLabjackDevice +from .sr830device import ( + SR830Device, DebugSR830Device, DebugPrologixGPIBPort, + AuxInput, AuxOutput, StreamChannel, +) \ No newline at end of file diff --git a/hardwarelibrary/daq/daqdevice.py b/hardwarelibrary/daq/daqdevice.py index 5c385ee..7f1a7d2 100644 --- a/hardwarelibrary/daq/daqdevice.py +++ b/hardwarelibrary/daq/daqdevice.py @@ -43,19 +43,19 @@ class AnalogInputStreamDevice(AnalogInputDevice): """Hardware-timed analog input (waveform acquisition). Combine with PhysicalDevice in a driver. The driver implements the four - streaming primitives; acquireWaveform is provided on top of them. scanRate is - the per-channel sample rate in Hz; readStream returns one block of samples as - {channel: [volts, ...]}. The aggregate rate (scanRate times the number of - channels) is the hardware limit, not scanRate alone. + streaming primitives; acquireWaveform is provided on top of them. sampleRate + is the per-channel sample rate in Hz; readStream returns one block of samples + as {channel: [volts, ...]}. The aggregate rate (sampleRate times the number + of channels) is the hardware limit, not sampleRate alone. One-shot acquisition (blocks until sampleCount samples are collected): - waveform = device.acquireWaveform(channels=[0], scanRate=5000, sampleCount=1000) + waveform = device.acquireWaveform(channels=[0], sampleRate=5000, sampleCount=1000) samples = waveform[0] # 1000 calibrated voltages from AIN0 Continuous acquisition with the primitives: - device.configureStream(channels=[0, 1], scanRate=2000) + device.configureStream(channels=[0, 1], sampleRate=2000) device.startStream() try: while acquiring: @@ -66,23 +66,33 @@ class AnalogInputStreamDevice(AnalogInputDevice): """ @abstractmethod - def configureStream(self, channels, scanRate): + def configureStream(self, channels, sampleRate): + """Set up a hardware-timed acquisition of channels at sampleRate (Hz).""" ... @abstractmethod def startStream(self): + """Start the configured acquisition.""" ... @abstractmethod def readStream(self): + """Return the samples acquired since the last read, as {channel: [volts, ...]}.""" ... @abstractmethod def stopStream(self): + """Stop the acquisition and release any hardware streaming resources.""" ... - def acquireWaveform(self, channels, scanRate, sampleCount): - self.configureStream(channels, scanRate) + def acquireWaveform(self, channels, sampleRate, sampleCount): + """Acquire exactly sampleCount samples per channel, blocking until done. + + Configures, starts, and drains the stream (looping readStream) on the + caller's behalf, then stops it; returns {channel: [volts, ...]} truncated + to sampleCount per channel. + """ + self.configureStream(channels, sampleRate) samples = {channel: [] for channel in channels} self.startStream() try: @@ -95,6 +105,162 @@ def acquireWaveform(self, channels, scanRate, sampleCount): return {channel: values[:sampleCount] for channel, values in samples.items()} +class InputSource(Enum): + """Signal input a lock-in demodulator measures, named instrument-agnostically. + + A driver maps each member to its hardware setting (for the SR830: SingleEnded + -> input A, Differential -> A-B, Current1M -> current input at 1 MOhm, + Current100M -> current input at 100 MOhm). + """ + + SingleEnded = "SingleEnded" + Differential = "Differential" + Current1M = "Current1M" + Current100M = "Current100M" + + +class PhaseLockedDetectionDevice(ABC): + """Phase-locked (lock-in) detection capability. Combine with PhysicalDevice. + + Reads the demodulated outputs (X, Y, R, theta) and reference frequency, and + configures the signal input source, sensitivity (full-scale, in volts), and + time constant (in seconds). Sensitivity and time constant are expressed in + physical units so no instrument's discrete step encoding leaks into the + contract; a driver snaps a requested value to its nearest supported step. + """ + + @abstractmethod + def getInPhaseVoltage(self): + """Returns the in-phase component X, in volts.""" + ... + + @abstractmethod + def getQuadratureVoltage(self): + """Returns the quadrature component Y, in volts.""" + ... + + @abstractmethod + def getMagnitude(self): + """Returns the magnitude R = sqrt(X^2 + Y^2), in volts.""" + ... + + @abstractmethod + def getPhase(self): + """Returns the phase theta, in degrees.""" + ... + + @abstractmethod + def getReferenceFrequency(self): + """Returns the reference frequency, in Hz.""" + ... + + @abstractmethod + def getInputSource(self) -> InputSource: + """Returns the signal input the demodulator currently measures.""" + ... + + @abstractmethod + def setInputSource(self, source: InputSource): + """Select which signal input (an InputSource member) the demodulator measures.""" + ... + + @abstractmethod + def getSensitivity(self): + """Returns the full-scale sensitivity, in volts.""" + ... + + @abstractmethod + def setSensitivity(self, volts): + """Set the full-scale sensitivity to the nearest supported step, in volts.""" + ... + + @abstractmethod + def getTimeConstant(self): + """Returns the time constant, in seconds.""" + ... + + @abstractmethod + def setTimeConstant(self, seconds): + """Set the time constant to the nearest supported step, in seconds.""" + ... + + def supportedInputSources(self): + """Optional: the InputSource members this instrument supports, or None.""" + return None + + def supportedSensitivities(self): + """Optional: the full-scale sensitivities (volts) this instrument supports, or None.""" + return None + + def supportedTimeConstants(self): + """Optional: the time constants (seconds) this instrument supports, or None.""" + return None + + def getDemodulatedValues(self): + """One reading of all demodulated outputs plus the reference frequency. + + Built on the individual getters; a driver may override it to read the + outputs atomically (a single coherent timepoint) when the hardware + supports it. + """ + return { + "X": self.getInPhaseVoltage(), + "Y": self.getQuadratureVoltage(), + "R": self.getMagnitude(), + "theta": self.getPhase(), + "referenceFrequency": self.getReferenceFrequency(), + } + + +class TriggerSource(Enum): + """Where a triggerable acquisition gets its start. + + A trigger starts (arms) an acquisition; it is not the sampling clock (see + SampleClock), which is a separate concept. + """ + + Internal = "Internal" # start immediately + External = "External" # start on an external hardware trigger + + +class SampleClock(Enum): + """What paces the samples of a stream acquisition: the device's own timebase + at a fixed rate (Internal), or one sample per external clock/trigger edge + (External). Distinct from TriggerSource, which only starts the acquisition.""" + + Internal = "Internal" + External = "External" + + +class TriggerableDevice(ABC): + """Capability for a device whose acquisition can be armed to a trigger. + + setTriggerSource selects an internal (immediate) start versus waiting for an + external hardware trigger, and softwareTrigger() issues a manual trigger edge + (equivalent to a pulse on the external trigger line). Combine with + PhysicalDevice in a driver. + """ + + @abstractmethod + def setTriggerSource(self, source: 'TriggerSource'): + """Select whether the acquisition starts immediately or on an external trigger.""" + ... + + @abstractmethod + def getTriggerSource(self) -> 'TriggerSource': + """Returns the currently selected TriggerSource.""" + ... + + @abstractmethod + def softwareTrigger(self): + """Issue a manual (software) trigger edge.""" + ... + + def supportedTriggerSources(self): + """Optional: the TriggerSource members this device supports, or None.""" + return None + + class DigitalInputDevice(ABC): """Digital input capability. Combine with PhysicalDevice in a driver.""" diff --git a/hardwarelibrary/daq/labjackdevice.py b/hardwarelibrary/daq/labjackdevice.py index 9fe482e..bd59549 100644 --- a/hardwarelibrary/daq/labjackdevice.py +++ b/hardwarelibrary/daq/labjackdevice.py @@ -1,7 +1,6 @@ from hardwarelibrary.physicaldevice import PhysicalDevice, DeviceState, PhysicalDeviceNotification from hardwarelibrary.daq import AnalogIODevice, DigitalIODevice, AnalogInputStreamDevice import inspect -import u3 class LabjackDevice(PhysicalDevice, AnalogIODevice, DigitalIODevice, AnalogInputStreamDevice): @@ -28,6 +27,7 @@ def doInitializeDevice(self): PhysicalDevice normalizes a "*" or None serialNumber to the regex ".*", so both spellings mean "first found"; any other value is an exact serial. """ + import u3 self.dev = u3.U3(autoOpen=False) if self.serialNumber in ("*", ".*"): self.dev.open() @@ -64,6 +64,7 @@ def _validateConfigIOParameters(parameters): # configureAnalogIO and configureDigitalIO both forward to the U3's single # configIO command; validate keys against its actual signature so an # unexpected key fails clearly here instead of deep inside configIO. + import u3 valid = set(inspect.signature(u3.U3.configIO).parameters) - {'self'} unexpected = set(parameters) - valid if unexpected: @@ -100,14 +101,18 @@ def getTemperature(self): def toggleLED(self): self.dev.toggleLED() - def configureStream(self, channels, scanRate): + def configureStream(self, channels, sampleRate=None, scanRate=None): + # scanRate is a deprecated synonym for sampleRate, kept temporarily for + # callers written against the old AnalogInputStreamDevice contract. + if sampleRate is None: + sampleRate = scanRate self._streamChannels = list(channels) self.dev.streamConfig( NumChannels=len(self._streamChannels), PChannels=self._streamChannels, NChannels=[31] * len(self._streamChannels), Resolution=3, - ScanFrequency=scanRate, + ScanFrequency=sampleRate, ) def startStream(self): @@ -186,7 +191,7 @@ def getTemperature(self): def toggleLED(self): pass - def configureStream(self, channels, scanRate): + def configureStream(self, channels, sampleRate=None, scanRate=None): self._streamChannels = list(channels) def startStream(self): diff --git a/hardwarelibrary/daq/sr830device.py b/hardwarelibrary/daq/sr830device.py new file mode 100644 index 0000000..93b5f93 --- /dev/null +++ b/hardwarelibrary/daq/sr830device.py @@ -0,0 +1,678 @@ +import math +import re +from enum import Enum + +from serial.tools.list_ports import comports + +from hardwarelibrary.physicaldevice import PhysicalDevice +from hardwarelibrary.communication.prologixgpibport import PrologixGPIBPort +from hardwarelibrary.communication.communicationport import ( + CommunicationPort, CommunicationReadTimeout, +) +from hardwarelibrary.daq.daqdevice import ( + AnalogInputStreamDevice, AnalogOutputDevice, PhaseLockedDetectionDevice, + TriggerableDevice, InputSource, TriggerSource, SampleClock, +) + +FLOAT_PATTERN = r"([-+]?\d*\.?\d+(?:[eE][-+]?\d+)?)" +INTEGER_PATTERN = r"(-?\d+)" + +# SR830 SENS command steps (full-scale sensitivity in volts) and OFLT command +# steps (time constant in seconds). The list index is the command argument. +SR830_SENSITIVITIES = [ + 2e-9, 5e-9, 10e-9, 20e-9, 50e-9, 100e-9, 200e-9, 500e-9, + 1e-6, 2e-6, 5e-6, 10e-6, 20e-6, 50e-6, 100e-6, 200e-6, 500e-6, + 1e-3, 2e-3, 5e-3, 10e-3, 20e-3, 50e-3, 100e-3, 200e-3, 500e-3, + 1.0, +] +SR830_TIME_CONSTANTS = [ + 10e-6, 30e-6, 100e-6, 300e-6, + 1e-3, 3e-3, 10e-3, 30e-3, + 100e-3, 300e-3, 1.0, 3.0, + 10.0, 30.0, 100.0, 300.0, + 1e3, 3e3, 10e3, 30e3, +] + +# SR830 SRAT command steps: data-buffer sample rate in Hz, index = argument. +# Index 14 ("Trigger") is the external per-sample clock, handled separately. +SR830_SAMPLE_RATES = [ + 0.0625, 0.125, 0.25, 0.5, 1, 2, 4, 8, 16, 32, 64, 128, 256, 512, +] + +class AuxInput(Enum): + """The SR830's four auxiliary A/D inputs (rear panel). The value is the + OAUX? channel number.""" + + Aux1 = 1 + Aux2 = 2 + Aux3 = 3 + Aux4 = 4 + + +class AuxOutput(Enum): + """The SR830's four auxiliary D/A outputs (rear panel). The value is the + AUXV channel number.""" + + Aux1 = 1 + Aux2 = 2 + Aux3 = 3 + Aux4 = 4 + + +_INPUT_SOURCE_TO_INDEX = { + InputSource.SingleEnded: 0, + InputSource.Differential: 1, + InputSource.Current1M: 2, + InputSource.Current100M: 3, +} +_INDEX_TO_INPUT_SOURCE = {index: source for source, index in _INPUT_SOURCE_TO_INDEX.items()} + +_TRIGGER_SOURCE_TO_INDEX = {TriggerSource.Internal: 0, TriggerSource.External: 1} +_INDEX_TO_TRIGGER_SOURCE = {index: source for source, index in _TRIGGER_SOURCE_TO_INDEX.items()} + + +class StreamChannel(Enum): + """A quantity the SR830 can record into its data buffer. The SR830 buffers + exactly the two front-panel displays (CH1, CH2), so X and R share CH1 while + Y and Theta share CH2: a stream may hold at most one CH1 and one CH2 + quantity.""" + + X = "X" + Y = "Y" + R = "R" + Theta = "Theta" + + +# Each StreamChannel maps to (display number, DDEF quantity index) per the SR830 +# DDEF command: CH1 j=0 X / j=1 R; CH2 j=0 Y / j=1 theta. +_STREAM_CHANNEL_TO_DISPLAY = { + StreamChannel.X: (1, 0), + StreamChannel.R: (1, 1), + StreamChannel.Y: (2, 0), + StreamChannel.Theta: (2, 1), +} + + +class SR830Device(PhysicalDevice, AnalogInputStreamDevice, AnalogOutputDevice, + PhaseLockedDetectionDevice, TriggerableDevice): + """Stanford Research Systems SR830 DSP lock-in amplifier over a Prologix + GPIB-USB controller. + + The SR830 is a GPIB instrument; the Prologix adaptor bridges it to a host + serial port (a generic FTDI cable, VID 0x0403 / PID 0x6001), so this driver + talks to a PrologixGPIBPort with the ordinary readString/writeString + primitives. It exposes several capabilities: AnalogInputStreamDevice for the + four rear-panel Aux A/D inputs (OAUX?, 1-4) plus hardware-timed buffered + acquisition of the demodulated outputs (the internal data buffer), + AnalogOutputDevice for the four rear-panel Aux D/A outputs (AUXV, 1-4), + PhaseLockedDetectionDevice for the demodulated outputs (X, Y, R, theta via + OUTP?/SNAP?), the signal input source (ISRC), and the sensitivity and + time-constant settings, and TriggerableDevice for the rear-panel TRIG IN. The + Aux inputs and outputs are general-purpose and independent of the lock-in + signal path. + """ + + classIdVendor = 0x0403 + classIdProduct = 0x6001 + usesGenericSerialConverter = True + + sensitivities = SR830_SENSITIVITIES + timeConstants = SR830_TIME_CONSTANTS + sampleRates = SR830_SAMPLE_RATES + minAuxOutputVoltage = -10.5 + maxAuxOutputVoltage = 10.5 + + def __init__(self, gpibAddress=8, portPath=None, serialNumber=None, + idProduct=0x6001, idVendor=0x0403): + """Create an SR830 driver. gpibAddress is the SR830's GPIB address (8 is + the factory default). portPath, if given, names the Prologix adaptor's + serial port directly; otherwise the adaptor is discovered by its FTDI + idVendor/idProduct, narrowed by serialNumber when one is provided.""" + super().__init__(serialNumber, idProduct=idProduct, idVendor=idVendor) + self.gpibAddress = gpibAddress + self.portPath = portPath + self.idn = None + self._streamChannels = [] + self._streamReadIndex = 0 + + def doInitializeDevice(self): + """Find the SR830 among the FTDI adaptors, open it, and confirm identity. + + Several instruments share the generic FTDI 0x0403:0x6001 identity (the + Prologix adaptor, plain RS-232 cables, ...), so VID/PID alone cannot pick + out the SR830. Confirming identity is exactly this method's job: it probes + each candidate adaptor with *IDN? and keeps the one that answers as an + SR830, raising UnableToInitialize if none does. Once found, the adaptor's + serial number is pinned so a later reconnect goes straight to it. + """ + candidates = self._candidateAdaptors() + if not candidates: + raise PhysicalDevice.UnableToInitialize( + "No Prologix GPIB-USB adaptor (FTDI {0:#06x}:{1:#06x}) found for the " + "SR830".format(self.idVendor, self.idProduct)) + + lastError = None + for candidatePath, candidateSerial in candidates: + port = PrologixGPIBPort(gpibAddress=self.gpibAddress, portPath=candidatePath) + try: + port.open() + self.port = port + self.readIdentity() + if self.idn is not None and "SR830" in self.idn: + # Pin the adaptor so a later reconnect selects it directly by + # VID/PID + serial number instead of probing every adaptor. + if candidateSerial is not None: + self.serialNumber = candidateSerial + return + lastError = "{0} answered *IDN?={1!r}".format(candidatePath, self.idn) + except Exception as error: + lastError = "{0}: {1}".format(candidatePath, error) + if port.isOpen: + port.close() + self.port = None + + raise PhysicalDevice.UnableToInitialize( + "No SR830 found at GPIB address {0} on any FTDI adaptor ({1})".format( + self.gpibAddress, lastError)) + + def _candidateAdaptors(self): + """The (portPath, serialNumber) adaptors to probe for the SR830. + + An explicit portPath is the only candidate. Otherwise every connected + FTDI adaptor matching this device's VID/PID is a candidate, narrowed by + serialNumber when one is known (PhysicalDevice turns the default into + ".*", which matches any).""" + if self.portPath is not None: + return [(self.portPath, None)] + + serialNumber = None if self.serialNumber in (None, ".*") else self.serialNumber + candidates = [] + for port in comports(): + if port.vid != self.idVendor or port.pid != self.idProduct: + continue + if serialNumber is not None and (port.serial_number is None + or not re.search(serialNumber, port.serial_number, re.IGNORECASE)): + continue + candidates.append((port.device, port.serial_number)) + return candidates + + def doShutdownDevice(self): + """Close the Prologix port and drop the reference to it.""" + if self.port is not None: + self.port.close() + self.port = None + + def writeCommand(self, command): + """Send a command that expects no reply, appending the GPIB terminator.""" + self.port.writeString(command + "\n") + + def query(self, command) -> str: + """Send command and return its reply as a stripped string. + + The write and the read are held under the port's transactionLock so a + concurrent caller cannot interleave and read this command's reply. + """ + with self.port.transactionLock: + self.port.writeString(command + "\n") + return self.port.readString().strip() + + def queryFloat(self, command) -> float: + """Send command and return the first floating-point number in the reply.""" + _, group = self.port.writeStringReadFirstMatchingGroup( + command + "\n", replyPattern=FLOAT_PATTERN) + return float(group) + + def queryInteger(self, command) -> int: + """Send command and return the first integer in the reply.""" + _, group = self.port.writeStringReadFirstMatchingGroup( + command + "\n", replyPattern=INTEGER_PATTERN) + return int(group) + + def readIdentity(self) -> str: + """Query *IDN?, cache it in self.idn, and return the identity string.""" + self.idn = self.query("*IDN?") + return self.idn + + def getAnalogVoltage(self, channel): + """Read an Aux Input, in volts (SR830 OAUX?). channel is an AuxInput + member; a bare 1-4 is also accepted and coerced, and anything else + raises ValueError.""" + channel = AuxInput(channel) + return self.queryFloat("OAUX? {0}".format(channel.value)) + + def setAnalogVoltage(self, value, channel): + """Set an Aux Output to value volts (SR830 AUXV). channel is an AuxOutput + member (a bare 1-4 is also accepted and coerced). value must be within + [-10.5, 10.5] V or ValueError is raised.""" + channel = AuxOutput(channel) + if not self.minAuxOutputVoltage <= value <= self.maxAuxOutputVoltage: + raise ValueError( + "SR830 Aux Output must be within [{0}, {1}] V, got {2}".format( + self.minAuxOutputVoltage, self.maxAuxOutputVoltage, value)) + self.writeCommand("AUXV {0},{1:.3f}".format(channel.value, value)) + + def getAnalogOutputVoltage(self, channel): + """Read back an Aux Output setpoint, in volts (SR830 AUXV?). channel is + an AuxOutput member (a bare 1-4 is also accepted and coerced).""" + channel = AuxOutput(channel) + return self.queryFloat("AUXV? {0}".format(channel.value)) + + def getInPhaseVoltage(self): + """Returns the in-phase component X, in volts (SR830 OUTP? 1).""" + return self.queryFloat("OUTP? 1") + + def getQuadratureVoltage(self): + """Returns the quadrature component Y, in volts (SR830 OUTP? 2).""" + return self.queryFloat("OUTP? 2") + + def getMagnitude(self): + """Returns the magnitude R = sqrt(X^2 + Y^2), in volts (SR830 OUTP? 3).""" + return self.queryFloat("OUTP? 3") + + def getPhase(self): + """Returns the phase theta, in degrees (SR830 OUTP? 4).""" + return self.queryFloat("OUTP? 4") + + def getReferenceFrequency(self): + """Returns the reference frequency, in Hz (SR830 FREQ?).""" + return self.queryFloat("FREQ?") + + def snap(self, *parameters): + """Read 2 to 6 outputs at one instant (SR830 SNAP?). Parameter codes: + 1=X, 2=Y, 3=R, 4=theta, 5-8=Aux1-4, 9=reference frequency. Returns a + tuple of floats in the requested order.""" + if not 2 <= len(parameters) <= 6: + raise ValueError("SNAP? takes 2 to 6 parameters, got {0}".format(len(parameters))) + command = "SNAP? " + ",".join(str(int(parameter)) for parameter in parameters) + return tuple(float(value) for value in self.query(command).split(",")) + + def getDemodulatedValues(self): + """Read X, Y, R, and theta at one instant, plus the reference frequency. + + Overrides the base implementation to read the four outputs atomically via + SNAP? (a single coherent timepoint) rather than four separate queries. + """ + x, y, r, theta = self.snap(1, 2, 3, 4) + return {"X": x, "Y": y, "R": r, "theta": theta, + "referenceFrequency": self.getReferenceFrequency()} + + # AnalogInputStreamDevice: hardware-timed acquisition into the SR830 data + # buffer. channels are StreamChannel members (at most one CH1 and one CH2 + # quantity). acquireWaveform (from the base) loops readStream for you. + + def configureStream(self, channels, sampleRate, sampleClock=SampleClock.Internal): + """Set up buffered acquisition. channels is a list of StreamChannel + members (1 or 2, not both on the same display). With an Internal + sampleClock, sampleRate (Hz) is snapped to the nearest SRAT step. With an + External sampleClock, one sample is taken per external clock/trigger edge + (SRAT 14) and sampleRate is ignored. The start (immediate vs external + trigger) is a separate concern, set via setTriggerSource.""" + channels = [StreamChannel(channel) for channel in channels] + if not 1 <= len(channels) <= 2: + raise ValueError("SR830 buffers 1 or 2 channels (the CH1/CH2 displays)") + usedDisplays = set() + for channel in channels: + display, quantityIndex = _STREAM_CHANNEL_TO_DISPLAY[channel] + if display in usedDisplays: + raise ValueError( + "channels sharing display {0} cannot be buffered together " + "(e.g. X and R, or Y and Theta)".format(display)) + usedDisplays.add(display) + self.writeCommand("DDEF {0},{1},0".format(display, quantityIndex)) + if sampleClock == SampleClock.External: + self.writeCommand("SRAT 14") + else: + self.writeCommand("SRAT {0}".format(self._sampleRateIndexFor(sampleRate))) + self.writeCommand("SEND 0") + self._streamChannels = channels + self._streamReadIndex = 0 + + def startStream(self): + """Clear the data buffer (REST) and start acquisition (STRT). + + With an External trigger source the SR830 arms here but does not record + until a trigger edge arrives (softwareTrigger or the TRIG IN line). + """ + self.writeCommand("REST") + self._streamReadIndex = 0 + self.writeCommand("STRT") + + def readStream(self): + """Return the samples buffered since the last read, as {channel: [volts, ...]}. + + Reads how many points the buffer holds (SPTS?) and transfers only the new + ones (TRCA?) for each configured channel, advancing the read cursor. An + empty list per channel means no new samples since the previous call. + """ + available = self.queryInteger("SPTS?") + newCount = available - self._streamReadIndex + if newCount <= 0: + return {channel: [] for channel in self._streamChannels} + block = {} + for channel in self._streamChannels: + display = _STREAM_CHANNEL_TO_DISPLAY[channel][0] + reply = self.query("TRCA? {0},{1},{2}".format( + display, self._streamReadIndex, newCount)) + block[channel] = [float(value) for value in reply.split(",") if value.strip()] + self._streamReadIndex = available + return block + + def stopStream(self): + """Pause acquisition into the data buffer (SR830 PAUS).""" + self.writeCommand("PAUS") + + def supportedSampleRates(self): + """Returns the SR830's discrete data-buffer sample rates, in Hz.""" + return list(self.sampleRates) + + def _sampleRateIndexFor(self, rate): + """Returns the SRAT index whose rate (Hz) is closest to rate.""" + return min(range(len(self.sampleRates)), + key=lambda index: abs(self.sampleRates[index] - rate)) + + # TriggerableDevice: the rear-panel TRIG IN. setTriggerSource(External) arms + # the scan to start on a trigger edge (TSTR); trigger() issues a software edge. + + def setTriggerSource(self, source: TriggerSource): + """Select immediate (Internal) or external-trigger (External) scan start + (SR830 TSTR). source must be a supported TriggerSource or ValueError is + raised.""" + if source not in _TRIGGER_SOURCE_TO_INDEX: + raise ValueError("Unsupported trigger source {0}".format(source)) + self.writeCommand("TSTR {0}".format(_TRIGGER_SOURCE_TO_INDEX[source])) + + def getTriggerSource(self) -> TriggerSource: + """Returns the current scan-start TriggerSource (SR830 TSTR?).""" + return _INDEX_TO_TRIGGER_SOURCE[self.queryInteger("TSTR?")] + + def softwareTrigger(self): + """Issue a software trigger edge (SR830 TRIG), as if TRIG IN pulsed.""" + self.writeCommand("TRIG") + + def supportedTriggerSources(self): + """Returns the trigger sources the SR830 supports (Internal, External).""" + return list(TriggerSource) + + def getInputSource(self) -> InputSource: + """Returns the demodulator's signal input source (SR830 ISRC?).""" + return _INDEX_TO_INPUT_SOURCE[self.queryInteger("ISRC?")] + + def setInputSource(self, source: InputSource): + """Select the demodulator's signal input source (SR830 ISRC). source must + be a supported InputSource or ValueError is raised.""" + if source not in _INPUT_SOURCE_TO_INDEX: + raise ValueError("Unsupported input source {0}".format(source)) + self.writeCommand("ISRC {0}".format(_INPUT_SOURCE_TO_INDEX[source])) + + def supportedInputSources(self): + """Returns the four input sources the SR830 supports.""" + return list(InputSource) + + def getSensitivity(self): + """Returns the current full-scale sensitivity, in volts (SR830 SENS?).""" + return self.sensitivities[self.queryInteger("SENS?")] + + def setSensitivity(self, volts): + """Set the full-scale sensitivity to the smallest step >= volts (SR830 SENS).""" + self.writeCommand("SENS {0}".format(self._sensitivityIndexFor(volts))) + + def getTimeConstant(self): + """Returns the current time constant, in seconds (SR830 OFLT?).""" + return self.timeConstants[self.queryInteger("OFLT?")] + + def setTimeConstant(self, seconds): + """Set the time constant to the nearest step, in seconds (SR830 OFLT).""" + self.writeCommand("OFLT {0}".format(self._timeConstantIndexFor(seconds))) + + def supportedSensitivities(self): + """Returns the SR830's discrete full-scale sensitivities, in volts.""" + return list(self.sensitivities) + + def supportedTimeConstants(self): + """Returns the SR830's discrete time constants, in seconds.""" + return list(self.timeConstants) + + def _sensitivityIndexFor(self, volts): + """Returns the SENS index for the smallest full-scale that contains volts. + + Rounding up rather than to nearest keeps a requested value that falls + between two steps from clipping; a value above the largest step clamps to + it. + """ + for index, value in enumerate(self.sensitivities): + if value >= volts: + return index + return len(self.sensitivities) - 1 + + def _timeConstantIndexFor(self, seconds): + """Returns the OFLT index whose time constant is closest to seconds.""" + return min(range(len(self.timeConstants)), + key=lambda index: abs(self.timeConstants[index] - seconds)) + + def doGetStatusUserInfo(self): + """Returns the demodulated values for the periodic status notification.""" + return self.getDemodulatedValues() + + +class DebugPrologixGPIBPort(CommunicationPort): + """Hardware-free stand-in for a Prologix controller wired to an SR830. + + Follows the CommunicationPort structure: it implements only the transport + primitives (open/close/isOpen/flush/bytesAvailable and readData/writeData + over an in-memory buffer, like USBPort), so the inherited readString and + writeStringReadFirstMatchingGroup drive it unchanged. writeData interprets a + line: '++' controller commands are consumed silently, an SR830 query enqueues + a reply, and an SR830 set mutates the simulated state. + """ + + def __init__(self): + """Create the fake port with a plausible initial SR830 state.""" + super().__init__() + self._isOpen = False + self._buffer = bytearray() + self._auxVoltages = {1: 0.10, 2: 0.20, 3: 0.30, 4: 0.40} + self._auxOutputs = {1: 0.0, 2: 0.0, 3: 0.0, 4: 0.0} + self._x = 1.0e-3 + self._y = 2.0e-3 + self._referenceFrequency = 1.0e3 + self._sensitivityIndex = 26 + self._timeConstantIndex = 10 + self._inputIndex = 0 + self._sampleRateIndex = 4 + self._bufferMode = 0 + self._triggerStartIndex = 0 + self._scanning = False + self._storedPoints = 0 + self._buffers = {1: [], 2: []} + + @property + def isOpen(self): + """True while the fake port is open.""" + return self._isOpen + + def open(self): + """Mark the port open and clear the reply buffer.""" + self._isOpen = True + self._buffer = bytearray() + + def close(self): + """Mark the port closed.""" + self._isOpen = False + + def bytesAvailable(self) -> int: + """Returns the number of bytes waiting to be read.""" + return len(self._buffer) + + def flush(self): + """Discard any buffered reply bytes.""" + self._buffer = bytearray() + + def writeData(self, data, endPoint=None) -> int: + """Interpret one written line and queue any reply it produces. + + Returns the number of bytes accepted. Controller ('++') lines are + consumed silently; an SR830 query enqueues its reply, and an SR830 set + mutates the simulated state. + """ + line = bytes(data).decode("utf-8").strip() + reply = self._process(line) + if reply is not None: + self._buffer += bytearray((reply + "\n").encode("utf-8")) + return len(data) + + def readData(self, length, endPoint=None) -> bytearray: + """Return length bytes from the reply buffer, or time out if too few.""" + if len(self._buffer) < length: + raise CommunicationReadTimeout("Only obtained {0}".format(self._buffer)) + data = self._buffer[:length] + self._buffer = self._buffer[length:] + return data + + def _process(self, line): + """Compute the reply for one command line (or None for no reply / a set).""" + if line.startswith("++"): + return None + if line == "*IDN?": + return "Stanford_Research_Systems,SR830,s/n12345,ver1.07" + if line.startswith("OAUX?"): + channel = int(line[len("OAUX?"):]) + return self._formatFloat(self._auxVoltages.get(channel, 0.0)) + if line.startswith("AUXV?"): + channel = int(line[len("AUXV?"):]) + return self._formatFloat(self._auxOutputs.get(channel, 0.0)) + if line.startswith("AUXV "): + channelText, voltageText = line[len("AUXV "):].split(",") + self._auxOutputs[int(channelText)] = float(voltageText) + return None + if line.startswith("OUTP?"): + return self._formatFloat(self._outputValue(int(line[len("OUTP?"):]))) + if line.startswith("SNAP?"): + codes = [int(code) for code in line[len("SNAP?"):].split(",")] + return ",".join(self._formatFloat(self._snapValue(code)) for code in codes) + if line == "FREQ?": + return self._formatFloat(self._referenceFrequency) + if line == "SENS?": + return str(self._sensitivityIndex) + if line.startswith("SENS "): + self._sensitivityIndex = int(line[len("SENS "):]) + return None + if line == "OFLT?": + return str(self._timeConstantIndex) + if line.startswith("OFLT "): + self._timeConstantIndex = int(line[len("OFLT "):]) + return None + if line == "ISRC?": + return str(self._inputIndex) + if line.startswith("ISRC "): + self._inputIndex = int(line[len("ISRC "):]) + return None + if line.startswith("DDEF "): + return None + if line == "SRAT?": + return str(self._sampleRateIndex) + if line.startswith("SRAT "): + self._sampleRateIndex = int(line[len("SRAT "):]) + return None + if line.startswith("SEND "): + self._bufferMode = int(line[len("SEND "):]) + return None + if line == "TSTR?": + return str(self._triggerStartIndex) + if line.startswith("TSTR "): + self._triggerStartIndex = int(line[len("TSTR "):]) + return None + if line == "TRIG": + self._storeStreamPoints(1) + return None + if line == "REST": + self._storedPoints = 0 + self._buffers = {1: [], 2: []} + self._scanning = False + return None + if line == "STRT": + # Internal start (TSTR 0) runs immediately; external start (TSTR 1) + # waits for a trigger edge, so no points accrue until trigger(). + self._scanning = self._triggerStartIndex == 0 + return None + if line == "PAUS": + self._scanning = False + return None + if line == "SPTS?": + # An internal timebase (SRAT 0-13) accrues samples on its own; an + # external clock (SRAT 14) only advances on a trigger edge. + if self._scanning and self._sampleRateIndex != 14: + self._storeStreamPoints(25) + return str(self._storedPoints) + if line.startswith("TRCA?"): + display, start, count = (int(field) for field in line[len("TRCA?"):].split(",")) + values = self._buffers.get(display, [])[start:start + count] + return ",".join(self._formatFloat(value) for value in values) + return None + + def _storeStreamPoints(self, count): + """Append count synthetic samples to each display buffer (a ramp so + successive reads return distinguishable values).""" + for display in (1, 2): + base = {1: 1.0e-3, 2: 2.0e-3}[display] + start = len(self._buffers[display]) + self._buffers[display].extend(base + 1e-6 * (start + i) for i in range(count)) + self._storedPoints += count + + def _outputValue(self, index): + """Returns the simulated demodulated output for an OUTP?/SNAP? code + (1=X, 2=Y, 3=R, 4=theta).""" + if index == 1: + return self._x + if index == 2: + return self._y + if index == 3: + return math.hypot(self._x, self._y) + if index == 4: + return math.degrees(math.atan2(self._y, self._x)) + return 0.0 + + def _snapValue(self, code): + """Returns the simulated value for a SNAP? code (1-4=X/Y/R/theta, + 5-8=Aux1-4, 9=reference frequency).""" + if code in (1, 2, 3, 4): + return self._outputValue(code) + if code in (5, 6, 7, 8): + return self._auxVoltages.get(code - 4, 0.0) + if code == 9: + return self._referenceFrequency + return 0.0 + + @staticmethod + def _formatFloat(value): + """Format a float the way the SR830 renders numeric replies.""" + return "{0:.6e}".format(value) + + +class DebugSR830Device(SR830Device): + """Hardware-free SR830 for tests. Swaps the transport (DebugPrologixGPIBPort) + so every parse/validate path in SR830Device runs unchanged.""" + + classIdVendor = 0xFFFF + classIdProduct = 0xFFF9 + usesGenericSerialConverter = False + + def __init__(self, serialNumber="debug"): + """Create a debug SR830 carrying the reserved debug USB identity.""" + super().__init__(gpibAddress=8, serialNumber=serialNumber, + idProduct=self.classIdProduct, idVendor=self.classIdVendor) + + def doInitializeDevice(self): + """Attach the in-memory DebugPrologixGPIBPort and confirm identity. + + Overrides discovery only; every other method runs the real SR830Device + code path against the fake port. + """ + self.port = DebugPrologixGPIBPort() + self.port.open() + self.readIdentity() + + def doShutdownDevice(self): + """Close the in-memory port and drop the reference to it.""" + if self.port is not None: + self.port.close() + self.port = None diff --git a/hardwarelibrary/tests/testLabjackU3.py b/hardwarelibrary/tests/testLabjackU3.py index 17b5c01..628aa4e 100644 --- a/hardwarelibrary/tests/testLabjackU3.py +++ b/hardwarelibrary/tests/testLabjackU3.py @@ -200,14 +200,14 @@ def testConfigureDigitalIO(self): self.device.configureDigitalIO({}) def testAcquireWaveformSampleCount(self): - waveform = self.device.acquireWaveform(channels=[0], scanRate=1000, sampleCount=200) + waveform = self.device.acquireWaveform(channels=[0], sampleRate=1000, sampleCount=200) self.assertEqual(len(waveform[0]), 200) def testAcquireWaveformReadsLoopback(self): self.skipUnlessAnalogLoopback(0) self.device.setAnalogVoltage(value=2.0, channel=0) time.sleep(self.analogSettlingTime) - waveform = self.device.acquireWaveform(channels=[0], scanRate=2000, sampleCount=500) + waveform = self.device.acquireWaveform(channels=[0], sampleRate=2000, sampleCount=500) self.assertEqual(len(waveform[0]), 500) mean = sum(waveform[0]) / len(waveform[0]) self.assertAlmostEqual(mean, 2.0, 1) @@ -307,14 +307,14 @@ def testConfigureDigitalIORejectsUnknownKey(self): def testAcquireWaveform(self): self.device.setAnalogVoltage(value=2.5, channel=0) - waveform = self.device.acquireWaveform(channels=[0], scanRate=1000, sampleCount=50) + waveform = self.device.acquireWaveform(channels=[0], sampleRate=1000, sampleCount=50) self.assertEqual(len(waveform[0]), 50) self.assertTrue(all(abs(value - 2.5) < 1e-9 for value in waveform[0])) def testAcquireWaveformMultiChannel(self): self.device.setAnalogVoltage(value=1.0, channel=0) self.device.setAnalogVoltage(value=2.0, channel=1) - waveform = self.device.acquireWaveform(channels=[0, 1], scanRate=1000, sampleCount=30) + waveform = self.device.acquireWaveform(channels=[0, 1], sampleRate=1000, sampleCount=30) self.assertEqual(len(waveform[0]), 30) self.assertEqual(len(waveform[1]), 30) self.assertAlmostEqual(waveform[0][0], 1.0) diff --git a/hardwarelibrary/tests/testSR830.py b/hardwarelibrary/tests/testSR830.py new file mode 100644 index 0000000..de1e501 --- /dev/null +++ b/hardwarelibrary/tests/testSR830.py @@ -0,0 +1,534 @@ +import env +import unittest +from unittest import mock + +from hardwarelibrary.physicaldevice import PhysicalDevice +from hardwarelibrary.communication import PrologixGPIBPort +from hardwarelibrary.communication.communicationport import CommunicationReadTimeout +from hardwarelibrary.daq import ( + AnalogInputDevice, AnalogOutputDevice, AnalogInputStreamDevice, + PhaseLockedDetectionDevice, TriggerableDevice, + InputSource, AuxInput, AuxOutput, StreamChannel, TriggerSource, SampleClock, + SR830Device, DebugSR830Device, DebugPrologixGPIBPort, +) + +SR830_IDN = "Stanford_Research_Systems,SR830,s/n86552,ver1.07" + + +class TestDebugSR830Device(unittest.TestCase): + def setUp(self): + self.device = DebugSR830Device() + self.device.initializeDevice() + + def tearDown(self): + self.device.shutdownDevice() + + def testCreate(self): + self.assertIsNotNone(self.device) + + def testInitializeAndShutdown(self): + device = DebugSR830Device() + device.initializeDevice() + device.shutdownDevice() + + def testIdentityIsSR830(self): + self.assertIn("SR830", self.device.idn) + + def testGetAnalogVoltageChannels(self): + for channel in AuxInput: + value = self.device.getAnalogVoltage(channel) + self.assertIsInstance(value, float) + self.assertAlmostEqual(self.device.getAnalogVoltage(AuxInput.Aux1), 0.10) + self.assertAlmostEqual(self.device.getAnalogVoltage(AuxInput.Aux4), 0.40) + + def testGetAnalogVoltageAcceptsBareInt(self): + self.assertAlmostEqual(self.device.getAnalogVoltage(1), 0.10) + + def testGetAnalogVoltageInvalidChannel(self): + with self.assertRaises(ValueError): + self.device.getAnalogVoltage(0) + with self.assertRaises(ValueError): + self.device.getAnalogVoltage(5) + + def testSetAndGetAnalogOutput(self): + self.device.setAnalogVoltage(1.5, AuxOutput.Aux1) + self.assertAlmostEqual(self.device.getAnalogOutputVoltage(AuxOutput.Aux1), 1.5) + self.device.setAnalogVoltage(-2.25, AuxOutput.Aux3) + self.assertAlmostEqual(self.device.getAnalogOutputVoltage(AuxOutput.Aux3), -2.25) + + def testSetAnalogVoltageAcceptsBareInt(self): + self.device.setAnalogVoltage(0.5, 2) + self.assertAlmostEqual(self.device.getAnalogOutputVoltage(2), 0.5) + + def testSetAnalogVoltageInvalidChannel(self): + with self.assertRaises(ValueError): + self.device.setAnalogVoltage(1.0, 5) + + def testSetAnalogVoltageOutOfRange(self): + with self.assertRaises(ValueError): + self.device.setAnalogVoltage(20.0, AuxOutput.Aux1) + with self.assertRaises(ValueError): + self.device.setAnalogVoltage(-11.0, AuxOutput.Aux1) + + def testDemodulatedReads(self): + self.assertAlmostEqual(self.device.getInPhaseVoltage(), 1.0e-3) + self.assertAlmostEqual(self.device.getQuadratureVoltage(), 2.0e-3) + self.assertAlmostEqual(self.device.getMagnitude(), + (1.0e-3 ** 2 + 2.0e-3 ** 2) ** 0.5) + self.assertIsInstance(self.device.getPhase(), float) + self.assertAlmostEqual(self.device.getReferenceFrequency(), 1.0e3) + + def testGetDemodulatedValues(self): + values = self.device.getDemodulatedValues() + self.assertEqual(set(values.keys()), + {"X", "Y", "R", "theta", "referenceFrequency"}) + self.assertAlmostEqual(values["X"], 1.0e-3) + self.assertAlmostEqual(values["referenceFrequency"], 1.0e3) + + def testSnapLength(self): + self.assertEqual(len(self.device.snap(1, 2, 3, 4)), 4) + self.assertEqual(len(self.device.snap(1, 2)), 2) + + def testSnapRejectsWrongParameterCount(self): + with self.assertRaises(ValueError): + self.device.snap(1) + with self.assertRaises(ValueError): + self.device.snap(1, 2, 3, 4, 5, 6, 7) + + def testSensitivityRoundTrip(self): + self.device.setSensitivity(1.0) + self.assertAlmostEqual(self.device.getSensitivity(), 1.0) + + def testSensitivitySnapsUpToAvoidClipping(self): + # 3 mV lands between the 2 mV and 5 mV steps; the smallest step that + # still contains it is 5 mV. + self.device.setSensitivity(3e-3) + self.assertAlmostEqual(self.device.getSensitivity(), 5e-3) + + def testTimeConstantRoundTrip(self): + self.device.setTimeConstant(1.0) + self.assertAlmostEqual(self.device.getTimeConstant(), 1.0) + + def testTimeConstantSnapsToNearest(self): + # 2 s is nearest the 3 s step (|3-2| < |1-2| is False -> 1 s and 3 s + # are equidistant; min picks the earlier 1 s). Use 2.5 s -> 3 s. + self.device.setTimeConstant(2.5) + self.assertAlmostEqual(self.device.getTimeConstant(), 3.0) + + def testSupportedListsAreMonotonic(self): + sensitivities = self.device.supportedSensitivities() + timeConstants = self.device.supportedTimeConstants() + self.assertEqual(len(sensitivities), 27) + self.assertEqual(len(timeConstants), 20) + self.assertEqual(sensitivities, sorted(sensitivities)) + self.assertEqual(timeConstants, sorted(timeConstants)) + + def testInputSourceRoundTrip(self): + self.device.setInputSource(InputSource.Differential) + self.assertEqual(self.device.getInputSource(), InputSource.Differential) + self.device.setInputSource(InputSource.Current100M) + self.assertEqual(self.device.getInputSource(), InputSource.Current100M) + + def testSupportedInputSources(self): + self.assertEqual(set(self.device.supportedInputSources()), set(InputSource)) + + def testStatusUserInfo(self): + info = self.device.doGetStatusUserInfo() + self.assertIn("R", info) + + def testTriggerSourceRoundTrip(self): + self.device.setTriggerSource(TriggerSource.External) + self.assertEqual(self.device.getTriggerSource(), TriggerSource.External) + self.device.setTriggerSource(TriggerSource.Internal) + self.assertEqual(self.device.getTriggerSource(), TriggerSource.Internal) + + def testSupportedTriggerSources(self): + self.assertEqual(set(self.device.supportedTriggerSources()), set(TriggerSource)) + + def testAcquireWaveform(self): + self.device.setTriggerSource(TriggerSource.Internal) + waveform = self.device.acquireWaveform( + channels=[StreamChannel.X, StreamChannel.Y], sampleRate=512, sampleCount=60) + self.assertEqual(len(waveform[StreamChannel.X]), 60) + self.assertEqual(len(waveform[StreamChannel.Y]), 60) + + def testStreamPrimitives(self): + self.device.configureStream(channels=[StreamChannel.R], sampleRate=64) + self.device.startStream() + try: + block = self.device.readStream() + finally: + self.device.stopStream() + self.assertIn(StreamChannel.R, block) + + def testExternalSampleClockAdvancesOnSoftwareTrigger(self): + # With an External sample clock, no samples accrue until each trigger edge. + self.device.setTriggerSource(TriggerSource.Internal) + self.device.configureStream( + channels=[StreamChannel.X], sampleRate=0, sampleClock=SampleClock.External) + self.device.startStream() + try: + self.assertEqual(self.device.readStream()[StreamChannel.X], []) + self.device.softwareTrigger() + self.device.softwareTrigger() + self.assertEqual(len(self.device.readStream()[StreamChannel.X]), 2) + finally: + self.device.stopStream() + + def testConfigureStreamRejectsSameDisplay(self): + with self.assertRaises(ValueError): + self.device.configureStream(channels=[StreamChannel.X, StreamChannel.R], sampleRate=64) + + def testConfigureStreamRejectsWrongChannelCount(self): + with self.assertRaises(ValueError): + self.device.configureStream(channels=[], sampleRate=64) + + def testSampleRateSnapsToNearest(self): + # 512 Hz is an exact SRAT step; 60 Hz snaps to the nearest (64). + self.assertEqual(self.device._sampleRateIndexFor(512), self.device.sampleRates.index(512)) + self.assertEqual(self.device._sampleRateIndexFor(60), self.device.sampleRates.index(64)) + + def testSupportedSampleRates(self): + rates = self.device.supportedSampleRates() + self.assertEqual(rates, sorted(rates)) + self.assertEqual(rates[-1], 512) + + def testSensitivityClampsToLargestStep(self): + # A request above the 1 V maximum full-scale clamps to that largest step + # rather than raising. + self.device.setSensitivity(100.0) + self.assertAlmostEqual(self.device.getSensitivity(), 1.0) + + def testSetTriggerSourceRejectsUnknown(self): + with self.assertRaises(ValueError): + self.device.setTriggerSource(SampleClock.Internal) + + def testSetInputSourceRejectsUnknown(self): + with self.assertRaises(ValueError): + self.device.setInputSource(SampleClock.Internal) + + def testSnapReadsAuxAndFrequencyCodes(self): + # Codes 5-8 are Aux1-4, 9 is the reference frequency, and an unknown code + # reads back as 0.0. + values = self.device.snap(5, 6, 7, 8, 9, 99) + self.assertEqual(len(values), 6) + self.assertAlmostEqual(values[0], 0.10) + self.assertAlmostEqual(values[3], 0.40) + self.assertAlmostEqual(values[4], 1.0e3) + self.assertAlmostEqual(values[5], 0.0) + + +class TestPhaseLockedDetectionContract(unittest.TestCase): + def testDeclaresAllCapabilities(self): + device = DebugSR830Device() + self.assertIsInstance(device, AnalogInputDevice) + self.assertIsInstance(device, AnalogOutputDevice) + self.assertIsInstance(device, AnalogInputStreamDevice) + self.assertIsInstance(device, PhaseLockedDetectionDevice) + self.assertIsInstance(device, TriggerableDevice) + + +class _MinimalLockIn(PhaseLockedDetectionDevice, TriggerableDevice): + """A bare capability implementation that supplies only the abstract hooks, so + the base-class optional hooks and the base getDemodulatedValues are exercised + (SR830Device overrides all of these).""" + + def getInPhaseVoltage(self): + return 0.1 + + def getQuadratureVoltage(self): + return 0.2 + + def getMagnitude(self): + return 0.3 + + def getPhase(self): + return 45.0 + + def getReferenceFrequency(self): + return 1000.0 + + def getInputSource(self): + return InputSource.SingleEnded + + def setInputSource(self, source): + pass + + def getSensitivity(self): + return 1.0 + + def setSensitivity(self, volts): + pass + + def getTimeConstant(self): + return 0.1 + + def setTimeConstant(self, seconds): + pass + + def setTriggerSource(self, source): + pass + + def getTriggerSource(self): + return TriggerSource.Internal + + def softwareTrigger(self): + pass + + +class TestCapabilityDefaults(unittest.TestCase): + def testOptionalHooksDefaultToNone(self): + device = _MinimalLockIn() + self.assertIsNone(device.supportedInputSources()) + self.assertIsNone(device.supportedSensitivities()) + self.assertIsNone(device.supportedTimeConstants()) + self.assertIsNone(device.supportedTriggerSources()) + + def testBaseGetDemodulatedValues(self): + values = _MinimalLockIn().getDemodulatedValues() + self.assertEqual(set(values), + {"X", "Y", "R", "theta", "referenceFrequency"}) + self.assertAlmostEqual(values["R"], 0.3) + self.assertAlmostEqual(values["referenceFrequency"], 1000.0) + + +class TestPrologixGPIBPort(unittest.TestCase): + def testControllerHandshakeUsesManualRead(self): + # Exercise the real PrologixGPIBPort.configureController (it only calls + # writeString, so no serial line is needed) and assert the manual-read + # handshake: ++auto 0 with no controller-appended reply terminator. + port = PrologixGPIBPort(gpibAddress=8) + sent = [] + port.writeString = lambda text: sent.append(text.strip()) + port.configureController() + self.assertEqual(sent[0], "++mode 1") + self.assertIn("++addr 8", sent) + self.assertIn("++auto 0", sent) + self.assertIn("++eot_enable 0", sent) + self.assertNotIn("++auto 1", sent) + + def testQueryRoundTripsThroughStringPrimitives(self): + port = DebugPrologixGPIBPort() + port.open() + port.writeString("*IDN?\n") + self.assertIn("SR830", port.readString()) + + def testFloatQueryMatches(self): + port = DebugPrologixGPIBPort() + port.open() + _, group = port.writeStringReadFirstMatchingGroup( + "FREQ?\n", replyPattern=r"([-+]?\d*\.?\d+(?:[eE][-+]?\d+)?)") + self.assertAlmostEqual(float(group), 1.0e3) + + +class TestSR830Device(unittest.TestCase): + def setUp(self): + # doInitializeDevice self-discovers: it probes the connected FTDI + # adaptors and confirms the SR830 by *IDN?, so no portPath is needed. + self.device = SR830Device() + try: + self.device.initializeDevice() + except Exception: + self.skipTest("No SR830 / Prologix adaptor connected") + + def tearDown(self): + self.device.shutdownDevice() + + def testIdentity(self): + self.assertIn("SR830", self.device.idn) + + def testPinsAdaptorSerialNumber(self): + # After a successful probe the winning adaptor's serial is recorded so a + # reconnect can select it directly by VID/PID + serial number. + self.assertNotIn(self.device.serialNumber, (None, ".*")) + + def testGetAnalogVoltage(self): + self.assertIsInstance(self.device.getAnalogVoltage(AuxInput.Aux1), float) + + def testGetReferenceFrequency(self): + self.assertIsInstance(self.device.getReferenceFrequency(), float) + + def testGetMagnitude(self): + self.assertIsInstance(self.device.getMagnitude(), float) + + def testInputSourceRoundTrip(self): + original = self.device.getInputSource() + self.device.setInputSource(InputSource.Differential) + self.assertEqual(self.device.getInputSource(), InputSource.Differential) + self.device.setInputSource(original) + self.assertEqual(self.device.getInputSource(), original) + + def testAnalogOutputRoundTrip(self): + original = self.device.getAnalogOutputVoltage(AuxOutput.Aux1) + try: + self.device.setAnalogVoltage(1.234, AuxOutput.Aux1) + self.assertAlmostEqual( + self.device.getAnalogOutputVoltage(AuxOutput.Aux1), 1.234, places=3) + finally: + self.device.setAnalogVoltage(original, AuxOutput.Aux1) + + def testTriggerSourceRoundTrip(self): + original = self.device.getTriggerSource() + try: + self.device.setTriggerSource(TriggerSource.External) + self.assertEqual(self.device.getTriggerSource(), TriggerSource.External) + finally: + self.device.setTriggerSource(original) + + def testAcquireWaveform(self): + self.device.setTriggerSource(TriggerSource.Internal) + waveform = self.device.acquireWaveform( + channels=[StreamChannel.X, StreamChannel.Y], sampleRate=512, sampleCount=64) + self.assertEqual(len(waveform[StreamChannel.X]), 64) + self.assertEqual(len(waveform[StreamChannel.Y]), 64) + self.assertTrue(all(isinstance(value, float) for value in waveform[StreamChannel.X])) + + +class TestDebugPrologixGPIBPort(unittest.TestCase): + def setUp(self): + self.port = DebugPrologixGPIBPort() + self.port.open() + + def testIsOpenReflectsState(self): + self.assertTrue(self.port.isOpen) + self.port.close() + self.assertFalse(self.port.isOpen) + + def testBytesAvailableAndFlush(self): + self.port.writeString("FREQ?\n") + self.assertGreater(self.port.bytesAvailable(), 0) + self.port.flush() + self.assertEqual(self.port.bytesAvailable(), 0) + + def testSampleRateRegisterRoundTrip(self): + self.port.writeString("SRAT 7\n") + self.port.writeString("SRAT?\n") + self.assertEqual(self.port.readString().strip(), "7") + + def testReadTimesOutWhenBufferShort(self): + with self.assertRaises(CommunicationReadTimeout): + self.port.readData(1) + + def testControllerLineProducesNoReply(self): + self.port.writeString("++mode 1\n") + self.assertEqual(self.port.bytesAvailable(), 0) + + def testUnknownCommandProducesNoReply(self): + self.port.writeString("BOGUS?\n") + self.assertEqual(self.port.bytesAvailable(), 0) + + def testOutputValueUnknownIndexReadsZero(self): + self.port.writeString("OUTP? 9\n") + self.assertAlmostEqual(float(self.port.readString()), 0.0) + + +class _FakeComPort: + """Stand-in for a serial.tools.list_ports entry.""" + + def __init__(self, device, serialNumber, vid=0x0403, pid=0x6001): + self.device = device + self.serial_number = serialNumber + self.vid = vid + self.pid = pid + + +class _FakeAdaptorPort(DebugPrologixGPIBPort): + """A Prologix port whose *IDN? reply depends on which serial port it opened, + so a discovery probe can be simulated without hardware. identitiesByPath maps + a portPath to its *IDN? reply; a missing/None entry means the adaptor never + answers (the read times out).""" + + identitiesByPath = {} + + def __init__(self, gpibAddress, portPath=None, **kwargs): + super().__init__() + self.gpibAddress = gpibAddress + self.portPath = portPath + + def _process(self, line): + if line == "*IDN?": + return type(self).identitiesByPath.get(self.portPath) + return super()._process(line) + + +class TestSR830Discovery(unittest.TestCase): + """Hardware-free coverage of SR830Device.doInitializeDevice / _candidateAdaptors, + which DebugSR830Device bypasses by overriding doInitializeDevice.""" + + def setUp(self): + _FakeAdaptorPort.identitiesByPath = {} + + def _patched(self, comportList): + return ( + mock.patch("hardwarelibrary.daq.sr830device.PrologixGPIBPort", _FakeAdaptorPort), + mock.patch("hardwarelibrary.daq.sr830device.comports", return_value=comportList), + ) + + def testDiscoversSR830AndPinsSerial(self): + _FakeAdaptorPort.identitiesByPath = { + "/dev/A": "Keithley,MODEL 2000,1234,A01", + "/dev/B": SR830_IDN, + } + comportList = [_FakeComPort("/dev/A", "AAAA"), _FakeComPort("/dev/B", "BBBB")] + portPatch, comportPatch = self._patched(comportList) + with portPatch, comportPatch: + device = SR830Device() + device.initializeDevice() + try: + self.assertIn("SR830", device.idn) + self.assertEqual(device.serialNumber, "BBBB") + finally: + device.shutdownDevice() + + def testRaisesWhenNoAdaptorPresent(self): + comportList = [_FakeComPort("/dev/Z", "ZZZZ", vid=0x1234, pid=0x5678)] + portPatch, comportPatch = self._patched(comportList) + with portPatch, comportPatch: + device = SR830Device() + with self.assertRaises(PhysicalDevice.UnableToInitialize): + device.initializeDevice() + + def testRaisesWhenNoSR830AmongAdaptors(self): + _FakeAdaptorPort.identitiesByPath = { + "/dev/A": None, + "/dev/B": "Keithley,MODEL 2000,1234,A01", + } + comportList = [_FakeComPort("/dev/A", "AAAA"), _FakeComPort("/dev/B", "BBBB")] + portPatch, comportPatch = self._patched(comportList) + with portPatch, comportPatch: + device = SR830Device() + with self.assertRaises(PhysicalDevice.UnableToInitialize): + device.initializeDevice() + + def testExplicitPortPathIsSoleCandidate(self): + _FakeAdaptorPort.identitiesByPath = {"/dev/X": SR830_IDN} + with mock.patch("hardwarelibrary.daq.sr830device.PrologixGPIBPort", _FakeAdaptorPort), \ + mock.patch("hardwarelibrary.daq.sr830device.comports", + side_effect=AssertionError("comports() must not be consulted with an explicit portPath")): + device = SR830Device(portPath="/dev/X") + device.initializeDevice() + try: + self.assertIn("SR830", device.idn) + finally: + device.shutdownDevice() + + def testSerialNumberNarrowsCandidates(self): + _FakeAdaptorPort.identitiesByPath = {"/dev/match": SR830_IDN} + comportList = [ + _FakeComPort("/dev/other", "OTHER"), + _FakeComPort("/dev/none", None), + _FakeComPort("/dev/match", "WANTED"), + ] + portPatch, comportPatch = self._patched(comportList) + with portPatch, comportPatch: + device = SR830Device(serialNumber="WANT") + device.initializeDevice() + try: + self.assertIn("SR830", device.idn) + self.assertEqual(device.serialNumber, "WANTED") + finally: + device.shutdownDevice() + + +if __name__ == '__main__': + unittest.main()