Use config server for OAV config files

conftest.py

@@ -14,7 +14,7 @@
 from tests.devices.test_daq_configuration import MOCK_DAQ_CONFIG_PATH
 from tests.test_data import (
     TEST_DISPLAY_CONFIG,
-    TEST_OAV_ZOOM_LEVELS_XML,
+    TEST_OAV_ZOOM_LEVELS,
 )
 
 from dodal.common.beamlines import beamline_utils
@@ -27,7 +27,7 @@
 
 mock_paths = [
     ("DAQ_CONFIGURATION_PATH", MOCK_DAQ_CONFIG_PATH),
-    ("ZOOM_PARAMS_FILE", TEST_OAV_ZOOM_LEVELS_XML),
+    ("ZOOM_PARAMS_FILE", TEST_OAV_ZOOM_LEVELS),
     ("DISPLAY_CONFIG", TEST_DISPLAY_CONFIG),
     ("LOOK_UPTABLE_DIR", LOOKUP_TABLE_PATH),
 ]

src/dodal/devices/oav/oav_parameters.py

@@ -1,11 +1,13 @@
-import json
 from abc import abstractmethod
 from collections import ChainMap
 from dataclasses import dataclass
-from typing import Any, Generic, TypeVar
+from typing import Any, Generic, Literal, TypedDict, TypeVar
 from xml.etree import ElementTree
 from xml.etree.ElementTree import Element
 
+from daq_config_server.client import ConfigServer
+from daq_config_server.models import DisplayConfig
+
 # GDA currently assumes this aspect ratio for the OAV window size.
 # For some beamline this doesn't affect anything as the actual OAV aspect ratio
 # matches. Others need to take it into account to rescale the values stored in
@@ -36,21 +38,25 @@ def __init__(
     ):
         self.oav_config_json: str = oav_config_json
         self.context = context
+        config_server = ConfigServer(url="https://daq-config.diamond.ac.uk")
 
-        self.global_params, self.context_dicts = self.load_json(self.oav_config_json)
+        self.global_params, self.context_dicts = self.load_json(
+            config_server, self.oav_config_json
+        )
         self.active_params: ChainMap = ChainMap(
             self.context_dicts[self.context], self.global_params
         )
         self.update_self_from_current_context()
 
     @staticmethod
-    def load_json(filename: str) -> tuple[dict[str, Any], dict[str, dict]]:
+    def load_json(
+        config_server: ConfigServer, filename: str
+    ) -> tuple[dict[str, Any], dict[str, dict]]:
         """
-        Loads the json from the specified file, and returns a dict with all the
+        Loads the specified file from the config server, and returns a dict with all the
         individual top-level k-v pairs, and one with all the subdicts.
         """
-        with open(filename) as f:
-            raw_params: dict[str, Any] = json.load(f)
+        raw_params: dict[str, Any] = config_server.get_file_contents(filename, dict)
         global_params = {
             k: raw_params.pop(k)
             for k, v in list(raw_params.items())
@@ -124,18 +130,21 @@ class OAVConfigBase(Generic[ParamType]):
     def __init__(self, zoom_params_file: str):
         self.zoom_params = self._get_zoom_params(zoom_params_file)
 
-    def _get_zoom_params(self, zoom_params_file: str):
-        tree = ElementTree.parse(zoom_params_file)
-        root = tree.getroot()
-        return root.findall(".//zoomLevel")
+    def _get_zoom_params(self, zoom_params_file: str) -> dict:
+        config_server = ConfigServer(url="https://daq-config.diamond.ac.uk")
+        return config_server.get_file_contents(zoom_params_file, dict)[
+            "JCameraManSettings"
+        ]
 
     def _read_zoom_params(self) -> dict:
         um_per_pix = {}
-        for node in self.zoom_params:
-            zoom = str(_get_element_as_float(node, "level"))
-            um_pix_x = _get_element_as_float(node, "micronsPerXPixel")
-            um_pix_y = _get_element_as_float(node, "micronsPerYPixel")
-            um_per_pix[zoom] = (um_pix_x, um_pix_y)
+        zoom_levels: list[dict] = self.zoom_params["levels"]["zoomLevel"]
+        for level in zoom_levels:
+            zoom = level["level"]
+            um_per_pix[zoom] = (
+                float(level["micronsPerXPixel"]),
+                float(level["micronsPerYPixel"]),
+            )
         return um_per_pix
 
     @abstractmethod
@@ -166,19 +175,14 @@ def __init__(
         self.display_config = self._get_display_config(display_config_file)
         super().__init__(zoom_params_file)
 
-    def _get_display_config(self, display_config_file: str):
-        with open(display_config_file) as f:
-            file_lines = f.readlines()
-        return file_lines
+    def _get_display_config(self, display_config_file: str) -> DisplayConfig:
+        config_server = ConfigServer(url="https://daq-config.diamond.ac.uk")
+        return config_server.get_file_contents(display_config_file, DisplayConfig)
 
     def _read_display_config(self) -> dict:
         crosshairs = {}
-        for i in range(len(self.display_config)):
-            if self.display_config[i].startswith("zoomLevel"):
-                zoom = self.display_config[i].split(" = ")[1].strip()
-                x = int(self.display_config[i + 1].split(" = ")[1])
-                y = int(self.display_config[i + 2].split(" = ")[1])
-                crosshairs[zoom] = (x, y)
+        for zoom, values in self.display_config.zoom_levels.items():
+            crosshairs[str(zoom)] = (values.crosshairX, values.crosshairY)
         return crosshairs
 
     def get_parameters(self) -> dict[str, ZoomParamsCrosshair]:

src/dodal/testing/__init__.py

@@ -0,0 +1,4 @@
+from .mock_config_server.mock_config_server import MockConfigServer
+from .mock_config_server.mock_oav_config import MockOavConfig
+
+__all__ = ["MockConfigServer", "MockOavConfig"]

src/dodal/testing/mock_config_server/mock_config_server.py

@@ -0,0 +1,47 @@
+import json
+from pathlib import Path
+from typing import Any
+
+from daq_config_server.client import ConfigServer
+from daq_config_server.models import (
+    ConfigModel,
+)
+
+from .mock_oav_config import MockOavConfig
+
+
+class MockConfigServer(ConfigServer):
+    def get_file_contents(
+        self,
+        file_path: str | Path,
+        desired_return_type: type[Any] = str,
+        reset_cached_result: bool = False,
+    ) -> Any:
+        print()
+        if str(file_path).endswith("display.configuration"):
+            return MockOavConfig.get_display_config()
+        elif str(file_path).endswith("jCameraManZoomLevels.xml"):
+            return MockOavConfig.get_zoom_params_file()
+        elif str(file_path).endswith("OAVCentring.json"):
+            return MockOavConfig.get_oav_config_json()
+        else:
+            return self._fake_config_server_read(
+                file_path, desired_return_type, reset_cached_result
+            )
+
+    def _fake_config_server_read(
+        self,
+        filepath: str | Path,
+        desired_return_type: type[str] | type[dict] | ConfigModel = str,
+        reset_cached_result: bool = False,
+    ) -> Any:
+        filepath = Path(filepath)
+        # Minimal logic required for unit tests
+        with filepath.open("r") as f:
+            contents = f.read()
+            if desired_return_type is str:
+                return contents
+            elif desired_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))

src/dodal/testing/mock_config_server/mock_oav_config.py

@@ -0,0 +1,182 @@
+from daq_config_server.models import (
+    DisplayConfig,
+    DisplayConfigData,
+)
+
+
+class MockOavConfig:
+    @staticmethod
+    def get_oav_config_json():
+        return {
+            "exposure": 0.075,
+            "acqPeriod": 0.05,
+            "gain": 1.0,
+            "minheight": 70,
+            "oav": "OAV",
+            "mxsc_input": "CAM",
+            "min_callback_time": 0.080,
+            "close_ksize": 11,
+            "direction": 0,
+            "pinTipCentring": {
+                "zoom": 1.0,
+                "preprocess": 8,
+                "preProcessKSize": 21,
+                "CannyEdgeUpperThreshold": 20.0,
+                "CannyEdgeLowerThreshold": 5.0,
+                "brightness": 20,
+                "max_tip_distance": 300,
+                "mxsc_input": "proc",
+                "minheight": 10,
+                "min_callback_time": 0.15,
+                "filename": "/dls_sw/prod/R3.14.12.7/support/adPython/2-1-11/adPythonApp/scripts/adPythonMxSampleDetect.py",
+            },
+            "loopCentring": {
+                "zoom": 5.0,
+                "preprocess": 8,
+                "preProcessKSize": 21,
+                "CannyEdgeUpperThreshold": 20.0,
+                "CannyEdgeLowerThreshold": 5.0,
+                "brightness": 20,
+                "filename": "/dls_sw/prod/R3.14.12.7/support/adPython/2-1-11/adPythonApp/scripts/adPythonMxSampleDetect.py",
+                "max_tip_distance": 300,
+                "minheight": 10,
+            },
+            "xrayCentring": {
+                "zoom": 7.5,
+                "preprocess": 8,
+                "preProcessKSize": 31,
+                "CannyEdgeUpperThreshold": 30.0,
+                "CannyEdgeLowerThreshold": 5.0,
+                "close_ksize": 3,
+                "filename": "/dls_sw/prod/R3.14.12.7/support/adPython/2-1-11/adPythonApp/scripts/adPythonMxSampleDetect.py",
+                "brightness": 80,
+            },
+            "rotationAxisAlign": {
+                "zoom": 10.0,
+                "preprocess": 8,
+                "preProcessKSize": 21,
+                "CannyEdgeUpperThreshold": 20.0,
+                "CannyEdgeLowerThreshold": 5.0,
+                "filename": "/dls_sw/prod/R3.14.12.7/support/adPython/2-1-11/adPythonApp/scripts/adPythonMxSampleDetect.py",
+                "brightness": 100,
+            },
+            "SmargonOffsets1": {
+                "zoom": 1.0,
+                "preprocess": 8,
+                "preProcessKSize": 21,
+                "CannyEdgeUpperThreshold": 50.0,
+                "CannyEdgeLowerThreshold": 5.0,
+                "brightness": 80,
+            },
+            "SmargonOffsets2": {
+                "zoom": 5.0,
+                "preprocess": 8,
+                "preProcessKSize": 11,
+                "CannyEdgeUpperThreshold": 50.0,
+                "CannyEdgeLowerThreshold": 5.0,
+                "brightness": 90,
+            },
+        }
+
+    @staticmethod
+    def get_zoom_params_file():
+        return {
+            "JCameraManSettings": {
+                "levels": {
+                    "zoomLevel": [
+                        {
+                            "level": "1.0",
+                            "position": "0",
+                            "micronsPerXPixel": "2.87",
+                            "micronsPerYPixel": "2.87",
+                        },
+                        {
+                            "level": "2.5",
+                            "position": "10",
+                            "micronsPerXPixel": "2.31",
+                            "micronsPerYPixel": "2.31",
+                        },
+                        {
+                            "level": "5.0",
+                            "position": "25",
+                            "micronsPerXPixel": "1.58",
+                            "micronsPerYPixel": "1.58",
+                        },
+                        {
+                            "level": "7.5",
+                            "position": "50",
+                            "micronsPerXPixel": "0.806",
+                            "micronsPerYPixel": "0.806",
+                        },
+                        {
+                            "level": "10.0",
+                            "position": "75",
+                            "micronsPerXPixel": "0.438",
+                            "micronsPerYPixel": "0.438",
+                        },
+                        {
+                            "level": "15.0",
+                            "position": "90",
+                            "micronsPerXPixel": "0.302",
+                            "micronsPerYPixel": "0.302",
+                        },
+                    ]
+                },
+                "tolerance": "1.0",
+            }
+        }
+
+    @staticmethod
+    def get_display_config():
+        return DisplayConfig(
+            zoom_levels={
+                1.0: DisplayConfigData(
+                    crosshairX=477,
+                    crosshairY=359,
+                    topLeftX=383,
+                    topLeftY=253,
+                    bottomRightX=410,
+                    bottomRightY=278,
+                ),
+                2.5: DisplayConfigData(
+                    crosshairX=493,
+                    crosshairY=355,
+                    topLeftX=340,
+                    topLeftY=283,
+                    bottomRightX=388,
+                    bottomRightY=322,
+                ),
+                5.0: DisplayConfigData(
+                    crosshairX=517,
+                    crosshairY=350,
+                    topLeftX=268,
+                    topLeftY=326,
+                    bottomRightX=354,
+                    bottomRightY=387,
+                ),
+                7.5: DisplayConfigData(
+                    crosshairX=549,
+                    crosshairY=437,
+                    topLeftX=248,
+                    topLeftY=394,
+                    bottomRightX=377,
+                    bottomRightY=507,
+                ),
+                10.0: DisplayConfigData(
+                    crosshairX=613,
+                    crosshairY=344,
+                    topLeftX=2,
+                    topLeftY=489,
+                    bottomRightX=206,
+                    bottomRightY=630,
+                ),
+                15.0: DisplayConfigData(
+                    crosshairX=693,
+                    crosshairY=339,
+                    topLeftX=1,
+                    topLeftY=601,
+                    bottomRightX=65,
+                    bottomRightY=767,
+                ),
+            },
+        )

system_tests/test_oav_system.py

@@ -4,7 +4,7 @@
 from ophyd_async.core import init_devices
 from tests.test_data import (
     TEST_DISPLAY_CONFIG,
-    TEST_OAV_ZOOM_LEVELS_XML,
+    TEST_OAV_ZOOM_LEVELS,
 )
 
 from dodal.devices.oav.oav_detector import OAV, OAVConfig
@@ -18,7 +18,7 @@
 
 @pytest.fixture
 async def oav() -> OAV:
-    oav_config = OAVConfig(TEST_OAV_ZOOM_LEVELS_XML, TEST_DISPLAY_CONFIG)
+    oav_config = OAVConfig(TEST_OAV_ZOOM_LEVELS, TEST_DISPLAY_CONFIG)
     async with init_devices(connect=True):
         oav = OAV("", config=oav_config, name="oav")
     return oav
@@ -39,7 +39,7 @@ def take_snapshot_with_grid(oav: OAV, snapshot_filename, snapshot_directory):
 @pytest.mark.skip(reason="Don't want to actually take snapshots during testing.")
 def test_grid_overlay(run_engine: RunEngine):
     beamline = "BL03I"
-    oav_params = OAVConfig(TEST_OAV_ZOOM_LEVELS_XML, TEST_DISPLAY_CONFIG)
+    oav_params = OAVConfig(TEST_OAV_ZOOM_LEVELS, TEST_DISPLAY_CONFIG)
     oav = OAV(name="oav", prefix=f"{beamline}", config=oav_params)
     snapshot_filename = "snapshot"
     snapshot_directory = "."