| 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
 | """
Implements partial support for Siglent SDS1000X-E series oscilloscopes.
"""
import socket
import re
from lab_control.oscilloscope import Oscilloscope
def _checkChannel(channel):
    assert channel in SDS1000XE.AVAILABLE_CHANNELS, "SDS1000X-E: Invalid channel {channel}"
class SDS1000XE(Oscilloscope):
    """
    Instances of this class connect to the SDS1000X-E IP
    and port and offer an API to control the device.
    """
    PORT = 5025
    TIMEOUT = 1.0
    AVAILABLE_CHANNELS = range(1, 5)
    def __init__(self, address):
        super().__init__()
        self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self._socket.connect((address, SDS1000XE.PORT))
        self._socket.settimeout(SDS1000XE.TIMEOUT)
    def _measure(self, channel: int, code: str) -> float:
        _checkChannel(channel)
        pattern = r"C(?P<responseChannel>\d):PAVA .+,(?P<rawMeasurement>[\d.E+-]+)\w+"
        query = f"C{channel}:PAVA? {code}\r\n"
        self._socket.sendall(query.encode())
        try:
            # TODO add code to regex
            response = self._socket.recv(4096).decode()
            matches = re.search(pattern, response)
            measurement = float(matches.group("rawMeasurement"))
        except TimeoutError:
            measurement = None
        return measurement
    def measureAmplitude(self, channel: int) -> float:
        return self._measure(channel, "AMPL")
    def measurePeakToPeak(self, channel: int) -> float:
        return self._measure(channel, "PKPK")
    def measureRMS(self, channel: int) -> float:
        return self._measure(channel, "RMS")
    def measureFrequency(self, channel: int) -> float:
        return self._measure(channel, "FREQ")
 |