Unify p38 and i22, swap out with env variables

src/dodal/beamlines/i22.py

@@ -11,7 +11,10 @@
 )
 from dodal.common.beamlines.beamline_utils import set_beamline as set_utils_beamline
 from dodal.common.beamlines.device_helpers import numbered_slits
-from dodal.common.visit import RemoteDirectoryServiceClient, StaticVisitPathProvider
+from dodal.common.visit import (
+    RemoteDirectoryServiceClient,
+    StaticVisitPathProvider,
+)
 from dodal.devices.focusing_mirror import FocusingMirror
 from dodal.devices.i22.dcm import CrystalMetadata, DoubleCrystalMonochromator
 from dodal.devices.i22.fswitch import FSwitch
@@ -28,13 +31,24 @@
 set_log_beamline(BL)
 set_utils_beamline(BL)
 
+_LAB_NAME = "p38"
+
+IS_LAB = BL == _LAB_NAME
+
+LINKAM_IS_IN_LAB = False
 # Currently we must hard-code the visit, determining the visit at runtime requires
 # infrastructure that is still WIP.
 # Communication with GDA is also WIP so for now we determine an arbitrary scan number
 # locally and write the commissioning directory. The scan number is not guaranteed to
 # be unique and the data is at risk - this configuration is for testing only.
 set_path_provider(
     StaticVisitPathProvider(
+        BL,
+        Path("/dls/p38/data/2024/cm37282-2/bluesky"),
+        client=RemoteDirectoryServiceClient("https://p38-control:8088/api"),
+    )
+    if IS_LAB
+    else StaticVisitPathProvider(
         BL,
         Path("/dls/i22/data/2024/cm37271-2/bluesky"),
         client=RemoteDirectoryServiceClient("http://i22-control:8088/api"),
@@ -44,7 +58,28 @@
 
 def saxs(
     wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
-) -> PilatusDetector:
+) -> NXSasOAV | NXSasPilatus:
+    if IS_LAB:
+        return device_instantiation(
+            NXSasOAV,
+            "saxs",
+            "-DI-DCAM-03:",
+            wait_for_connection,
+            fake_with_ophyd_sim,
+            drv_suffix="DET:",
+            hdf_suffix="HDF5:",
+            metadata_holder=NXSasMetadataHolder(
+                x_pixel_size=(1.72e-1, "mm"),
+                y_pixel_size=(1.72e-1, "mm"),
+                description="Dectris Pilatus3 2M",
+                type="Photon Counting Hybrid Pixel",
+                sensor_material="silicon",
+                sensor_thickness=(0.45, "mm"),
+                distance=(4711.833684146172, "mm"),
+            ),
+            directory_provider=get_path_provider(),
+        )
+
     return device_instantiation(
         NXSasPilatus,
         "saxs",
@@ -66,8 +101,9 @@ def saxs(
     )
 
 
+@skip_device(lambda: BL == _LAB_NAME)
 def synchrotron(
-    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = IS_LAB
 ) -> Synchrotron:
     return device_instantiation(
         Synchrotron,
@@ -78,9 +114,31 @@ def synchrotron(
     )
 
 
+# d12 at p38
 def waxs(
     wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
 ) -> PilatusDetector:
+    if IS_LAB:
+        return device_instantiation(
+            AravisDetector,
+            "waxs",
+            "-DI-DCAM-04:",
+            wait_for_connection,
+            fake_with_ophyd_sim,
+            drv_suffix="DET:",
+            hdf_suffix="HDF5:",
+            metadata_holder=NXSasMetadataHolder(
+                x_pixel_size=(1.72e-1, "mm"),
+                y_pixel_size=(1.72e-1, "mm"),
+                description="Dectris Pilatus3 2M",
+                type="Photon Counting Hybrid Pixel",
+                sensor_material="silicon",
+                sensor_thickness=(0.45, "mm"),
+                distance=(175.4199417092314, "mm"),
+            ),
+            directory_provider=get_path_provider(),
+        )
+
     return device_instantiation(
         NXSasPilatus,
         "waxs",
@@ -109,17 +167,24 @@ def i0(
     return device_instantiation(
         TetrammDetector,
         "i0",
-        "-EA-XBPM-02:",
+        f"-EA-XBPM-0{1 if IS_LAB else 2}:",
         wait_for_connection,
         fake_with_ophyd_sim,
         type="Cividec Diamond XBPM",
         path_provider=get_path_provider(),
     )
 
 
+#
+# The following devices are fake by default since P38 has no optics,
+# but having mock devices here means they will be reflected in downstream data
+# processing, where they may be required.
+#
+
+
 def it(
     wait_for_connection: bool = True,
-    fake_with_ophyd_sim: bool = False,
+    fake_with_ophyd_sim: bool = IS_LAB,
 ) -> TetrammDetector:
     return device_instantiation(
         TetrammDetector,
@@ -134,7 +199,7 @@ def it(
 
 def vfm(
     wait_for_connection: bool = True,
-    fake_with_ophyd_sim: bool = False,
+    fake_with_ophyd_sim: bool = IS_LAB,
 ) -> FocusingMirror:
     return device_instantiation(
         FocusingMirror,
@@ -147,7 +212,7 @@ def vfm(
 
 def hfm(
     wait_for_connection: bool = True,
-    fake_with_ophyd_sim: bool = False,
+    fake_with_ophyd_sim: bool = IS_LAB,
 ) -> FocusingMirror:
     return device_instantiation(
         FocusingMirror,
@@ -160,7 +225,7 @@ def hfm(
 
 def dcm(
     wait_for_connection: bool = True,
-    fake_with_ophyd_sim: bool = False,
+    fake_with_ophyd_sim: bool = IS_LAB,
 ) -> DoubleCrystalMonochromator:
     return device_instantiation(
         DoubleCrystalMonochromator,
@@ -188,7 +253,7 @@ def dcm(
 
 def undulator(
     wait_for_connection: bool = True,
-    fake_with_ophyd_sim: bool = False,
+    fake_with_ophyd_sim: bool = IS_LAB,
 ) -> Undulator:
     return device_instantiation(
         Undulator,
@@ -205,7 +270,7 @@ def undulator(
 
 def slits_1(
     wait_for_connection: bool = True,
-    fake_with_ophyd_sim: bool = False,
+    fake_with_ophyd_sim: bool = IS_LAB,
 ) -> Slits:
     return numbered_slits(
         1,
@@ -216,7 +281,7 @@ def slits_1(
 
 def slits_2(
     wait_for_connection: bool = True,
-    fake_with_ophyd_sim: bool = False,
+    fake_with_ophyd_sim: bool = IS_LAB,
 ) -> Slits:
     return numbered_slits(
         2,
@@ -227,7 +292,7 @@ def slits_2(
 
 def slits_3(
     wait_for_connection: bool = True,
-    fake_with_ophyd_sim: bool = False,
+    fake_with_ophyd_sim: bool = IS_LAB,
 ) -> Slits:
     return numbered_slits(
         3,
@@ -238,7 +303,7 @@ def slits_3(
 
 def slits_4(
     wait_for_connection: bool = True,
-    fake_with_ophyd_sim: bool = False,
+    fake_with_ophyd_sim: bool = IS_LAB,
 ) -> Slits:
     return numbered_slits(
         4,
@@ -249,7 +314,7 @@ def slits_4(
 
 def slits_5(
     wait_for_connection: bool = True,
-    fake_with_ophyd_sim: bool = False,
+    fake_with_ophyd_sim: bool = IS_LAB,
 ) -> Slits:
     return numbered_slits(
         5,
@@ -260,7 +325,7 @@ def slits_5(
 
 def slits_6(
     wait_for_connection: bool = True,
-    fake_with_ophyd_sim: bool = False,
+    fake_with_ophyd_sim: bool = IS_LAB,
 ) -> Slits:
     return numbered_slits(
         6,
@@ -271,7 +336,7 @@ def slits_6(
 
 def fswitch(
     wait_for_connection: bool = True,
-    fake_with_ophyd_sim: bool = False,
+    fake_with_ophyd_sim: bool = IS_LAB,
 ) -> FSwitch:
     return device_instantiation(
         FSwitch,
@@ -287,6 +352,7 @@ def fswitch(
 
 # Must document what PandAs are physically connected to
 # See: https://github.com/bluesky/ophyd-async/issues/284
+@skip_device(lambda: IS_LAB)
 def panda1(
     wait_for_connection: bool = True,
     fake_with_ophyd_sim: bool = False,
@@ -301,7 +367,7 @@ def panda1(
     )
 
 
-@skip_device()
+@skip_device(lambda: IS_LAB)
 def panda2(
     wait_for_connection: bool = True,
     fake_with_ophyd_sim: bool = False,
@@ -316,7 +382,7 @@ def panda2(
     )
 
 
-@skip_device()
+@skip_device(lambda: IS_LAB)
 def panda3(
     wait_for_connection: bool = True,
     fake_with_ophyd_sim: bool = False,
@@ -331,7 +397,7 @@ def panda3(
     )
 
 
-@skip_device()
+@skip_device(lambda: BL == _LAB_NAME)
 def panda4(
     wait_for_connection: bool = True,
     fake_with_ophyd_sim: bool = False,
@@ -346,31 +412,42 @@ def panda4(
     )
 
 
+# d3 at p38
 def oav(
     wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
 ) -> AravisDetector:
     return device_instantiation(
-        NXSasOAV,
+        AravisDetector if IS_LAB else NXSasOAV,
         "oav",
-        "-DI-OAV-01:",
+        f"-DI-{'DCAM' if IS_LAB else 'OAV'}-01:",
         wait_for_connection,
         fake_with_ophyd_sim,
         drv_suffix="DET:",
         hdf_suffix="HDF5:",
-        metadata_holder=NXSasMetadataHolder(
-            x_pixel_size=(3.45e-3, "mm"),  # Double check this figure
+        metadata_holder=None
+        if IS_LAB
+        else NXSasMetadataHolder(
+            x_pixel_size=(3.45e-3, "mm"),
             y_pixel_size=(3.45e-3, "mm"),
             description="AVT Mako G-507B",
             distance=(-1.0, "m"),
         ),
-        path_provider=get_path_provider(),
+        directory_provider=get_path_provider(),
     )
 
 
-@skip_device()
+@skip_device(lambda: IS_LAB != LINKAM_IS_IN_LAB)
 def linkam(
     wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
 ) -> Linkam3:
+    if IS_LAB:
+        return device_instantiation(
+            Linkam3,
+            "linkam",
+            "-EA-LINKM-02",
+            wait_for_connection,
+            fake_with_ophyd_sim,
+        )
     return device_instantiation(
         Linkam3,
         "linkam",

src/dodal/cli.py

@@ -4,7 +4,10 @@
 from bluesky.run_engine import RunEngine
 from ophyd_async.core import NotConnected
 
-from dodal.beamlines import all_beamline_names, module_name_for_beamline
+from dodal.beamlines import (
+    all_beamline_names,
+    module_name_for_beamline,
+)
 from dodal.utils import make_all_devices
 
 from . import __version__
@@ -51,7 +54,6 @@ def connect(beamline: str, all: bool, sim_backend: bool) -> None:
     # We need to make a RunEngine to allow ophyd-async devices to connect.
     # See https://blueskyproject.io/ophyd-async/main/explanations/event-loop-choice.html
     RunEngine()
-
     print(f"Attempting connection to {beamline} (using {full_module_path})")
     devices, exceptions = make_all_devices(
         full_module_path,

tests/beamlines/unit_tests/test_i22.py

@@ -0,0 +1,14 @@
+from dodal.beamlines import i22
+from dodal.common.beamlines import beamline_utils
+from dodal.devices.i22.nxsas import NXSasPilatus
+
+
+def test_list():
+    beamline_utils.clear_devices()
+    i22.synchrotron(wait_for_connection=False, fake_with_ophyd_sim=True)
+    saxs = i22.saxs(wait_for_connection=False, fake_with_ophyd_sim=True)
+    assert (
+        saxs.__class__ == NXSasPilatus
+    ), f"Expected NXSasPilatus, got {saxs.__class__}"
+
+    assert beamline_utils.list_active_devices() == ["synchrotron", "saxs"]

tests/beamlines/unit_tests/test_p38.py

@@ -0,0 +1,32 @@
+import os
+
+import pytest
+
+from dodal.common.beamlines import beamline_utils
+from dodal.devices.i22.nxsas import NXSasOAV
+
+os.environ["BEAMLINE"] = "p38"
+from dodal.beamlines import i22
+
+
+def test_devices_diff_when_in_lab():
+    beamline_utils.clear_devices()
+    saxs = i22.saxs(wait_for_connection=False, fake_with_ophyd_sim=True)
+    assert saxs.__class__ == NXSasOAV, f"Expected NXSasOav, got {saxs.__class__}"
+
+
+# todo fix the following tests
+# FAILED tests/beamlines/unit_tests/test_i03.py::test_list - KeyError: "No beamline parameter path found, maybe 'BEAMLINE' environment variable is not set!"
+# FAILED tests/beamlines/unit_tests/test_p38.py::test_devices_diff_when_in_lab - AssertionError: Expected NXSasOav, got <class 'dodal.devices.i22.nxsas.NXSasPilatus'>
+# assert <class 'dodal.devices.i22.nxsas.NXSasPilatus'> == NXSasOAV
+#  +  where <class 'dodal.devices.i22.nxsas.NXSasPilatus'> = <dodal.devices.i22.nxsas.NXSasPilatus object at 0x7f016ca5b340>.__class__
+
+
+@pytest.mark.parametrize("module_and_devices_for_beamline", ["p38"], indirect=True)
+def test_device_creation(RE, module_and_devices_for_beamline):
+    _, devices = module_and_devices_for_beamline
+    saxs: NXSasOAV = devices["saxs"]  # type: ignore
+
+    print(saxs)
+    assert saxs.prefix == "BL24I-MO-VGON-01:"
+    assert saxs.kappa.prefix == "BL24I-MO-VGON-01:KAPPA"