Add spectrograph routes and improve typing

pyproject.toml

@@ -25,13 +25,16 @@ packages = [
 include = []
 
 [tool.poetry.dependencies]
-python = "^3.11,<4.0"
+python = "^3.11,<3.13"
 sdsstools = "^1.2.0"
 typing-extensions = "^4.5.0"
 fastapi = ">=0.100.0"
 gunicorn = "^21.2.0"
 uvicorn = {extras = ["standard"], version = ">=0.24.0"}
 sdss-clu = "^2.2.1"
+aiohttp = "^3.9.0b1"
+influxdb-client = {extras = ["async"], version = "^1.38.0"}
+pandas = "^2.1.3"
 
 [tool.poetry.group.dev.dependencies]
 ipython = ">=8.0.0"

src/lvmapi/app.py

@@ -10,11 +10,12 @@
 
 from fastapi import FastAPI
 
-from lvmapi.routers import telescopes
+from lvmapi.routers import spectrographs, telescopes
 
 
 app = FastAPI()
 app.include_router(telescopes.router)
+app.include_router(spectrographs.router)
 
 
 @app.get("/")

src/lvmapi/config.yaml

@@ -2,3 +2,8 @@
 rabbitmq:
   host: lvm-hub.lco.cl
   port: 5672
+
+influxdb:
+  host: lvm-webapp.lco.cl
+  port: 9999
+  org: LVM

src/lvmapi/routers/spectrographs.py

@@ -0,0 +1,134 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# @Author: José Sánchez-Gallego (gallegoj@uw.edu)
+# @Date: 2023-11-11
+# @Filename: cryostats.py
+# @License: BSD 3-clause (http://www.opensource.org/licenses/BSD-3-Clause)
+
+from __future__ import annotations
+
+from typing import get_args
+
+import pandas
+from fastapi import APIRouter, HTTPException, Query
+from pydantic import BaseModel
+
+from lvmapi.tools import (
+    get_spectrograph_mechanics,
+    get_spectrograph_pressures,
+    get_spectrograph_temperature_label,
+    get_spectrograph_temperatures,
+    query_influxdb,
+)
+from lvmapi.types import Cameras, CamSpec, Sensors, Spectrographs
+
+
+class Temperatures(BaseModel):
+    time: str
+    camera: str
+    sensor: str
+    temperature: float
+
+
+router = APIRouter(prefix="/spectrographs", tags=["spectrographs"])
+
+
+@router.get("/", summary="List of spectrographs")
+async def get_cryostats() -> list[str]:
+    """Returns the list of cryostats."""
+
+    return list(get_args(Spectrographs))
+
+
+@router.get(
+    "/temperatures",
+    summary="Cryostat temperatures",
+    response_model=list[Temperatures],
+)
+async def get_temperatures(
+    start: str = Query("-30m", description="Flux-compatible start time"),
+    stop: str = Query("now()", description="Flux-compatible stop time"),
+    camera: CamSpec | None = Query(None, description="Camera to return, or all"),
+    sensor: Sensors | None = Query(None, description="Sensor to return, or all"),
+):
+    """Returns the temperatures of one or multiple cryostats."""
+
+    time_range = f"|> range(start: {start}, stop: {stop})"
+
+    spec_filter = r'|> filter(fn: (r) => r["_measurement"] =~ /lvmscp\.sp[1-3]/)'
+    if camera is not None:
+        spec_filter = (
+            f'|> filter(fn: (r) => r["_measurement"] == "lvmscp.sp{camera[-1]}")'
+        )
+
+    sensor_filter = r'|> filter(fn: (r) => r["_field"] =~ /mod(2|12)\/temp[abc]/)'
+    if camera is not None and sensor is not None:
+        field = get_spectrograph_temperature_label(camera[0], sensor)
+        sensor_filter = f'|> filter(fn: (r) => r["_field"] == "status.{field}")'
+
+    query = rf"""
+    from(bucket: "actors")
+        {time_range}
+        {spec_filter}
+        {sensor_filter}
+    """
+
+    try:
+        results = await query_influxdb(query)
+    except Exception:
+        raise HTTPException(500, detail="Failed querying InfluxDB.")
+
+    if len(results) == 0:
+        return []
+
+    results.loc[:, "time"] = results["_time"].map(lambda tt: tt.isoformat())
+    results.loc[:, "camera"] = pandas.Series("", dtype="S3")
+    results.loc[:, "sensor"] = pandas.Series("", dtype="S3")
+    results.loc[:, "temperature"] = results._value
+
+    for spec in ["sp1", "sp2", "sp3"]:
+        for cc in get_args(Cameras):
+            for ss in get_args(Sensors):
+                label = get_spectrograph_temperature_label(cc, ss)
+
+                results.loc[
+                    (results._measurement == f"lvmscp.{spec}")
+                    & (results._field == f"status.{label}"),
+                    "camera",
+                ] = f"{cc}{spec[-1]}"
+
+                results.loc[results._field == f"status.{label}", "sensor"] = ss
+
+    results = results.loc[:, ["time", "camera", "sensor", "temperature"]]
+
+    if camera:
+        results = results.loc[results.camera == camera, :]
+
+    if sensor:
+        results = results.loc[results.sensor == sensor, :]
+
+    return list(results.itertuples(index=False))
+
+
+@router.get("/{spectrograph}", summary="Cryostat basic information")
+@router.get("/{spectrograph}/summary", summary="Cryostat basic information")
+async def get_summary(
+    spectrograph: Spectrographs,
+    mechs: bool = Query(False, description="Return mechanics information?"),
+) -> dict[str, float | str]:
+    """Retrieves current spectrograph information (temperature, pressure, etc.)"""
+
+    try:
+        temps_response = await get_spectrograph_temperatures(spectrograph)
+        pres_reponse = await get_spectrograph_pressures(spectrograph)
+
+        if mechs:
+            mechs_response = await get_spectrograph_mechanics(spectrograph)
+        else:
+            mechs_response = {}
+
+    except Exception:
+        raise HTTPException(500, detail="Error retrieving cryostat information.")
+
+    return temps_response | pres_reponse | mechs_response

src/lvmapi/routers/telescopes.py

@@ -8,60 +8,61 @@
 
 from __future__ import annotations
 
-from typing import Literal, get_args
+from typing import get_args
 
 from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel
 
 from lvmapi.tools import CluClient
+from lvmapi.types import Coordinates, Telescopes
 
 
-Telescopes = Literal["sci", "spec", "skye", "skyw"]
-Frames = Literal["radec", "altaz"]
-Coordinates = Literal["ra", "dec", "alt", "az"]
+class PointingResponse(BaseModel):
+    ra: float
+    dec: float
+    alt: float
+    az: float
 
 
 router = APIRouter(prefix="/telescopes", tags=["telescopes"])
 
 
-@router.get("/")
-async def get_telescopes():
+@router.get("/", summary="List of telescopes")
+async def get_telescopes() -> list[str]:
     """Returns the list of telescopes."""
 
     return list(get_args(Telescopes))
 
 
-@router.get("/{telescope}/pointing")
-async def get_pointing(telescope: Telescopes, frame: Frames = "radec"):
+@router.get(
+    "/{telescope}/pointing",
+    summary="Telescope pointing",
+    response_model=PointingResponse,
+)
+async def get_pointing(telescope: Telescopes) -> PointingResponse:
     """Gets the pointing of a telescope."""
 
     try:
         async with CluClient() as client:
             status_cmd = await client.send_command(f"lvm.{telescope}.pwi", "status")
 
-        if frame == "radec":
-            ax0 = status_cmd.replies.get("ra_apparent_hours") * 15
-            ax1 = status_cmd.replies.get("dec_apparent_degs")
-        elif frame == "altaz":
-            ax0 = status_cmd.replies.get("altitude_degs")
-            ax1 = status_cmd.replies.get("azimuth_degs")
+        ra = status_cmd.replies.get("ra_apparent_hours") * 15
+        dec = status_cmd.replies.get("dec_apparent_degs")
+        alt = status_cmd.replies.get("altitude_degs")
+        az = status_cmd.replies.get("azimuth_degs")
 
     except Exception:
         raise HTTPException(
             status_code=500,
             detail="Error retrieving telescope information.",
         )
 
-    if frame == "radec":
-        return {"ra": ax0, "dec": ax1}
-    else:
-        return {"alt": ax0, "az": ax1}
+    return PointingResponse(ra=ra, dec=dec, alt=alt, az=az)
 
 
-@router.get("/{telescope}/{coordinate}")
-async def get_ra(telescope: Telescopes, coordinate: Coordinates):
+@router.get("/{telescope}/{coordinate}", summary="Telescope coordinates")
+async def get_ra(telescope: Telescopes, coordinate: Coordinates) -> float:
     """Returns a given coordinate for a telescope."""
 
-    frame = "radec" if coordinate in ["ra", "dec"] else "altaz"
-
-    pointing = await get_pointing(telescope, frame=frame)
-    return pointing[coordinate]
+    pointing = await get_pointing(telescope)
+    return getattr(pointing, coordinate)