Initial ophyd-async support

.github/workflows/code.yml

@@ -74,7 +74,7 @@ jobs:
       - name: Install project code and developer dependencies
         run: |
           set -vxeuo pipefail
-          pip install -e .[dev]
+          pip install -e .[dev,async]
 
       - name: Start EPICS IOCs in Docker
         run: |

.gitignore

@@ -1,3 +1,6 @@
+# Emacs temporary files
+*~
+
 # local developer code
 dev_*.py
 dev_*.ipynb

apstools/_devices_async/README.md

@@ -0,0 +1,9 @@
+# Awaitable Devices Using *ophyd-async*
+
+The devices defined here are subject to change. Feedback is welcome.
+
+This folder contains **experimental** devices written using the
+*ophyd-async* library instead of classic (threaded) *ophyd*. Using
+*ophyd-async* requires installing apstools with the ``[async]``
+extra. As such, it should not be imported from the rest of *apstools*
+without making *ophyd_async* a default dependency.

apstools/_devices_async/aps_machine.py

@@ -0,0 +1,58 @@
+import logging
+from typing import Annotated as A
+
+from ophyd_async.core import SignalR, StandardReadable
+from ophyd_async.core import StandardReadableFormat as Format
+from ophyd_async.core import StrictEnum, SubsetEnum
+from ophyd_async.epics.core import EpicsDevice, PvSuffix, epics_signal_r
+
+log = logging.getLogger(__name__)
+
+
+class ApsMachine(EpicsDevice, StandardReadable):
+    _ophyd_labels_ = {"synchrotrons"}
+
+    class MachineStatus(SubsetEnum):
+        UNKNOWN = "State Unknown"
+        USER_OPERATIONS = "USER OPERATIONS"
+        SUPPLEMENTAL_TIME = "SUPLEMENTAL TIME"
+        ASD_STUDIES = "ASD Studies"
+        NO_BEAM = "NO BEAM"
+        MAINTENANCE = "MAINTENANCE"
+
+    class OperatingMode(StrictEnum):
+        STATE_UNKNOWN = "State Unknown"
+        NO_BEAM = "NO BEAM"
+        INJECTING = "Injecting"
+        STORED_BEAM = "Stored Beam"
+        DELIVERED_BEAM = "Delivered Beam"
+        MAINTENANCE = "MAINTENANCE"
+
+    current: A[SignalR[float], PvSuffix("S-DCCT:CurrentM"), Format.CHILD]
+    operators: A[SignalR[str], PvSuffix("OPS:message1"), Format.CONFIG_SIGNAL]
+    floor_coordinator: A[SignalR[str], PvSuffix("OPS:message2"), Format.CONFIG_SIGNAL]
+    fill_pattern: A[SignalR[str], PvSuffix("OPS:message3"), Format.CONFIG_SIGNAL]
+    last_problem_message: A[SignalR[str], PvSuffix("OPS:message4"), Format.CONFIG_SIGNAL]
+    last_trip_message: A[SignalR[str], PvSuffix("OPS:message5"), Format.CONFIG_SIGNAL]
+    message6: A[SignalR[str], PvSuffix("OPS:message6"), Format.CONFIG_SIGNAL]
+    message7: A[SignalR[str], PvSuffix("OPS:message7"), Format.CONFIG_SIGNAL]
+    message8: A[SignalR[str], PvSuffix("OPS:message8"), Format.CONFIG_SIGNAL]
+    machine_status: A[SignalR[MachineStatus], PvSuffix("S:DesiredMode"), Format.CONFIG_SIGNAL]
+    operating_mode: A[SignalR[OperatingMode], PvSuffix("S:ActualMode"), Format.CONFIG_SIGNAL]
+    shutter_status: A[SignalR[bool], PvSuffix("XFD:ShutterPermit"), Format.CONFIG_SIGNAL]
+    shutters_open: A[SignalR[int], PvSuffix("NoOfShuttersOpenA"), Format.CONFIG_SIGNAL]
+    fill_number: A[SignalR[int], PvSuffix("S:FillNumber"), Format.CONFIG_SIGNAL]
+    orbit_correction: A[SignalR[float], PvSuffix("S:OrbitCorrection:CC"), Format.CONFIG_SIGNAL]
+
+
+# -----------------------------------------------------------------------------
+# :author:    Mark Wolfman
+# :email:     wolfman@anl.gov
+# :copyright: Copyright © 2023, UChicago Argonne, LLC
+#
+# Distributed under the terms of the Argonne National Laboratory Open
+# Source License.
+#
+# The full license is in the file LICENSE.txt, distributed with this
+# software.
+# -----------------------------------------------------------------------------

apstools/_devices_async/tests/test_aps_machine.py

@@ -0,0 +1,47 @@
+import pytest
+
+from ..aps_machine import ApsMachine
+
+
+@pytest.mark.asyncio
+async def test_read_attrs():
+    aps = ApsMachine(name="aps")
+    await aps.connect(mock=True)
+    # Reading
+    read_signals = {"aps-current"}
+    reading = await aps.read()
+    assert set(reading.keys()) == read_signals
+    # Configuration
+    config_signals = {
+        "aps-fill_number",
+        "aps-fill_pattern",
+        "aps-floor_coordinator",
+        "aps-last_problem_message",
+        "aps-last_trip_message",
+        "aps-machine_status",
+        "aps-message6",
+        "aps-message7",
+        "aps-message8",
+        "aps-operating_mode",
+        "aps-operators",
+        "aps-orbit_correction",
+        "aps-shutter_status",
+        "aps-shutters_open",
+    }
+    config = await aps.read_configuration()
+    assert set(config.keys()) == config_signals
+    # Hints
+    assert aps.hints == {}
+
+
+# -----------------------------------------------------------------------------
+# :author:    Mark Wolfman
+# :email:     wolfman@anl.gov
+# :copyright: Copyright © 2023, UChicago Argonne, LLC
+#
+# Distributed under the terms of the Argonne National Laboratory Open
+# Source License.
+#
+# The full license is in the file LICENSE.txt, distributed with this
+# software.
+# -----------------------------------------------------------------------------

pyproject.toml

@@ -80,6 +80,7 @@ dev = [
   "ophyd-registry",
   "pre-commit",
   "pytest",
+  "pytest-asyncio",
   "ruff",
   # databroker 2.0.0b57 needs
   "doct",
@@ -104,7 +105,10 @@ doc = [
   "sphinx-autoapi",
   "sphinx-design",
 ]
-all = ["apstools[dev,doc]"]
+async = [
+  "ophyd-async[ca]",
+]
+all = ["apstools[dev,doc,async]"]
 
 [project.scripts]
 spec2ophyd = "apstools.migration.spec2ophyd:main"