Add transparency endpoing

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## Next version
 
+### 🚀 New
+
+* [#14](https://github.com/sdss/lvmapi/pull/14) Add `/transparency` endpoint to retrieve transparency data.
+
 ### 🔧 Fixed
 
 * Correctly calculate time lost in night metrics while the night is ongoing.

src/lvmapi/app.py

@@ -29,6 +29,7 @@
     spectrographs,
     tasks,
     telescopes,
+    transparency,
     weather,
 )
 from lvmapi.tools.kubernetes import Kubernetes
@@ -51,6 +52,7 @@
 app.include_router(actors.router)
 app.include_router(logs.router)
 app.include_router(notifications.router)
+app.include_router(transparency.router)
 
 
 @app.get("/id")

src/lvmapi/routers/transparency.py

@@ -0,0 +1,77 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# @Author: José Sánchez-Gallego (gallegoj@uw.edu)
+# @Date: 2024-11-11
+# @Filename: transparency.py
+# @License: BSD 3-clause (http://www.opensource.org/licenses/BSD-3-Clause)
+
+from __future__ import annotations
+
+from datetime import datetime
+from time import time
+
+from typing import Annotated
+
+from fastapi import APIRouter, Query
+from pydantic import BaseModel, Field
+
+from lvmapi.tools.transparency import get_transparency
+
+
+router = APIRouter(prefix="/transparency", tags=["transparency"])
+
+
+class TransparencyResponse(BaseModel):
+    """Response model for transparency measurements."""
+
+    start_time: Annotated[
+        float,
+        Field(description="Start time of the measurements as a UNIX timestamp"),
+    ]
+    end_time: Annotated[
+        float,
+        Field(description="End time of the measurements as a UNIX timestamp"),
+    ]
+    data: Annotated[
+        list[TransparencyData],
+        Field(description="Transparency data"),
+    ]
+
+
+class TransparencyData(BaseModel):
+    """Model for transparency data."""
+
+    time: Annotated[datetime, Field(description="Time of the measurement")]
+    telescope: Annotated[str, Field(description="Telescope name")]
+    zero_point: Annotated[float, Field(description="Zero-point value")]
+
+
+@router.get("/", summary="Transparency measurements")
+async def route_get_transparency(
+    start_time: Annotated[
+        float | None,
+        Query(description="Start time as a UNIX timestamp"),
+    ] = None,
+    end_time: Annotated[
+        float | None,
+        Query(description="End time as a UNIX timestamp"),
+    ] = None,
+) -> TransparencyResponse:
+    """Returns transparency measurements.
+
+    Without any parameters, returns the transparency measurements for the last hour.
+
+    """
+
+    if start_time is None or end_time is None:
+        end_time = time()
+        start_time = end_time - 3600
+
+    data = await get_transparency(start_time, end_time)
+
+    return TransparencyResponse(
+        start_time=start_time,
+        end_time=end_time,
+        data=[TransparencyData(**row) for row in data.to_dicts()],
+    )

src/lvmapi/tools/transparency.py

@@ -0,0 +1,36 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# @Author: José Sánchez-Gallego (gallegoj@uw.edu)
+# @Date: 2024-11-11
+# @Filename: transparency.py
+# @License: BSD 3-clause (http://www.opensource.org/licenses/BSD-3-Clause)
+
+from __future__ import annotations
+
+import polars
+
+from lvmapi.tools.influxdb import query_influxdb
+
+
+async def get_transparency(start_time: float, end_time: float):
+    """Returns transparency measurements."""
+
+    query = rf"""
+from(bucket: "actors")
+  |> range(start: {int(start_time)}, stop: {int(end_time)})
+  |> filter(fn: (r) => (r["_measurement"] =~ /lvm\.[a-z]+\.guider/) and
+                        (r["_field"] == "measured_pointing.zero_point"))
+  |> yield(name: "mean")
+"""
+
+    data = await query_influxdb(query)
+
+    # Clean up the dataframe.
+    data = data.select(
+        time=polars.col._time,
+        telescope=polars.col._measurement.str.extract(r"lvm\.([a-z]+)\.guider"),
+        zero_point=polars.col._value,
+    )
+
+    return data