Refactored attenuator controlling code to a separate module

3c9a71c6 · Projects · e3aed1f5 · 3c9a71c6 · 3c9a71c6
Commit 3c9a71c6 authored Mar 05, 2024 by Projects
Expand all Hide whitespace changes
Inline Side-by-side

Showing with 194 additions and 150 deletions

oattnusbtest software/lib/oattnusbtest +53 -150

pyoattnusb.py software/lib/pyoattnusb.py +141 -0

No files found.
--- a/software/lib/oattnusbtest
+++ b/software/lib/oattnusbtest
--- a/software/lib/pyoattnusb.py
+++ b/software/lib/pyoattnusb.py
+# SPDX-FileCopyrightText: 2024 CERN (home.cern)
+#
+# SPDX-License-Identifier: LGPL-2.1-or-later
+
+from ctypes import *
+import os.path
+
+class Dev(Structure):
+    _fields_ = [
+    ( 'lun',	c_int ),
+    ( 'busnum',	c_char * 6 ),
+    ( 'devnum',	c_char * 6 ),
+    ( 'portnum',	c_char * 6 ),
+    ( 'busport',	c_char * 16 ),
+    ( 'devno',	c_char * 10 ),
+    ( 'serial',	c_char * 48 ),
+    ( 'path',	c_char * 256 ),
+    ( 'unused',	c_int * 4 ),
+    ]
+
+class DevArray(Structure):
+    _fields_ = [
+        ( 'ndevs',  c_int ),
+        ( 'devs',   POINTER(Dev) ),
+    ]
+
+class Manager:
+    VENDOR_ID = 0x1556 # CERN USB Vendor ID
+    DEVICE_ID = 0x0443 # oau USB Device ID
+
+    FTDI_VENDOR_ID = 0x403  # FTDI Vendor ID
+    FTDI_DEVICE_ID = 0x6010 # FTDI Product ID
+
+    def __init__(self):
+        self._load_solib()
+        self._devices = self._get_devarray(self.VENDOR_ID, self.DEVICE_ID)
+
+    def _load_solib(self):
+        """try to load attn dyn library from script dir or ../lib."""
+
+        self.solib = './liboattnusb.so'
+        scriptdir = os.path.dirname(os.path.realpath(__file__))
+        rootdir = os.path.dirname(scriptdir)
+        libdir = os.path.join(rootdir, 'lib', self.solib)
+        try:
+            self.api = CDLL(self.solib)
+        except OSError as e:
+            pass
+        try:
+            self.api = CDLL(libdir)
+        except OSError as e:
+            raise RuntimeError('cannot load oattn solib, exiting')
+
+        self.api.oau_factory_program_eeprom.argtypes = [c_int, c_int, c_char_p]
+
+    def _get_devarray(self, vid, pid):
+        self.api.oattnusb_get_devarray.restype = POINTER(DevArray)
+        self.devarray = self.api.oattnusb_get_devarray(vid, pid)
+        ndevs = self.devarray.contents.ndevs
+        devs = self.devarray.contents.devs
+        devices = []
+        for i in range(ndevs):
+            dev = devs[i]
+            devices.append({
+                'lun' :     dev.lun,
+                'busnum' :	str(dev.busnum, 'utf8'),
+                'devnum' :	str(dev.devnum, 'utf8'),
+                'portnum' :	str(dev.portnum, 'utf8'),
+                'busport' :	str(dev.busport, 'utf8'),
+                'devno'	:	str(dev.devno, 'utf8'),
+                'serial' :	str(dev.serial, 'utf8'),
+                'path'	 :	str(dev.path , 'utf8')})
+        return devices
+
+    def devices(self):
+        return self._devices
+
+    def get_device(self, serial):
+        return Device(self.api, serial)
+
+    def version(self):
+        version = c_char_p.in_dll(self.api, "liboattnusb_version_s")
+        return str(version.value, 'utf8')
+
+    def eeprom_program(self, bus, dev, serial):
+        ret = self.api.oau_factory_program_eeprom(bus, dev, bytes(str(serial), 'ascii'))
+
+        if (ret != 0):
+            raise RuntimeError('EEPROM programming error detected')
+
+class Device:
+    attn_values = {
+        '-40dB' : 0,
+        '-20dB' : 1,
+          '0dB' : 3,
+    }
+    attn_labels = {
+        0 : '-40dB',
+        1 : '-20dB',
+        3 :   '0dB',
+    }
+
+    def __init__(self, api, serial):
+        self.api = api
+
+        self.fd = self.api.oattnusb_open_by_serial(bytes(serial, 'utf8'))
+        if self.fd < 0:
+            raise RuntimeError('cannot open device {}'.format(serial))
+
+        self.nchannels = self.api.oattnusb_get_nchannels(self.fd)
+        if self.nchannels <= 0:
+            raise RuntimeError('error {} getting number of channels'.format(err))
+
+        self.serial = serial
+
+    def __del__(self):
+        if self.fd is not None and self.fd > 0:
+            err = self.api.oattnusb_close(self.fd)
+            if err < 0:
+                raise RuntimeError('could not close fd {}'.format(self.fd))
+
+    def get_attn(self, ch):
+        if ch < 1 or ch > self.nchannels:
+            raise RuntimeError('channels must be in range 1-{0}'.format(self.nchannels))
+
+        attn = self.api.oattnusb_get_relay(self.fd, ch)
+        if attn < 0:
+            raise RuntimeError('error getting attenuation for channel {}'.format(ch))
+
+        return attn
+
+    def set_attn(self, ch, attn):
+        if ch < 1 or ch > self.nchannels:
+            raise RuntimeError('channels must be in range 1-{0}'.format(self.nchannels))
+
+        if attn not in self.attn_values.values():
+            raise RuntimeError('invalid attenuation value: {0}'.format(attn))
+
+        err = self.api.oattnusb_set_relay(self.fd, ch, attn);
+        if err < 0:
+            raise RuntimeError('error setting attenuation {} for channel {}'.format(attn_labels[attn], ch))