Source code for pelagos_py.steps.input_output.load_data
# This file is part of pelagos_py.
#
# Copyright 2025-2026 National Oceanography Centre and The Contributors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Class definition for loading data steps."""
#### Mandatory imports ####
from pelagos_py.steps.base_step import BaseStep, register_step
import pelagos_py.utils.diagnostics as diag
import xarray as xr
import pandas as pd
import numpy as np
from pathlib import Path
MIN_YEAR_FILTER = "1990-01-01"
@register_step
[docs]
class LoadOG1(BaseStep):
"""
Loads NetCDF files from ``file_path``.
If ``filter_bad_times`` is set then measurements made
outside of the time range specified by ``data_start`` and ``data_end`` are removed before
further processing.
Time values are expected to be stored under the "TIME" variable name in the NetCDF file
(corresponding to OG1 format). If "TIME" is not monotonically increasing, or has missing values
(NaT) then this will raise an error.
Parameters
----------
filter_bad_time : bool, optional
If True (default), removes all timestamps outside the expected time window.
data_start : str or np.datetime64, optional
The minimum valid timestamp for the data. If not provided, the filter defaults
to the DEPLOYMENT_TIME found in the dataset, or 1990-01-01T00:00:00 if no
deployment time is found.
data_end : str or np.datetime64, optional
The maximum valid timestamp for the data. If not provided, it defaults to
the current system time when the pipeline is run.
Examples
--------
Example usage in a pipeline configuration:
.. code-block:: yaml
steps:
- name: Load OG1
parameters:
file_path: "/path/to/your/dataset.nc"
filter_bad_time: false
data_start: "2023-05-01T00:00:00"
data_end: "2024-05-01T00:00:00"
"""
step_name = "Load OG1"
required_variables = []
provided_variables = ["TIME", "LATITUDE", "LONGITUDE", "PRES", "TEMP", "CNDC"]
parameter_schema = {
"file_path": {
"type": str,
"required": True,
"description": "Path to the OG1 data file."
},
"filter_bad_time": {
"type": bool,
"default": True,
"description": "If True, removes all timestamps outside the expected time window."
},
"data_start": {
"type": str,
"default": None,
"description": "Minimum valid timestamp (e.g. '2023-05-01T00:00:00'). Defaults to deployment time or 1990."
},
"data_end": {
"type": str,
"default": None,
"description": "Maximum valid timestamp. Defaults to current system time."
}
}
def run(self):
# load data from xarray
self.data = xr.open_dataset(self.file_path)
self.log(f"Loaded data from {self.file_path}")
self.log("Loading dataset to RAM...")
self.data.load()
# Parameters are resolved from parameter_schema in BaseStep.__init__,
# so every declared attribute is guaranteed to be set.
filter_bad_time = self.filter_bad_time
data_start = self.data_start
data_end = self.data_end
# Filter data to the specified time window
if filter_bad_time and (
"TIME" in self.data.variables or "TIME" in self.data.coords
):
orig_len = len(self.data["TIME"])
time_array = self.data["TIME"]
start_val = (
np.datetime64(data_start)
if data_start
else np.datetime64(MIN_YEAR_FILTER)
)
end_val = (
np.datetime64(data_end)
if data_end
else np.datetime64(pd.Timestamp.now())
)
valid_mask = time_array >= start_val
valid_mask &= time_array <= end_val
if not data_start and "DEPLOYMENT_TIME" in self.data.variables:
deploy_time = pd.to_datetime(self.data["DEPLOYMENT_TIME"].values)
if isinstance(deploy_time, pd.DatetimeIndex):
deploy_time = deploy_time[0]
valid_mask &= time_array >= np.datetime64(deploy_time)
time_dim = self.data["TIME"].dims[0]
self.data = self.data.isel({time_dim: valid_mask.values})
new_len = len(self.data["TIME"])
if new_len < orig_len:
self.log_warn(
f"Removed {orig_len - new_len} records containing invalid or pre-deployment timestamps."
)
if "TIME" in self.data.coords:
self.data = self.data.reset_coords("TIME", drop=False)
self.data = self.data.reset_coords("LATITUDE", drop=False)
self.data = self.data.reset_coords("LONGITUDE", drop=False)
if "TIME" not in self.data.data_vars:
raise ValueError(
"\n'TIME' could not be found in the dataset. Pipelines cannot be run without this variable.\n"
"If TIME is listed under another name, please rename it to conform to the OG1 format."
)
# Check that the "TIME" variable is monotonic and has no missing values
if np.any(np.isnat(self.data["TIME"].values)):
raise ValueError(
"\n'TIME' has NaT values. Pipelines cannot be run without a continuous monotonic time coordinate.\n"
"Please remove these values (and their concurrent measurements) from the input."
)
if not np.all(np.diff(self.data["TIME"]) >= 0):
self.log_warn(
"'TIME' is not monotonically increasing. This may cause fatal issues in processing. "
"Please check the quality of your input data."
)
# Generate diagnostics if enabled
self.log(f"Loaded data from {self.file_path}")
self.context["global_parameters"]["filename_core"] = Path(self.file_path).stem # Make this available to other steps
if self.diagnostics:
self.generate_diagnostics()
self.context["data"] = self.data
return self.context
[docs]
def generate_diagnostics(self):
"""
Print a structural summary of the loaded dataset.
Called automatically at the end of :meth:`run` when ``diagnostics`` is
enabled. Delegates to
:func:`pelagos_py.utils.diagnostics.generate_info`, which prints the
dataset's dimensions, variables and global attributes (via
:meth:`xarray.Dataset.info`) to stdout — a quick check that the data
was loaded as expected.
"""
self.log("Generating diagnostics...")
diag.generate_info(self.data)
