Split converters and models into modules

Author: jacob720

src/daq_config_server/converters/_converter_utils.py

@@ -1,8 +1,6 @@
 import ast
 from typing import Any
 
-from daq_config_server.converters.models import GenericLookupTable
-
 ALLOWED_BEAMLINE_PARAMETER_STRINGS = ["FB", "FULL", "deadtime"]
 
 
@@ -25,24 +23,3 @@ def parse_value(value: str, convert_to: type | None = None) -> Any:
     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

@@ -3,12 +3,16 @@
 
 import xmltodict
 
-from daq_config_server.converters._converters import (
+from daq_config_server.converters.beamline_parameters._converters import (
     beamline_parameters_to_dict,
+)
+from daq_config_server.converters.display_config._converters import (
+    display_config_to_model,
+)
+from daq_config_server.converters.lookup_tables._converters import (
     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

src/daq_config_server/converters/_file_converter_map.py

@@ -0,0 +1,28 @@
+from typing import Any
+
+from daq_config_server.converters._converter_utils import parse_value, remove_comments
+
+ALLOWED_BEAMLINE_PARAMETER_STRINGS = ["FB", "FULL", "deadtime"]
+
+
+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)

src/daq_config_server/converters/beamline_parameters/__init__.py

@@ -0,0 +1,33 @@
+from daq_config_server.converters._converter_utils import parse_value, remove_comments
+from daq_config_server.converters.display_config.models import (
+    DisplayConfig,
+    DisplayConfigData,
+)
+
+
+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)

src/daq_config_server/converters/beamline_parameters/_converters.py

@@ -0,0 +1,30 @@
+from pydantic import model_validator
+
+from daq_config_server.converters.models import ConfigModel
+
+
+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