Start on plans for controlling backlight

pyproject.toml

@@ -27,7 +27,7 @@ dependencies = [
     "pandablocks",
     "scipy",
     "tiled==0.1.0a91",
-    "ophyd_async>=0.3.4",
+    "ophyd_async>=0.9.0a2",
 ]
 dynamic = ["version"]
 license.file = "LICENSE"
@@ -44,6 +44,7 @@ dev = [
     "pydata-sphinx-theme>=0.12",
     "pytest",
     "pytest-cov",
+    "pytest-asyncio",
     "ruff",
     "sphinx-autobuild",
     "sphinx-copybutton",
@@ -80,9 +81,8 @@ addopts = """
 filterwarnings = "error"
 # Doctest python code in docs, python code in src docstrings, test functions in tests
 testpaths = "docs tests"
-env= [
-    "BEAMLINE=",
-]
+env = ["BEAMLINE="]
+asyncio_mode = "auto"
 
 [tool.coverage.run]
 data_file = "/tmp/htss_rig_bluesky.coverage"

src/htss_rig_bluesky/plans/backlight.py

@@ -0,0 +1,47 @@
+import bluesky.plan_stubs as bps
+from bluesky.utils import MsgGenerator
+from ophyd_async.fastcs.panda import HDFPanda
+
+
+def set_backlight_intensity(
+    panda: HDFPanda,
+    intensity: float,
+    wait: bool = True,
+    group: str | None = None,
+) -> MsgGenerator[None]:
+    if 0.0 <= intensity <= 1.0:
+        duty_cycle = int(round(intensity * 100.0))
+        if duty_cycle == 0:
+            port = "ZERO"
+        elif duty_cycle == 100:
+            port = "ONE"
+        else:
+            port = yield from create_pwm(panda, duty_cycle, wait, group)
+
+        yield from set_backlight_control_port(panda, port, wait, group)
+    else:
+        raise ValueError(f"Given intensity {intensity} should be between 0.0 and 1.0")
+
+
+def set_backlight_control_port(
+    panda: HDFPanda,
+    port: str,
+    wait: bool = True,
+    group: str | None = None,
+) -> MsgGenerator[None]:
+    ttl_out_2 = panda.ttlout[2]
+    yield from bps.abs_set(
+        ttl_out_2.val,
+        port,
+        wait=wait,
+        group=group,
+    )
+
+
+def create_pwm(
+    panda: HDFPanda,
+    duty_cycle: int,
+    wait: bool = True,
+    group: str | None = None,
+) -> MsgGenerator[str]:
+    return NotImplemented

tests/unit_tests/plans/test_backlight.py

@@ -0,0 +1,44 @@
+import pytest
+from bluesky import RunEngine
+from dodal.beamlines import training_rig as training_rig
+from ophyd_async.core import (
+    Device,
+    DeviceVector,
+)
+from ophyd_async.epics.core import epics_signal_rw
+from ophyd_async.fastcs.panda import HDFPanda
+from ophyd_async.plan_stubs import ensure_connected
+
+from htss_rig_bluesky.plans.backlight import (
+    set_backlight_control_port,
+)
+
+
+@pytest.fixture
+def run_engine() -> RunEngine:
+    return RunEngine()
+
+
+@pytest.fixture
+async def mock_panda(
+    run_engine: RunEngine,
+):
+    class TtlOutBlock(Device):
+        def __init__(self, name: str = ""):
+            self.val = epics_signal_rw(str, "VAL")
+            super().__init__(name)
+
+    mock_panda = training_rig.panda()
+    mock_panda.phase_1_signal_units = epics_signal_rw(int, "")
+    mock_panda.ttlout = DeviceVector({i: TtlOutBlock() for i in range(1, 5)})
+
+    run_engine(ensure_connected(mock_panda, mock=True))
+    yield mock_panda
+
+
+async def test_set_control_port_sets_control_port(
+    run_engine: RunEngine, mock_panda: HDFPanda
+):
+    assert (await mock_panda.ttlout[2].val.get_value()) == ""
+    run_engine(set_backlight_control_port(mock_panda, "FOO"), wait=True)
+    assert (await mock_panda.ttlout[2].val.get_value()) == "FOO"