Skip to content

feat: update the dscovr api and restrict the end time #78

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sahiljhawar merged 1 commit into from
Jul 9, 2026
Merged

feat: update the dscovr api and restrict the end time #78

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
206 changes: 105 additions & 101 deletions swvo/io/solar_wind/dscovr.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
Module for handling DSCOVR Solar Wind data.
"""

import json
import logging
import warnings
from datetime import datetime, timedelta, timezone
Expand Down Expand Up @@ -46,48 +47,64 @@ class DSCOVR(BaseIO):

ENV_VAR_NAME = "SW_DSCOVR_STREAM_DIR"

URL = "https://services.swpc.noaa.gov/products/solar-wind/"
NAME_MAG = "mag-1-day.json"
NAME_SWEPAM = "plasma-1-day.json"
URL = "https://www.ncei.noaa.gov/cloud-access/space-weather-portal/api/v1/values"
NAME_DATA = "dscovr.json"

SWEPAM_FIELDS = ["speed", "proton_density", "temperature"]
MAG_FIELDS = ["bx_gsm", "by_gsm", "bz_gsm", "bavg"]
MAG_PARAMETER_FIELDS = {
"bavg": "bt",
"bx_gsm": "bx_gse",
"by_gsm": "by_gse",
"bz_gsm": "bz_gse",
}
SWEPAM_PARAMETER_FIELDS = {
"speed": "proton_speed",
"proton_density": "proton_density",
"temperature": "proton_temperature",
}

LABEL = "dscovr"

def download_and_process(self, request_time: datetime) -> None:
def download_and_process(self, start_time: datetime, end_time: datetime) -> None:
"""
Download and process DSCOVR data, splitting data across midnight into appropriate day files.

Parameters
----------
request_time : datetime
The time for which the data is requested. Must be in the past and within the last two hours.
start_time : datetime
Start time of the data to download. Must be timezone-aware. If `end_time` is not provided, this is treated
as a nowcast request time and the previous 24 hours are downloaded.
end_time : datetime | None
End time of the data to download. Must be timezone-aware.

Raises
------
AssertionError
If the request_time is in the future.
If the request_time is in the future or if start_time is after end_time.
FileNotFoundError
If the downloaded files are empty.
ValueError
If the end_time is after 2026-06-30, since DSCOVR data is only available until that date.
Comment on lines +84 to +88

Returns
-------
None
"""
current_time = datetime.now(timezone.utc)
assert request_time < current_time, (
f"Request time: {request_time} cannot be in the future. Current time: {current_time}!"
)
if current_time - request_time > timedelta(hours=24):
logger.debug("We can only download DSCOVR data for the last 23 hours and a hour in past!")
return
start_time = enforce_utc_timezone(start_time)
end_time = enforce_utc_timezone(end_time)
Comment on lines +94 to +95

if end_time > datetime(2026, 6, 29, 23, 59, 59, tzinfo=timezone.utc):
raise ValueError(
"DSCOVR data is only available until 2026-06-29 23:59:59 UTC. Please choose an earlier end_time."
)

assert start_time < end_time, "Start time must be before end time!"

temporary_dir = Path("./temp_sw_dscovr_wget")
temporary_dir.mkdir(exist_ok=True, parents=True)

self._download(temporary_dir, self.NAME_MAG)
self._download(temporary_dir, self.NAME_SWEPAM)
self._download(temporary_dir, start_time, end_time)

logger.debug("Processing file ...")
processed_df = self._process_single_file(temporary_dir)
Expand Down Expand Up @@ -118,21 +135,23 @@ def download_and_process(self, request_time: datetime) -> None:

except Exception as e:
logger.error(f"Failed to process file for {date}: {e}")
if tmp_path.exists():
tmp_path.unlink()
# if tmp_path.exists():
# tmp_path.unlink()
continue
Comment on lines 136 to 140

rmtree(temporary_dir, ignore_errors=True)

def _download(self, temporary_dir: Path, file_name: str) -> None:
"""Download a file from DSCOVR server.
def _download(self, temporary_dir: Path, start_time: datetime, end_time: datetime) -> None:
"""Download a DSCOVR data file from the NOAA NCEI Space Weather Portal API.

Parameters
----------
temporary_dir : Path
Temporary directory to store the downloaded file.
file_name : str
Name of the file to download.
start_time : datetime
Start time of the data to download.
end_time : datetime
End time of the data to download.

Raises
------
Expand All @@ -141,15 +160,28 @@ def _download(self, temporary_dir: Path, file_name: str) -> None:
FileNotFoundError
If the downloaded file is empty.
"""
logger.debug(f"Downloading file {self.URL + file_name} ...")
response = requests.get(self.URL + file_name, timeout=10)
params = {
"start_time": start_time.strftime("%Y-%m-%dT%H:%M:%S"),
"end_time": end_time.strftime("%Y-%m-%dT%H:%M:%S"),
"parameters": ";".join(self._portal_parameters()),
"format": "json",
"time_format": "iso",
}
logger.debug(f"Downloading DSCOVR data from {self.URL} for {start_time} - {end_time} ...")
response = requests.get(self.URL, params=params, timeout=30)
response.raise_for_status()

with open(temporary_dir / file_name, "wb") as f:
with open(temporary_dir / self.NAME_DATA, "wb") as f:
f.write(response.content)

if (temporary_dir / file_name).stat().st_size == 0:
raise FileNotFoundError(f"Error while downloading file: {self.URL + file_name}!")
if (temporary_dir / self.NAME_DATA).stat().st_size == 0:
raise FileNotFoundError(f"Error while downloading file: {self.URL}!")

def _portal_parameters(self) -> list[str]:
"""Build the parameter list for the NCEI portal values API."""
mag_parameters = [f"DSCOVR:m1m_dscovr:{field}" for field in self.MAG_PARAMETER_FIELDS.values()]
swepam_parameters = [f"DSCOVR:f1m_dscovr:{field}" for field in self.SWEPAM_PARAMETER_FIELDS.values()]
return mag_parameters + swepam_parameters

def read(
self,
Expand Down Expand Up @@ -183,19 +215,15 @@ def read(
Raises
------
AssertionError
Raises `AssertionError` if the start time is before the end time.
Raises `AssertionError` if the end time is before the start time.
"""
if start_time > end_time:
msg = "start_time must be before end_time"
logger.error(msg)
raise ValueError(msg)

start_time = enforce_utc_timezone(start_time)
end_time = enforce_utc_timezone(end_time)

if propagation:
logger.info("Shiting start day by -1 day to account for propagation")
logger.info("Shifting start day by -1 day to account for propagation")
start_time = start_time - timedelta(days=1)
assert start_time < end_time, "Start time must be before end time!"

file_paths, _ = self._get_processed_file_list(start_time, end_time)

Expand All @@ -220,14 +248,13 @@ def read(
},
)

for file_path in file_paths:
if not file_path.exists() and download:
file_date = enforce_utc_timezone(datetime.strptime(file_path.stem.split("_")[-1], "%Y%m%d"))
try:
self.download_and_process(file_date)
except AssertionError as e:
logger.error(f"`download_and_process` failed because: {e}")
if download and any(not file_path.exists() for file_path in file_paths):
try:
self.download_and_process(start_time, end_time)
except AssertionError as e:
logger.error(f"`download_and_process` failed because: {e}")

for file_path in file_paths:
if not file_path.exists():
warnings.warn(f"File {file_path} not found")
continue
Expand Down Expand Up @@ -306,83 +333,60 @@ def _read_single_file(self, file_path) -> pd.DataFrame:
return df

def _process_single_file(self, temporary_dir: Path) -> pd.DataFrame:
"""Process mag and swepam DSCOVR file to a DataFrame.
"""Process combined MAG and SWEPAM DSCOVR data to a DataFrame.

Returns
-------
pd.DataFrame
DSCOVR data.
"""
data_mag = self._process_mag_file(temporary_dir)
data_swepam = self._process_swepam_file(temporary_dir)
with open(temporary_dir / self.NAME_DATA, "r") as file:
payload = json.load(file)

status_code = payload.get("status", {}).get("code")
if status_code is not None and status_code != 200:
msg = payload.get("status", {}).get("message", f"NOAA DSCOVR API returned status code {status_code}.")
logger.error(msg)
raise FileNotFoundError(msg)

data = pd.concat([data_swepam, data_mag], axis=1)
data = pd.DataFrame(payload.get("data", {}))

if len(data["time"]) == 0:
raise FileNotFoundError("No data found in the downloaded DSCOVR file.")
Comment on lines +352 to +355

data["t"] = pd.to_datetime(data["time"], utc=True)
data.index = data["t"]
data.drop(["time", "t"], axis=1, inplace=True)

data.rename(columns=self._portal_column_map(), inplace=True)
expected_columns = [*self.MAG_FIELDS, *self.SWEPAM_FIELDS]
for column in expected_columns:
if column not in data.columns:
data[column] = np.nan
data = data[expected_columns].apply(pd.to_numeric, errors="coerce")
self._replace_invalid_values(data)

start_time = data.index.min()
end_time = data.index.max()
complete_range = pd.date_range(start=start_time, end=end_time, freq="1min", tz="UTC")

data = data.reindex(complete_range)
data.index.name = "t"
data["pdyn"] = 2e-6 * data["proton_density"].values * data["speed"].values ** 2

return data

def _process_mag_file(self, temporary_dir: Path) -> pd.DataFrame:
"""
Reads magnetic instrument last available real time DSCOVR data.

Returns
-------

pd.DataFrame
Dataframe with magnetic field components and timestamp sampled every minute.
"""

data_mag = pd.read_json(temporary_dir / self.NAME_MAG)
data_mag.columns = data_mag.iloc[0]
data_mag = data_mag.iloc[1:].reset_index(drop=True)
data_mag["t"] = pd.to_datetime(data_mag["time_tag"])
data_mag.index = data_mag["t"]
data_mag.index = enforce_utc_timezone(data_mag.index)
data_mag.drop(
["lon_gsm", "lat_gsm", "time_tag", "t"],
axis=1,
inplace=True,
)

data_mag.rename(columns={"bt": "bavg"}, inplace=True)

return data_mag

def _process_swepam_file(self, temporary_dir: Path) -> pd.DataFrame:
"""
This method reads faraday cup SWEPAM instrument daily file from DSCOVR original data.


Returns
-------

pd.DataFrame
Dataframe with solar wind speed, proton density, temperature and timestamp, sampled every minute.
"""

data_plasma = pd.read_json(temporary_dir / self.NAME_SWEPAM)
data_plasma.columns = data_plasma.iloc[0]
data_plasma = data_plasma.iloc[1:].reset_index(drop=True)
data_plasma["t"] = data_plasma["time_tag"]
data_plasma.index = pd.to_datetime(data_plasma["t"])
data_plasma.index = enforce_utc_timezone(data_plasma.index)
data_plasma.drop(
["time_tag", "t"],
axis=1,
inplace=True,
)

data_plasma.rename(columns={"bt": "bavg", "density": "proton_density"}, inplace=True)
data_plasma = data_plasma.astype(float)
data_plasma["pdyn"] = 2e-6 * data_plasma["proton_density"].values * data_plasma["speed"].values ** 2

return data_plasma
def _portal_column_map(self) -> dict[str, str]:
"""Map NCEI portal response columns to SWVO DSCOVR columns."""
mag_map = {f"m1m_dscovr.{source}": target for target, source in self.MAG_PARAMETER_FIELDS.items()}
swepam_map = {f"f1m_dscovr.{source}": target for target, source in self.SWEPAM_PARAMETER_FIELDS.items()}
return mag_map | swepam_map

def _replace_invalid_values(self, data: pd.DataFrame) -> None:
"""Replace known DSCOVR missing-value sentinels with NaN."""
for k in [*self.MAG_FIELDS, *self.SWEPAM_FIELDS]:
mask = (data[k] < -99999.0) | (data[k] == 0.0)
data.loc[mask, k] = np.nan
Comment on lines +385 to +389

def _update_filename(self, row: pd.Series) -> str:
"""Update the filename in the row.
Expand Down
Loading
Loading