Migrate BeamlineParameters to use config server

pyproject.toml

@@ -30,7 +30,7 @@ dependencies = [
     "scanspec>=0.7.3",
     "pyzmq==26.3.0",                  # Until we can move to RHEL 8 https://github.com/DiamondLightSource/mx-bluesky/issues/1139
     "deepdiff",
-    "daq-config-server>=v1.0.0", # For getting Configuration settings.
+    "daq-config-server>=v1.1.2", # For getting Configuration settings.
 ]
 
 dynamic = ["version"]

src/dodal/common/beamlines/beamline_parameters.py

@@ -1,7 +1,7 @@
-import ast
-from typing import Any, cast
+from typing import Any
+
+from daq_config_server.client import ConfigServer
 
-from dodal.log import LOGGER
 from dodal.utils import get_beamline_name
 
 BEAMLINE_PARAMETER_KEYWORDS = ["FB", "FULL", "deadtime"]
@@ -24,41 +24,12 @@ def __repr__(self) -> str:
     def __getitem__(self, item: str):
         return self.params[item]
 
-    @classmethod
-    def from_lines(cls, file_name: str, config_lines: list[str]):
-        config_lines_nocomments = [line.split("#", 1)[0] for line in config_lines]
-        config_lines_sep_key_and_value = [
-            # XXX removes all whitespace instead of just trim
-            line.translate(str.maketrans("", "", " \n\t\r")).split("=")
-            for line in config_lines_nocomments
-        ]
-        config_pairs: list[tuple[str, Any]] = [
-            cast(tuple[str, Any], param)
-            for param in config_lines_sep_key_and_value
-            if len(param) == 2
-        ]
-        for i, (param, value) in enumerate(config_pairs):
-            try:
-                # BEAMLINE_PARAMETER_KEYWORDS effectively raw string but whitespace removed
-                if value not in BEAMLINE_PARAMETER_KEYWORDS:
-                    config_pairs[i] = (
-                        param,
-                        cls.parse_value(value),
-                    )
-            except Exception as e:
-                LOGGER.warning(f"Unable to parse {file_name} line {i}: {e}")
-
-        return cls(params=dict(config_pairs))
-
     @classmethod
     def from_file(cls, path: str):
-        with open(path) as f:
-            config_lines = f.readlines()
-        return cls.from_lines(path, config_lines)
-
-    @classmethod
-    def parse_value(cls, value: str):
-        return ast.literal_eval(value.replace("Yes", "True").replace("No", "False"))
+        config_server = ConfigServer(url="https://daq-config.diamond.ac.uk")
+        return cls(
+            params=config_server.get_file_contents(path, dict, reset_cached_result=True)
+        )
 
 
 def get_beamline_parameters(beamline_param_path: str | None = None):

tests/common/beamlines/test_beamline_parameters.py

@@ -8,7 +8,6 @@
     get_beamline_parameters,
 )
 from tests.test_data import (
-    BAD_BEAMLINE_PARAMETERS,
     I04_BEAMLINE_PARAMETERS,
     TEST_BEAMLINE_PARAMETERS_TXT,
 )
@@ -34,40 +33,6 @@ def test_i03_beamline_parameters():
     ]
 
 
-@patch("dodal.common.beamlines.beamline_parameters.LOGGER")
-def test_parse_exception_causes_warning(mock_logger):
-    params = GDABeamlineParameters.from_file(BAD_BEAMLINE_PARAMETERS)
-    assert params["flux_predict_polynomial_coefficients_5"] == [
-        -0.0000707134131045123,
-        7.0205491504418,
-        -194299.6440518530,
-        1835805807.3974800,
-        -3280251055671.100,
-    ]
-    mock_logger.warning.assert_called_once()
-
-    params = GDABeamlineParameters.from_file(BAD_BEAMLINE_PARAMETERS)
-    assert params["flux_predict_polynomial_coefficients_5"] == [
-        -0.0000707134131045123,
-        7.0205491504418,
-        -194299.6440518530,
-        1835805807.3974800,
-        -3280251055671.100,
-    ]
-
-
-def test_parse_list():
-    test_data = [([1, 2, 3], "[1, 2, 3]"), ([1, True, 3], "[1, Yes, 3]")]
-    for expected, input in test_data:
-        actual = GDABeamlineParameters.parse_value(input)
-        assert expected == actual, f"Actual:{actual}, expected: {expected}\n"
-
-
-def test_parse_list_raises_exception():
-    with pytest.raises(SyntaxError):
-        GDABeamlineParameters.parse_value("[1, 2")
-
-
 def test_get_beamline_parameters_works_with_no_environment_variable_set():
     if environ.get("BEAMLINE"):
         del environ["BEAMLINE"]
@@ -108,32 +73,6 @@ def test_get_beamline_parameters_raises_error_when_beamline_not_found(
         get_beamline_parameters()
 
 
-def test_parse_nested_list():
-    actual = GDABeamlineParameters.parse_value("[[1, 2], [3, 4]]")
-    expected = [[1, 2], [3, 4]]
-    assert actual == expected
-
-
-def test_parse_nested_nested_list():
-    actual = GDABeamlineParameters.parse_value("[[[1, 2], [3, 4]], [[5, 6], [7, 8]]]")
-    expected = [[[1, 2], [3, 4]], [[5, 6], [7, 8]]]
-    assert actual == expected
-
-
-def test_leading_comma_in_list_causes_error():
-    with pytest.raises(SyntaxError):
-        GDABeamlineParameters.parse_value("[,1, 2, 3, 4]")
-    with pytest.raises(SyntaxError):
-        GDABeamlineParameters.parse_value("[[1, 2], [ ,3, 4]]")
-
-
-def test_Yes_and_No_replaced_with_bool_values():  # noqa: N802
-    value = "[Yes, No, True, False, 0, 1]"
-    expected = [True, False, True, False, 0, 1]
-    actual = GDABeamlineParameters.parse_value(value)
-    assert actual == expected
-
-
 @pytest.fixture(autouse=True)
 def i03_beamline_parameters():
     with patch.dict(

tests/conftest.py

@@ -1,4 +1,5 @@
 import importlib
+import json
 import logging
 import os
 import sys
@@ -8,6 +9,7 @@
 from unittest.mock import MagicMock, patch
 
 import pytest
+from daq_config_server.models import ConfigModel
 from ophyd_async.core import PathProvider
 
 from dodal.common.beamlines import beamline_parameters, beamline_utils
@@ -159,3 +161,33 @@ def eiger_params(tmp_path: Path) -> DetectorParams:
         det_dist_to_beam_converter_path=TEST_LUT_TXT,
         detector_size_constants=EIGER2_X_16M_SIZE.det_type_string,  # type: ignore
     )
+
+
+def _fake_config_server_get_file_contents(
+    filepath: str | Path,
+    desired_return_type: type[str] | type[dict] | ConfigModel = str,
+    reset_cached_result: bool = True,
+):
+    filepath = Path(filepath)
+    # Minimal logic required for unit tests
+    with filepath.open("r") as f:
+        contents = f.read()
+        print(contents)
+        if desired_return_type is str:
+            return contents
+        elif desired_return_type is dict:
+            print("return type is dict")
+            return json.loads(contents)
+        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
+
+    with patch(
+        "daq_config_server.client.ConfigServer.get_file_contents",
+        side_effect=_fake_config_server_get_file_contents,
+    ):
+        yield