DAQ: added functionality to set global variables for aerotech controller in epics

src/aare/devices/aerotech.py

@@ -3,9 +3,10 @@ import sys
 import threading
 import time
 from enum import Enum
+from typing import Union
 
 #import automation1 as a1
-from epics import PV, Motor
+from epics import PV, Motor, caput, caget
 
 from aare.common.beamline import MXBeamline, mx_beamline
 from aare.devices.my_motor import MyMotor
@@ -17,6 +18,7 @@ class AerotechController:
             self.controller = None
         else:
             self.controller = a1.Controller.connect(controller_ip)
+
         self.start_controller()
 
     def start_controller(self):
@@ -29,6 +31,8 @@ class AerotechController:
         self.controller.runtime.commands.motion.enable(axis.upper())
 
     def home_motor(self, axis: str):
+        if not self.controller.runtime.commands.motion.is_enabled(axis.upper()):
+            self.enable_motion(axis)
         self.controller.runtime.commands.motion.home(axis.upper())
 
 
@@ -40,26 +44,30 @@ class AerotechRunEnum(Enum):
     PAUSE = 4
     RESET = 5
 
+class VariableTypeEnum(Enum):
+    INT = 0
+    REAL = 1
+    STRING = 2
 
 class AerotechControllerEpics:
     def __init__(self, beamline: MXBeamline):
         BEAMLINE = beamline.value.upper()
-        aerotech_pv_prefix = f"{BEAMLINE}-ES-DF1"
-        trx_prefix = f"{aerotech_pv_prefix}:TRX1"
-        try_prefix = f"{aerotech_pv_prefix}:TRY1"
-        trz_prefix = f"{aerotech_pv_prefix}:TRZ1"
-        rotu_prefix = f"{aerotech_pv_prefix}:ROTU"
+        self.aerotech_pv_prefix = f"{BEAMLINE}-ES-DF1"
+        trx_prefix = f"{self.aerotech_pv_prefix}:TRX1"
+        try_prefix = f"{self.aerotech_pv_prefix}:TRY1"
+        trz_prefix = f"{self.aerotech_pv_prefix}:TRZ1"
+        rotu_prefix = f"{self.aerotech_pv_prefix}:ROTU"
         self.gmx = MyMotor(trx_prefix)
         self.gmy = MyMotor(try_prefix)
         self.gmz = MyMotor(trz_prefix)
         self.omega = MyMotor(rotu_prefix)
-        self.__enable_all = PV(f"{aerotech_pv_prefix}:EnableAll")
-        self.__disable_all = PV(f"{aerotech_pv_prefix}:DisableAll")
-        self.__acknowledge_all = PV(f"{aerotech_pv_prefix}:AckAll")
-        self.__stop_all = PV(f"{aerotech_pv_prefix}:StopAll")
-        self.__task_filename = PV(f"{aerotech_pv_prefix}:TASK:FILENAME")
-        self.__task_id = PV(f"{aerotech_pv_prefix}:TASK:TASKIDX")  # values can be 1 to 4 DO NOT USE 0!!!!
-        self.__task_run_enum = PV(f"{aerotech_pv_prefix}:TASK:SWITCH")  # 0 Stop, 1 Start, 2 Run,
+        self.__enable_all = PV(f"{self.aerotech_pv_prefix}:EnableAll")
+        self.__disable_all = PV(f"{self.aerotech_pv_prefix}:DisableAll")
+        self.__acknowledge_all = PV(f"{self.aerotech_pv_prefix}:AckAll")
+        self.__stop_all = PV(f"{self.aerotech_pv_prefix}:StopAll")
+        self.__task_filename = PV(f"{self.aerotech_pv_prefix}:TASK:FILENAME")
+        self.__task_id = PV(f"{self.aerotech_pv_prefix}:TASK:TASKIDX")  # values can be 1 to 4 DO NOT USE 0!!!!
+        self.__task_run_enum = PV(f"{self.aerotech_pv_prefix}:TASK:SWITCH")  # 0 Stop, 1 Start, 2 Run,
         # 3 Load, 4 Pause, 5 Reset
         self.enable_all()
 
@@ -124,6 +132,53 @@ class AerotechControllerEpics:
         self.__task_run()
         print(self.__task_run_enum.get())
 
+    def __set_global(self, index:int, value: Union[int, float, str],
+                     timeout:float=10.0):
+        """Set a global variable in aerotech, it takes ~200 ms for the value to be set
+        :param index: select a variable to change. an integer between 0..256 for int and real variable for 0..31 for strings
+        :param value: the value to set can be int, float (REAL) or string
+        :param timeout: timeout for vairable change, default 10.0 seconds """
+        var_type = self.__get_var_type(value)
+        caput(f"{self.aerotech_pv_prefix}:VAR:{var_type.upper()}-ADDR", index)
+        if var_type == 'STRING':
+            var_type = 'STRING-SHORT'
+        caput(f"{self.aerotech_pv_prefix}:VAR:{var_type.upper()}", value)
+        start = time.perf_counter()
+        while time.perf_counter() - start < timeout:
+            rbv = self.__read_global_from_index(index, var_type)
+            if rbv == str(value):
+                return
+            time.sleep(0.1)
+
+        raise TimeoutError(f"Timeout setting global variable {index} to {value}")
+
+
+    def __read_global_feedback(self, index, var_type: str):
+        caput(f"{self.aerotech_pv_prefix}:VAR:{var_type.upper()}-RBV.PROC", 1)
+        time.sleep(0.5)
+        return caget(f"{self.aerotech_pv_prefix}:VAR:{var_type.upper()}-RBV", as_string=True)
+
+    def __read_global_from_index(self, index, var_type: str):
+        return caget(f"{self.aerotech_pv_prefix}:VAR:{var_type.upper()}{index}_RBV", as_string=True)
+
+    def __get_var_type(self, value):
+        if type(value) is int:
+            return VariableTypeEnum.INT.name
+        elif type(value) is float:
+            return VariableTypeEnum.REAL.name
+        elif type(value) is str:
+            return VariableTypeEnum.STRING.name
+        else:
+            raise ValueError(f"Invalid type {type(value)} for global variable")
+
+    def get_global_variable(self, index: int, var_type:VariableTypeEnum):
+        return self.__read_global_from_index(index, var_type.name)
+
+    def set_global_variable(self, index: int, value: Union[int, float, str]):
+        self.__set_global(index, value)
+
+
+
 
 if __name__ == "__main__":
     beamline = mx_beamline()
@@ -135,5 +190,10 @@ if __name__ == "__main__":
     #aerotech.disconnect()
 
     aerotech_epics = AerotechControllerEpics(beamline)
+    #aerotech_epics.omega.speed = 80.0
+    print(aerotech_epics.get_global_variable(0, VariableTypeEnum.REAL))
+    aerotech_epics.set_global_variable(0, 100.0)
+    print(aerotech_epics.get_global_variable(0, VariableTypeEnum.REAL))
+    #aerotech_epics.omega.move(90.0, wait=True)
     #aerotech_epics.home_all()
-    #aerotech_epics.omega.move(90.0)
+