DM-53787: Extend the API to serve data ready for a React SPA

.gitignore

@@ -166,3 +166,6 @@ local_scripts/
 
 # jest
 coverage/
+
+# github prompts
+.github/prompts/*

README.md

@@ -13,6 +13,11 @@ The codebase aims to be modular and easy to extend: a backend built with FastAPI
 
 This repository contains the core services and documentation to get started quickly and to follow project conventions.
 
+## API discovery
+
+As per every FastAPI project, the full API can be found at `/docs`
+e.g. for the [dev deployment at USDF](https://usdf-rsp-dev.slac.stanford.edu/rubintv/docs)
+
 ## Project layout (typical)
 
 - `python/lsst/ts/rubintv` - FastAPI application and service code

pyproject.toml

@@ -68,5 +68,28 @@ dev = [
   "types-PyYAML",
   "beautifulsoup4",
   "types-python-dateutil",
-  "types-redis"
+  "types-redis",
+  "flake8"
   ]
+
+[tool.mypy]
+python_version = "3.12"
+strict = false
+warn_return_any = false
+warn_unused_configs = true
+disallow_untyped_defs = false
+disallow_incomplete_defs = false
+check_untyped_defs = true
+disallow_untyped_decorators = false
+no_implicit_optional = true
+warn_redundant_casts = true
+warn_unused_ignores = true
+warn_no_return = true
+warn_unreachable = true
+ignore_missing_imports = true
+show_error_codes = true
+disable_error_code = "type-arg"
+
+[[tool.mypy.overrides]]
+module = "tests.*"
+ignore_errors = true

python/lsst/ts/rubintv/background/background_helpers.py

@@ -1,5 +1,5 @@
 from lsst.ts.rubintv.config import rubintv_logger
-from lsst.ts.rubintv.models.models import Event
+from lsst.ts.rubintv.models.models import Event, ExtensionInfo, StructuredData
 
 logger = rubintv_logger()
 
@@ -33,12 +33,17 @@ async def get_next_previous_from_table(
     if chan_table == {}:
         return (None, None)
 
+    # this should never happen
+    if isinstance(event.seq_num, str):
+        logger.warning("Looking for prev/next for Per Day Event")
+        return (None, None)
+
     # creates a 'None' padded list of seq. nums
-    all_seqs = sorted(set(chan_table.keys() | {event.seq_num_force_int()}))
+    all_seqs = sorted(set(chan_table.keys() | {event.seq_num}))
     padded_seqs = [None, *all_seqs, None]
 
     # find the index of event's seq num in that padded list
-    index = padded_seqs.index(event.seq_num_force_int())
+    index = padded_seqs.index(event.seq_num)
 
     next_seq = padded_seqs[index + 1]
     prev_seq = padded_seqs[index - 1]
@@ -49,3 +54,90 @@ async def get_next_previous_from_table(
     )
 
     return nxt_prv
+
+
+async def make_structured_data(
+    events: list[Event],
+) -> StructuredData:
+    """Build structured data from events.
+
+    Structure: {channel_name: {seq_num1, seq_num2, ...}}
+
+    This is a compressed format suitable for WebSocket transmission.
+
+    Parameters
+    ----------
+    events : list[Event]
+        List of events to build structured data from.
+
+    Returns
+    -------
+    structured: StructuredData
+        Structured data mapping channel names to sets of sequence numbers.
+    """
+    structured: StructuredData = {}
+    for event in events:
+        channel_name = event.channel_name
+        seq_num = event.seq_num
+        if channel_name not in structured:
+            structured[channel_name] = set()
+        structured[channel_name].add(seq_num)
+    return structured
+
+
+async def make_extension_info(
+    events: list[Event],
+) -> ExtensionInfo:
+    """Build extension info from events.
+
+    Structure:
+        {channel_name: {"default": "ext", "exceptions": {seq_num: "ext"}}}
+
+    This is needed for the frontend to reconstruct filenames from
+    structuredData.
+
+    Parameters
+    ----------
+    events : list[Event]
+        List of events to build extension info from.
+
+    Returns
+    -------
+    extension_info: ExtensionInfo
+        Extension info mapping channel names to extension metadata.
+        Each channel has a default extension and exceptions for specific
+        seq_nums.
+    """
+    extension_info: ExtensionInfo = {}
+    extension_counts: dict[str, dict[str, int]] = {}
+
+    for event in events:
+        channel_name = event.channel_name
+        ext = event.ext
+        if channel_name not in extension_info:
+            extension_info[channel_name] = {"default": "jpg", "exceptions": {}}
+            extension_counts[channel_name] = {}
+
+        # Track extension frequency to determine default
+        if ext not in extension_counts[channel_name]:
+            extension_counts[channel_name][ext] = 0
+        extension_counts[channel_name][ext] += 1
+
+    # Set default to most common extension, record others as exceptions
+    for channel_name, ext_counts in extension_counts.items():
+        if ext_counts:
+            default_ext = max(ext_counts.items(), key=lambda x: x[1])[0]
+            extension_info[channel_name]["default"] = default_ext
+
+    # Mark extensions different from default as exceptions
+    for event in events:
+        channel_name = event.channel_name
+        ext = event.ext
+        default_ext = extension_info[channel_name]["default"]
+        if ext != default_ext:
+            seq_num = (
+                int(event.seq_num) if isinstance(event.seq_num, str) else event.seq_num
+            )
+            extension_info[channel_name]["exceptions"][seq_num] = ext
+
+    return extension_info

python/lsst/ts/rubintv/background/currentpoller.py

@@ -2,21 +2,26 @@
 from asyncio import Event as AsyncioEvent
 from asyncio import sleep
 from time import time
-from typing import AsyncGenerator
+from typing import TYPE_CHECKING, AsyncGenerator
 
-from lsst.ts.rubintv.background.background_helpers import get_next_previous_from_table
+from lsst.ts.rubintv.background.background_helpers import (
+    get_next_previous_from_table,
+    make_extension_info,
+    make_structured_data,
+)
 from lsst.ts.rubintv.config import rubintv_logger
 from lsst.ts.rubintv.handlers.websocket_notifiers import notify_ws_clients
 from lsst.ts.rubintv.models.models import (
     Camera,
     Event,
+    ExtensionInfo,
     Location,
     NightReport,
     NightReportData,
 )
 from lsst.ts.rubintv.models.models import ServiceMessageTypes as MessageType
 from lsst.ts.rubintv.models.models import ServiceTypes as Service
-from lsst.ts.rubintv.models.models import get_current_day_obs
+from lsst.ts.rubintv.models.models import StructuredData, get_current_day_obs
 from lsst.ts.rubintv.models.models_helpers import (
     all_objects_to_events,
     make_table_from_event_list,
@@ -25,6 +30,9 @@
 from lsst.ts.rubintv.s3_connection_pool import get_shared_s3_client
 from lsst.ts.rubintv.s3client import S3Client
 
+if TYPE_CHECKING:
+    from lsst.ts.rubintv.background.historicaldata import HistoricalPoller
+
 logger = rubintv_logger()
 
 
@@ -44,10 +52,13 @@ def __init__(
         test_mode: bool = False,
     ) -> None:
         self._s3clients: dict[str, S3Client] = {}
+        self._historical_poller: HistoricalPoller | None = None
         self._objects: dict[str, list] = {}
         self._events: dict[str, list[Event]] = {}
         self._metadata: dict[str, dict] = {}
         self._table: dict[str, dict[int, dict[str, dict]]] = {}
+        self._structured_events: dict[str, StructuredData] = {}
+        self._extension_info: dict[str, ExtensionInfo] = {}
         self._per_day: dict[str, dict[str, dict]] = {}
         self._yesterday_prefixes: dict[str, list[str]] = {}
         self._most_recent_events: dict[str, Event] = {}
@@ -61,17 +72,23 @@ def __init__(
         self.completed_first_poll_event = first_pass_event
 
         self.locations = locations
-        self._current_day_obs = get_current_day_obs()
+        self._last_day_obs = get_current_day_obs()
         for location in locations:
             self._s3clients[location.name] = get_shared_s3_client(
                 location.profile_name, location.bucket_name, location.endpoint_url
             )
 
+    def set_historical_poller(self, hp: "HistoricalPoller") -> None:
+        """Set reference to HistoricalPoller for day rollover integration."""
+        self._historical_poller = hp
+
     async def clear_todays_data(self) -> None:
         self._objects = {}
         self._events = {}
         self._metadata = {}
         self._table = {}
+        self._structured_events = {}
+        self._extension_info = {}
         self._per_day = {}
         self._most_recent_events = {}
         self._nr_metadata = {}
@@ -102,18 +119,22 @@ async def check_for_empty_per_day_channels(self) -> None:
                 ]
                 loc_prefixes = self._yesterday_prefixes[location.name]
                 for chan in missing_chans:
-                    prefix = f"{camera.name}/{self._current_day_obs}/{chan.name}"
+                    prefix = f"{camera.name}/{self._last_day_obs}/{chan.name}"
                     loc_prefixes.append(prefix)
 
     async def poll_buckets_for_todays_data(self, test_day: str = "") -> None:
         time_total = 0.0
         while True:
             timer_start = time()
             try:
-                if self._current_day_obs != get_current_day_obs():
+                if self._last_day_obs != get_current_day_obs():
+                    # Day has changed - integrate with historical before
+                    # clearing
+                    if self._historical_poller:
+                        await self._historical_poller.integrate_todays_data(self)
                     await self.check_for_empty_per_day_channels()
                     await self.clear_todays_data()
-                day_obs = self._current_day_obs = get_current_day_obs()
+                day_obs = self._last_day_obs = get_current_day_obs()
 
                 for location in self.locations:
                     client = self._s3clients[location.name]
@@ -237,8 +258,18 @@ async def process_channel_objects(
 
             table = await self.make_channel_table(camera, events)
             self._table[loc_cam] = table
+
+            structured = await make_structured_data(events)
+            self._structured_events[loc_cam] = structured
+
+            ext_info = await make_extension_info(events)
+            self._extension_info[loc_cam] = ext_info
+            ws_payload = {
+                "structuredData": structured,
+                "extensionInfo": ext_info,
+            }
             await notify_ws_clients(
-                Service.CAMERA, MessageType.CAMERA_TABLE, loc_cam, table
+                Service.CAMERA, MessageType.CAMERA_TABLE, loc_cam, ws_payload
             )
 
         # clear all relevant prefixes from the store looking for
@@ -473,6 +504,21 @@ async def get_current_channel_table(
         table = self._table.get(loc_cam, {})
         return table
 
+    async def get_current_structured_data(
+        self, location_name: str, camera: Camera
+    ) -> dict[str, set[int | str]]:
+        """Get compressed structured data for current day.
+
+        Returns:
+            {channel_name: [seq_num1, seq_num2, ...]}
+        """
+        loc_cam = self._get_loc_cam(location_name, camera)
+        structured = self._structured_events.get(loc_cam, {})
+        # replaced_sets: dict[str, list[int | str]] = {
+        #     chan: list(seq_nums) for chan, seq_nums in structured.items()
+        # }
+        return structured
+
     async def get_current_per_day_data(
         self, location_name: str, camera: Camera
     ) -> dict[str, dict[str, dict]]:
@@ -563,10 +609,17 @@ async def get_latest_data(
     ) -> AsyncGenerator:
         match service:
             case Service.CAMERA:
-                channel_data = await self.get_current_channel_table(
+                structured = await self.get_current_structured_data(
                     location.name, camera
                 )
-                yield MessageType.CAMERA_TABLE, channel_data
+                ext_info = self._extension_info.get(
+                    f"{location.name}/{camera.name}", {}
+                )
+                ws_payload = {
+                    "structuredData": structured,
+                    "extensionInfo": ext_info,
+                }
+                yield MessageType.CAMERA_TABLE, ws_payload
 
                 metadata = await self.get_current_metadata(location.name, camera)
                 yield MessageType.CAMERA_METADATA, metadata
@@ -620,7 +673,7 @@ async def get_latest_data(
                     yield MessageType.CAMERA_PER_DAY, latest_per_day
 
     async def get_all_channel_names_for_seq_num(
-        self, location_name: str, camera_name: str, seq_num: int
+        self, location_name: str, camera_name: str, seq_num: int | str
     ) -> list[str]:
         """Get all channel names for a given sequence number.
         Parameters