7
\$\begingroup\$

A recorder to asynchronously record both scalar data and images and then synchronize them. The images should be cut with dynamics ROI.

Here's my attempt on it. Please feel free to comment upon it and show me how to improve it. I tried to follow PEP8 with the following exceptions:

  • I used tabs instead of spaces
  • For hanging indent I used one tap instead of aligning with (
  • Within the docstrings I mixed tabs and spaces

Other than that it should conform to it.

from collections import deque
from time import sleep, time
from copy import deepcopy
from numbers import Integral
from threading import Lock
from pickle import dump

import numpy as np
from scipy.signal import medfilt
from PyQt5.QtCore import QThread, pyqtSignal

from epics import PV


''' A simple recorde which can be used to record BSA data (including
cameras) and applying a ROI to the camera image, note that the returned
image is not processed in any other way. '''


class Recorder(QThread):
    ''' Public facing API. This is the main recorder. Note that its
    properties can only be set at creation. In order to change it
    discard the Recorder and create a new one.
Methods:
    start/stop - Start and stops both data recording and synchronisation.
                 Note that data from a stopped recorder is still valid
                 and can be read.
    clear      - Clears the buffer and sets full to False. Can be called
                 with a running buffer.
    __len__    - Amount of stored synchronized Data.
Signals:
    filled     - Signal is sent once the buffer is full
    new_row    - Signal is sent once new data arrived
Properties:
    data       - Holds the actual synchronized buffer.
    maxlen     - Maximal buffer size.
    lock       - Mutex protecting the buffer. '''

    filled = pyqtSignal()
    new_row = pyqtSignal(int)
    breaker = object()

    def __init__(self, maxlen=1, matching_maxlen=50,
        timestamp_threshold=1 / 160, **pvs):
        ''' Setup of the buffer. Note that redoing the setup is not
        possible. One needs to destroy and create a new setup.
Input:
    maxlen              - Maximal buffer size.
    matching_maxlen     - Maximal matching buffer size. Change this value
                          if the total rate is much lower than slowest PV.
    timestamp_threshold - Threshold on what timedifference is considered
                          equal.
    pv:
        - Scalar: {address: None}
        - Image: {address: {
            'shape': [1024, 1024], # Shape of original image
            'threshold': 0.2,      # Image cutting threshold
            'expand': [20, 10]}}   # Enlarge the ROI by this dimensions '''

        super().__init__()

        self.maxlen = int(maxlen)
        self.matching_maxlen = int(matching_maxlen)
        self.timestamp_thresh = float(timestamp_threshold)
        self.pv = pvs

        self.data = Table(self.maxlen, pvs)
        self.lock = Lock()
        self.synchr_buffer = Buffer(consumer=self)

        self.rate = Rate(*pvs, 'Total')
        self.matching_thread = MatchingThread(self, self.matching_maxlen,
            self.timestamp_thresh, self.synchr_buffer, self.rate)
        self.save_thread = SaveThread(self)

        self.clear()

    def __len__(self):
        return self.maxlen if self.full else self.index

    def run(self):
        for row in self.synchr_buffer:
            if row is self.breaker:
                break
            else:
                self.append_new_row(row)

    def append_new_row(self, row):
        self.rate.tick('Total')

        with self.lock:
            self.data[self.index] = row
            self.new_row.emit(self.index)
            self.index += 1

        if self.index == self.maxlen:
            if not self.full:
                self.filled.emit()
                self.full = True
            self.index = 0

    def start(self):
        self.matching_thread.start()
        super().start()

    def stop(self):
        self.matching_thread.requestInterruption()
        self.requestInterruption()

    def save(self, path):
        with self.lock:
            self.save_thread.start(deepcopy(self.data), path)

    def clear(self):
        with self.lock:
            self.data.clear()
            self.index = 0
            self.full = False


class SaveThread(QThread):
    ''' Asynchronous saving of the data. '''

    def start(self, raw, path):
        self.raw = raw
        self.path = path
        super().start()

    def run(self):
        with open(self.path, 'wb') as fid:
            dump(self.raw, fid)


class MatchingThread(QThread):
    ''' Synchronizes the incomming data according to timestamp. '''

    breaker = object()

    def __init__(self, parent, maxlen, timestamp_thresh, synchr_buffer, rate):
        super().__init__(parent)

        self.maxlen = maxlen
        self.timestamp_thresh = timestamp_thresh
        self.synchr_buffer = synchr_buffer
        self.rate = rate

    def start(self):
        ''' Creation of PVs is done here instead of the initialization to
        save network bandwidth when idle. '''

        self.index = 0
        self.synchr_table = Table(self.maxlen, self.parent().data.meta())
        self.non_synchr_buffer = Buffer(consumer=self)
        self.pv = [PV(name, auto_monitor=True) for name in self.parent().pv]

        for pv in self.pv:
            pv.add_callback(self.non_synchr_buffer.append_pv_callback)
        super().start()

    def run(self):
        for raw in self.non_synchr_buffer:
            if raw is self.breaker:
                break
            else:
                self.synchronize(*raw)
        self.cleanup()

    def cleanup(self):
        for pv in self.pv:
            pv.clear_callbacks()
        del self.non_synchr_buffer
        del self.synchr_table
        del self.pv

    def timestamp_index(self, timestamp):
        ''' Returns timestamp index. In case it does not exist it
        cleans a new slot. '''

        dt = np.abs(self.synchr_table['timestamp'] - timestamp)
        ind = np.argmin(dt)

        if dt[ind] > self.timestamp_thresh:
            ind = self.index
            self.synchr_table['valid'][ind] = 0
            self.synchr_table['timestamp'][ind] = timestamp
            self.index = (self.index + 1) % self.maxlen
        return ind

    def write_to_table(self, name, ind, value):
        if name in self.synchr_table.cam_data:
            img, start = img_cut(value, self.synchr_table.cam_data[name])
            self.synchr_table[name][ind] = img
            self.synchr_table[f'{name} start'][ind] = start
        else:
            self.synchr_table[name][ind] = value
        self.synchr_table['valid'][ind] += 1

    def synchronize(self, name, value, timestamp):
        self.rate.tick(name)
        ind = self.timestamp_index(timestamp)
        self.write_to_table(name, ind, value)

        if self.synchr_table['valid'][ind] == len(self.pv):
            self.synchr_buffer.append(self.synchr_table[ind])


class Table(dict):
    ''' The buffer is a dict which creates a np.array for each PV. For
    camera pv it creates a np.array(dtype='O') together with a start
    value, as well as a start value entrie. The buffer allows indexing
    by int which returns a copy of the value. '''

    def __init__(self, size, items):
        super().__init__()
        self.cam_data = {}

        for name, arg in items.items():
            if arg is None:
                self[name] = np.zeros(size)
            else:
                self[name] = np.zeros(size, dtype='O')
                self[f'{name} start'] = np.zeros(size,
                    dtype=[('x', 'u2'), ('y', 'u2')])
                self.cam_data[name] = arg
        self['valid'] = np.zeros(size, dtype='u2')
        self['timestamp'] = np.zeros(size)

    def meta(self):
        return {key: self.cam_data.get(key, None)
            for key in self if not key.count(' ')}

    def __getitem__(self, key):
        if isinstance(key, Integral):
            return {name: deepcopy(val[key]) for name, val in self.items()}
        else:
            return super().__getitem__(key)

    def __setitem__(self, key, val):
        if isinstance(key, Integral):
            for name in self:
                self[name][key] = val[name]
        else:
            return super().__setitem__(key, val)

    def clear(self):
        for val in self.values():
            val[:] = 0


class Buffer(deque):
    ''' Deques are thread-safe, this class is used to create a
    producer-consumer queue. A full queue silently drops the oldest
    element. '''

    def __init__(self, consumer, maxlen=100, sleep=0.001):
        super().__init__(maxlen=maxlen)
        self.sleep = sleep
        self.consumer = consumer

    def __iter__(self):
        return self

    def __next__(self):
        ''' Pauses if empty. In case interruption is requested it
        returns the consumer.breaker object. '''

        while not self.consumer.isInterruptionRequested():
            try:
                return super().popleft()
            except IndexError:
                sleep(self.sleep)
        return self.consumer.breaker

    def append_pv_callback(self, pvname, value, timestamp, **kwargs):
        self.append((pvname, value, timestamp))


class Rate(dict):
    ''' Simple Object tracking the rate of all items, needs to be
    started before tick and rate can be called. '''

    def __init__(self, *names, maxlen=100):
        super().__init__({name: deque(maxlen=maxlen) for name in names})
        self.maxlen = maxlen

    def tick(self, name):
        self[name].append(time())

    def rate(self, name):
        return (self.maxlen - 1) / (self[name][-1] - self[name][0] + 1e-15)


def img_cut(raw, meta):
    ''' Cut a ROI and returns starting point as well as center of mass. '''

    img = raw.reshape(meta['shape'])

    start_x, stop_x = find_roi_slice(img.sum(1), meta['threshold'],
        meta['expand'][0])
    start_y, stop_y = find_roi_slice(img.sum(0), meta['threshold'],
        meta['expand'][1])
    return img[start_x:stop_x, start_y:stop_y].T, (start_x, start_y)


def find_roi_slice(line, thresh, enlarge):
    line = medfilt(line, 5)
    mask = line > (line.min() * (1 - thresh) + line.max() * thresh)
    hist = np.nonzero(mask)[0]
    radius = hist.size // 2
    center = hist[radius]

    start = center - radius - enlarge
    end = center + radius + enlarge

    return max(start, 0), min(end, line.size - 1)

Here is a GUI which uses the recorder:

#!/usr/bin/env python3
import sys
from contextlib import suppress

from PyQt5.QtWidgets import (QApplication, QWidget, QPushButton, QGridLayout,
    QVBoxLayout, QHBoxLayout, QLabel, QProgressBar, QLineEdit)
import pyqtgraph as pg
import numpy as np

from recorder import Recorder


class Main(QWidget):
    ''' A simple gui, it only creates an empty shell of a gui. Linking
    has to be done in a separate step. '''

    def __init__(self, model):
        super().__init__()
        layout = QHBoxLayout(self)
        self._camera_shift = 0
        self.model = model

        # One pannel showing the latest image
        pan = pg.PlotWidget(self)
        layout.addWidget(pan)
        self.image = pg.ImageItem()
        pan.addItem(self.image)

        # Another pannel showing the correlations
        pan = pg.PlotWidget(self)
        layout.addWidget(pan)
        self.scr_693 = pg.ScatterPlotItem(x=[], y=[], symbol='+', pen='r')
        pan.addItem(self.scr_693)
        self.scr_250 = pg.ScatterPlotItem(x=[], y=[], symbol='o', pen='c')
        pan.addItem(self.scr_250)

        # Showing a few infos and offer ctrl
        ctr_layout = QVBoxLayout()
        layout.addLayout(ctr_layout)
        self.start = QPushButton('Start', self)
        ctr_layout.addWidget(self.start)
        self.stop = QPushButton('Stop', self)
        self.stop.setEnabled(False)
        ctr_layout.addWidget(self.stop)
        self.clear = QPushButton('Clear', self)
        ctr_layout.addWidget(self.clear)
        self.save = QPushButton('Save', self)
        ctr_layout.addWidget(self.save)

        grid = QGridLayout()
        ctr_layout.addLayout(grid)
        grid.addWidget(QLabel('Camera shift'), 0, 0)
        self.camera_shift = QLineEdit('0', self)
        self.camera_shift.returnPressed.connect(lambda:
            setattr(self, '_camera_shift', int(self.camera_shift.text())))
        grid.addWidget(self.camera_shift, 0, 1)
        grid.addWidget(QLabel('Save Path'), 1, 0)
        self.path = QLineEdit('test.npy', self)
        grid.addWidget(self.path, 0, 2)

        self.rate = {}
        for name in model.rate:
            bar = QProgressBar(self)
            bar.setFormat(f'{name}: %v Hz')
            bar.setRange(0, 120)
            ctr_layout.addWidget(bar)
            self.rate[name] = bar

        self.full = QLabel('Not full', self)
        ctr_layout.addWidget(self.full)
        self.bar = QProgressBar(self)
        self.bar.setRange(0, model.maxlen)
        ctr_layout.addWidget(self.bar)
        self.setGeometry(0, 0, 800, 300)

        self.show()

    def update(self, ind):
        data = self.model.data

        shifted_start = np.roll(data['OTRS:DMP1:695:CAMERA.IRAW start'],
            self._camera_shift)
        self.scr_693.setData(x=shifted_start['x'], y=data['BPMS:DMP1:693:Y'])
        self.scr_250.setData(x=shifted_start['x'], y=data['BPMS:LTU1:250:X'])

        # Due to Recorder.clear being dirty
        with suppress(AttributeError, IndexError):
            self.image.setImage(data['OTRS:DMP1:695:CAMERA.IRAW'][ind])

        self.bar.setValue(len(self.model))
        for name, bar in self.rate.items():
            bar.setValue(self.model.rate.rate(name))

        QApplication.processEvents


app = QApplication(sys.argv)

# ## setup buffer ## #
PVs = {
    'BPMS:DMP1:693:Y': None,
    'BPMS:LTU1:250:X': None,
    'OTRS:DMP1:695:CAMERA.IRAW': {
        'shape': [1024, 1024],
        'threshold': 0.2,
        'expand': [20, 10]}
}

recorder = Recorder(maxlen=100, **PVs)
app.aboutToQuit.connect(recorder.stop)

# ## setup the main gui ## #
main = Main(recorder)
main.start.clicked.connect(recorder.start)
main.stop.clicked.connect(recorder.stop)
main.clear.clicked.connect(recorder.clear)
main.clear.clicked.connect(lambda: main.full.setText('Not Full'))
main.save.clicked.connect(lambda: recorder.save(main.path.text()))

recorder.filled.connect(lambda: main.full.setText('Full'))
recorder.new_row.connect(main.update)
recorder.save_thread.started.connect(lambda: main.save.setEnabled(False))
recorder.save_thread.finished.connect(lambda: main.save.setEnabled(True))
recorder.started.connect(lambda:
    main.start.setEnabled(False) or
    main.stop.setEnabled(True))
recorder.finished.connect(lambda:
    main.start.setEnabled(True) or
    main.stop.setEnabled(False))

sys.exit(app.exec_())

And for people unfamiliar with epics I created a dummy class simulating its behaviour:

''' Dummy file. Proper files can be found at
http://cars9.uchicago.edu/software/python/pyepics3/. Returns value over network
therefore delay is network dependent. '''


from time import time

from PyQt5.QtCore import QTimer
from numpy.random import random


class PV:
    def __init__(self, addr, auto_monitor=None):
        self.addr = addr
        self.timer = QTimer()

    def add_callback(self, fun):
        value = img if self.addr == 'OTRS:DMP1:695:CAMERA.IRAW' else random

        self.timer.timeout.connect(lambda: fun(
            pvname=self.addr, timestamp=time(), value=value()))
        self.timer.start(1 / 120)

    def clear_callbacks(self):
        self.timer.stop()
        self.timer.timeout.disconnect()


def img(size=1024, dia=20, signal=50):
    val = random((size, size))
    x_dia = (int(dia * random()) + 1) * 2
    y_dia = (int(dia * random()) + 1) * 2

    x = int(random() * (size - x_dia))
    y = int(random() * (size - y_dia))

    val[x:x + x_dia, y:y + y_dia] = random((x_dia, y_dia)) * signal
    return val.reshape((1, size * size))
\$\endgroup\$
1
\$\begingroup\$

Some readers are not steeped in computer vision techniques when they approach your code. Nowhere do you explain in prose or comments about Region Of Interest, to disambiguate from the more common Return On Investment. We don't need a heavyweight definition; a mere mention would suffice.

The imports are nicely grouped. Consider applying M-x sort within each group. When multiple people edit the code there should be one obvious place to insert each import.

Offer a mention of what BSA denotes, please.

typo: simple recorde

This would be a fine comment:

# Public facing API.

But putting that in the docstring is entirely redundant.

Thank you for the helpful Recorder overview. Consider discarding the Properties overview, as it may be redundant with / conflicts with the ctor docstring. What we're examining here is "what is the public API?", that is, do you expect caller to directly interact with properties like data or maxlen?

The class-level assignments to {filled, new_row, breaker} are odd. Is this a singleton, I can only create one instance? Why not assign them in the constructor?

In the ctor docstring, consider describing rate and other properties, if you expect callers to interact with them. Do spell out the obscure QT pv in a mention somewhere. Please add a comment or two to class PV. There's more, but that's enough for now.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.