jansbox: Data acquisition from Jan's box

This device is used for acquiring seismometer and tiltmeter data and
for augmenting commercial AVI systems with low frequency feedback.

jansbox.py

+import click
+import enum
+import os
+import re
+import serial
+import signal
+import struct
+import time
+
+from datetime import datetime, UTC
+from ptblab import datalogger, terminal, win32time
+
+
+class Mode(enum.IntEnum):
+    AVERAGE = 0
+    SAMPLE = 1
+
+
+class Device:
+
+    # Commands are terminated by CR. Most commands do not send a reply, some
+    # commands reply with a LF terminated string, some commands reply with an
+    # unterminated strings, the ``rec.info`` command replies with multiple LF
+    # terminated lines, the data transfer initiated by ``rec.start`` results in
+    # fixed lenght records without termination to be emitted.
+    #
+    # For all commands that are not handled specifically, attempt to read a LF
+    # terminated reply with a timeout of 0.1 seconds. For commands that provide
+    # a reply, this ensure that the commands has completed successfully, fot all
+    # other commands, this provides a 0.1 seconds implicit delay.
+
+    def __init__(self, device):
+        self.serial = serial.Serial(device, baudrate=115200, timeout=0.1)
+        self.debug = False
+
+    def write(self, cmd):
+        if self.debug:
+            print('<', cmd)
+        self.serial.write(cmd + b'\r')
+
+    def read(self):
+        r = self.serial.read_until(b'\n')
+        if self.debug:
+            print('>', r)
+        return r
+
+    def query(self, cmd):
+        self.write(cmd)
+        return self.read()
+
+    def _get_available_channels_count(self):
+        self.write(b'rec.info')
+        r = self.serial.read(2048).rstrip(b'\n')
+
+        # use the explicit channel count field, if present
+        if m := re.search(rb'^Channel count: ([0-9]+)$', r, re.MULTILINE):
+            return int(m.group(1))
+        # otherwise return the index of the last entry
+        if m := re.match(rb'^\[([0-9]+)\] ', r.split(b'\n')[-1]):
+            return int(m.group(1))
+
+        raise ValueError
+
+    def _configure_channel(self, idx, name, mode):
+        self.write(b'rec.assign[%i] %s %i' % (idx, name.encode('ascii') if name else b' ', mode))
+        r = self.read()
+
+        if name is not None and b'assigned' in r:
+            return True
+        if name is None and b'cleared' in r:
+            return True
+
+        raise ValueError(r)
+
+    def _get_record_format(self, nchannels):
+        self.write(b'rec.typehdr')
+
+        r = self.read().decode('ascii').rstrip('\n')
+        if len(r) != nchannels:
+            raise ValueError(r)
+
+        # record format is the ASCII string ``rec``, an unsigned char record number, and the data values
+        return '<3sB' + r
+
+    def configure(self, channels):
+        nchannels = len(channels)
+        names = [name for name, mode in channels]
+
+        navailable = self._get_available_channels_count()
+        if nchannels > navailable:
+            raise ValueError(f'Too many channels: {nchannels} requested but {navailable} available')
+
+        # pad channels definition to disable data recording on unused channels
+        channels = channels + [(None, Mode.AVERAGE), ] * (navailable - nchannels)
+
+        for i, (name, mode) in enumerate(channels):
+            self._configure_channel(i, name, mode)
+
+        frmt = self._get_record_format(nchannels)
+
+        return names, frmt
+
+
+class Datalogger(datalogger.Datalogger):
+
+    # Use filenames in the format used by the old data acquisition script,
+    # record data in a format compatible with the old data acquisition script,
+    # and add header with column names to new files.
+
+    def __init__(self, datadir, mode='a', header=None):
+        super().__init__(datadir, 'data_{:%Y-%m-%d}.dat', mode)
+        self.header = header
+
+    def write(self, *data):
+        timestamp, *values = data
+        key = self.key(*data)
+        if self.fd is None or key != self.current:
+            filename = os.path.join(self.datadir, self.template.format(key))
+            new = not os.path.exists(filename)
+            self.fd = open(filename, self.mode, encoding='utf8')
+            if new and self.header is not None:
+                print(*self.header, file=self.fd, sep='\t')
+            self.current = key
+        print(*values, file=self.fd, sep='\t')
+
+
+SIGNALS = [
+    # ("ttl.input", Mode.SAMPLE),
+    ("adc1.data[0]", Mode.AVERAGE),
+    ("adc1.data[1]", Mode.AVERAGE),
+    ("adc1.data[2]", Mode.AVERAGE),
+    ("adc1.data[3]", Mode.AVERAGE),
+    ("adc1.data[4]", Mode.AVERAGE),
+    ("adc1.data[5]", Mode.AVERAGE),
+    # ("adc1.data[6]", Mode.AVERAGE),
+    # ("adc1.data[7]", Mode.AVERAGE),
+    # ("adc2.data[0]", Mode.AVERAGE),
+    # ("adc2.data[1]", Mode.AVERAGE),
+    # ("adc2.data[2]", Mode.AVERAGE),
+    # ("adc2.data[3]", Mode.AVERAGE),
+    # ("adc2.data[4]", Mode.AVERAGE),
+    # ("adc2.data[5]", Mode.AVERAGE),
+    ("tracker.freq[0]", Mode.AVERAGE),
+    # ("tracker.freq[1]", Mode.AVERAGE),
+    # ("tracker.freq[1]", Mode.AVERAGE),
+]
+
+
+@click.command()
+@click.option('--device', '-d', metavar='DEV')
+@click.option('--fsampl', type=int, default=20, metavar='HZ', help='Sampling frequency.')
+@click.option('--datadir', type=click.Path(), help='Data folder for datalogger mode.')
+def main(device, fsampl, datadir):
+    """Data acquisition for Yan's box."""
+
+    win32time.set_resolution(win32time.get_resolution().max)
+    terminal.setup()
+
+    shift0 = 8
+    shift1 = 4
+    div = 10000 / fsampl                        # sample rate divider. fsampl = fs / div, fs = 10 kHz
+    ftw0 = int(7.6 * 2**48/100)                 # tracker center frequency in MHz
+    ftw1 = int(10.86 * 2**48/100)               # tracker center frequency in MHz
+    offs0 = int(7.60 * 2**48/100)               # tracker offset frequency in MHz
+    offs1 = int(10.847 * 2**48/100)             # tracker offset frequency in MHz
+    dec = 100000000 / fsampl                    # tracker output data rate. fsampl = fFPGA / dec, fFPGA= 100 MHz
+
+    dev = Device(device)
+
+    dev.query(b'tracker.cmd stop')
+    dev.query(b'rec.stop')                      # stop recording
+    dev.query(b'rec.pause=0')                   # unpause if paused
+    dev.query(b'rec.div=%i' % div)              # set rate divider
+    dev.query(b'tracker.cmd ftw1=%i' % ftw0)    # set tracker frequency
+    dev.query(b'tracker.cmd ftw2=%i' % ftw1)    # set tracker frequency
+    dev.query(b'tracker.cmd offs1=%i' % offs0)  # set tracker offset
+    dev.query(b'tracker.cmd offs2=%i' % offs1)  # set tracker offset
+    dev.query(b'tracker.cmd dacen=1')           # enable tracker outputs
+    dev.query(b'tracker.cmd mode=0')
+    dev.query(b'tracker.cmd dec=%i' % dec)
+    dev.query(b'tracker.cmd sendmask=3')
+    dev.query(b'tracker.cmd shift1=%i' % shift0)
+    dev.query(b'tracker.cmd shift2=%i' % shift1)
+    dev.query(b'tracker.cmd start')
+    dev.query(b'tracker.cmd reset')
+
+    # configure data acquisition
+    names, frmt = dev.configure(SIGNALS)
+
+    # compute record size
+    size = struct.calcsize(frmt)
+
+    # data file header
+    header = ('counter', 'offset', 'offset1', 'shift', 'shift1', *names, 'timestamp')
+
+    logger = Datalogger(datadir, header=header) if datadir else datalogger.File(None)
+
+    dev.query(b'rec.mode=3')                    # continuous mode
+    dev.query(b'rec.start')                     # start recording
+
+    n = 0
+    acquiring = True
+
+    def terminate(*args):
+        nonlocal acquiring
+        acquiring = False
+
+    signal.signal(signal.SIGINT, terminate)
+
+    while acquiring:
+        # data rates lower than 10 Hz require increasing the serial device read timeout
+        data = dev.serial.read(size)
+        timestamp = time.time()
+
+        header, count, *values = struct.unpack(frmt, data)
+        if header != b'rec':
+            raise ValueError(data)
+
+        t = datetime.fromtimestamp(timestamp, UTC).replace(tzinfo=None).isoformat(' ', 'milliseconds')
+        logger.write(timestamp, count, offs0, offs1, shift0, shift1, *values, t)
+
+        # output console log and flush data to disk every second
+        n = (n + 1) % fsampl
+        if n == 0:
+            print(f'{timestamp:.3f}  {t:s}', *values, end='\033[0K\r')
+            logger.fd.flush()
+
+    dev.query(b'rec.stop')
+
+
+if __name__ == '__main__':
+    main()