53 add blanco facility to aeonlib

codegen/lco/generator.py

@@ -7,7 +7,7 @@
 import textcase
 from jinja2 import Environment, FileSystemLoader
 
-VALID_FACILITIES = ["SOAR", "LCO", "SAAO"]
+VALID_FACILITIES = ["SOAR", "LCO", "SAAO", "BLANCO"]
 
 
 def get_modes(ins: dict, type: str) -> list[str]:
@@ -43,11 +43,19 @@ def generate_instrument_configs(ins_s: str, facility: str) -> str:
         # Soar instruments look like SoarGhtsBluecam, already prefixed, so no need to add a prefix.
         prefix = ""
         filtered = {k: v for k, v in ins_data.items() if "soar" in k.lower()}
+    elif facility == "BLANCO":
+        # Blanco instrument(s) look like BLANCO_NEWFIRM
+        prefix = ""
+        filtered = {k: v for k, v in ins_data.items() if "blanco" in k.lower()}
     elif facility == "LCO":
         # We add a prefix for LCO because some instruments start with a number,
         # which is not allowed in Python class names. For example: Lco0M4ScicamQhy600
         prefix = "Lco"
-        filtered = {k: v for k, v in ins_data.items() if "soar" not in k.lower()}
+        filtered = {
+            k: v
+            for k, v in ins_data.items()
+            if "soar" not in k.lower() and "blanco" not in k.lower()
+        }
     elif facility == "SAAO":
         # SAAO config doesn't share any instruments with other facilities so we don't need
         # to filter it

pyproject.toml

@@ -50,4 +50,5 @@ exclude = [
     "src/aeonlib/ocs/lco/instruments.py",
     "src/aeonlib/ocs/soar/instruments.py",
     "src/aeonlib/ocs/saao/instruments.py",
+    "src/aeonlib/ocs/blanco/instruments.py",
 ]

src/aeonlib/conf.py

@@ -12,6 +12,10 @@ class Settings(BaseSettings):
     soar_token: str = ""
     soar_api_root: str = "https://observe.lco.global/api/"
 
+    # BLANCO
+    blanco_token: str = ""
+    blanco_api_root: str = "https://observe.lco.global/api/"
+
     # South African Astronomical Observatory
     saao_token: str = ""
     saao_api_root: str = "https://ocsio.saao.ac.za/api/"

src/aeonlib/ocs/blanco/facility.py

@@ -0,0 +1,27 @@
+from logging import getLogger
+
+from aeonlib.conf import Settings
+from aeonlib.ocs.facility import OCSFacility
+
+logger = getLogger(__name__)
+
+
+class BlancoFacility(OCSFacility):
+    """
+    BLANCO Facility
+    The BLANCO API interface goes through the LCO OCS API, so this
+    class is essentially a wrapper around the LCO Facility.
+    Configuration:
+        - AEON_BLANCO_TOKEN: API token for authentication
+        - AEON_BLANCO_API_ROOT: Root URL of the API
+    """
+
+    def api_key(self, settings: Settings) -> str:
+        if not settings.blanco_token:
+            logger.warn("AEON_BLANCO_TOKEN setting is missing, trying LCO credentials")
+            return settings.lco_token
+        else:
+            return settings.blanco_token
+
+    def api_root(self, settings: Settings) -> str:
+        return settings.blanco_api_root

src/aeonlib/ocs/blanco/instruments.py

@@ -0,0 +1,70 @@
+# This file is generated automatically and should not be edited by hand.
+
+from typing import Any, Annotated, Literal, Union
+
+from annotated_types import Le
+from pydantic import BaseModel, ConfigDict
+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 BlancoNewfirmOpticalElements(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+    filter: Literal["JX", "HX", "KXs"]
+
+
+class BlancoNewfirmGuidingConfig(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+    mode: Literal["ON"]
+    optional: bool
+    """Whether the guiding is optional or not"""
+    exposure_time: Annotated[int, NonNegativeInt, Le(120)] | None = None
+    """Guiding exposure time"""
+    extra_params: dict[Any, Any] = {}
+
+
+class BlancoNewfirmAcquisitionConfig(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+    mode: Literal["MANUAL"]
+    exposure_time: Annotated[int, NonNegativeInt, Le(60)] | None = None
+    """Acquisition exposure time"""
+    extra_params: dict[Any, Any] = {}
+
+
+class BlancoNewfirmConfig(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+    exposure_count: PositiveInt
+    """The number of exposures to take. This field must be set to a value greater than 0"""
+    exposure_time: NonNegativeInt
+    """ Exposure time in seconds"""
+    mode: Literal["fowler1", "fowler2"]
+    rois: list[Roi] | None = None
+    extra_params: dict[Any, Any] = {}
+    optical_elements: BlancoNewfirmOpticalElements
+
+
+class BlancoNewfirm(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+    type: Literal["EXPOSE", "SKY_FLAT", "STANDARD", "DARK"]
+    instrument_type: Literal["BLANCO_NEWFIRM"] = "BLANCO_NEWFIRM"
+    repeat_duration: NonNegativeInt | None = None
+    extra_params: dict[Any, Any] = {}
+    instrument_configs: list[BlancoNewfirmConfig] = []
+    acquisition_config: BlancoNewfirmAcquisitionConfig
+    guiding_config: BlancoNewfirmGuidingConfig
+    target: TARGET_TYPES
+    constraints: Constraints
+
+    config_class = BlancoNewfirmConfig
+    guiding_config_class = BlancoNewfirmGuidingConfig
+    acquisition_config_class = BlancoNewfirmAcquisitionConfig
+    optical_elements_class = BlancoNewfirmOpticalElements
+
+
+# Export a type that encompasses all instruments
+BLANCO_INSTRUMENTS = Union[
+    BlancoNewfirm,
+]

src/aeonlib/ocs/lco/instruments.py

@@ -279,65 +279,11 @@ class Lco2M0ScicamMuscat(BaseModel):
     optical_elements_class = Lco2M0ScicamMuscatOpticalElements
 
 
-class LcoBlancoNewfirmOpticalElements(BaseModel):
-    model_config = ConfigDict(validate_assignment=True)
-    filter: Literal["JX", "HX", "KXs"]
-
-
-class LcoBlancoNewfirmGuidingConfig(BaseModel):
-    model_config = ConfigDict(validate_assignment=True)
-    mode: Literal["ON"]
-    optional: bool
-    """Whether the guiding is optional or not"""
-    exposure_time: Annotated[int, NonNegativeInt, Le(120)] | None = None
-    """Guiding exposure time"""
-    extra_params: dict[Any, Any] = {}
-
-
-class LcoBlancoNewfirmAcquisitionConfig(BaseModel):
-    model_config = ConfigDict(validate_assignment=True)
-    mode: Literal["MANUAL"]
-    exposure_time: Annotated[int, NonNegativeInt, Le(60)] | None = None
-    """Acquisition exposure time"""
-    extra_params: dict[Any, Any] = {}
-
-
-class LcoBlancoNewfirmConfig(BaseModel):
-    model_config = ConfigDict(validate_assignment=True)
-    exposure_count: PositiveInt
-    """The number of exposures to take. This field must be set to a value greater than 0"""
-    exposure_time: NonNegativeInt
-    """ Exposure time in seconds"""
-    mode: Literal["fowler1", "fowler2"]
-    rois: list[Roi] | None = None
-    extra_params: dict[Any, Any] = {}
-    optical_elements: LcoBlancoNewfirmOpticalElements
-
-
-class LcoBlancoNewfirm(BaseModel):
-    model_config = ConfigDict(validate_assignment=True)
-    type: Literal["EXPOSE", "SKY_FLAT", "STANDARD", "DARK"]
-    instrument_type: Literal["BLANCO_NEWFIRM"] = "BLANCO_NEWFIRM"
-    repeat_duration: NonNegativeInt | None = None
-    extra_params: dict[Any, Any] = {}
-    instrument_configs: list[LcoBlancoNewfirmConfig] = []
-    acquisition_config: LcoBlancoNewfirmAcquisitionConfig
-    guiding_config: LcoBlancoNewfirmGuidingConfig
-    target: TARGET_TYPES
-    constraints: Constraints
-
-    config_class = LcoBlancoNewfirmConfig
-    guiding_config_class = LcoBlancoNewfirmGuidingConfig
-    acquisition_config_class = LcoBlancoNewfirmAcquisitionConfig
-    optical_elements_class = LcoBlancoNewfirmOpticalElements
-
-
 # Export a type that encompasses all instruments
 LCO_INSTRUMENTS = Union[
     Lco0M4ScicamQhy600,
     Lco1M0NresScicam,
     Lco1M0ScicamSinistro,
     Lco2M0FloydsScicam,
     Lco2M0ScicamMuscat,
-    LcoBlancoNewfirm,
 ]

src/aeonlib/ocs/request_models.py

@@ -11,6 +11,7 @@
 )
 
 from aeonlib.models import Window
+from aeonlib.ocs.blanco.instruments import BLANCO_INSTRUMENTS
 from aeonlib.ocs.lco.instruments import LCO_INSTRUMENTS
 from aeonlib.ocs.saao.instruments import SAAO_INSTRUMENTS
 from aeonlib.ocs.soar.instruments import SOAR_INSTRUMENTS
@@ -41,7 +42,7 @@ class Cadence(BaseModel):
 
 # Informs Pydantic which instrument configuration type should be used during parsing
 Configuration = Annotated[
-    Union[LCO_INSTRUMENTS, SOAR_INSTRUMENTS, SAAO_INSTRUMENTS],
+    Union[LCO_INSTRUMENTS, SOAR_INSTRUMENTS, SAAO_INSTRUMENTS, BLANCO_INSTRUMENTS],
     Field(discriminator="instrument_type"),
 ]
 

tests/ocs/blanco_requests.py

@@ -0,0 +1,61 @@
+from datetime import datetime, timedelta
+
+from aeonlib.models import SiderealTarget, Window
+from aeonlib.ocs import (
+    Constraints,
+    Location,
+    Request,
+    RequestGroup,
+)
+from aeonlib.ocs.blanco.instruments import BlancoNewfirm
+
+target = SiderealTarget(
+    name="M10",
+    type="ICRS",
+    ra=254.287,
+    dec=-4.72,
+)
+
+window = Window(
+    start=datetime.now(),
+    end=datetime.now() + timedelta(days=60),
+)
+
+blanco_newfirm = RequestGroup(
+    name="blanco_test",
+    observation_type="NORMAL",
+    operator="SINGLE",
+    proposal="TEST_PROPOSAL",
+    ipp_value=1.0,
+    requests=[
+        Request(
+            location=Location(telescope_class="4m0"),
+            configurations=[
+                BlancoNewfirm(
+                    type="EXPOSE",
+                    target=target,
+                    constraints=Constraints(max_airmass=3.0),
+                    instrument_configs=[
+                        BlancoNewfirm.config_class(
+                            exposure_count=1,
+                            exposure_time=2,
+                            mode="fowler1",
+                            optical_elements=BlancoNewfirm.optical_elements_class(
+                                filter="HX"
+                            ),
+                        )
+                    ],
+                    acquisition_config=BlancoNewfirm.acquisition_config_class(
+                        mode="MANUAL"
+                    ),
+                    guiding_config=BlancoNewfirm.guiding_config_class(
+                        mode="ON", optional=True
+                    ),
+                )
+            ],
+            windows=[window],
+        )
+    ],
+)
+
+BLANCO_REQUESTS = {"blanco_newfirm": blanco_newfirm}

tests/ocs/test_online.py

@@ -14,11 +14,13 @@
 
 import pytest
 
+from aeonlib.ocs.blanco.facility import BlancoFacility
 from aeonlib.ocs.lco.facility import LcoFacility
 from aeonlib.ocs.request_models import RequestGroup
 from aeonlib.ocs.saao.facility import SAAOFacility
 from aeonlib.ocs.soar.facility import SoarFacility
 
+from .blanco_requests import BLANCO_REQUESTS
 from .lco_requests import LCO_REQUESTS
 from .saao_requests import SAAO_REQUESTS
 from .soar_requests import SOAR_REQUESTS
@@ -52,8 +54,6 @@ def test_submit_lco_request(lco_facility: LcoFacility):
 
 
 # Soar requests work exactly like LCO requests
-
-
 @pytest.fixture
 def soar_facility() -> SoarFacility:
     return SoarFacility()
@@ -79,8 +79,6 @@ def test_submit_soar_request(soar_facility: SoarFacility):
 
 
 # SAAO works the same as LCO and SOAR
-
-
 @pytest.fixture
 def saao_facility() -> SAAOFacility:
     return SAAOFacility()
@@ -94,3 +92,30 @@ def test_valid_saao_requests(saao_facility: SAAOFacility, request_group: Request
     if not valid:
         logger.error("Online validation failed. Server response: %s", errors)
     assert valid
+
+
+# BLANCO tests
+@pytest.fixture
+def blanco_facility() -> BlancoFacility:
+    return BlancoFacility()
+
+
+@pytest.mark.parametrize(
+    "request_group", BLANCO_REQUESTS.values(), ids=BLANCO_REQUESTS.keys()
+)
+def test_valid_blanco_requests(
+    blanco_facility: BlancoFacility, request_group: RequestGroup
+):
+    valid, errors = blanco_facility.validate_request_group(request_group)
+    if not valid:
+        logger.error("Online validation failed. Server response: %s", errors)
+    assert valid
+
+
+@pytest.mark.side_effect
+def test_submit_blanco_request(blanco_facility: BlancoFacility):
+    request_group_in = BLANCO_REQUESTS["blanco_newfirm"]
+    request_group_out = blanco_facility.submit_request_group(request_group_in)
+    assert request_group_out.id
+    assert request_group_out.state == "PENDING"
+    assert request_group_out.created