Merge pull request #57 from AEONplus/feature/ocs_validation_schema_support

Add in support for pulling out extra params fields defined in instrum…

Author: Fingel

README.md

@@ -90,20 +90,20 @@ This ensures regular users of the library do not need to install these dependenc
 The `generate.py` script takes as input JSON as produced by the instruments endpoint:
 
 ```bash
-codegen/lco/generator.py instruments.json
+codegen/lco/generator.py {facility} instruments.json
 ```
 
 Or directly from stdin using a pipe:
 
 ```bash
-curl https://observe.lco.global/api/instruments/ | codegen/lco/generator.py
+curl https://observe.lco.global/api/instruments/ | codegen/lco/generator.py {facility}
 ```
 
 If the output looks satisfactory, you can redirect the output to overwrite the
 LCO instruments definition file:
 
 ```bash
-curl https://observe.lco.global/api/instruments/ | codegen/lco/generator.py > src/aeonlib/ocs/lco/instruments.py
+curl https://observe.lco.global/api/instruments/ | codegen/lco/generator.py {facility} > src/aeonlib/ocs/lco/instruments.py
 ```
 # Supported Facilities
 

codegen/lco/generator.py

@@ -11,6 +11,53 @@
 VALID_FACILITIES = ["SOAR", "LCO", "SAAO", "BLANCO"]
 
 
+def get_extra_params_fields(extra_params_validation_schema: dict) -> dict:
+    """Loops over the "extra_params" section of a validation_schema dict and creates a dictionary of
+    field to aeonlib field_class to place into the template
+    """
+    fields = {}
+    for field, properties in extra_params_validation_schema.items():
+        field_class = ""
+        # If a set of allowed values is present, use that to make a Literal unless this is a boolean variable
+        if "allowed" in properties and properties.get("type") != "boolean":
+            allowed_values = [
+                f'"{val}"' if properties["type"] == "string" else val
+                for val in properties["allowed"]
+            ]
+            field_class += f"Literal[{', '.join(allowed_values)}]"
+        else:
+            # Otherwise form an Annotated field based on its datatype, with min/max validation if present
+            field_class += "Annotated["
+            match properties["type"]:
+                case "string":
+                    field_class += "str"
+                case "integer":
+                    field_class += "int"
+                case "float":
+                    field_class += "float"
+                case "boolean":
+                    field_class += "bool"
+            if "min" in properties:
+                field_class += f", Ge({properties['min']})"
+            if "max" in properties:
+                field_class += f", Le({properties['max']})"
+            # Add description to Annotated field. Annotated fields must have at least 2 properties.
+            field_class += f', "{properties.get("description", "")}"]'
+        if not properties.get("required", False) and "default" not in properties:
+            # The field is considered optional if it doesn't have a default or required is not set to True
+            field_class += " | None = None"
+        elif "default" in properties:
+            # If a default value is present, provide it
+            default = (
+                f'"{properties["default"]}"'
+                if properties["type"] == "string"
+                else properties["default"]
+            )
+            field_class += f" = {default}"
+        fields[field] = field_class
+    return fields
+
+
 def get_modes(ins: dict[str, Any], type: str) -> list[str]:
     try:
         return [m["code"] for m in ins["modes"][type]["modes"]]
@@ -84,6 +131,17 @@ def generate_instrument_configs(ins_s: str, facility: str) -> str:
                     k.rstrip("s"): v
                     for k, v in ins["optical_elements"].items()
                 },
+                "configuration_extra_params": get_extra_params_fields(
+                    ins["validation_schema"].get("extra_params", {}).get("schema", {})
+                ),
+                "instrument_config_extra_params": get_extra_params_fields(
+                    ins["validation_schema"]
+                    .get("instrument_configs", {})
+                    .get("schema", {})
+                    .get("schema", {})
+                    .get("extra_params", {})
+                    .get("schema", {})
+                ),
             }
         )
 

codegen/lco/templates/instruments.jinja

@@ -3,8 +3,8 @@
 
 from typing import Any, Annotated, Literal
 
-from annotated_types import Le
-from pydantic import BaseModel, ConfigDict
+from annotated_types import Le, Ge
+from pydantic import BaseModel, ConfigDict, Field
 from pydantic.types import NonNegativeInt, PositiveInt
 
 from aeonlib.models import TARGET_TYPES
@@ -13,6 +13,22 @@ from aeonlib.ocs.config_models import Roi
 
 
 {% for ctx in instruments %}
+
+
+class {{ ctx.class_name}}ConfigExtraParams(BaseModel):
+    model_config = ConfigDict(validate_assignment=True, extra='allow')
+    {% for field, field_class in ctx.configuration_extra_params.items() %}
+    {{ field }}: {{ field_class }}
+    {% endfor %}
+
+
+class {{ ctx.class_name}}InstrumentConfigExtraParams(BaseModel):
+    model_config = ConfigDict(validate_assignment=True, extra='allow')
+    {% for field, field_class in ctx.instrument_config_extra_params.items() %}
+    {{ field }}: {{ field_class }}
+    {% endfor %}
+
+
 class {{ ctx.class_name }}OpticalElements(BaseModel):
     model_config = ConfigDict(validate_assignment=True)
     {% for key, values in ctx.optical_elements.items() %}
@@ -49,7 +65,7 @@ class {{ ctx.class_name }}Config(BaseModel):
     rotator_mode: Literal[{% for m in ctx.rotator_modes %}"{{ m }}"{% if not loop.last %}, {% endif %}{% endfor %}]
     {% endif %}
     rois: list[Roi] | None = None
-    extra_params: dict[Any, Any] = {}
+    extra_params: {{ ctx.class_name }}InstrumentConfigExtraParams = Field(default_factory={{ ctx.class_name }}InstrumentConfigExtraParams)
     optical_elements: {{ ctx.class_name}}OpticalElements
 
 
@@ -58,7 +74,7 @@ class {{ ctx.class_name }}(BaseModel):
     type: Literal[{% for t in ctx.config_types %}"{{ t }}"{% if not loop.last %}, {% endif %}{% endfor %}]
     instrument_type: Literal["{{ ctx.instrument_type }}"] = "{{ ctx.instrument_type }}"
     repeat_duration: NonNegativeInt | None = None
-    extra_params: dict[Any, Any] = {}
+    extra_params: {{ ctx.class_name }}ConfigExtraParams = Field(default_factory={{ ctx.class_name }}ConfigExtraParams)
     instrument_configs: list[{{ ctx.class_name }}Config] = []
     acquisition_config: {{ ctx.class_name }}AcquisitionConfig
     guiding_config: {{ ctx.class_name }}GuidingConfig

src/aeonlib/ocs/blanco/instruments.py

@@ -3,15 +3,31 @@
 
 from typing import Any, Annotated, Literal
 
-from annotated_types import Le
-from pydantic import BaseModel, ConfigDict
+from annotated_types import Le, Ge
+from pydantic import BaseModel, ConfigDict, Field
 from pydantic.types import NonNegativeInt, PositiveInt
 
 from aeonlib.models import TARGET_TYPES
 from aeonlib.ocs.target_models import Constraints
 from aeonlib.ocs.config_models import Roi
 
 
+
+
+class BlancoNewfirmConfigExtraParams(BaseModel):
+    model_config = ConfigDict(validate_assignment=True, extra='allow')
+    dither_value: Annotated[int, Ge(0), Le(1600), "The amount in arc seconds between dither points"] = 80
+    dither_sequence: Literal["2x2", "3x3", "4x4", "5-point"] = "2x2"
+    detector_centering: Literal["none", "det_1", "det_2", "det_3", "det_4"] = "det_1"
+    dither_sequence_random_offset: Annotated[bool, "Implements a random offset between dither patterns if repeating the dither pattern, i.e. when sequence repeats > 1"] = True
+
+
+class BlancoNewfirmInstrumentConfigExtraParams(BaseModel):
+    model_config = ConfigDict(validate_assignment=True, extra='allow')
+    coadds: Annotated[int, Ge(1), Le(100), "This reduces data volume with short integration times necessary for broadband H and Ks observations. Coadding is digital summation of the images to avoid long integrations that could cause saturation of the detector."] = 1
+    sequence_repeats: Annotated[int, Ge(1), Le(500), "The number of times to repeat the dither sequence"] = 1
+
+
 class BlancoNewfirmOpticalElements(BaseModel):
     model_config = ConfigDict(validate_assignment=True)
     filter: Literal["JX", "HX", "KXs"]
@@ -43,7 +59,7 @@ class BlancoNewfirmConfig(BaseModel):
     """ Exposure time in seconds"""
     mode: Literal["fowler1", "fowler2"]
     rois: list[Roi] | None = None
-    extra_params: dict[Any, Any] = {}
+    extra_params: BlancoNewfirmInstrumentConfigExtraParams = Field(default_factory=BlancoNewfirmInstrumentConfigExtraParams)
     optical_elements: BlancoNewfirmOpticalElements
 
 
@@ -52,7 +68,7 @@ class BlancoNewfirm(BaseModel):
     type: Literal["EXPOSE", "SKY_FLAT", "STANDARD", "DARK"]
     instrument_type: Literal["BLANCO_NEWFIRM"] = "BLANCO_NEWFIRM"
     repeat_duration: NonNegativeInt | None = None
-    extra_params: dict[Any, Any] = {}
+    extra_params: BlancoNewfirmConfigExtraParams = Field(default_factory=BlancoNewfirmConfigExtraParams)
     instrument_configs: list[BlancoNewfirmConfig] = []
     acquisition_config: BlancoNewfirmAcquisitionConfig
     guiding_config: BlancoNewfirmGuidingConfig

src/aeonlib/ocs/lco/instruments.py

@@ -3,15 +3,28 @@
 
 from typing import Any, Annotated, Literal
 
-from annotated_types import Le
-from pydantic import BaseModel, ConfigDict
+from annotated_types import Le, Ge
+from pydantic import BaseModel, ConfigDict, Field
 from pydantic.types import NonNegativeInt, PositiveInt
 
 from aeonlib.models import TARGET_TYPES
 from aeonlib.ocs.target_models import Constraints
 from aeonlib.ocs.config_models import Roi
 
 
+
+
+class Lco0M4ScicamQhy600ConfigExtraParams(BaseModel):
+    model_config = ConfigDict(validate_assignment=True, extra='allow')
+    sub_expose: Annotated[bool, "Whether or not to split your exposures into sub_exposures to guide during the observation, and stack them together at the end for the final data product."] = False
+    sub_exposure_time: Annotated[float, Ge(15.0), "Exposure time for the sub-exposures in seconds, if sub_expose mode is set"] | None = None
+
+
+class Lco0M4ScicamQhy600InstrumentConfigExtraParams(BaseModel):
+    model_config = ConfigDict(validate_assignment=True, extra='allow')
+    defocus: Annotated[float, Ge(-5.0), Le(5.0), "Observations may be defocused to prevent the CCD from saturating on bright targets. This term describes the offset (in mm) of the secondary mirror from its default (focused) position. The limits are ± 5mm."] | None = None
+
+
 class Lco0M4ScicamQhy600OpticalElements(BaseModel):
     model_config = ConfigDict(validate_assignment=True)
     filter: Literal["OIII", "SII", "Astrodon-Exo", "w", "opaque", "up", "rp", "ip", "gp", "zs", "V", "B", "H-Alpha"]
@@ -43,7 +56,7 @@ class Lco0M4ScicamQhy600Config(BaseModel):
     """ Exposure time in seconds"""
     mode: Literal["central30x30", "full_frame"]
     rois: list[Roi] | None = None
-    extra_params: dict[Any, Any] = {}
+    extra_params: Lco0M4ScicamQhy600InstrumentConfigExtraParams = Field(default_factory=Lco0M4ScicamQhy600InstrumentConfigExtraParams)
     optical_elements: Lco0M4ScicamQhy600OpticalElements
 
 
@@ -52,7 +65,7 @@ class Lco0M4ScicamQhy600(BaseModel):
     type: Literal["EXPOSE", "REPEAT_EXPOSE", "AUTO_FOCUS", "BIAS", "DARK", "STANDARD", "SKY_FLAT"]
     instrument_type: Literal["0M4-SCICAM-QHY600"] = "0M4-SCICAM-QHY600"
     repeat_duration: NonNegativeInt | None = None
-    extra_params: dict[Any, Any] = {}
+    extra_params: Lco0M4ScicamQhy600ConfigExtraParams = Field(default_factory=Lco0M4ScicamQhy600ConfigExtraParams)
     instrument_configs: list[Lco0M4ScicamQhy600Config] = []
     acquisition_config: Lco0M4ScicamQhy600AcquisitionConfig
     guiding_config: Lco0M4ScicamQhy600GuidingConfig
@@ -65,6 +78,17 @@ class Lco0M4ScicamQhy600(BaseModel):
     optical_elements_class = Lco0M4ScicamQhy600OpticalElements
 
 
+
+
+class Lco1M0NresScicamConfigExtraParams(BaseModel):
+    model_config = ConfigDict(validate_assignment=True, extra='allow')
+
+
+class Lco1M0NresScicamInstrumentConfigExtraParams(BaseModel):
+    model_config = ConfigDict(validate_assignment=True, extra='allow')
+    defocus: Annotated[float, Ge(-5.0), Le(5.0), "Observations may be defocused to prevent the CCD from saturating on bright targets. This term describes the offset (in mm) of the secondary mirror from its default (focused) position. The limits are ± 5mm."] | None = None
+
+
 class Lco1M0NresScicamOpticalElements(BaseModel):
     model_config = ConfigDict(validate_assignment=True)
 
@@ -95,7 +119,7 @@ class Lco1M0NresScicamConfig(BaseModel):
     """ Exposure time in seconds"""
     mode: Literal["default"]
     rois: list[Roi] | None = None
-    extra_params: dict[Any, Any] = {}
+    extra_params: Lco1M0NresScicamInstrumentConfigExtraParams = Field(default_factory=Lco1M0NresScicamInstrumentConfigExtraParams)
     optical_elements: Lco1M0NresScicamOpticalElements
 
 
@@ -104,7 +128,7 @@ class Lco1M0NresScicam(BaseModel):
     type: Literal["NRES_SPECTRUM", "REPEAT_NRES_SPECTRUM", "NRES_EXPOSE", "NRES_TEST", "SCRIPT", "ENGINEERING", "ARC", "LAMP_FLAT", "NRES_BIAS", "NRES_DARK", "AUTO_FOCUS"]
     instrument_type: Literal["1M0-NRES-SCICAM"] = "1M0-NRES-SCICAM"
     repeat_duration: NonNegativeInt | None = None
-    extra_params: dict[Any, Any] = {}
+    extra_params: Lco1M0NresScicamConfigExtraParams = Field(default_factory=Lco1M0NresScicamConfigExtraParams)
     instrument_configs: list[Lco1M0NresScicamConfig] = []
     acquisition_config: Lco1M0NresScicamAcquisitionConfig
     guiding_config: Lco1M0NresScicamGuidingConfig
@@ -117,6 +141,17 @@ class Lco1M0NresScicam(BaseModel):
     optical_elements_class = Lco1M0NresScicamOpticalElements
 
 
+
+
+class Lco1M0ScicamSinistroConfigExtraParams(BaseModel):
+    model_config = ConfigDict(validate_assignment=True, extra='allow')
+
+
+class Lco1M0ScicamSinistroInstrumentConfigExtraParams(BaseModel):
+    model_config = ConfigDict(validate_assignment=True, extra='allow')
+    defocus: Annotated[float, Ge(-5.0), Le(5.0), "Observations may be defocused to prevent the CCD from saturating on bright targets. This term describes the offset (in mm) of the secondary mirror from its default (focused) position. The limits are ± 5mm."] | None = None
+
+
 class Lco1M0ScicamSinistroOpticalElements(BaseModel):
     model_config = ConfigDict(validate_assignment=True)
     filter: Literal["I", "R", "U", "w", "Y", "up", "rp", "ip", "gp", "zs", "V", "B", "400um-Pinhole", "150um-Pinhole", "CN"]
@@ -148,7 +183,7 @@ class Lco1M0ScicamSinistroConfig(BaseModel):
     """ Exposure time in seconds"""
     mode: Literal["full_frame", "central_2k_2x2"]
     rois: list[Roi] | None = None
-    extra_params: dict[Any, Any] = {}
+    extra_params: Lco1M0ScicamSinistroInstrumentConfigExtraParams = Field(default_factory=Lco1M0ScicamSinistroInstrumentConfigExtraParams)
     optical_elements: Lco1M0ScicamSinistroOpticalElements
 
 
@@ -157,7 +192,7 @@ class Lco1M0ScicamSinistro(BaseModel):
     type: Literal["EXPOSE", "REPEAT_EXPOSE", "BIAS", "DARK", "STANDARD", "SCRIPT", "AUTO_FOCUS", "ENGINEERING", "SKY_FLAT"]
     instrument_type: Literal["1M0-SCICAM-SINISTRO"] = "1M0-SCICAM-SINISTRO"
     repeat_duration: NonNegativeInt | None = None
-    extra_params: dict[Any, Any] = {}
+    extra_params: Lco1M0ScicamSinistroConfigExtraParams = Field(default_factory=Lco1M0ScicamSinistroConfigExtraParams)
     instrument_configs: list[Lco1M0ScicamSinistroConfig] = []
     acquisition_config: Lco1M0ScicamSinistroAcquisitionConfig
     guiding_config: Lco1M0ScicamSinistroGuidingConfig
@@ -170,6 +205,17 @@ class Lco1M0ScicamSinistro(BaseModel):
     optical_elements_class = Lco1M0ScicamSinistroOpticalElements
 
 
+
+
+class Lco2M0FloydsScicamConfigExtraParams(BaseModel):
+    model_config = ConfigDict(validate_assignment=True, extra='allow')
+
+
+class Lco2M0FloydsScicamInstrumentConfigExtraParams(BaseModel):
+    model_config = ConfigDict(validate_assignment=True, extra='allow')
+    defocus: Annotated[float, Ge(-5.0), Le(5.0), ""] | None = None
+
+
 class Lco2M0FloydsScicamOpticalElements(BaseModel):
     model_config = ConfigDict(validate_assignment=True)
     slit: Literal["slit_6.0as", "slit_1.6as", "slit_2.0as", "slit_1.2as"]
@@ -202,7 +248,7 @@ class Lco2M0FloydsScicamConfig(BaseModel):
     mode: Literal["default"]
     rotator_mode: Literal["VFLOAT", "SKY"]
     rois: list[Roi] | None = None
-    extra_params: dict[Any, Any] = {}
+    extra_params: Lco2M0FloydsScicamInstrumentConfigExtraParams = Field(default_factory=Lco2M0FloydsScicamInstrumentConfigExtraParams)
     optical_elements: Lco2M0FloydsScicamOpticalElements
 
 
@@ -211,7 +257,7 @@ class Lco2M0FloydsScicam(BaseModel):
     type: Literal["SPECTRUM", "REPEAT_SPECTRUM", "ARC", "ENGINEERING", "SCRIPT", "LAMP_FLAT"]
     instrument_type: Literal["2M0-FLOYDS-SCICAM"] = "2M0-FLOYDS-SCICAM"
     repeat_duration: NonNegativeInt | None = None
-    extra_params: dict[Any, Any] = {}
+    extra_params: Lco2M0FloydsScicamConfigExtraParams = Field(default_factory=Lco2M0FloydsScicamConfigExtraParams)
     instrument_configs: list[Lco2M0FloydsScicamConfig] = []
     acquisition_config: Lco2M0FloydsScicamAcquisitionConfig
     guiding_config: Lco2M0FloydsScicamGuidingConfig
@@ -224,6 +270,17 @@ class Lco2M0FloydsScicam(BaseModel):
     optical_elements_class = Lco2M0FloydsScicamOpticalElements
 
 
+
+
+class Lco2M0ScicamMuscatConfigExtraParams(BaseModel):
+    model_config = ConfigDict(validate_assignment=True, extra='allow')
+
+
+class Lco2M0ScicamMuscatInstrumentConfigExtraParams(BaseModel):
+    model_config = ConfigDict(validate_assignment=True, extra='allow')
+    defocus: Annotated[float, Ge(-8.0), Le(8.0), "Observations may be defocused to prevent the CCD from saturating on bright targets. This term describes the offset (in mm) of the secondary mirror from its default (focused) position. The limits are ± 8mm."] | None = None
+
+
 class Lco2M0ScicamMuscatOpticalElements(BaseModel):
     model_config = ConfigDict(validate_assignment=True)
     narrowband_g_position: Literal["out", "in"]
@@ -258,7 +315,7 @@ class Lco2M0ScicamMuscatConfig(BaseModel):
     """ Exposure time in seconds"""
     mode: Literal["MUSCAT_SLOW", "MUSCAT_FAST"]
     rois: list[Roi] | None = None
-    extra_params: dict[Any, Any] = {}
+    extra_params: Lco2M0ScicamMuscatInstrumentConfigExtraParams = Field(default_factory=Lco2M0ScicamMuscatInstrumentConfigExtraParams)
     optical_elements: Lco2M0ScicamMuscatOpticalElements
 
 
@@ -267,7 +324,7 @@ class Lco2M0ScicamMuscat(BaseModel):
     type: Literal["EXPOSE", "REPEAT_EXPOSE", "BIAS", "DARK", "STANDARD", "SCRIPT", "AUTO_FOCUS", "ENGINEERING", "SKY_FLAT"]
     instrument_type: Literal["2M0-SCICAM-MUSCAT"] = "2M0-SCICAM-MUSCAT"
     repeat_duration: NonNegativeInt | None = None
-    extra_params: dict[Any, Any] = {}
+    extra_params: Lco2M0ScicamMuscatConfigExtraParams = Field(default_factory=Lco2M0ScicamMuscatConfigExtraParams)
     instrument_configs: list[Lco2M0ScicamMuscatConfig] = []
     acquisition_config: Lco2M0ScicamMuscatAcquisitionConfig
     guiding_config: Lco2M0ScicamMuscatGuidingConfig