Add file converters

docs/how-to/config-server-guide.md

@@ -15,7 +15,14 @@ You can then make a request through this client through its `get_file_contents`
 ```python
 config_server.get_file_contents(FILE_PATH,reset_cached_result=True)
 ```
-By default, this will return the file's raw string output - which includes things like linebreaks. It is up to you to format this output - using `splitlines()` will get you a list where each item is a line in the file. `get_file_contents` has a `desired_return_type` parameter where you can instead ask for a `dict` or `bytes`. The server will try and do the conversion, and respond with an HTTP error if the requested file is incompatible with that type.
+By default, this will return the file's raw string output - which includes things like linebreaks. It is up to you to format this output - using `splitlines()` will get you a list where each item is a line in the file. `get_file_contents` has a `desired_return_type` parameter where you can instead ask for a `dict`, `bytes` or a one of the `pydantic models` defined in the server. The server will try and do the conversion, and respond with an HTTP error if the requested file is incompatible with that type, or a pydantic validation error if the data cannot be converted to the pydantic model provided. Custom [file converters](#file-converters) can be used to specify how a file should be converted to a dict or pydantic model.
+
+(file-converters)=
+# File converters
+
+Converters can be used to turn a file into a standard format server-side, reducing the complexity of reading config files client-side. Converters can convert config files to a `dict` or `pydantic model`, and the same `pydantic model` can be reconstructed client-side by the `get_file_contents` method. Available converters exist [here](https://github.com/DiamondLightSource/daq-config-server/blob/main/converters._converters.py) - see if there's a suitable converter you can use before adding your own. [This dictionary](https://github.com/DiamondLightSource/daq-config-server/blob/main/converters._file_converter_map.py) maps files to converters. Add the path of your config file and a suitable converter to this dictionary and it will automatically be used by the config server when a request for that file is made.
+
+A request for `str` or `bytes` will fetch the raw file with no conversion.
 
 # Adding files to the whitelist
 

docs/reference/current_and_planned_features.md

@@ -6,11 +6,11 @@
 - Periodically check the main branch's to update the whitelist.
 - Provide a client module for users to easily communicate with the server, with caching.
 - Have this service hosted on diamond's central argus cluster - with url `https://daq-config-server.diamond.ac.uk`
+- Provide server-side json and pydantic model formatting for commonly used configuration files - eg `beamline_parameters.txt` should be returned as a dictionary or pydantic model.
 
 
 ## Future features
 Note that this is not actively ongoing work, but features that we are aware will be needed in the future.
-- Provide server-side formatting for commonly used configuration files - eg `beamline_parameters.txt` should be returned as a dictionary.
 - Remove absolute filepath as a user dependancy. For example, once we have a good picture of which files are being read, the client should be able to request the beamline parameters with something like `client.get_file_contents(SUPPORTED_CONFIG_FILES.BeamlineParameters, beamline=i03)`
 - Add authorisation + authentication. This is a pre-requisite for file-writing.
 - Add endpoints for configuration file-writing. At this point, we can begin to remove any configuration from the filesystem. The config-server will use a redis database to store these values, and we can have a simple web-interface to change the values.

pyproject.toml

@@ -28,6 +28,7 @@ dependencies = [
     "pydantic",
     "urllib3",
     "requests",
+    "xmltodict",
 ]
 
 [project.optional-dependencies]

src/daq_config_server/app.py

@@ -1,4 +1,3 @@
-import json
 import logging
 import os
 from collections.abc import Awaitable, Callable
@@ -16,6 +15,7 @@
 from daq_config_server.constants import (
     ENDPOINTS,
 )
+from daq_config_server.converters import get_converted_file_contents
 from daq_config_server.log import set_up_logging
 from daq_config_server.whitelist import get_whitelist
 
@@ -134,8 +134,7 @@ def get_configuration(
     try:
         match accept:
             case ValidAcceptHeaders.JSON:
-                with file_path.open("r", encoding="utf-8") as f:
-                    content = json.loads(f.read())
+                content = get_converted_file_contents(file_path)
                 return JSONResponse(
                     content=content,
                 )

src/daq_config_server/client.py

@@ -1,7 +1,7 @@
 import operator
 from logging import Logger, getLogger
 from pathlib import Path
-from typing import Any, TypeVar, get_origin
+from typing import Any, TypeVar, get_origin, overload
 
 import requests
 from cachetools import TTLCache, cachedmethod
@@ -10,18 +10,26 @@
 from requests.exceptions import HTTPError
 
 from daq_config_server.app import ValidAcceptHeaders
+from daq_config_server.converters.models import ConfigModel
 
 from .constants import ENDPOINTS
 
-T = TypeVar("T", str, bytes, dict[Any, Any])
+TModel = TypeVar("TModel", bound=ConfigModel)
+TNonModel = TypeVar("TNonModel", str, bytes, dict[str, Any])
 
 
 class TypeConversionException(Exception): ...
 
 
-def _get_mime_type(requested_return_type: type[T]) -> ValidAcceptHeaders:
+def _get_mime_type(
+    requested_return_type: type[TModel | TNonModel],
+) -> ValidAcceptHeaders:
     # Get correct mapping for typed dict or plain dict
-    if get_origin(requested_return_type) is dict or requested_return_type is dict:
+    if (
+        get_origin(requested_return_type) is dict
+        or requested_return_type is dict
+        or issubclass(requested_return_type, ConfigModel)
+    ):
         return ValidAcceptHeaders.JSON
     elif requested_return_type is bytes:
         return ValidAcceptHeaders.RAW_BYTES
@@ -131,12 +139,28 @@ def _get(
 
         return content
 
+    @overload
+    def get_file_contents(
+        self,
+        file_path: str | Path,
+        desired_return_type: type[TNonModel] = str,
+        reset_cached_result: bool = False,
+    ) -> TNonModel: ...
+
+    @overload
+    def get_file_contents(
+        self,
+        file_path: str | Path,
+        desired_return_type: type[TModel],
+        reset_cached_result: bool = False,
+    ) -> TModel: ...
+
     def get_file_contents(
         self,
-        file_path: Path | str,
-        desired_return_type: type[T] = str,
+        file_path: str | Path,
+        desired_return_type: type[Any] = str,
         reset_cached_result: bool = False,
-    ) -> T:
+    ) -> Any:
         """
         Get contents of a file from the config server in the format specified.
         Optionally look for cached result before making request.

src/daq_config_server/converters/__init__.py

@@ -0,0 +1,3 @@
+from daq_config_server.converters.convert import get_converted_file_contents
+
+__all__ = ["get_converted_file_contents"]

src/daq_config_server/converters/_converter_utils.py

@@ -0,0 +1,48 @@
+import ast
+from typing import Any
+
+from daq_config_server.converters.models import GenericLookupTable
+
+ALLOWED_BEAMLINE_PARAMETER_STRINGS = ["FB", "FULL", "deadtime"]
+
+
+class ConverterParseError(Exception): ...
+
+
+def remove_comments(lines: list[str]) -> list[str]:
+    return [
+        line.strip().split("#", 1)[0].strip()
+        for line in lines
+        if line.strip().split("#", 1)[0]
+    ]
+
+
+def parse_value(value: str, convert_to: type | None = None) -> Any:
+    """Convert a string value into an appropriate Python type. Optionally provide a type
+    to convert to. If not given, the type will be inferred.
+    """
+    value = ast.literal_eval(value.replace("Yes", "True").replace("No", "False"))
+    if convert_to:
+        value = convert_to(value)
+    return value
+
+
+def parse_lut(contents: str, *params: tuple[str, type | None]) -> GenericLookupTable:
+    """Converts a lookup table to a pydantic model, containing the names of each column
+    and the rows as a 2D list.
+
+    Any args after the contents provide the column names and optionally, python types
+    for values in a column to be converted to. e.g: (energy_EV, float), (pixels, int).
+    If a type is provided, the values in that column will be converted to that type.
+    Otherwise, the type will be inferred. Units should be included in the column name.
+    """
+    rows: list[list[Any]] = []
+    column_names = [param[0] for param in params]
+    types = [param[1] for param in params]
+    for line in remove_comments(contents.splitlines()):
+        if line.startswith("Units"):
+            continue
+        rows.append(
+            [parse_value(value, types[i]) for i, value in enumerate(line.split())]
+        )
+    return GenericLookupTable(column_names=column_names, rows=rows)

src/daq_config_server/converters/_converters.py

@@ -0,0 +1,82 @@
+from typing import Any
+
+from daq_config_server.converters._converter_utils import (
+    ALLOWED_BEAMLINE_PARAMETER_STRINGS,
+    GenericLookupTable,
+    parse_lut,
+    parse_value,
+    remove_comments,
+)
+from daq_config_server.converters.models import DisplayConfig, DisplayConfigData
+
+
+def beamline_parameters_to_dict(contents: str) -> dict[str, Any]:
+    """Extracts the key value pairs from a beamline parameters file. If the value
+    is in ALLOWED_BEAMLINE_PARAMETER_STRINGS, it leaves it as a string. Otherwise, it is
+    converted to a number or bool."""
+    lines = contents.splitlines()
+    config_pairs: dict[str, Any] = {}
+
+    # Get dict of parameter keys and values
+    for line in remove_comments(lines):
+        splitline = line.split("=")
+        if len(splitline) >= 2:
+            param, value = line.split("=")
+            if param.strip() in config_pairs:
+                raise ValueError(f"Repeated key in parameters: {param}")
+            config_pairs[param.strip()] = value.strip()
+
+    # Parse each value
+    for param, value in config_pairs.items():
+        if value not in ALLOWED_BEAMLINE_PARAMETER_STRINGS:
+            config_pairs[param] = parse_value(value)
+    return dict(config_pairs)
+
+
+def display_config_to_model(contents: str) -> DisplayConfig:
+    lines = contents.splitlines()
+    config_dict: dict[float, dict[str, int | float]] = {}
+    zoom_level = None
+
+    for line in remove_comments(lines):
+        key, value = (item.strip() for item in line.split("=", 1))
+        if key == "zoomLevel":
+            zoom_level = float(value)
+            assert zoom_level not in config_dict.keys(), (
+                f"Multiple instances of zoomLevel {zoom_level}"
+            )
+            config_dict[zoom_level] = {}
+            continue
+
+        assert zoom_level is not None, "File must start with a zoom level"
+        assert key not in config_dict[zoom_level].keys(), (
+            "File can't have repeated keys for a given zoom level"
+        )
+        config_dict[zoom_level][key] = parse_value(value)
+    zoom_levels = {
+        key: DisplayConfigData.model_validate(value)
+        for key, value in config_dict.items()
+    }
+
+    return DisplayConfig(zoom_levels=zoom_levels)
+
+
+def detector_xy_lut(contents: str) -> GenericLookupTable:
+    return parse_lut(
+        contents,
+        ("detector_distances_mm", float),
+        ("beam_centre_x_mm", float),
+        ("beam_centre_y_mm", float),
+    )
+
+
+def beamline_pitch_lut(contents: str) -> GenericLookupTable:
+    return parse_lut(contents, ("bragg_angle_deg", float), ("pitch_mrad", float))
+
+
+def beamline_roll_lut(contents: str) -> GenericLookupTable:
+    return parse_lut(contents, ("bragg_angle_deg", float), ("roll_mrad", float))
+
+
+def undulator_energy_gap_lut(contents: str) -> GenericLookupTable:
+    return parse_lut(contents, ("energy_eV", int), ("gap_mm", float))

src/daq_config_server/converters/_file_converter_map.py

@@ -0,0 +1,38 @@
+from collections.abc import Callable
+from typing import Any
+
+import xmltodict
+
+from daq_config_server.converters._converters import (
+    beamline_parameters_to_dict,
+    beamline_pitch_lut,
+    beamline_roll_lut,
+    detector_xy_lut,
+    display_config_to_model,
+    undulator_energy_gap_lut,
+)
+from daq_config_server.converters.models import ConfigModel
+
+FILE_TO_CONVERTER_MAP: dict[str, Callable[[str], ConfigModel | dict[str, Any]]] = {  # type: ignore
+    "/tests/test_data/test_good_lut.txt": undulator_energy_gap_lut,  # For system tests # noqa
+    "/dls_sw/i23/software/aithre/aithre_display.configuration": display_config_to_model,
+    "/dls_sw/i03/software/gda_versions/var/display.configuration": display_config_to_model,  # noqa
+    "/dls_sw/i04/software/bluesky/scratch/display.configuration": display_config_to_model,  # noqa
+    "/dls_sw/i19-1/software/daq_configuration/domain/display.configuration": display_config_to_model,  # noqa
+    "/dls_sw/i24/software/gda_versions/var/display.configuration": display_config_to_model,  # noqa
+    "/dls_sw/i03/software/gda/configurations/i03-config/xml/jCameraManZoomLevels.xml": xmltodict.parse,  # noqa
+    "/dls_sw/i04/software/bluesky/scratch/jCameraManZoomLevels.xml": xmltodict.parse,
+    "/dls_sw/i19-1/software/gda_versions/gda/config/xml/jCameraManZoomLevels.xml": xmltodict.parse,  # noqa
+    "/dls_sw/i24/software/gda_versions/gda/config/xml/jCameraManZoomLevels.xml": xmltodict.parse,  # noqa
+    "/dls_sw/i03/software/daq_configuration/domain/beamlineParameters": beamline_parameters_to_dict,  # noqa
+    "/dls_sw/i04/software/daq_configuration/domain/beamlineParameters": beamline_parameters_to_dict,  # noqa
+    "/dls_sw/i03/software/daq_configuration/lookup/DetDistToBeamXYConverter.txt": detector_xy_lut,  # noqa
+    "/dls_sw/i04/software/daq_configuration/lookup/DetDistToBeamXYConverter.txt": detector_xy_lut,  # noqa
+    "/dls_sw/i04-1/software/daq_configuration/lookup/DetDistToBeamXYConverter.txt": detector_xy_lut,  # noqa
+    "/dls_sw/i19-1/software/daq_configuration/lookup/DetDistToBeamXYConverter.txt": detector_xy_lut,  # noqa
+    "/dls_sw/i23/software/daq_configuration/lookup/DetDistToBeamXYConverter.txt": detector_xy_lut,  # noqa
+    "/dls_sw/i24/software/daq_configuration/lookup/DetDistToBeamXYConverter.txt": detector_xy_lut,  # noqa
+    "/dls_sw/i03/software/daq_configuration/lookup/BeamLineEnergy_DCM_Pitch_converter.txt": beamline_pitch_lut,  # noqa
+    "/dls_sw/i03/software/daq_configuration/lookup/BeamLineEnergy_DCM_Roll_converter.txt": beamline_roll_lut,  # noqa
+    "/dls_sw/i03/software/daq_configuration/lookup/BeamLine_Undulator_toGap.txt": undulator_energy_gap_lut,  # noqa
+}

src/daq_config_server/converters/convert.py

@@ -0,0 +1,24 @@
+import json
+from pathlib import Path
+from typing import Any
+
+import daq_config_server.converters._file_converter_map as file_converter_map
+from daq_config_server.converters._converter_utils import ConverterParseError
+from daq_config_server.converters.models import ConfigModel
+
+
+def get_converted_file_contents(file_path: Path) -> dict[str, Any]:
+    with file_path.open("r", encoding="utf-8") as f:
+        raw_contents = f.read()
+    if converter := file_converter_map.FILE_TO_CONVERTER_MAP.get(str(file_path)):
+        try:
+            contents = converter(raw_contents)
+            if isinstance(contents, ConfigModel):
+                return contents.model_dump()
+            return contents
+        except Exception as e:
+            raise ConverterParseError(
+                f"Unable to parse {str(file_path)} due to the following exception: \
+                {type(e).__name__}: {e}"
+            ) from e
+    return json.loads(raw_contents)

src/daq_config_server/converters/models.py

@@ -0,0 +1,47 @@
+from pydantic import BaseModel, model_validator
+
+
+class ConfigModel(BaseModel): ...
+
+
+class DisplayConfigData(ConfigModel):
+    crosshairX: int
+    crosshairY: int
+    topLeftX: int
+    topLeftY: int
+    bottomRightX: int
+    bottomRightY: int
+
+
+class DisplayConfig(ConfigModel):
+    zoom_levels: dict[float, DisplayConfigData]
+    required_zoom_levels: set[float] | None = None
+
+    @model_validator(mode="after")
+    def check_zoom_levels_match_required(self):
+        existing_keys = set(self.zoom_levels.keys())
+        if (
+            self.required_zoom_levels is not None
+            and self.required_zoom_levels != existing_keys
+        ):
+            raise ValueError(
+                f"Zoom levels {existing_keys} "
+                f"do not match required zoom levels: {self.required_zoom_levels}"
+            )
+        return self
+
+
+class GenericLookupTable(ConfigModel):
+    column_names: list[str]
+    rows: list[list[int | float]]
+
+    @model_validator(mode="after")
+    def check_row_length_matches_n_columns(self):
+        n_columns = len(self.column_names)
+        for row in self.rows:
+            if len(row) != n_columns:
+                raise ValueError(
+                    f"Length of row {row} does not match number "
+                    f"of columns: {self.column_names}"
+                )
+        return self