From 278b668f259104db476dbd8d3f68cdba503e9d44 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Wed, 8 Jul 2026 21:37:45 -0400 Subject: [PATCH 1/5] Add DAQ lock-in/trigger capabilities; rename stream rate; decouple u3 Problem: The DAQ family had no capability contract for lock-in (phase-locked) detection or for triggered acquisition, so a lock-in driver would have to bolt ad-hoc methods onto one class. The stream sample-rate parameter was named scanRate, and importing hardwarelibrary.daq eagerly imported u3 (LabJackPython), so the whole package failed to import on hosts without the LabJack driver. Solution: - Add PhaseLockedDetectionDevice (X/Y/R/theta, reference frequency, input source, sensitivity, time constant, all in physical units) and TriggerableDevice (setTriggerSource/getTriggerSource/softwareTrigger) with the TriggerSource and SampleClock enums, following the interface-segregated mixin convention. - Rename the stream sample-rate parameter scanRate -> sampleRate on AnalogInputStreamDevice.configureStream/acquireWaveform. LabjackDevice keeps scanRate as a temporary deprecated synonym. - Defer the u3 import in labjackdevice to point of use so the daq package imports without LabJackPython installed. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01T2DmAfLHSiK1fb3PiXDyJP --- hardwarelibrary/daq/daqdevice.py | 170 +++++++++++++++++++++++-- hardwarelibrary/daq/labjackdevice.py | 13 +- hardwarelibrary/tests/testLabjackU3.py | 8 +- 3 files changed, 174 insertions(+), 17 deletions(-) diff --git a/hardwarelibrary/daq/daqdevice.py b/hardwarelibrary/daq/daqdevice.py index 5c385ee..bda0029 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,7 +66,7 @@ class AnalogInputStreamDevice(AnalogInputDevice): """ @abstractmethod - def configureStream(self, channels, scanRate): + def configureStream(self, channels, sampleRate): ... @abstractmethod @@ -81,8 +81,8 @@ def readStream(self): def stopStream(self): ... - def acquireWaveform(self, channels, scanRate, sampleCount): - self.configureStream(channels, scanRate) + def acquireWaveform(self, channels, sampleRate, sampleCount): + self.configureStream(channels, sampleRate) samples = {channel: [] for channel in channels} self.startStream() try: @@ -95,6 +95,158 @@ 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: + ... + + @abstractmethod + def setInputSource(self, source: InputSource): + ... + + @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'): + ... + + @abstractmethod + def getTriggerSource(self) -> '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/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) From 1fa3e133c92227fd00ce222c0f312b7093e834f3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Wed, 8 Jul 2026 21:37:57 -0400 Subject: [PATCH 2/5] Add PrologixGPIBPort GPIB-USB transport Problem: The library had no way to reach a GPIB (IEEE-488) instrument. The common lab bridge is a Prologix GPIB-USB controller, which enumerates as a generic FTDI serial port and speaks a "++" controller protocol on top of the serial link. Solution: Add PrologixGPIBPort, a SerialPort subclass that reuses the FTDI discovery and read/write primitives and adds only the controller handshake in open() (++mode 1 / ++addr / ++auto 0 / ++eoi 1 / ++eos 2 / ++eot_enable 0 / ++read_tmo_ms). Manual-read mode (++auto 0) is used because auto mode addresses the instrument to talk after every command, including no-reply commands, wedging the bus; the required ++read eoi is encapsulated in readString so instruments talk to the port with the ordinary readString/writeString primitives. Export it from the communication package. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01T2DmAfLHSiK1fb3PiXDyJP --- hardwarelibrary/communication/__init__.py | 1 + .../communication/prologixgpibport.py | 58 +++++++++++++++++++ 2 files changed, 59 insertions(+) create mode 100644 hardwarelibrary/communication/prologixgpibport.py 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..be49f79 --- /dev/null +++ b/hardwarelibrary/communication/prologixgpibport.py @@ -0,0 +1,58 @@ +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): + super().__init__(idVendor=idVendor, idProduct=idProduct, + serialNumber=serialNumber, portPath=portPath) + self.gpibAddress = gpibAddress + + def open(self, baudRate=115200, timeout=1.0): + 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: + # Manual-read mode: ask the controller to read the addressed instrument's + # reply (until EOI), then read the forwarded bytes with the inherited + # serial readString. Keeping this here lets the driver use the ordinary + # readString/writeStringReadMatch primitives unchanged. + self.writeString("++read eoi\n") + return super().readString(endPoint) From d0bd991fdf0128131f16d24c716d026b5632df44 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Wed, 8 Jul 2026 21:38:14 -0400 Subject: [PATCH 3/5] Add Stanford Research SR830 lock-in amplifier driver Problem: The lab's SR830 DSP lock-in amplifier, reached over a Prologix GPIB-USB adaptor, had no driver in PyHardwareLibrary. Solution: Add SR830Device (and DebugSR830Device) combining PhysicalDevice, AnalogInputStreamDevice, AnalogOutputDevice, PhaseLockedDetectionDevice, and TriggerableDevice: - Aux A/D inputs (OAUX?) via getAnalogVoltage(AuxInput); Aux D/A outputs (AUXV) via setAnalogVoltage(value, AuxOutput). - Demodulated reads X/Y/R/theta (OUTP?/SNAP?), reference frequency (FREQ?), input source (ISRC), sensitivity (SENS) and time constant (OFLT) in physical units with nearest-step snapping. - Buffered acquisition of the internal data buffer (DDEF/SRAT/SEND/REST/STRT/ PAUS/SPTS?/TRCA?) via the AnalogInputStreamDevice primitives, with a StreamChannel enum and a SampleClock (internal rate vs external per-edge). - Trigger via TSTR (setTriggerSource) and TRIG (softwareTrigger). - doInitializeDevice self-discovers the Prologix among the connected FTDI adaptors by confirming *IDN? contains SR830, and pins the adaptor serial for reconnects. - DebugSR830Device swaps in a DebugPrologixGPIBPort (an in-memory CommunicationPort that follows the readData/writeData structure) so all parse/validate paths run without hardware. Verified against the lab SR830 (s/n 86552): tests/testSR830.py has 43 tests, including 8 that talk to the real instrument (reads, aux out round-trip, buffered acquisition, trigger) and skip cleanly when no SR830 is attached. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01T2DmAfLHSiK1fb3PiXDyJP --- CHANGELOG.md | 28 ++ hardwarelibrary/daq/__init__.py | 8 +- hardwarelibrary/daq/sr830device.py | 587 +++++++++++++++++++++++++++++ hardwarelibrary/tests/testSR830.py | 295 +++++++++++++++ 4 files changed, 917 insertions(+), 1 deletion(-) create mode 100644 hardwarelibrary/daq/sr830device.py create mode 100644 hardwarelibrary/tests/testSR830.py 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/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/sr830device.py b/hardwarelibrary/daq/sr830device.py new file mode 100644 index 0000000..fb07044 --- /dev/null +++ b/hardwarelibrary/daq/sr830device.py @@ -0,0 +1,587 @@ +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): + super().__init__(serialNumber, idProduct=idProduct, idVendor=idVendor) + self.gpibAddress = gpibAddress + self.portPath = portPath + self.idn = None + self._streamChannels = [] + self._streamReadIndex = 0 + + def doInitializeDevice(self): + # 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: + # probe each candidate adaptor with *IDN? and keep the one that answers + # as an SR830; if none does, raise. serialNumber (once known) narrows the + # candidates so a reconnect goes straight to the right adaptor. + 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): + if self.port is not None: + self.port.close() + self.port = None + + def writeCommand(self, command): + self.port.writeString(command + "\n") + + def query(self, command) -> str: + with self.port.transactionLock: + self.port.writeString(command + "\n") + return self.port.readString().strip() + + def queryFloat(self, command) -> float: + _, group = self.port.writeStringReadFirstMatchingGroup( + command + "\n", replyPattern=FLOAT_PATTERN) + return float(group) + + def queryInteger(self, command) -> int: + _, group = self.port.writeStringReadFirstMatchingGroup( + command + "\n", replyPattern=INTEGER_PATTERN) + return int(group) + + def readIdentity(self) -> str: + 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): + return self.queryFloat("OUTP? 1") + + def getQuadratureVoltage(self): + return self.queryFloat("OUTP? 2") + + def getMagnitude(self): + return self.queryFloat("OUTP? 3") + + def getPhase(self): + return self.queryFloat("OUTP? 4") + + def getReferenceFrequency(self): + 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): + 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): + self.writeCommand("REST") + self._streamReadIndex = 0 + self.writeCommand("STRT") + + def readStream(self): + 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): + self.writeCommand("PAUS") + + def supportedSampleRates(self): + return list(self.sampleRates) + + def _sampleRateIndexFor(self, 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): + 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: + return _INDEX_TO_TRIGGER_SOURCE[self.queryInteger("TSTR?")] + + def softwareTrigger(self): + self.writeCommand("TRIG") + + def supportedTriggerSources(self): + return list(TriggerSource) + + def getInputSource(self) -> InputSource: + return _INDEX_TO_INPUT_SOURCE[self.queryInteger("ISRC?")] + + def setInputSource(self, source: InputSource): + 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): + return list(InputSource) + + def getSensitivity(self): + return self.sensitivities[self.queryInteger("SENS?")] + + def setSensitivity(self, volts): + self.writeCommand("SENS {0}".format(self._sensitivityIndexFor(volts))) + + def getTimeConstant(self): + return self.timeConstants[self.queryInteger("OFLT?")] + + def setTimeConstant(self, seconds): + self.writeCommand("OFLT {0}".format(self._timeConstantIndexFor(seconds))) + + def supportedSensitivities(self): + return list(self.sensitivities) + + def supportedTimeConstants(self): + return list(self.timeConstants) + + def _sensitivityIndexFor(self, volts): + # Smallest full-scale that still contains the signal, so a requested + # value between two steps does not clip; clamp to the largest step. + for index, value in enumerate(self.sensitivities): + if value >= volts: + return index + return len(self.sensitivities) - 1 + + def _timeConstantIndexFor(self, seconds): + return min(range(len(self.timeConstants)), + key=lambda index: abs(self.timeConstants[index] - seconds)) + + def doGetStatusUserInfo(self): + 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): + 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): + return self._isOpen + + def open(self): + self._isOpen = True + self._buffer = bytearray() + + def close(self): + self._isOpen = False + + def bytesAvailable(self) -> int: + return len(self._buffer) + + def flush(self): + self._buffer = bytearray() + + def writeData(self, data, endPoint=None) -> int: + 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: + 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): + 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): + 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): + 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): + 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): + 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"): + super().__init__(gpibAddress=8, serialNumber=serialNumber, + idProduct=self.classIdProduct, idVendor=self.classIdVendor) + + def doInitializeDevice(self): + self.port = DebugPrologixGPIBPort() + self.port.open() + self.readIdentity() + + def doShutdownDevice(self): + if self.port is not None: + self.port.close() + self.port = None diff --git a/hardwarelibrary/tests/testSR830.py b/hardwarelibrary/tests/testSR830.py new file mode 100644 index 0000000..8433f6c --- /dev/null +++ b/hardwarelibrary/tests/testSR830.py @@ -0,0 +1,295 @@ +import env +import unittest + +from hardwarelibrary.physicaldevice import PhysicalDevice +from hardwarelibrary.daq import ( + AnalogInputDevice, AnalogOutputDevice, AnalogInputStreamDevice, + PhaseLockedDetectionDevice, TriggerableDevice, + InputSource, AuxInput, AuxOutput, StreamChannel, TriggerSource, SampleClock, + SR830Device, DebugSR830Device, DebugPrologixGPIBPort, +) + + +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)) + + +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 TestPrologixGPIBPort(unittest.TestCase): + def testControllerHandshakeOnOpen(self): + port = DebugPrologixGPIBPort() + sent = [] + original = port.writeData + + def recordingWriteData(data, endPoint=None): + sent.append(bytes(data).decode("utf-8").strip()) + return original(data, endPoint) + + port.writeData = recordingWriteData + port.open() + # Simulate the same handshake PrologixGPIBPort.open() sends. + for command in ("++mode 1", "++addr 8", "++auto 1", "++eoi 1", "++eos 2", + "++eot_enable 1", "++eot_char 10", "++read_tmo_ms 1500"): + port.writeString(command + "\n") + self.assertIn("++mode 1", sent) + self.assertIn("++addr 8", sent) + self.assertIn("++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])) + + +if __name__ == '__main__': + unittest.main() From c3da636c796165070c5403afe1bd5cf622dd607c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Wed, 8 Jul 2026 21:50:45 -0400 Subject: [PATCH 4/5] Document SR830 driver, Prologix port, and DAQ capabilities Problem: several methods added in this branch lacked docstrings, and a few method-level explanatory comments read as inline "what this does" comments rather than the docstrings the project style calls for. Solution: add concise docstrings to every method and capability contract added in this branch (PrologixGPIBPort, SR830Device/DebugSR830Device, DebugPrologixGPIBPort, and the new PhaseLockedDetectionDevice/TriggerableDevice abstract hooks and touched AnalogInputStreamDevice methods). Method-level comments explaining behavior are converted to docstrings; genuine "why" comments inside method bodies are kept inline. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01T2DmAfLHSiK1fb3PiXDyJP --- .../communication/prologixgpibport.py | 25 +++- hardwarelibrary/daq/daqdevice.py | 14 +++ hardwarelibrary/daq/sr830device.py | 107 ++++++++++++++++-- 3 files changed, 134 insertions(+), 12 deletions(-) diff --git a/hardwarelibrary/communication/prologixgpibport.py b/hardwarelibrary/communication/prologixgpibport.py index be49f79..a4302bb 100644 --- a/hardwarelibrary/communication/prologixgpibport.py +++ b/hardwarelibrary/communication/prologixgpibport.py @@ -22,11 +22,24 @@ class PrologixGPIBPort(SerialPort): 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() @@ -50,9 +63,13 @@ def configureController(self): self.writeString(command + "\n") def readString(self, endPoint=None) -> str: - # Manual-read mode: ask the controller to read the addressed instrument's - # reply (until EOI), then read the forwarded bytes with the inherited - # serial readString. Keeping this here lets the driver use the ordinary - # readString/writeStringReadMatch primitives unchanged. + """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/daqdevice.py b/hardwarelibrary/daq/daqdevice.py index bda0029..7f1a7d2 100644 --- a/hardwarelibrary/daq/daqdevice.py +++ b/hardwarelibrary/daq/daqdevice.py @@ -67,21 +67,31 @@ class AnalogInputStreamDevice(AnalogInputDevice): @abstractmethod 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, 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() @@ -146,10 +156,12 @@ def getReferenceFrequency(self): @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 @@ -231,10 +243,12 @@ class TriggerableDevice(ABC): @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 diff --git a/hardwarelibrary/daq/sr830device.py b/hardwarelibrary/daq/sr830device.py index fb07044..93b5f93 100644 --- a/hardwarelibrary/daq/sr830device.py +++ b/hardwarelibrary/daq/sr830device.py @@ -124,6 +124,10 @@ class SR830Device(PhysicalDevice, AnalogInputStreamDevice, AnalogOutputDevice, 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 @@ -132,12 +136,15 @@ def __init__(self, gpibAddress=8, portPath=None, serialNumber=None, self._streamReadIndex = 0 def doInitializeDevice(self): - # 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: - # probe each candidate adaptor with *IDN? and keep the one that answers - # as an SR830; if none does, raise. serialNumber (once known) narrows the - # candidates so a reconnect goes straight to the right adaptor. + """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( @@ -190,29 +197,39 @@ def _candidateAdaptors(self): 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 @@ -241,18 +258,23 @@ def getAnalogOutputVoltage(self, 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): @@ -265,6 +287,11 @@ def snap(self, *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()} @@ -301,11 +328,22 @@ def configureStream(self, channels, sampleRate, sampleClock=SampleClock.Internal 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: @@ -320,12 +358,15 @@ def readStream(self): 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)) @@ -333,61 +374,83 @@ def _sampleRateIndexFor(self, rate): # 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): - # Smallest full-scale that still contains the signal, so a requested - # value between two steps does not clip; clamp to the largest step. + """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() @@ -403,6 +466,7 @@ class DebugPrologixGPIBPort(CommunicationPort): """ def __init__(self): + """Create the fake port with a plausible initial SR830 state.""" super().__init__() self._isOpen = False self._buffer = bytearray() @@ -423,22 +487,33 @@ def __init__(self): @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: @@ -446,6 +521,7 @@ def writeData(self, data, endPoint=None) -> int: 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] @@ -453,6 +529,7 @@ def readData(self, length, endPoint=None) -> bytearray: 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?": @@ -533,6 +610,8 @@ def _process(self, line): 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]) @@ -540,6 +619,8 @@ def _storeStreamPoints(self, 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: @@ -551,6 +632,8 @@ def _outputValue(self, index): 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): @@ -561,6 +644,7 @@ def _snapValue(self, code): @staticmethod def _formatFloat(value): + """Format a float the way the SR830 renders numeric replies.""" return "{0:.6e}".format(value) @@ -573,15 +657,22 @@ class DebugSR830Device(SR830Device): 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 From 2eba383148c55f89bcfa92acc89b9cf9d2432c8d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Wed, 8 Jul 2026 22:03:59 -0400 Subject: [PATCH 5/5] Cover SR830 discovery, guards, and debug-port paths with tests Problem: SR830Device.doInitializeDevice and _candidateAdaptors (the FTDI adaptor probe/confirm/pin logic) were only exercised on real hardware, because DebugSR830Device overrides doInitializeDevice. Several defensive branches (unsupported-enum guards, the sensitivity clamp, base-capability defaults, and DebugPrologixGPIBPort conveniences) were also untested, and the Prologix handshake test asserted the pre-fix command set without exercising the real port. Solution: add hardware-free tests. TestSR830Discovery patches comports() and PrologixGPIBPort to cover discovery, serial pinning, serial-number narrowing, explicit portPath, and both failure raises. Add TestDebugPrologixGPIBPort for the port primitives, TestCapabilityDefaults for the base optional hooks and base getDemodulatedValues, and debug-device tests for the enum guards, the sensitivity clamp, supportedSampleRates, and the aux/frequency SNAP codes. Rewrite the handshake test to exercise the real configureController and assert the manual-read (++auto 0, ++eot_enable 0) handshake. This takes sr830device.py and prologixgpibport.py to 100% line coverage (62 tests, all passing). Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01T2DmAfLHSiK1fb3PiXDyJP --- hardwarelibrary/tests/testSR830.py | 271 +++++++++++++++++++++++++++-- 1 file changed, 255 insertions(+), 16 deletions(-) diff --git a/hardwarelibrary/tests/testSR830.py b/hardwarelibrary/tests/testSR830.py index 8433f6c..de1e501 100644 --- a/hardwarelibrary/tests/testSR830.py +++ b/hardwarelibrary/tests/testSR830.py @@ -1,7 +1,10 @@ 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, @@ -9,6 +12,8 @@ SR830Device, DebugSR830Device, DebugPrologixGPIBPort, ) +SR830_IDN = "Stanford_Research_Systems,SR830,s/n86552,ver1.07" + class TestDebugSR830Device(unittest.TestCase): def setUp(self): @@ -183,6 +188,35 @@ def testSampleRateSnapsToNearest(self): 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): @@ -194,25 +228,84 @@ def testDeclaresAllCapabilities(self): self.assertIsInstance(device, TriggerableDevice) -class TestPrologixGPIBPort(unittest.TestCase): - def testControllerHandshakeOnOpen(self): - port = DebugPrologixGPIBPort() - sent = [] - original = port.writeData +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 recordingWriteData(data, endPoint=None): - sent.append(bytes(data).decode("utf-8").strip()) - return original(data, endPoint) + def getInPhaseVoltage(self): + return 0.1 - port.writeData = recordingWriteData - port.open() - # Simulate the same handshake PrologixGPIBPort.open() sends. - for command in ("++mode 1", "++addr 8", "++auto 1", "++eoi 1", "++eos 2", - "++eot_enable 1", "++eot_char 10", "++read_tmo_ms 1500"): - port.writeString(command + "\n") - self.assertIn("++mode 1", sent) + 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 1", sent) + self.assertIn("++auto 0", sent) + self.assertIn("++eot_enable 0", sent) + self.assertNotIn("++auto 1", sent) def testQueryRoundTripsThroughStringPrimitives(self): port = DebugPrologixGPIBPort() @@ -291,5 +384,151 @@ def testAcquireWaveform(self): 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()