add tmp file write/read step

src/virtualship/instruments/adcp.py

@@ -5,7 +5,7 @@
 import numpy as np
 from parcels import ParticleFile, ParticleSet
 
-from virtualship.instruments.base import Instrument
+from virtualship.instruments.base import FetchSpec, Instrument
 from virtualship.instruments.sensors import SensorType
 from virtualship.instruments.types import InstrumentType
 from virtualship.utils import build_particle_class_from_sensors, register_instrument
@@ -61,18 +61,14 @@ class ADCPInstrument(Instrument):
     def __init__(self, expedition, from_data):
         """Initialize ADCPInstrument."""
         variables = expedition.instruments_config.adcp_config.active_variables()
-        limit_spec = {
-            "spatial": True
-        }  # spatial limits; lat/lon constrained to waypoint locations + buffer
 
         super().__init__(
             expedition,
             variables,
             add_bathymetry=False,
             allow_time_extrapolation=True,
             verbose_progress=False,
-            spacetime_buffer_size=None,
-            limit_spec=limit_spec,
+            fetch_spec=FetchSpec(),
             from_data=from_data,
         )
 

src/virtualship/instruments/argo_float.py

@@ -7,7 +7,7 @@
 from parcels import ParticleFile, ParticleSet, StatusCode, Variable
 from parcels.kernels import AdvectionRK2
 
-from virtualship.instruments.base import Instrument
+from virtualship.instruments.base import FetchSpec, Instrument
 from virtualship.instruments.sensors import SensorType
 from virtualship.instruments.types import InstrumentType
 from virtualship.models.spacetime import Spacetime
@@ -253,23 +253,19 @@ def __init__(self, expedition, from_data):
             "V": "vo",
             **sensor_variables,
         }  # advection variables (U and V) are always required for argo float simulation; sensor variables come from config
-        spacetime_buffer_size = {
-            "latlon": 3.0,  # [degrees]
-            "time": expedition.instruments_config.argo_float_config.lifetime.total_seconds()
+        fetch_spec = FetchSpec(
+            latlon_buffer=3.0,  # [degrees]
+            time_buffer=expedition.instruments_config.argo_float_config.lifetime.total_seconds()
             / (24 * 3600),  # [days]
-        }
-        limit_spec = {
-            "spatial": True,  # spatial limits; lat/lon constrained to waypoint locations + buffer
-        }
+        )
 
         super().__init__(
             expedition,
             variables,
             add_bathymetry=True,
             allow_time_extrapolation=False,
             verbose_progress=True,
-            spacetime_buffer_size=spacetime_buffer_size,
-            limit_spec=limit_spec,
+            fetch_spec=fetch_spec,
             from_data=from_data,
         )
 

src/virtualship/instruments/base.py

@@ -2,6 +2,8 @@
 
 import abc
 import collections
+import tempfile
+from dataclasses import dataclass
 from datetime import timedelta
 from itertools import pairwise
 from pathlib import Path
@@ -28,6 +30,17 @@
     from virtualship.models import Expedition
 
 
+@dataclass
+class FetchSpec:
+    """Fetch constraints and parameters for dataset retrieval."""
+
+    spatial: bool = True
+    latlon_buffer: float = 0.25  # degrees
+    time_buffer: float = 0.0  # days
+    depth_min: float | None = None
+    depth_max: float | None = None
+
+
 class Instrument(abc.ABC):
     """Base class for instruments and their simulation."""
 
@@ -50,8 +63,7 @@ def __init__(
         allow_time_extrapolation: bool,
         verbose_progress: bool,
         from_data: Path | None,
-        spacetime_buffer_size: dict | None = None,
-        limit_spec: dict | None = None,
+        fetch_spec: FetchSpec | None = None,
     ):
         """Initialise instrument."""
         self.expedition = expedition
@@ -67,8 +79,7 @@ def __init__(
         self.add_bathymetry = add_bathymetry
         self.allow_time_extrapolation = allow_time_extrapolation
         self.verbose_progress = verbose_progress
-        self.spacetime_buffer_size = spacetime_buffer_size
-        self.limit_spec = limit_spec
+        self.fetch_spec = fetch_spec or FetchSpec()
 
         wp_lats, wp_lons = _get_waypoint_latlons(expedition.schedule.waypoints)
         wp_times = [
@@ -152,12 +163,10 @@ def _get_copernicus_ds(
             variable=var if not physical else None,
         )
 
-        latlon_buffer = self._get_spec_value(
-            "buffer", "latlon", 0.25
-        )  # [degrees]; default 0.25 deg buffer to ensure coverage in field cell edge cases
-        depth_min = self._get_spec_value("limit", "depth_min", None)
-        depth_max = self._get_spec_value("limit", "depth_max", None)
-        spatial_constraint = self._get_spec_value("limit", "spatial", True)
+        latlon_buffer = self.fetch_spec.latlon_buffer
+        depth_min = self.fetch_spec.depth_min
+        depth_max = self.fetch_spec.depth_max
+        spatial_constraint = self.fetch_spec.spatial
 
         min_lon_bound = self.min_lon - latlon_buffer if spatial_constraint else None
         max_lon_bound = self.max_lon + latlon_buffer if spatial_constraint else None
@@ -176,18 +185,22 @@ def _get_copernicus_ds(
             minimum_depth=depth_min,
             maximum_depth=depth_max,
             coordinates_selection_method="outside",
+            vertical_axis="elevation",
         )
 
     def _generate_fieldset(self) -> parcels.FieldSet:
         """
         Create and combine FieldSets for each variable, supporting both local and Copernicus Marine data sources.
 
         N.B. Per variable avoids issues when using copernicusmarine and creating directly one FieldSet of ds's sourced from different Copernicus Marine product IDs (which can also have different temporal resolutions), which is often the case for BGC variables.
+
+        Includes an intermediate step of writing to tmp files, as per https://github.com/Parcels-code/parcels-benchmarks/pull/49
+        TODO: the need for this step may be removed as Parcels x copernicusmarine integration improves, tracked in https://github.com/Parcels-code/Parcels/issues/2756 and xref'd in VirtualShip #357 (https://github.com/Parcels-code/virtualship/issues/357)
         """
         fieldsets_list = []
         keys = list(self.variables.keys())
 
-        time_buffer = self._get_spec_value("buffer", "time", 0.0)
+        time_buffer = self.fetch_spec.time_buffer
 
         for key in keys:
             var = self.variables[key]
@@ -218,15 +231,12 @@ def _generate_fieldset(self) -> parcels.FieldSet:
                 )
                 field_var_name = var
 
-            # negate depth and reindex (to suit Parcels XGrid strictly increasing depth convention)
-            ds["depth"] = -ds["depth"]
-            ds = ds.reindex(depth=ds["depth"][::-1])
-
-            # TODO: update when decision on handling of nans/0s in v4 is made (i.e. https://github.com/Parcels-code/Parcels/issues/2393)
+            # TODO: to be removed when Parcels #2746 is merged (i.e. https://github.com/Parcels-code/Parcels/pull/2746)
             ds = ds.fillna(0)
 
             fields = {key: ds[field_var_name]}
             ds_fset = parcels.convert.copernicusmarine_to_sgrid(fields=fields)
+            ds_fset = self._via_tmp_ds(ds_fset)
 
             fs = parcels.FieldSet.from_sgrid_conventions(ds_fset)
 
@@ -249,7 +259,11 @@ def _generate_fieldset(self) -> parcels.FieldSet:
 
         return base_fieldset
 
-    def _get_spec_value(self, spec_type: str, key: str, default=None):
-        """Helper to extract a value from spacetime_buffer_size or limit_spec."""
-        spec = self.spacetime_buffer_size if spec_type == "buffer" else self.limit_spec
-        return spec.get(key) if spec and spec.get(key) is not None else default
+    @staticmethod
+    def _via_tmp_ds(ds) -> xr.Dataset:
+        """Create and re-load a temporary local dataset."""
+        tmpdir = tempfile.TemporaryDirectory()
+        tmp_fpath = Path(tmpdir.name).joinpath("tmp.nc")
+        ds.to_netcdf(tmp_fpath)
+        del ds
+        return xr.open_dataset(tmp_fpath)

src/virtualship/instruments/ctd.py

@@ -7,7 +7,7 @@
 from parcels import ParticleFile, ParticleSet, Variable
 from parcels._core.statuscodes import StatusCode
 
-from virtualship.instruments.base import Instrument
+from virtualship.instruments.base import FetchSpec, Instrument
 from virtualship.instruments.sensors import SensorType
 from virtualship.instruments.types import InstrumentType
 from virtualship.utils import (
@@ -142,18 +142,14 @@ class CTDInstrument(Instrument):
     def __init__(self, expedition, from_data):
         """Initialize CTDInstrument."""
         variables = expedition.instruments_config.ctd_config.active_variables()
-        limit_spec = {
-            "spatial": True
-        }  # spatial limits; lat/lon constrained to waypoint locations + buffer
 
         super().__init__(
             expedition,
             variables,
             add_bathymetry=True,
             allow_time_extrapolation=True,
             verbose_progress=False,
-            spacetime_buffer_size=None,
-            limit_spec=limit_spec,
+            fetch_spec=FetchSpec(),
             from_data=from_data,
         )
 
@@ -237,7 +233,7 @@ def simulate(self, measurements, out_path) -> None:
         )
 
         # there should be no particles left, as they delete themselves when they resurface
-        if len(ctd_particleset.lon) != 0:
+        if len(ctd_particleset.x) != 0:
             raise ValueError(
                 "Simulation ended before CTD resurfaced. This most likely means the field time dimension did not match the simulation time span."
             )

src/virtualship/instruments/drifter.py

@@ -8,7 +8,7 @@
 from parcels._core.statuscodes import StatusCode
 from parcels.kernels import AdvectionRK2
 
-from virtualship.instruments.base import Instrument
+from virtualship.instruments.base import FetchSpec, Instrument
 from virtualship.instruments.sensors import SensorType
 from virtualship.instruments.types import InstrumentType
 from virtualship.models.spacetime import Spacetime
@@ -88,29 +88,25 @@ def __init__(self, expedition, from_data):
             "V": "vo",
             **sensor_variables,
         }  # advection variables (U and V) are always required for drifter simulation; sensor variables come from config
-        spacetime_buffer_size = {
-            "latlon": None,
-            "time": expedition.instruments_config.drifter_config.lifetime.total_seconds()
+        fetch_spec = FetchSpec(
+            latlon_buffer=30.0,  # TODO: generous buffer to reduce tmp file footprint, can potentially be removed in the future as/when Parcels streaming performance improves (see #358)
+            time_buffer=expedition.instruments_config.drifter_config.lifetime.total_seconds()
             / (24 * 3600),  # [days]
-        }
-        limit_spec = {
-            "spatial": False,  # no spatial limits; generate global fieldset
-            "depth_min": abs(
+            depth_min=abs(
                 expedition.instruments_config.drifter_config.depth_meter
             ),  # [meters]
-            "depth_max": abs(
+            depth_max=abs(
                 expedition.instruments_config.drifter_config.depth_meter
             ),  # [meters]
-        }
+        )
 
         super().__init__(
             expedition,
             variables,
             add_bathymetry=False,
             allow_time_extrapolation=False,
             verbose_progress=True,
-            spacetime_buffer_size=spacetime_buffer_size,
-            limit_spec=limit_spec,
+            fetch_spec=fetch_spec,
             from_data=from_data,
         )
 

src/virtualship/instruments/ship_underwater_st.py

@@ -5,7 +5,7 @@
 import numpy as np
 from parcels import ParticleFile, ParticleSet
 
-from virtualship.instruments.base import Instrument
+from virtualship.instruments.base import FetchSpec, Instrument
 from virtualship.instruments.sensors import SensorType
 from virtualship.instruments.types import InstrumentType
 from virtualship.utils import (
@@ -67,22 +67,14 @@ def __init__(self, expedition, from_data):
         variables = (
             expedition.instruments_config.ship_underwater_st_config.active_variables()
         )
-        spacetime_buffer_size = {
-            "latlon": 0.25,  # [degrees]
-            "time": 0.0,  # [days]
-        }
-        limit_spec = {
-            "spatial": True
-        }  # spatial limits; lat/lon constrained to waypoint locations + buffer
 
         super().__init__(
             expedition,
             variables,
             add_bathymetry=False,
             allow_time_extrapolation=True,
             verbose_progress=False,
-            spacetime_buffer_size=spacetime_buffer_size,
-            limit_spec=limit_spec,
+            fetch_spec=FetchSpec(),
             from_data=from_data,
         )
 

src/virtualship/instruments/xbt.py

@@ -7,7 +7,7 @@
 from parcels import ParticleFile, ParticleSet, Variable
 from parcels._core.statuscodes import StatusCode
 
-from virtualship.instruments.base import Instrument
+from virtualship.instruments.base import FetchSpec, Instrument
 from virtualship.instruments.sensors import SensorType
 from virtualship.instruments.types import InstrumentType
 from virtualship.models.spacetime import Spacetime
@@ -95,18 +95,14 @@ class XBTInstrument(Instrument):
     def __init__(self, expedition, from_data):
         """Initialize XBTInstrument."""
         variables = expedition.instruments_config.xbt_config.active_variables()
-        limit_spec = {
-            "spatial": True
-        }  # spatial limits; lat/lon constrained to waypoint locations + buffer
 
         super().__init__(
             expedition,
             variables,
             add_bathymetry=True,
             allow_time_extrapolation=True,
             verbose_progress=False,
-            spacetime_buffer_size=None,
-            limit_spec=limit_spec,
+            fetch_spec=FetchSpec(),
             from_data=from_data,
         )