Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 1 addition & 68 deletions imap_processing/ialirt/generate_coverage.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,8 @@
"""Coverage time for each station."""

import logging
from pathlib import Path

import numpy as np
import pandas as pd

from imap_processing.ialirt.constants import STATIONS, StationProperties
from imap_processing.ialirt.process_ephemeris import calculate_azimuth_and_elevation
Expand All @@ -27,75 +25,10 @@
"DSS-56",
"DSS-74",
"DSS-75",
"DSS-43",
]


def parse_uksa_schedule_xlsx(xlsx_path: Path) -> list[tuple[str, str]]:
"""
Parse the UKSA (GHY-6) availability sheet and return a list of contacts.

Parameters
----------
xlsx_path : Path
Path to the UKSA (GHY-6) availability sheet.

Returns
-------
contacts : list[tuple[str, str]]
Available contacts for UKSA (GHY-6) availability sheet.
"""
data = pd.read_excel(xlsx_path)

# Import start and stop times.
start_dt = (
data["Date"]
+ pd.to_timedelta(
data["GHY-6 Start Availability Times (5degrees) (UTC)"].astype(str)
)
).to_numpy("datetime64[s]")

stop_dt = (
data["Date"]
+ pd.to_timedelta(
data["GHY-6 Stop Availability Times (5degrees) (UTC)"].astype(str)
)
).to_numpy("datetime64[s]")

# Indicates whether or not setup or teardown should be taken from contact window.
notes = data["Short due to existing booking "].fillna("")

truncate_setup = (
notes.eq("Yes- setup needs to be included with the window")
| notes.eq("Yes- setup and teardown needs to be included with the window")
).to_numpy()

truncate_teardown = (
notes.eq("Yes- tear down needs to be included within the window")
| notes.eq("Yes- setup and teardown needs to be included with the window")
).to_numpy()

setup_time = data["Setup time"].iloc[0]
teardown_time = data["Tear down time"].iloc[0]

setup_seconds = setup_time.hour * 3600 + setup_time.minute * 60 + setup_time.second
teardown_seconds = (
teardown_time.hour * 3600 + teardown_time.minute * 60 + teardown_time.second
)

setup_delta = np.timedelta64(setup_seconds, "s")
teardown_delta = np.timedelta64(teardown_seconds, "s")

# Apply adjustments
start_dt[truncate_setup] += setup_delta
stop_dt[truncate_teardown] -= teardown_delta

# Format to strings with ms, append Z
start_str = np.datetime_as_string(start_dt, unit="ms")
stop_str = np.datetime_as_string(stop_dt, unit="ms")

return list(zip(start_str, stop_str, strict=False))


def create_schedule_mask(
station: StationProperties, time_range: np.ndarray
) -> np.ndarray:
Expand Down
32 changes: 0 additions & 32 deletions imap_processing/tests/ialirt/unit/test_generate_coverage.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,6 @@
create_schedule_mask,
format_coverage_summary,
generate_coverage,
parse_uksa_schedule_xlsx,
)


Expand Down Expand Up @@ -182,34 +181,3 @@ def test_create_schedule_mask(mock_et_to_utc):
)

np.testing.assert_array_equal(mask, expected)


def test_parse_uksa_schedule_xlsx(schedule_path):
"Test parse_uksa_schedule_xlsx."

uksa_contacts = parse_uksa_schedule_xlsx(schedule_path)

# Verify that setup time and teardown time are properly accounted for.
assert uksa_contacts[1] == ("2026-01-29T14:40:00.000", "2026-01-29T16:54:26.000")
assert uksa_contacts[2] == ("2026-01-30T08:54:52.000", "2026-01-30T12:54:00.000")


@pytest.mark.external_kernel
def test_incorporate_uksa_coverage(schedule_path, furnish_kernels):
"Test to parse UKSA schedule."
kernels = [
"naif0012.tls",
"pck00011.tpc",
"de440s.bsp",
"imap_spk_demo.bsp",
]

uksa_contacts = parse_uksa_schedule_xlsx(schedule_path)

with furnish_kernels(kernels):
coverage_dict, outage_dict = generate_coverage(
"2026-01-29T00:00:00Z", uksa=uksa_contacts
)

assert coverage_dict["UKSA"][0] == "2026-01-29T14:45:00.000"
assert coverage_dict["UKSA"][-1] == "2026-01-29T16:50:00.000"
Loading