Pyepics migration

pixi.toml

@@ -6,7 +6,6 @@ platforms = ["linux-64"]
 python = ">=3.11,<3.14"
 algotom = ">=1.7.0,<2"
 ipython = ">=9.16.1,<10"
-bluesky-base = ">=1.15.1,<2"
 tiled-client = ">=0.2.14,<0.3"
 bluesky-tiled-plugins = ">=2.0.9,<3"
 rich = ">=15.0.0,<16"
@@ -20,12 +19,13 @@ scikit-image = ">=0.26.0,<0.27"
 [pypi-dependencies]
 hextools = { path = ".", editable = true }
 nslsii = { git = "https://github.com/NSLS2/nslsii", rev = "main" }
+bluesky = { git = "https://github.com/bluesky/bluesky", rev = "main" }
 ophyd-async = { git = "https://github.com/jwlodek/ophyd-async", rev = "add-codec-and-process-plugin-allow-for-filtering" }
 
 [feature.profile.dependencies]
 redis-json-dict = ">=0.2.2,<0.3"
 
-[feature.qs.dependencies]
+[feature.qs.pypi-dependencies]
 bluesky-queueserver = ">=0.0.25,<0.0.26"
 bluesky-queueserver-api = ">=0.0.13,<0.0.14"
 bluesky-httpserver = ">=0.0.12,<0.0.13"

pyproject.toml

@@ -32,16 +32,15 @@ dependencies = [
   "algotom",
   "ipython",
   "numpy",
+  "bluesky",
+  "nslsii",
   "ophyd-async[ca, pva]",
   "rich>=15.0.0",
   "scikit-image>=0.26.0",
   "scipy>=1.17.1",
 ]
 
 [project.optional-dependencies]
-iocs = [
-  "caproto",
-]
 gui = [
 #   "ntnda-qt-viewer",
 ]

src/hextools/detectors/kinetix.py

@@ -1 +1,13 @@
 """Kinetix detector support for HEX beamline."""
+
+from ophyd_async.epics.adkinetix import KinetixDetector
+from ophyd_async.epics.adcore import ADWriterFactory
+
+def kinetix_factory(num: int, path_provider, name: str):
+    """Factory function to create a KinetixDetector with HDF writer."""
+    return KinetixDetector(
+        f"XF:27ID1-BI{{Kinetix-Det:{num}}}",
+        ADWriterFactory.hdf(path_provider),
+        proc_suffix="Proc1:",
+        name=name,
+    )

src/hextools/motors.py

@@ -21,6 +21,7 @@
 from ophyd_async.epics.core import (
     EpicsDevice,
     epics_signal_r,
+    epics_signal_rw,
     epics_triggerable_command,
 )
 from ophyd_async.epics.motor import Motor as AsyncEpicsMotor
@@ -56,11 +57,11 @@ def __init__(self, prefix: str, name="optics_table"):
         with self.add_children_as_readables(Format.CHILD):
             self.x2 = AsyncEpicsMotor(prefix + "X2}Mtr", name="x2")
             self.y2 = AsyncEpicsMotor(prefix + "Y2}Mtr", name="y2")
-            self.rx3 = AsyncEpicsMotor(prefix + "RX3}Mtr", name="rx3")
-            self.ry3 = AsyncEpicsMotor(prefix + "RY3}Mtr", name="ry3")
+            self.rx3 = AsyncEpicsMotor(prefix + "Rx3}Mtr", name="rx3")
+            self.ry3 = AsyncEpicsMotor(prefix + "Ry3}Mtr", name="ry3")
             self.x3 = AsyncEpicsMotor(prefix + "X3}Mtr", name="x3")
             self.y3 = AsyncEpicsMotor(prefix + "Y3}Mtr", name="y3")
-            self.ry4 = AsyncEpicsMotor(prefix + "RY4}Mtr", name="ry4")
+            self.ry4 = AsyncEpicsMotor(prefix + "Ry4}Mtr", name="ry4")
             self.x4 = AsyncEpicsMotor(prefix + "X4}Mtr", name="x4")
             self.a1 = AsyncEpicsMotor(prefix + "A1}Mtr", name="a1")
             self.z0 = AsyncEpicsMotor(prefix + "Z0}Mtr", name="z0")
@@ -161,8 +162,8 @@ def __init__(self, prefix: str, name: str = "sample_tower"):
 class CameraObjective(StrictEnum):
     """Represents the camera objective in use."""
 
-    LEFT_4MM = "left_4mm"
-    RIGHT_2MM = "right_2mm"
+    RIGHT_2MM = "Right Objective"
+    LEFT_4MM = "Left Objective"
 
 
 class HomeStatus(StrictEnum):
@@ -172,7 +173,7 @@ class HomeStatus(StrictEnum):
     HOMED = "Homed"
 
 
-class DoubleObjCamera(StandardReadable, EpicsDevice, AsyncMovable[CameraObjective]):
+class FOV_2_4_mm_Camera(StandardReadable, EpicsDevice, AsyncMovable[CameraObjective | str]):
     """HEX double objective camera."""
 
     def __init__(self, prefix: str, name: str = "double_obj_camera"):
@@ -189,21 +190,18 @@ def __init__(self, prefix: str, name: str = "double_obj_camera"):
             prefix + "ObjSel}Sts:HomeCmplt-Sts",
             name="obj_selector_home_sts",
         )
+        self.objective = epics_signal_rw(CameraObjective, prefix + "ObjSel}Objective", name="objective")
+
         self._at_right_objective = epics_signal_r(
             bool, prefix + "ObjSel}AtRightObj", name="at_right_objective"
         )
         self._at_left_objective = epics_signal_r(
             bool, prefix + "ObjSel}AtLeftObj", name="at_left_objective"
         )
-        self._goto_right_objective = epics_triggerable_command(
-            prefix + "ObjSel}Cmd:GotoRight-Cmd", name="goto_right_objective"
-        )
-        self._goto_left_objective = epics_triggerable_command(
-            prefix + "ObjSel}Cmd:GotoLeft-Cmd", name="goto_left_objective"
-        )
+
 
     @AsyncStatus.wrap
-    async def set(self, value: CameraObjective):
+    async def set(self, value: CameraObjective | str):
         """Move the camera to the specified objective.
 
         Parameters
@@ -222,20 +220,32 @@ async def set(self, value: CameraObjective):
                 "Please home it before moving to a specific objective."
             )
 
-        if value == CameraObjective.LEFT_4MM:
-            await self._goto_left_objective.execute()
-            rb_check = self._at_left_objective
-        else:
-            await self._goto_right_objective.execute()
-            rb_check = self._at_right_objective
+        # Attempt to auto-deduce the CameraObjective from a string input.
+        if isinstance(value, str):
+            for possible_value in CameraObjective:
+                if value.upper() in possible_value.name:
+                    value = possible_value
+                    break
+
+        if not isinstance(value, CameraObjective):
+            raise ValueError(
+                f"Invalid objective value: {value}. "
+                "Must be a CameraObjective or a string matching" \
+                "one of its names."
+            )
 
+        await self.objective.set(value)
+        if value == CameraObjective.RIGHT_2MM:
+            rb_check = self._at_right_objective
+        else:
+            rb_check = self._at_left_objective
         await wait_for_value(rb_check, True, timeout=None)
 
 
-class WideFOVCamera(StandardReadable, EpicsDevice):
+class FOV_20_40_mm_Camera(StandardReadable, EpicsDevice):
     """HEX wide field of view camera."""
 
-    def __init__(self, prefix: str, name: str = "wide_fov_camera"):
+    def __init__(self, prefix: str, name: str = "fov_20_40_mm_camera"):
         super().__init__(prefix, name=name)
         with self.add_children_as_readables(Format.CHILD):
             self.focus = AsyncEpicsMotor(prefix + "Focus}Mtr", name="focus")