Add exposure_list DF to lvmcore in post-observing recipe

Author: albireox

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## Next release
+
+### ✨ Improved
+
+* Add an `exposure_list_<MJD>.parquet` file to `LVMCORE_DIR/exposure_list` at the end of the night with the list of exposures taken.
+
+
 ## 1.11.5 - December 26, 2025
 
 ### ✨ Improved

scripts/exposure_list_to_lvmcore.py

@@ -0,0 +1,68 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# @Author: José Sánchez-Gallego (gallegoj@uw.edu)
+# @Date: 2026-02-27
+# @Filename: exposure_list_to_lvmcore.py
+# @License: BSD 3-clause (http://www.opensource.org/licenses/BSD-3-Clause)
+
+from __future__ import annotations
+
+import asyncio
+import os
+import pathlib
+
+from rich.progress import BarColumn, MofNCompleteColumn, Progress, TextColumn
+
+from sdsstools.time import get_sjd
+
+from gort.tools import get_exposure_list
+
+
+async def exposure_list_to_lvmcore(overwrite: bool = False):
+    """Populates ``lvmcore`` with ``exposure_list`` files for all MJDs."""
+
+    lvmcore_dir = os.environ.get("LVMCORE_DIR", None)
+    if not lvmcore_dir:
+        raise RuntimeError("LVMCORE_DIR environment variable is not set.")
+
+    lvmcore_dir = pathlib.Path(lvmcore_dir)
+    if not lvmcore_dir.exists():
+        raise RuntimeError(f"LVMCORE_DIR {lvmcore_dir} does not exist.")
+
+    exp_list_dir = lvmcore_dir / "exposure_list"
+    exp_list_dir.mkdir(exist_ok=True)
+
+    progress = Progress(
+        TextColumn("[progress.description]{task.description}"),
+        BarColumn(bar_width=None),
+        MofNCompleteColumn(),
+        expand=True,
+        auto_refresh=True,
+    )
+
+    mjd0 = 60007
+    mjd1 = get_sjd("LCO")
+
+    task = progress.add_task(f"Processing MJD {mjd0}", total=(mjd1 - mjd0 + 1))
+
+    with progress:
+        for mjd in range(mjd0, mjd1 + 1):
+            exp_list_file = exp_list_dir / f"exposure_list_{mjd}.parquet"
+            if exp_list_file.exists() and not overwrite:
+                progress.update(task, advance=1)
+                continue
+
+            progress.update(task, description=f"Processing MJD {mjd}")
+
+            try:
+                df = await get_exposure_list(mjd)
+                df.write_parquet(exp_list_file)
+            except Exception:
+                pass
+
+            progress.update(task, advance=1)
+
+
+if __name__ == "__main__":
+    asyncio.run(exposure_list_to_lvmcore())

src/gort/recipes/operations.py

@@ -10,13 +10,22 @@
 
 import asyncio
 import json
+import os
+import pathlib
 
 from typing import TYPE_CHECKING, ClassVar, Coroutine
 
 from rich.prompt import Confirm
 
+from sdsstools.time import get_sjd
+
 from gort.overwatcher.helpers import get_actor_ping, restart_actors
-from gort.tools import decap, get_lvmapi_route, overwatcher_is_running
+from gort.tools import (
+    decap,
+    get_exposure_list,
+    get_lvmapi_route,
+    overwatcher_is_running,
+)
 
 from .base import BaseRecipe
 
@@ -459,6 +468,29 @@ async def recipe(self, send_email: bool = True, force_park: bool = False):
             else:
                 self.gort.log.info("Overwatcher has been disabled.")
 
+        # Add the list of exposures to lvmcore
+        lvmcore_dir = os.environ.get("LVMCORE_DIR", None)
+        if not lvmcore_dir:
+            self.gort.log.warning("LVMCORE_DIR environment variable is not set.")
+            return
+
+        lvmcore_dir = pathlib.Path(lvmcore_dir)
+        if not lvmcore_dir.exists():
+            self.gort.log.warning(f"LVMCORE_DIR {lvmcore_dir} does not exist.")
+            return
+
+        exp_list_dir = lvmcore_dir / "exposure_list"
+        exp_list_dir.mkdir(exist_ok=True)
+
+        mjd = get_sjd("LCO")
+        exp_list_file = exp_list_dir / f"exposure_list_{mjd}.parquet"
+
+        try:
+            df = await get_exposure_list(mjd)
+            df.write_parquet(exp_list_file)
+        except Exception as err:
+            self.gort.log.warning(f"Failed to write exposure list: {err}")
+
 
 class RebootAGsRecipe(BaseRecipe):
     """Reboots the AG cameras."""

src/gort/tools.py

@@ -36,6 +36,7 @@
     TypedDict,
     TypeVar,
     cast,
+    overload,
 )
 
 import httpx
@@ -106,6 +107,7 @@
     "record_overheads",
     "OverheadDict",
     "ping_host",
+    "get_exposure_list",
 ]
 
 AnyPath = str | os.PathLike
@@ -1224,3 +1226,62 @@ async def ping_host(host: str):
 
     return_code = await cmd.wait()
     return return_code == 0
+
+
+@overload
+async def get_exposure_list(
+    mjd: int | None = None,
+    as_dataframe: bool = True,
+) -> polars.DataFrame: ...
+
+
+@overload
+async def get_exposure_list(
+    mjd: int | None = None,
+    as_dataframe: bool = False,
+) -> dict: ...
+
+
+async def get_exposure_list(
+    mjd: int | None = None,
+    as_dataframe: bool = True,
+) -> dict | polars.DataFrame:
+    """Retrieves the list of exposures for a given MJD from the LVM API."""
+
+    mjd = mjd or get_sjd("LCO")
+    data: dict = await get_lvmapi_route(f"/logs/exposures/data/{mjd}", timeout=15)
+
+    if not as_dataframe:
+        return data
+
+    data_df: list[dict[str, Any]] = []
+
+    for exposure in data.values():
+        lamps = exposure.pop("lamps")
+        data_df.append({**exposure, **lamps})
+
+    schema = {
+        "exposure_no": polars.Int32,
+        "mjd": polars.Int32,
+        "obstime": polars.String,
+        "image_type": polars.String,
+        "exposure_time": polars.Float32,
+        "ra": polars.Float64,
+        "dec": polars.Float64,
+        "airmass": polars.Float32,
+        "n_standards": polars.Int16,
+        "n_cameras": polars.Int16,
+        "object": polars.String,
+        "dpos": polars.Int16,
+        "Argon": polars.Boolean,
+        "Neon": polars.Boolean,
+        "LDLS": polars.Boolean,
+        "Quartz": polars.Boolean,
+        "HgNe": polars.Boolean,
+        "Xenon": polars.Boolean,
+    }
+
+    if len(data_df) == 0:
+        return polars.DataFrame(schema=schema)
+
+    return polars.DataFrame(data_df, schema=schema)