Read beamline parameters through config server

pyproject.toml

@@ -52,12 +52,12 @@ dependencies = [
     # These dependencies may be issued as pre-release versions and should have a pin constraint
     # as by default pip-install will not upgrade to a pre-release.
     #
-    "daq-config-server>=v1.0.0-rc.2",
+    "daq-config-server>=v1.1.2",
     "blueapi >= 1.8.0",
     "ophyd >= 1.10.5",
     "ophyd-async >= 0.14.0",
     "bluesky >= 1.14.6",
-    "dls-dodal @ git+https://github.com/DiamondLightSource/dodal.git@main",
+    "dls-dodal @ git+https://github.com/DiamondLightSource/dodal.git@mx_bluesky_1504_migrate_beamline_parameters_to_config_server",
 ]
 
 

src/mx_bluesky/common/experiment_plans/beamstop_check.py

@@ -1,7 +1,8 @@
+from typing import Any
+
 import pydantic
 from bluesky import plan_stubs as bps
 from bluesky.utils import MsgGenerator
-from dodal.common.beamlines.beamline_parameters import GDABeamlineParameters
 from dodal.devices.aperturescatterguard import ApertureScatterguard, ApertureValue
 from dodal.devices.attenuator.attenuator import BinaryFilterAttenuator
 from dodal.devices.backlight import Backlight
@@ -50,7 +51,7 @@ class BeamObstructedError(BeamlineCheckFailureError): ...
 
 def move_beamstop_in_and_verify_using_diode(
     devices: BeamstopCheckDevices,
-    beamline_parameters: GDABeamlineParameters,
+    beamline_parameters: dict[str, Any],
     detector_min_z_mm: float,
     detector_max_z_mm: float,
 ) -> MsgGenerator:

src/mx_bluesky/hyperion/experiment_plans/udc_default_state.py

@@ -1,6 +1,7 @@
 import bluesky.plan_stubs as bps
 import pydantic
 from bluesky.utils import MsgGenerator
+from dodal.beamlines.i03 import BL
 from dodal.common.beamlines.beamline_parameters import (
     get_beamline_parameters,
 )
@@ -112,7 +113,7 @@ def move_to_udc_default_state(devices: UDCDefaultDevices):
         get_hyperion_config_client().get_feature_flags()
     )
     if feature_flags.BEAMSTOP_DIODE_CHECK:
-        beamline_parameters = get_beamline_parameters()
+        beamline_parameters = get_beamline_parameters(BL)
         config_client = get_hyperion_config_client()
         features_settings: HyperionFeatureSettings = config_client.get_feature_flags()
         detector_min_z = features_settings.DETECTOR_DISTANCE_LIMIT_MIN_MM

tests/conftest.py

@@ -18,11 +18,10 @@
 import pytest
 from bluesky.simulators import RunEngineSimulator
 from bluesky.utils import Msg, MsgGenerator
+from daq_config_server.models import ConfigModel
 from dodal.beamlines import aithre, i03
 from dodal.common.beamlines import beamline_utils
-from dodal.common.beamlines.beamline_parameters import (
-    GDABeamlineParameters,
-)
+from dodal.common.beamlines.beamline_parameters import get_beamline_parameters
 from dodal.common.beamlines.beamline_utils import clear_devices
 from dodal.common.beamlines.commissioning_mode import set_commissioning_signal
 from dodal.devices.aperturescatterguard import (
@@ -363,8 +362,8 @@ def _pass_on_mock(value: float, wait: bool):
 
 @pytest.fixture
 def beamline_parameters():
-    return GDABeamlineParameters.from_file(
-        "tests/test_data/test_beamline_parameters.txt"
+    return get_beamline_parameters(
+        "test_beamline", "tests/test_data/test_beamline_parameters.txt"
     )
 
 
@@ -574,7 +573,7 @@ async def fake_attenuator_set(val):
 
 @pytest.fixture
 def beamstop_phase1(
-    beamline_parameters: GDABeamlineParameters,
+    beamline_parameters: dict[str, Any],
     sim_run_engine: RunEngineSimulator,
 ) -> Generator[Beamstop, Any, Any]:
     with patch(
@@ -1706,10 +1705,30 @@ def assert_images_pixelwise_equal(actual, expected):
             )
 
 
-def _fake_config_server_read(
+IMPLEMENTED_CONFIG_CLIENTS: list[Callable] = [
+    get_hyperion_config_client,
+    get_i04_config_client,
+]
+
+
+@pytest.fixture(autouse=True)
+def mock_mx_config_server():
+    # Don't actually talk to central service during unit tests, and reset caches between test
+
+    for client in IMPLEMENTED_CONFIG_CLIENTS:
+        client.cache_clear()  # type: ignore - currently no option for "cachable" static type
+
+    with patch(
+        "mx_bluesky.common.external_interaction.config_server.MXConfigClient.get_file_contents",
+        side_effect=_fake_config_server_get_file_contents,
+    ):
+        yield
+
+
+def _fake_config_server_get_file_contents(
     filepath: str | Path,
-    desired_return_type: type[str] | type[dict] = str,
-    reset_cached_result=False,
+    desired_return_type: type[str] | type[dict] | ConfigModel = str,
+    reset_cached_result: bool = True,
 ):
     filepath = Path(filepath)
     # Minimal logic required for unit tests
@@ -1719,24 +1738,17 @@ def _fake_config_server_read(
             return contents
         elif desired_return_type is dict:
             return json.loads(contents)
-
-
-IMPLEMENTED_CONFIG_CLIENTS: list[Callable] = [
-    get_hyperion_config_client,
-    get_i04_config_client,
-]
+        elif issubclass(desired_return_type, ConfigModel):  # type: ignore
+            return desired_return_type.model_validate(json.loads(contents))
 
 
 @pytest.fixture(autouse=True)
 def mock_config_server():
     # Don't actually talk to central service during unit tests, and reset caches between test
 
-    for client in IMPLEMENTED_CONFIG_CLIENTS:
-        client.cache_clear()  # type: ignore - currently no option for "cachable" static type
-
     with patch(
-        "mx_bluesky.common.external_interaction.config_server.MXConfigClient.get_file_contents",
-        side_effect=_fake_config_server_read,
+        "daq_config_server.client.ConfigServer.get_file_contents",
+        side_effect=_fake_config_server_get_file_contents,
     ):
         yield