Skip RE tests when not on beamline network

pyproject.toml

@@ -11,9 +11,7 @@ classifiers = [
     "Programming Language :: Python :: 3.12",
 ]
 description = "Bluesky plans to be run on Diamond's test rigs e.g. ViSR, P45, etc"
-dependencies = [
-    "dls-dodal @ git+https://github.com/DiamondLightSource/dodal.git"
-]
+dependencies = ["dls-dodal>=1.55.1"]
 dynamic = ["version"]
 license.file = "LICENSE"
 readme = "README.md"

src/test_rig_bluesky/plans.py

@@ -1,3 +1,4 @@
+from dataclasses import dataclass
 from typing import Any
 
 from bluesky import plan_stubs as bps
@@ -9,13 +10,25 @@
 from dodal.plans import spec_scan
 from ophyd_async.core import TriggerInfo
 from ophyd_async.epics.adaravis import AravisDetector
+from ophyd_async.epics.adcore import NDAttributePv, NDAttributePvDbrType
+from ophyd_async.epics.adcore._core_io import NDROIStatNIO
+from ophyd_async.plan_stubs import setup_ndattributes
 from scanspec.specs import Line, Spec
 
 imaging_detector = inject("imaging_detector")
 spectroscopy_detector = inject("spectroscopy_detector")
 sample_stage = inject("sample_stage")
 
 
+@dataclass
+class ROI:
+    channel: int
+    name: str
+    start_x: int
+    start_y: int
+    size: int
+
+
 @attach_data_session_metadata_decorator()
 def snapshot(
     imaging_detector: AravisDetector = imaging_detector,
@@ -37,6 +50,52 @@ def spectroscopy(
     yield from bps.prepare(
         spectroscopy_detector, TriggerInfo(livetime=exposure_time), wait=True
     )
+    # TODO: This would be nicer if NDArrayBaseIO had a PortName signal
+    yield from bps.abs_set(
+        spectroscopy_detector.fileio.nd_array_port, "D2.roistat", wait=True
+    )
+
+    rois = [
+        ROI(2, "Green", 880, 605, 150),
+        ROI(3, "Blue", 1665, 600, 150),
+        ROI(1, "Red", 95, 610, 150),
+    ]
+
+    params: list[NDAttributePv] = []
+    for roi in rois:
+        roistatn = spectroscopy_detector.roistat.channels[roi.channel]  # type: ignore
+        assert isinstance(roistatn, NDROIStatNIO)
+
+        yield from bps.mv(
+            *(roistatn.name_, roi.name),
+            *(roistatn.min_x, roi.start_x),
+            *(roistatn.min_y, roi.start_y),
+            *(roistatn.size_x, roi.size),
+            *(roistatn.size_y, roi.size),
+            *(roistatn.use, True),
+            wait=True,
+        )
+
+        # We can't include all the channels as it makes the xml longer than 256
+        # params.append(
+        params = [
+            NDAttributePv(
+                name=f"{roi.name}Total",
+                signal=roistatn.total,
+                dbrtype=NDAttributePvDbrType.DBR_LONG,
+                description=f"Sum of {roi.name} channel",
+            )
+        ]
+        # )
+
+    yield from setup_ndattributes(spectroscopy_detector.roistat, params)  # type: ignore
+
+    for motor in [sample_stage.x, sample_stage.y]:
+        yield from bps.mv(
+            *(motor.acceleration_time, 0.001),
+            *(motor.velocity, 100),
+            wait=True,
+        )
 
     spec = spec or Line(sample_stage.x, 0, 5, 5)
 

tests/system_tests/conftest.py

@@ -1,4 +1,6 @@
 import os
+import socket
+import textwrap
 from collections.abc import Generator
 from pathlib import Path
 
@@ -12,6 +14,35 @@
 
 PROJECT_ROOT = Path(__file__).parent.parent.parent
 
+BEAMLINE_HOSTS = [
+    "b01-1-ws001.diamond.ac.uk",
+    "b01-1-control.diamond.ac.uk",
+    "bl01c-ea-serv-01.diamond.ac.uk",
+    "bl01c-di-serv-01.diamond.ac.uk",
+]
+
+
+def pytest_configure(config: pytest.Config):
+    config.addinivalue_line(
+        "markers", "control_system: test requires direct access to the control system"
+    )
+
+
+def pytest_runtest_setup(item: pytest.Item):
+    if next(item.iter_markers(name="control_system"), None) is not None:
+        if not on_controllable_machine():
+            pytest.skip(
+                reason=textwrap.dedent(f"""
+                    This test needs direct access to the control system and can only
+                    be run from one of the following machines: {BEAMLINE_HOSTS}
+                """)
+            )
+
+
+def on_controllable_machine() -> bool:
+    hostname = socket.gethostname()
+    return hostname in BEAMLINE_HOSTS
+
 
 @pytest.fixture
 def instrument() -> str:

tests/system_tests/test_plans_system.py

@@ -1,5 +1,10 @@
+import pytest
 from blueapi.service.model import TaskRequest
+from bluesky import RunEngine
+from dodal.beamlines.b01_1 import sample_stage, spectroscopy_detector
+from scanspec.specs import Line
 
+from test_rig_bluesky.plans import spectroscopy
 from test_rig_bluesky.testing import BlueskyPlanRunner
 
 
@@ -29,3 +34,14 @@ def test_spectroscopy(
         timeout=10,
     )
     assert events["FINISHED"][0]["scanDimensions"] == [5]
+
+
+@pytest.mark.control_system
+def test_spectroscopy_re():
+    RE = RunEngine()
+    _spectroscopy_detector = spectroscopy_detector(connect_immediately=True)
+    _sample_stage = sample_stage(connect_immediately=True)
+
+    scan_spec = Line(_sample_stage.x, 2, 5, 30) * Line(_sample_stage.y, 0, 5, 50)
+
+    RE(spectroscopy(_spectroscopy_detector, _sample_stage, scan_spec))