I-ALiRT - Coordinated coverage schedule

imap_processing/ialirt/generate_coverage.py

@@ -26,9 +26,13 @@
     "DSS-74",
     "DSS-75",
 ]
+# Non-DSN stations must be listed in order of priority.
+NON_DSN_STATIONS = {
+    "Kiel": STATIONS["Kiel"],
+}
 
 
-def generate_coverage(
+def generate_coverage(  # noqa: PLR0912
     start_time: str,
     outages: dict | None = None,
     dsn: dict | None = None,
@@ -55,9 +59,6 @@ def generate_coverage(
     duration_seconds = 24 * 60 * 60  # 86400 seconds in 24 hours
     time_step = 5 * 60  # 5 min in seconds
 
-    stations = {
-        "Kiel": STATIONS["Kiel"],
-    }
     coverage_dict = {}
     outage_dict = {}
 
@@ -67,20 +68,33 @@ def generate_coverage(
     time_range = np.arange(start_et_input, stop_et_input, time_step)
     total_visible_mask = np.zeros(time_range.shape, dtype=bool)
 
-    # Precompute DSN outage mask for non-DSN stations
+    # Precompute DSN occupied mask for non-DSN stations
+    dsn_contact_mask = np.zeros(time_range.shape, dtype=bool)
     dsn_outage_mask = np.zeros(time_range.shape, dtype=bool)
     if dsn:
-        for dsn_contacts in dsn.values():
-            for start, end in dsn_contacts:
-                start_et = str_to_et(start)
-                end_et = str_to_et(end)
-                dsn_outage_mask |= (time_range >= start_et) & (time_range <= end_et)
+        for dsn_station, dsn_contacts in dsn.items():
+            for contact_start, contact_end in dsn_contacts:
+                contact_start_et = str_to_et(contact_start)
+                contact_end_et = str_to_et(contact_end)
+                dsn_contact_mask |= (time_range >= contact_start_et) & (
+                    time_range <= contact_end_et
+                )
+
+            if outages and dsn_station in outages:
+                for outage_start, outage_end in outages[dsn_station]:
+                    dsn_outage_mask |= (time_range >= str_to_et(outage_start)) & (
+                        time_range <= str_to_et(outage_end)
+                    )
+
+    dsn_occupied_mask = dsn_contact_mask & ~dsn_outage_mask
 
-    for station_name, (lon, lat, alt, min_elevation) in stations.items():
+    # Blocks later stations.
+    non_dsn_occupied_mask = np.zeros(time_range.shape, dtype=bool)
+    for station_name, (lon, lat, alt, min_elevation) in NON_DSN_STATIONS.items():
         _azimuth, elevation = calculate_azimuth_and_elevation(
             lon, lat, alt, time_range, obsref="IAU_EARTH"
         )
-        visible = elevation > min_elevation
+        visible_unblocked = elevation > min_elevation
 
         outage_mask = np.zeros(time_range.shape, dtype=bool)
         if outages and station_name in outages:
@@ -89,9 +103,13 @@ def generate_coverage(
                 end_et = str_to_et(end)
                 outage_mask |= (time_range >= start_et) & (time_range <= end_et)
 
-        visible[outage_mask] = False
-        # DSN contacts block other stations
-        visible[dsn_outage_mask] = False
+        # Block this station if DSN is active OR already-occupied by earlier stations
+        unavailable_mask = outage_mask | dsn_occupied_mask | non_dsn_occupied_mask
+        visible = visible_unblocked & ~unavailable_mask
+
+        # This station now occupies these times and will block later stations
+        non_dsn_occupied_mask |= visible
+
         total_visible_mask |= visible
 
         coverage_dict[station_name] = et_to_utc(time_range[visible], format_str="ISOC")

imap_processing/tests/ialirt/unit/test_generate_coverage.py

@@ -1,10 +1,12 @@
 """Test processEphemeris functions."""
 
 from datetime import datetime
+from unittest.mock import patch
 
 import numpy as np
 import pytest
 
+from imap_processing.ialirt.constants import STATIONS
 from imap_processing.ialirt.generate_coverage import (
     format_coverage_summary,
     generate_coverage,
@@ -108,4 +110,107 @@ def test_dsn(furnish_kernels):
         )
 
         assert "I-ALiRT Coverage Summary" in output["summary"]
-        assert 40.6 == output["total_coverage_percent"]
+        assert 42.0 == output["total_coverage_percent"]
+
+
+@pytest.mark.external_kernel
+def test_non_dsn_priority_blocking_with_kernels(furnish_kernels):
+    "Test that non-dsn station block other non-dsn stations."
+    kernels = ["naif0012.tls", "pck00011.tpc", "de440s.bsp", "imap_spk_demo.bsp"]
+    start_time = "2026-09-22T00:00:00Z"
+
+    with furnish_kernels(kernels):
+        # Kiel-only coverage
+        with patch(
+            "imap_processing.ialirt.generate_coverage.NON_DSN_STATIONS",
+            new={"Kiel": STATIONS["Kiel"]},
+        ):
+            coverage_kiel, _ = generate_coverage(start_time)
+
+        kiel_times = coverage_kiel["Kiel"]
+
+        # Manaus-only coverage
+        with patch(
+            "imap_processing.ialirt.generate_coverage.NON_DSN_STATIONS",
+            new={"Manaus": STATIONS["Manaus"]},
+        ):
+            coverage_manaus_only, _ = generate_coverage(start_time)
+
+        manaus_only_times = coverage_manaus_only["Manaus"]
+
+        overlap = np.intersect1d(kiel_times, manaus_only_times)
+        # Assert the times overlap.
+        assert overlap.size > 0
+
+        # Kiel first, then Manaus
+        with patch(
+            "imap_processing.ialirt.generate_coverage.NON_DSN_STATIONS",
+            new={
+                "Kiel": STATIONS["Kiel"],
+                "Manaus": STATIONS["Manaus"],
+            },
+        ):
+            coverage, _ = generate_coverage(start_time)
+
+        manaus_coverage = coverage["Manaus"]
+
+        # Manaus should have no overlap with Kiel.
+        blocked_overlap = np.intersect1d(kiel_times, manaus_coverage)
+        assert blocked_overlap.size == 0
+        assert manaus_coverage[0] > kiel_times[-1]
+
+
+@pytest.mark.external_kernel
+def test_dsn_outage_allows_ground_station_coverage(furnish_kernels):
+    """
+    DSN contacts block non-DSN stations, but DSN outages remove blocking.
+    """
+    kernels = ["naif0012.tls", "pck00011.tpc", "de440s.bsp", "imap_spk_demo.bsp"]
+    start_time = "2026-09-22T00:00:00Z"
+
+    with furnish_kernels(kernels):
+        # Baseline Kiel-only coverage (no DSN)
+        with patch(
+            "imap_processing.ialirt.generate_coverage.NON_DSN_STATIONS",
+            new={"Kiel": STATIONS["Kiel"]},
+        ):
+            coverage_base, _ = generate_coverage(start_time)
+
+        kiel_times = coverage_base["Kiel"]
+
+        contact_start = kiel_times[10]  # inside Kiel coverage
+        contact_end = kiel_times[16]  # 30 min later (inclusive logic in your code)
+
+        # Outage inside the DSN contact
+        outage_start = kiel_times[12]
+        outage_end = kiel_times[14]
+
+        dsn = {"DSS-75": [(contact_start, contact_end)]}
+
+        # DSN contact, no DSN outage
+        # Kiel should be blocked during the full contact window
+        with patch(
+            "imap_processing.ialirt.generate_coverage.NON_DSN_STATIONS",
+            new={"Kiel": STATIONS["Kiel"]},
+        ):
+            coverage_blocked, _ = generate_coverage(start_time, dsn=dsn, outages=None)
+
+        kiel_blocked = coverage_blocked["Kiel"]
+
+        assert not np.any(np.isin(kiel_times[10:16], kiel_blocked))
+
+        # DSN contact + DSN outage
+        # Kiel allowed during outage sub-window
+        outages = {"DSS-75": [(outage_start, outage_end)]}
+
+        with patch(
+            "imap_processing.ialirt.generate_coverage.NON_DSN_STATIONS",
+            new={"Kiel": STATIONS["Kiel"]},
+        ):
+            coverage_punched, _ = generate_coverage(
+                start_time, dsn=dsn, outages=outages
+            )
+
+        kiel_punched = coverage_punched["Kiel"]
+
+        assert np.all(np.isin(kiel_times[12:14], kiel_punched))