wip

debye_bec/devices/mo1_bragg/acscontroller.py

@@ -0,0 +1,62 @@
+import socket
+import threading
+
+import numpy as np
+from ophyd.signal import Signal
+
+
+class ACSSignal(Signal):
+
+    def __init__(self, controller, tag, **kwargs):
+        self.controller = controller
+        self.tag = tag
+        super().__init__(**kwargs)
+
+    def get(self):
+        value = self.controller.read_tag(self.tag)
+        self._readback = value
+        return value
+
+    def put(self, value, **kwargs):
+        self.controller.write_tag(self.tag, value)
+        self._readback = value
+        return super().put(value, **kwargs)
+
+
+class ACSController:
+
+    def __init__(self, host: str, port: int, timeout: float = 2.0):
+        self.host = host
+        self.port = port
+        self.timeout = timeout
+        self._socket = None
+        self._lock = threading.Lock()
+
+        self.precision = 12
+
+    def connect(self):
+        if self._socket is not None:
+            return
+        self._socket = socket.create_connection((self.host, self.port), timeout=self.timeout)
+
+    def disconnect(self):
+        if self._socket is not None:
+            self._socket.close()
+            self._socket = None
+
+    def _query(self, command: str) -> str:
+        with self._lock:
+            self.connect()
+            msg = (command + "\r\n").encode()
+            self._socket.sendall(msg)
+            reply = self._socket.recv(4096)
+            return reply.decode().strip()
+
+    def read_tag(self, tag: int):
+        i = 0  # For arrays, to be implemented
+        response = self._query(f"?{{%0.{self.precision:0.0f}f}}GETVAR({tag},{i}")
+        return float(response)
+
+    def write_tag(self, tag: int, value):
+        i = 0  # For arrays, to be implemented
+        self._query(f"SETVAR({np.round(value, self.precision)},{tag},{i}")

debye_bec/devices/mo1_bragg/mo1_bragg_devices.py

@@ -18,6 +18,7 @@ from ophyd import (
 )
 from ophyd.utils import LimitError
 
+from debye_bec.devices.mo1_bragg.acs_controller import ACSSignal
 from debye_bec.devices.mo1_bragg.mo1_bragg_enums import MoveType
 
 # Initialise logger
@@ -256,6 +257,8 @@ class Mo1BraggPositioner(Device, PositionerBase):
 
     angle = Cpt(EpicsSignalRO, suffix="feedback_pos_angle_RBV", kind="normal", auto_monitor=True)
 
+    test = Cpt(ACSSignal, tag=53000)  # s_scan_angle_low
+
     ########## Move Command PVs   ##########
 
     move_abs = Cpt(EpicsSignal, suffix="move_abs", kind="config", put_complete=True)
@@ -278,6 +281,9 @@ class Mo1BraggPositioner(Device, PositionerBase):
         self._stopped = False
         self.readback.name = self.name
 
+        self.controller = ACSController(host, port)
+        kwargs["controller"] = self.controller
+
     def stop(self, *, success=False) -> None:
         """Stop any motion on the positioner