Cfht

.gitignore

@@ -176,3 +176,4 @@ cython_debug/
 # Direnv
 .envrc
 .zed/
+build/

Justfile

@@ -26,6 +26,10 @@ update-blanco:
 update-saao:
     curl https://ocsio.saao.ac.za/api/instruments/ | codegen/lco/generator.py SAAO > src/aeonlib/ocs/saao/instruments.py
 
+# Update the CFHT generated models via swagger
+update-cfht:
+    codegen/cfht/generator.py > src/aeonlib/cfht/models.py
+
 # Update all generated instrument files
 update-all: update-lco update-soar update-saao update-blanco
     @echo "All updates completed"

codegen/cfht/generator.py

@@ -0,0 +1,66 @@
+#!/usr/bin/env python3
+import sys
+from typing import Any
+
+import httpx
+import yaml
+from datamodel_code_generator import (
+    DataModelType,
+    Formatter,
+    GenerateConfig,
+    InputFileType,
+    OpenAPIScope,
+    TargetPydanticVersion,
+    generate,
+)
+from datamodel_code_generator import (
+    Error as DataModelCodegenError,
+)
+
+OPENAPI_SPEC_URL = "https://hou-stage.cfht.hawaii.edu/api-docs/pi_api.yaml"
+GENERATED_HEADER = "# Auto generated file, do not edit\n# ruff: noqa: E741\n"
+
+
+def generate_models(openapi_document: dict[str, Any]) -> str:
+    config = GenerateConfig(
+        input_file_type=InputFileType.OpenAPI,
+        input_filename="cfhtopenapi.yaml",
+        openapi_scopes=[OpenAPIScope.Schemas],
+        output_model_type=DataModelType.PydanticV2BaseModel,
+        base_class="aeonlib.cfht.base_model.CFHTBaseModel",
+        target_pydantic_version=TargetPydanticVersion.V2_11,
+        use_annotated=True,
+        field_constraints=True,
+        set_default_enum_member=True,
+        snake_case_field=True,
+        allow_population_by_field_name=True,
+        type_overrides={"DoubleValue": "aeonlib.cfht.types.DoubleValue"},
+        formatters=[Formatter.BUILTIN],
+    )
+
+    result = generate(openapi_document, config=config)
+    if not isinstance(result, str):
+        raise TypeError(
+            "Expected string output from datamodel-code-generator "
+            f"got {type(result).__name__}."
+        )
+
+    return GENERATED_HEADER + result + "\n"
+
+
+def main() -> int:
+    response = httpx.get(OPENAPI_SPEC_URL)
+    response.raise_for_status()
+    document = yaml.safe_load(response.content)
+    generated = generate_models(document)
+    _ = sys.stdout.write(generated)
+
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        raise SystemExit(main())
+    except (yaml.YAMLError, DataModelCodegenError, OSError, TypeError) as exc:
+        print(f"CFHT model generation failed: {exc}", file=sys.stderr)
+        raise SystemExit(1)

examples/cfht.ipynb

@@ -0,0 +1,173 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "id": "5accbf56-995b-4a45-9d64-701be9fd9b49",
+   "metadata": {},
+   "source": [
+    "# CFHT AEONLib Demonstration Notebook\n",
+    "The online tests are also a good reference and are more extensive. See tests/cfht/test_online.py"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 1,
+   "id": "08a7f23c-7c4e-4a1f-9481-562a2a391ad0",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import random\n",
+    "# Import generic Aeonlib models, as well as CFHT facilties\n",
+    "from aeonlib.cfht.facility import CFHTFacility\n",
+    "from aeonlib.cfht.conversions import target_data_from_aeon\n",
+    "from aeonlib.cfht.models import Instrument, TargetDataMagnitude\n",
+    "from aeonlib.conf import settings\n",
+    "from aeonlib.models import SiderealTarget"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "a7097b42-3a61-4a13-bc61-1543c231037f",
+   "metadata": {},
+   "source": [
+    "## Initialize the CFHT facility and select an observing program"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 2,
+   "id": "c10a85c8-613b-4a57-8de6-2c0c107f2b51",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# Normally credentials are automatically picked up by the environment. For this notebook we supply them directly\n",
+    "settings.cfht_access_token = \"<your api token here>\"\n",
+    "settings.cfht_api_root = \"https://api-stage.cfht.hawaii.edu/\"\n",
+    "facility = CFHTFacility(settings=settings)\n",
+    "\n",
+    "# Get a list of Programs available to us, and select the one we wish to use. Alternatively, the facility can be\n",
+    "# instatiated with a program token directly, which is probably what you'd want to use in normal circumstances.\n",
+    "programs = facility.programs()\n",
+    "if not programs:\n",
+    "    raise ValueError(\"No programs found. Check the Kealahou Phase2 Tool\")\n",
+    "# Set the first available program as the active one\n",
+    "program = programs[0]\n",
+    "facility.select_program(program)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "44a88310-a8b5-4bd2-a0d2-5885728572d1",
+   "metadata": {},
+   "source": [
+    "## Define a target\n",
+    "We use a generic AEONLib target as a starting point here, as it makes it easier to share between other non-CFHT facilities. If CFHT is the only facility you will be using, it might be easier to construct an aeonlib.cfht.models.TargetData instance directly."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 3,
+   "id": "8707a1a0-fddc-41f0-b931-049c56703129",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# Start with the AEONLib target\n",
+    "sidereal_target = SiderealTarget(\n",
+    "    name=\"AEONLib CFHT Notebook Target\",\n",
+    "    type=\"ICRS\",\n",
+    "    ra=320.11,\n",
+    "    dec=-42.0\n",
+    ")\n",
+    "# Use it to bootstrap a full CFHT TargetData\n",
+    "target_data = target_data_from_aeon(sidereal_target)\n",
+    "# Fill in CFHT specific data\n",
+    "target_data.token = f\"{facility.program_token}-{random.randint(1000000000, 9999999999)}\"\n",
+    "target_data.magnitude = TargetDataMagnitude(ab=10.0)\n",
+    "target_data.temperature_effective = 1234.5\n",
+    "target_data.standard_star = False\n",
+    "target_data.pointing_offset_token = f\"00AZ00-PO+{Instrument.megacam.value}+1\""
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "2ac0a48f-bd0f-43a0-912f-6c4e7d0340a9",
+   "metadata": {},
+   "source": [
+    "## Create and list targets"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 4,
+   "id": "7df70290-4fd6-4958-b59e-cb6aee52d32c",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "token='25BE25-4399809431' name='AEONLib CFHT Notebook Target' label=5 version=1 fixed_target=TargetDataFixedTarget(coordinate=SkyCoordinate(ra=320.11, dec=-42.0), proper_motion=FixedTargetProperMotion(ra_mas=None, dec_mas=None), computed_coordinate=None, estimated_radial_velocity_kmps=None) moving_target=None magnitude=TargetDataMagnitude(u=None, b=None, v=None, r=None, i=None, g=None, j=None, h=None, k=None, uu=None, gg=None, rr=None, ii=None, zz=None, ab=10.0) temperature_effective=1234.5 standard_star=None linked_target_identifiers=None finding_chart=[] pointing_offset_token='00AZ00-PO+MEGACAM+1' pointing_offset=PointingOffsetData(token='00AZ00-PO+MEGACAM+1', name='1', offset=OffsetCoordinate(ra_offset=None, dec_offset=None, exposure_number=None), label=None, version=None, user_token='SYSTEM', instrument=<Instrument.megacam: 'MEGACAM'>, is_system=True)\n"
+     ]
+    }
+   ],
+   "source": [
+    "target = facility.create_or_update_target(target_data, Instrument.megacam)\n",
+    "all_targets = facility.targets()\n",
+    "try:\n",
+    "    found_target = next(t for t in all_targets if t.token == target.token)\n",
+    "    print(found_target)\n",
+    "except StopIteration:\n",
+    "    raise ValueError(\"Created target did not appear in target list!\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "728dc77e-24ce-411d-a549-c00e3eb85492",
+   "metadata": {},
+   "source": [
+    "## Delete created target\n",
+    "Now would be a good time to check the K2 tool in your browser to confirm the target appears in the targets section before it is deleted."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 5,
+   "id": "828e4200-232c-4e82-b69b-f10ed839a6eb",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "if target.token is not None:\n",
+    "    facility.delete_target(target.token)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "51eb8bf2-0f7c-4a01-b990-04d6bcc4a402",
+   "metadata": {},
+   "outputs": [],
+   "source": []
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3 (ipykernel)",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.14.7"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}

pyproject.toml

@@ -20,6 +20,7 @@ build-backend = "hatchling.build"
 codegen = [
     "jinja2>=3.1.6",
     "textcase>=0.2.1",
+    "datamodel-code-generator>=0.66.3",
 ]
 
 dev = [
@@ -55,6 +56,7 @@ log_cli_format = "%(levelname)s [%(name)s %(filename)s:%(lineno)s %(funcName)s()
 extend-exclude = [
     "*.ipynb",
     "src/aeonlib/ocs/**/instruments.py",
+    "src/aeonlib/cfht/models.py"
 ]
 force-exclude = true
 

src/aeonlib/cfht/base_model.py

@@ -0,0 +1,18 @@
+from pydantic import BaseModel, ConfigDict
+
+
+class CFHTBaseModel(BaseModel):
+    """Base model for CFHT generated schemas.
+    This allows us to configure all derived classes
+    if necessary
+    """
+
+    model_config = ConfigDict(validate_assignment=True)
+
+    def api_dump(self, **kwargs):
+        return self.model_dump(
+            mode="json",
+            by_alias=True,
+            exclude_none=True,
+            **kwargs,
+        )

src/aeonlib/cfht/conversions.py

@@ -0,0 +1,60 @@
+from functools import singledispatch
+from typing import Any, overload
+
+from aeonlib.models import TARGET_TYPES, SiderealTarget
+
+from .models import TargetData, TargetDataFixedTarget
+
+# This is a bit over-engineered at the moment. I started writing
+# it before I realized that non-sidereal targets were going to require
+# a lot more work. Well, the infrastructure is here if we ever implement
+# non-sidereal target conversions.
+
+
+class FixedTargetData(TargetData):
+    """CFHT target data guaranteed to contain sidereal target fields."""
+
+    fixed_target: TargetDataFixedTarget
+
+
+def _sidereal_target_payload(target: SiderealTarget) -> dict[str, Any]:
+    return {
+        "name": target.name,
+        "fixed_target": {
+            "coordinate": {
+                "ra": target.ra.to_value("deg"),
+                "dec": target.dec.to_value("deg"),
+            },
+            "proper_motion": {
+                "ra_mas": target.proper_motion_ra,
+                "dec_mas": target.proper_motion_dec,
+            },
+        },
+    }
+
+
+@singledispatch
+def _target_data_from_aeon(target: object) -> TargetData:
+    raise TypeError(f"Cannot convert {type(target).__name__} to CFHT TargetData")
+
+
+@_target_data_from_aeon.register
+def _convert_sidereal_target_data(target: SiderealTarget) -> FixedTargetData:
+    return FixedTargetData.model_validate(_sidereal_target_payload(target))
+
+
+# TODO: Register non-sidereal target conversions here.
+# They will require a JPL Horizons client or API call because CFHT
+# uses a series of astrometric coordinates instead of orbital elements.
+
+
+@overload
+def target_data_from_aeon(target: SiderealTarget) -> FixedTargetData: ...
+
+
+@overload
+def target_data_from_aeon(target: TARGET_TYPES) -> TargetData: ...
+
+
+def target_data_from_aeon(target: TARGET_TYPES) -> TargetData:
+    return _target_data_from_aeon(target)