I-ALiRT - Coordinated coverage schedule (IMAP-Science-Operations-Center#2628)

Author: laspsandoval

imap_processing/ialirt/constants.py

@@ -1,6 +1,7 @@
 """Module for constants and useful shared classes used in I-ALiRT processing."""
 
 from dataclasses import dataclass
+from datetime import time
 from typing import NamedTuple
 
 import numpy as np
@@ -55,6 +56,8 @@ class StationProperties(NamedTuple):
     latitude: float  # latitude in degrees
     altitude: float  # altitude in kilometers
     min_elevation_deg: float  # minimum elevation angle in degrees
+    schedule_start: time | None = None  # station schedule start
+    schedule_end: time | None = None  # station schedule end
 
 
 # Verified by Observatory staff.
@@ -70,23 +73,31 @@ class StationProperties(NamedTuple):
         latitude=54.2632,  # degrees North
         altitude=0.1,  # approx 100 meters
         min_elevation_deg=5,  # 5 degrees is the requirement
+        schedule_start=None,
+        schedule_end=None,
     ),
     "Korea": StationProperties(
         longitude=126.2958,  # degrees East
         latitude=33.4273,  # degrees North
         altitude=0.1,  # approx 100 meters
         min_elevation_deg=5,  # 5 degrees is the requirement
+        schedule_start=None,
+        schedule_end=None,
     ),
     "Manaus": StationProperties(
         longitude=-59.969319,  # degrees East (negative = West)
         latitude=-2.891215,  # degrees North (negative = South)
         altitude=0.9578,  # approx 957.8 meters
         min_elevation_deg=5,  # 5 degrees is the requirement
+        schedule_start=None,
+        schedule_end=None,
     ),
     "SANSA": StationProperties(
         longitude=27.714,  # degrees East (negative = West)
         latitude=-25.888,  # degrees North (negative = South)
         altitude=1.542,  # approx 1542 meters
         min_elevation_deg=2,  # 5 degrees is the requirement
+        schedule_start=None,
+        schedule_end=None,
     ),
 }

imap_processing/ialirt/generate_coverage.py

@@ -4,7 +4,7 @@
 
 import numpy as np
 
-from imap_processing.ialirt.constants import STATIONS
+from imap_processing.ialirt.constants import STATIONS, StationProperties
 from imap_processing.ialirt.process_ephemeris import calculate_azimuth_and_elevation
 from imap_processing.spice.time import et_to_utc, str_to_et
 
@@ -28,6 +28,55 @@
 ]
 
 
+def create_schedule_mask(
+    station: StationProperties, time_range: np.ndarray
+) -> np.ndarray:
+    """
+    Create a boolean mask based on the static daily operating schedule.
+
+    Parameters
+    ----------
+    station : StationProperties
+        Ground station configuration.
+    time_range : np.ndarray
+        Array of ephemeris time (ET) values corresponding to the
+        coverage time.
+
+    Returns
+    -------
+    schedule_mask : np.ndarray
+        Boolean array True is operating window.
+    """
+    if station.schedule_start is None and station.schedule_end is None:
+        return np.ones(time_range.shape, dtype=bool)
+
+    utc_times = et_to_utc(time_range, format_str="ISOC")
+    utc_dt = utc_times.astype("datetime64[s]")
+
+    # seconds since midnight (UTC), vectorized
+    sec_of_day = (utc_dt - utc_dt.astype("datetime64[D]")) / np.timedelta64(1, "s")
+
+    schedule_mask = np.ones(time_range.shape, dtype=bool)
+
+    if station.schedule_start is not None:
+        start_sec = (
+            station.schedule_start.hour * 3600
+            + station.schedule_start.minute * 60
+            + station.schedule_start.second
+        )
+        schedule_mask &= sec_of_day >= start_sec
+
+    if station.schedule_end is not None:
+        end_sec = (
+            station.schedule_end.hour * 3600
+            + station.schedule_end.minute * 60
+            + station.schedule_end.second
+        )
+        schedule_mask &= sec_of_day <= end_sec
+
+    return schedule_mask
+
+
 def generate_coverage(
     start_time: str,
     outages: dict | None = None,
@@ -76,11 +125,18 @@ def generate_coverage(
                 end_et = str_to_et(end)
                 dsn_outage_mask |= (time_range >= start_et) & (time_range <= end_et)
 
-    for station_name, (lon, lat, alt, min_elevation) in stations.items():
+    for station_name, station in stations.items():
         _azimuth, elevation = calculate_azimuth_and_elevation(
-            lon, lat, alt, time_range, obsref="IAU_EARTH"
+            station.longitude,
+            station.latitude,
+            station.altitude,
+            time_range,
+            obsref="IAU_EARTH",
         )
-        visible = elevation > min_elevation
+        visible = elevation > station.min_elevation_deg
+
+        schedule_mask = create_schedule_mask(station, time_range)
+        visible &= schedule_mask
 
         outage_mask = np.zeros(time_range.shape, dtype=bool)
         if outages and station_name in outages:
@@ -133,9 +189,9 @@ def generate_coverage(
     coverage_dict["total_coverage_percent"] = total_coverage_percent
 
     # Ensure all stations are present in both dicts
-    for station in ALL_STATIONS:
-        coverage_dict.setdefault(station, np.array([], dtype="<U23"))
-        outage_dict.setdefault(station, np.array([], dtype="<U23"))
+    for station_name in ALL_STATIONS:
+        coverage_dict.setdefault(station_name, np.array([], dtype="<U23"))
+        outage_dict.setdefault(station_name, np.array([], dtype="<U23"))
 
     return coverage_dict, outage_dict
 

imap_processing/tests/ialirt/unit/test_generate_coverage.py

@@ -1,11 +1,14 @@
 """Test processEphemeris functions."""
 
-from datetime import datetime
+from datetime import datetime, time
+from types import SimpleNamespace
+from unittest.mock import patch
 
 import numpy as np
 import pytest
 
 from imap_processing.ialirt.generate_coverage import (
+    create_schedule_mask,
     format_coverage_summary,
     generate_coverage,
 )
@@ -109,3 +112,58 @@ def test_dsn(furnish_kernels):
 
         assert "I-ALiRT Coverage Summary" in output["summary"]
         assert 40.6 == output["total_coverage_percent"]
+
+
+@patch("imap_processing.ialirt.generate_coverage.et_to_utc")
+def test_create_schedule_mask(mock_et_to_utc):
+    """
+    Test create_schedule_mask.
+    """
+
+    mock_et_to_utc.return_value = np.array(
+        [
+            "2026-09-22T11:30:00.000",
+            "2026-09-22T11:35:00.000",
+            "2026-09-22T11:40:00.000",
+            "2026-09-22T11:45:00.000",
+            "2026-09-22T11:50:00.000",
+            "2026-09-22T11:55:00.000",
+            "2026-09-22T12:00:00.000",
+            "2026-09-22T12:05:00.000",
+            "2026-09-22T12:10:00.000",
+            "2026-09-22T12:15:00.000",
+            "2026-09-22T12:20:00.000",
+            "2026-09-22T12:25:00.000",
+            "2026-09-22T12:30:00.000",
+        ]
+    )
+
+    time_range = np.arange(13)
+
+    station = SimpleNamespace(
+        schedule_start=time(12, 0),
+        schedule_end=None,
+    )
+
+    mask = create_schedule_mask(station, time_range)
+
+    expected = np.array(
+        [
+            False,
+            False,
+            False,
+            False,
+            False,
+            False,
+            True,
+            True,
+            True,
+            True,
+            True,
+            True,
+            True,
+        ],
+        dtype=bool,
+    )
+
+    np.testing.assert_array_equal(mask, expected)