first prototypes for Sensors

slic/core/sensor/__init__.py

@@ -0,0 +1,6 @@
+
+from .sensor import Sensor
+from .bssensor import BSSensor
+from .pvsensor import PVSensor
+
+

slic/core/sensor/basesensor.py

@@ -0,0 +1,31 @@
+from abc import abstractmethod
+
+from slic.utils.metaclasses import RegistryABC
+
+
+class BaseSensor(RegistryABC):
+
+    @abstractmethod
+    def start(self):
+        raise NotImplementedError
+
+    @abstractmethod
+    def stop(self):
+        raise NotImplementedError
+
+
+
+#s = Sensor("DEVICE:NAME", aggregation="mean")
+
+#for step in scan:
+#    target = values[step]
+#    mot.set(target).wait()
+
+#    s.start() # start recording values (for PV add collecting callback)
+#    daq.acquire(n_pulses=1000)
+#    s.stop() # stop recording values (for PV remove callback)
+
+#    avg_val_during_step = s.get()
+
+
+

slic/core/sensor/bssensor.py

@@ -0,0 +1,56 @@
+from threading import Thread, Event
+from bsread import source, dispatcher
+from .sensor import Sensor
+
+
+class BSSensor(Sensor):
+
+    def start(self):
+        self.thread = thread = BSSourceThread(self.name, self._collect)
+        thread.start()
+
+    def stop(self):
+        self.thread.stop()
+
+
+
+class BSSourceThread(Thread):
+
+    def __init__(self, name, callback):
+        super().__init__()
+        self.name = name
+        self.callback = callback
+        self.running = Event()
+
+    def run(self):
+        running = self.running
+        running.set()
+        name = self.name
+        channels = [name]
+        with source(channels=channels, receive_timeout=-1) as src:
+            while running.is_set():
+                msg = src.receive()
+                data = msg.data.data
+                value = data[name].value
+                self.callback(value)
+        running.clear()
+
+    def stop(self):
+        self.running.clear()
+        self.join()
+
+
+
+if __name__ == "__main__":
+    from time import sleep
+
+    s = BSSensor("SARES11-CVME-EVR0:CALCS")
+    s.start()
+    sleep(1)
+    print(1, len(s._cache))
+    sleep(1)
+    s.stop()
+    print(2, len(s._cache))
+
+
+

slic/core/sensor/pvsensor.py

@@ -0,0 +1,47 @@
+from epics import PV
+from .sensor import Sensor
+
+
+class PVSensor(Sensor):
+
+    def __init__(self, name, *args, **kwargs):
+        super().__init__(name, *args, **kwargs)
+        self.pv = PV(name)
+        self._cb_index = None
+
+
+    #TODO: using a similar block in PVAdjustable, Motor, ... already
+
+    @property
+    def units(self):
+        units = self._units
+        if units is not None:
+            return units
+        return self.pv.units
+
+    @units.setter
+    def units(self, value):
+        self._units = value
+
+
+    #TODO: might be better to use the default from Sensor
+    def get_current_value(self):
+        return self.pv.get()
+
+
+    def start(self):
+        if self._cb_index is not None:
+            print("already running")
+        self._cb_index = self.pv.add_callback(self._collect_cb)
+
+    def stop(self):
+        if self._cb_index is None:
+            print("not started yet")
+        self.pv.remove_callback(self._cb_index)
+        self._cb_index = None
+
+    def _collect_cb(self, value=None, **kwargs):
+        self._collect(value)
+
+
+

slic/core/sensor/sensor.py

@@ -0,0 +1,68 @@
+import numpy as np
+from slic.utils import typename
+from .basesensor import BaseSensor
+
+
+class Sensor(BaseSensor):
+
+    def __init__(self, name, units=None, aggregation=np.mean):
+        self.name = name
+        self.units = units
+        self.aggregation = aggregation
+        self._cache = []
+
+
+    #TODO: should this be get_LAST_value?
+    def get_current_value(self):
+        try:
+            return self._cache[-1]
+        except IndexError:
+            return None
+
+    def get_aggregate(self):
+        try:
+            return self.aggregation(self._cache)
+        except Exception:
+            return None
+
+    get = get_aggregate #TODO: should get return the aggregation result or the current value!?
+
+
+    def _collect(self, value):
+        self._cache.append(value)
+
+
+    def __enter__(self, ):
+        self.start()
+        return self
+
+    def __exit__(self, _exc_type, _exc_val, _exc_tb):
+        self.stop()
+
+
+    #TODO: pull this out of adjustable and make it mixin:
+
+    def __repr__(self):
+        name  = self._printable_name()
+        value = self._printable_value()
+        return f"{name} at {value}"
+
+    def __str__(self):
+        return self._printable_value()
+
+    def _printable_name(self):
+        tname = typename(self)
+        name = self.name
+        return f"{tname} \"{name}\"" if name is not None else tname
+
+    def _printable_value(self):
+        value = self.get()
+        units = self.units
+        if units is None:
+            return str(value)
+        if units.casefold() in ["deg", "°"]:
+            return f"{value}°"
+        return f"{value} {units}"
+
+
+