Added 'field_validator' functions for the 'calibrations' and 'software_versions' keys in the 'MachineConfig' model; added helper model to correctly validate the magnification calibration table

Author: tieneupin

src/murfey/util/config.py

@@ -4,14 +4,23 @@
 import socket
 from functools import lru_cache
 from pathlib import Path
-from typing import Literal, Optional, Union
+from typing import Any, Literal, Optional
 
 import yaml
 from backports.entry_points_selectable import entry_points
-from pydantic import BaseModel, ConfigDict, field_validator
+from pydantic import BaseModel, ConfigDict, RootModel, ValidationInfo, field_validator
 from pydantic_settings import BaseSettings
 
 
+class MagnificationTable(RootModel[dict[int, float]]):
+    pass
+
+
+CALIBRATIONS_VALIDATION_SCHEMAS = {
+    "magnification": MagnificationTable,
+}
+
+
 class MachineConfig(BaseModel):  # type: ignore
     """
     Keys that describe the type of workflow conducted on the client side, and how
@@ -27,7 +36,7 @@ class MachineConfig(BaseModel):  # type: ignore
     # Hardware and software -----------------------------------------------------------
     camera: str = "FALCON"
     superres: bool = False
-    calibrations: dict[str, dict[str, Union[dict, float]]]
+    calibrations: dict[str, Any]
     acquisition_software: list[str]
     software_versions: dict[str, str] = {}
     software_settings_output_directories: dict[str, list[str]] = {}
@@ -94,8 +103,44 @@ class MachineConfig(BaseModel):  # type: ignore
     node_creator_queue: str = "node_creator"
     notifications_queue: str = "pato_notification"
 
+    # Pydantic BaseModel settings
     model_config = ConfigDict(extra="allow")
 
+    @field_validator("calibrations", mode="before")
+    @classmethod
+    def validate_calibration_data(
+        cls, v: dict[str, dict[Any, Any]]
+    ) -> dict[str, dict[Any, Any]]:
+        # Pass the calibration dictionaries through their matching Pydantic models, if any are set
+        if isinstance(v, dict):
+            validated = {}
+            for (
+                key,
+                value,
+            ) in v.items():
+                model_cls = CALIBRATIONS_VALIDATION_SCHEMAS.get(key)
+                if model_cls:
+                    try:
+                        # Validate and store as a dict object with the corrected types
+                        validated[key] = model_cls.model_validate(value).root
+                    except Exception as e:
+                        raise ValueError(f"Validation failed for key '{key}': {e}")
+                else:
+                    validated[key] = value
+            return validated
+        # Let it validate and fail as-is
+        return v
+
+    @field_validator("software_versions", mode="before")
+    @classmethod
+    def validate_software_versions(cls, v: dict[str, Any]) -> dict[str, str]:
+        # Software versions should be numerical strings, even if they appear int- or float-like
+        if isinstance(v, dict):
+            validated = {key: str(value) for key, value in v.items()}
+            return validated
+        # Let it validate and fail as-is
+        return v
+
 
 def from_file(config_file_path: Path, instrument: str = "") -> dict[str, MachineConfig]:
     with open(config_file_path, "r") as config_stream:
@@ -141,9 +186,9 @@ class Security(BaseModel):
 
     @field_validator("graylog_port")
     def check_port_present_if_host_is(
-        cls, v: Optional[int], values: dict, **kwargs
+        cls, v: Optional[int], info: ValidationInfo, **kwargs
     ) -> Optional[int]:
-        if values["graylog_host"] and v is None:
+        if info.data.get("graylog_host") and v is None:
             raise ValueError("The Graylog port must be set if the Graylog host is")
         return v