Get CS info from motor link field. Only works for compound motors

Author: tomtrafford

src/ophyd_async/epics/pmac/_pmacMotor.py

@@ -2,9 +2,22 @@
 
 from ophyd_async.epics.motion import Motor
 
+from ..signal.signal import epics_signal_r
+
 
 class PmacCSMotor(Motor, Movable, Stoppable):
-    def __init__(self, prefix: str, csNum: int, csAxis: str, name="") -> None:
-        self.csNum = csNum
-        self.csAxis = csAxis
+    def __init__(self, prefix: str, name="") -> None:
+        self.output_link = epics_signal_r(str, prefix + ".OUT")
+        self.cs_axis = ""
+        self.cs_port = ""
         super().__init__(prefix=prefix, name=name)
+
+    async def get_cs_info(self):
+        output_link = await self.output_link.get_value()
+        # Split "@asyn(PORT,num)" into ["PORT", "num"]
+        split = output_link.split("(")[1].rstrip(")").split(",")
+        self.cs_port = split[0].strip()
+        assert (
+            "CS" in self.cs_port
+        ), f"{self.name} not in a CS. It is not a compound motor."
+        self.cs_axis = "abcuvwxyz"[int(split[1].strip()) - 1]

src/ophyd_async/epics/pmac/_pmacTrajectory.py

@@ -37,31 +37,37 @@ async def prepare(self, stack: list[Frames[PmacCSMotor]]):
         scanSize = len(stack[0])
         scanAxes = stack[0].axes()
 
+        cs_ports = set()
         self.profile = {}
         for axis in scanAxes:
             if axis != "DURATION":
-                self.profile[axis.csAxis] = []
-                self.profile[axis.csAxis + "_velocity"] = []
+                await axis.get_cs_info()
+                self.profile[axis.cs_axis] = []
+                self.profile[axis.cs_axis + "_velocity"] = []
+                cs_ports.add(axis.cs_port)
             else:
                 self.profile[axis.lower()] = []
+        cs_port = cs_ports.pop()
 
         # Calc Velocity
 
         for axis in scanAxes:
             for i in range(scanSize - 1):
                 if axis != "DURATION":
-                    self.profile[axis.csAxis + "_velocity"].append(
+                    self.profile[axis.cs_axis + "_velocity"].append(
                         (stack[0].midpoints[axis][i + 1] - stack[0].midpoints[axis][i])
                         / (stack[0].midpoints["DURATION"][i])
                     )
-                    self.profile[axis.csAxis].append(stack[0].midpoints[axis][i])
+                    self.profile[axis.cs_axis].append(stack[0].midpoints[axis][i])
                 else:
                     self.profile[axis.lower()].append(
                         int(stack[0].midpoints[axis][i] / TICK_S)
                     )
             if axis != "DURATION":
-                self.profile[axis.csAxis].append(stack[0].midpoints[axis][scanSize - 1])
-                self.profile[axis.csAxis + "_velocity"].append(0)
+                self.profile[axis.cs_axis].append(
+                    stack[0].midpoints[axis][scanSize - 1]
+                )
+                self.profile[axis.cs_axis + "_velocity"].append(0)
             else:
                 self.profile[axis.lower()].append(
                     int(stack[0].midpoints[axis][scanSize - 1] / TICK_S)
@@ -75,30 +81,30 @@ async def prepare(self, stack: list[Frames[PmacCSMotor]]):
                 run_up_disp, run_up_time = await self._ramp_up_velocity_pos(
                     0,
                     axis,
-                    self.profile[axis.csAxis + "_velocity"][0],
+                    self.profile[axis.cs_axis + "_velocity"][0],
                 )
-                self.initial_pos[axis.csAxis] = (
-                    self.profile[axis.csAxis][0] - run_up_disp
+                self.initial_pos[axis.cs_axis] = (
+                    self.profile[axis.cs_axis][0] - run_up_disp
                 )
         self.profile["duration"][0] += run_up_time / TICK_S
 
         # Send trajectory to brick
         for axis in scanAxes:
             if axis != "DURATION":
-                self.profile_cs_name.set(self.cs)
+                self.profile_cs_name.set(cs_port)
                 self.points_to_build.set(scanSize)
-                getattr(self, "use_" + axis.csAxis).set(True)
-                getattr(self, axis.csAxis).set(self.profile[axis.csAxis])
-                getattr(self, axis.csAxis + "_vel").set(
-                    self.profile[axis.csAxis + "_velocity"]
+                getattr(self, "use_" + axis.cs_axis).set(True)
+                getattr(self, axis.cs_axis).set(self.profile[axis.cs_axis])
+                getattr(self, axis.cs_axis + "_vel").set(
+                    self.profile[axis.cs_axis + "_velocity"]
                 )
             else:
                 self.timeArray.set(self.profile["duration"])
 
         # MOVE TO START
         for axis in scanAxes:
             if axis != "DURATION":
-                await axis.set(self.initial_pos[axis.csAxis])
+                await axis.set(self.initial_pos[axis.cs_axis])
 
         # Set No Of Points
 

tests/epics/pmac/test_pmac.py

@@ -8,15 +8,14 @@
 @pytest.fixture
 async def sim_x_motor():
     async with DeviceCollector(mock=True):
-        sim_motor = PmacCSMotor(
-            "BLxxI-MO-STAGE-01:X", "BRICK1.CS3", "z", name="sim_x_motor"
-        )
+        sim_motor = PmacCSMotor("BLxxI-MO-STAGE-01:X", name="sim_x_motor")
 
     set_mock_value(sim_motor.motor_egu, "mm")
     set_mock_value(sim_motor.precision, 3)
     set_mock_value(sim_motor.acceleration_time, 0.5)
     set_mock_value(sim_motor.max_velocity, 5)
     set_mock_value(sim_motor.velocity, 0.5)
+    set_mock_value(sim_motor.output_link, "@asyn(BRICK1.CS3,9)")
 
     yield sim_motor