to_windowed_arrays only for non-underway instruments

src/virtualship/instruments/base.py

@@ -15,8 +15,10 @@
 from yaspin import yaspin
 
 from virtualship.errors import CopernicusCatalogueError
+from virtualship.instruments.types import InstrumentType
 from virtualship.utils import (
     COPERNICUSMARINE_PHYS_VARIABLES,
+    INSTRUMENT_CLASS_MAP,
     _find_files_in_timerange,
     _find_nc_file_with_variable,
     _get_bathy_data,
@@ -239,7 +241,11 @@ def _generate_fieldset(self) -> parcels.FieldSet:
             ds_fset = self._via_tmp_ds(ds_fset)
 
             fs = parcels.FieldSet.from_sgrid_conventions(ds_fset)
-            fs.to_windowed_arrays()  # always to windowed arrays, just in case any ds is Dask backed
+
+            # non-underway instruments to windowed arrays, just in case any ds is Dask backed
+            # underway instruments should not to converted to windowed arrays, as they use one direct fieldset.eval() call which could cause a big memory usage if the fieldset is windowed
+            if not self.instrument_type.is_underway:
+                fs = fs.to_windowed_arrays()
 
             fieldsets_list.append(fs)
 
@@ -268,3 +274,8 @@ def _via_tmp_ds(ds) -> xr.Dataset:
         ds.to_netcdf(tmp_fpath)
         del ds
         return xr.open_dataset(tmp_fpath)
+
+    @property
+    def instrument_type(self) -> InstrumentType:
+        """Return the InstrumentType for this instrument instance."""
+        return next(k for k, v in INSTRUMENT_CLASS_MAP.items() if type(self) is v)

tests/instruments/test_adcp.py

@@ -1,32 +1,64 @@
 """Test the simulation of ADCP instruments."""
 
 import datetime
+from typing import ClassVar
 
 import numpy as np
 import pydantic
 import pytest
 import xarray as xr
-
 from parcels import FieldSet
+
 from virtualship.instruments.adcp import ADCPInstrument
 from virtualship.instruments.sensors import SensorType
 from virtualship.instruments.types import InstrumentType
 from virtualship.models import Location, Spacetime, Waypoint
 from virtualship.models.expedition import ADCPConfig, InstrumentsConfig, SensorConfig
 
+# =====================================================
+# Shared constants and fixtures
+# =====================================================
 
-def test_simulate_adcp(tmpdir) -> None:
-    MAX_DEPTH = -1000
-    MIN_DEPTH = -5
-    NUM_BINS = 40
+BASE_TIME = datetime.datetime.strptime(
+    "1950-01-01", "%Y-%m-%d"
+)  # arbitrary time offset for the dummy fieldset
+MAX_DEPTH = -1000
+NUM_BINS = 40
+
+
+@pytest.fixture
+def adcp_expedition():
+    """Minimal Expedition for ADCPInstrument instantiation."""
+
+    class DummyExpedition:
+        class schedule:
+            waypoints: ClassVar[list] = [
+                Waypoint(
+                    location=Location(1, 2),
+                    time=BASE_TIME,
+                    instrument=InstrumentType.ADCP,
+                ),
+            ]
+
+        instruments_config = InstrumentsConfig(
+            adcp_config=ADCPConfig(
+                max_depth_meter=MAX_DEPTH,
+                num_bins=NUM_BINS,
+                period_minutes=5.0,
+                sensors=[SensorConfig(sensor_type=SensorType.VELOCITY)],
+            )
+        )
+
+    return DummyExpedition()
 
-    # arbitrary time offset for the dummy fieldset
-    base_time = datetime.datetime.strptime("1950-01-01", "%Y-%m-%d")
+
+def test_simulate_adcp(tmpdir, adcp_expedition) -> None:
+    MIN_DEPTH = -5
 
     # where to sample
     sample_points = [
-        Spacetime(Location(1, 2), base_time + datetime.timedelta(seconds=0)),
-        Spacetime(Location(3, 4), base_time + datetime.timedelta(seconds=1)),
+        Spacetime(Location(1, 2), BASE_TIME + datetime.timedelta(seconds=0)),
+        Spacetime(Location(3, 4), BASE_TIME + datetime.timedelta(seconds=1)),
     ]
 
     # expected observations at sample points
@@ -36,14 +68,14 @@ def test_simulate_adcp(tmpdir) -> None:
             "U": {"surface": 7, "max_depth": 8},
             "lat": sample_points[0].location.lat,
             "lon": sample_points[0].location.lon,
-            "time": base_time + datetime.timedelta(seconds=0),
+            "time": BASE_TIME + datetime.timedelta(seconds=0),
         },
         {
             "V": {"surface": 9, "max_depth": 10},
             "U": {"surface": 11, "max_depth": 12},
             "lat": sample_points[1].location.lat,
             "lon": sample_points[1].location.lon,
-            "time": base_time + datetime.timedelta(seconds=1),
+            "time": BASE_TIME + datetime.timedelta(seconds=1),
         },
     ]
 
@@ -79,31 +111,7 @@ def test_simulate_adcp(tmpdir) -> None:
         },
     )
 
-    # dummy expedition for ADCPInstrument
-    class DummyExpedition:
-        class schedule:
-            # ruff: noqa
-            waypoints = [
-                Waypoint(
-                    location=Location(1, 2),
-                    time=base_time,
-                    instrument=InstrumentType.ADCP,
-                ),
-            ]
-
-        instruments_config = InstrumentsConfig(
-            adcp_config=ADCPConfig(
-                max_depth_meter=MAX_DEPTH,
-                num_bins=NUM_BINS,
-                period_minutes=5.0,
-                sensors=[SensorConfig(sensor_type=SensorType.VELOCITY)],
-            )
-        )
-
-    expedition = DummyExpedition()
-    from_data = None
-
-    adcp_instrument = ADCPInstrument(expedition, from_data)
+    adcp_instrument = ADCPInstrument(adcp_expedition, from_data=None)
     out_path = tmpdir.join("out.zarr")
 
     adcp_instrument.load_input_data = lambda: fieldset
@@ -183,3 +191,10 @@ def test_adcp_config_unsupported_sensor_rejected():
             period_minutes=30.0,
             sensors=[SensorConfig(sensor_type=SensorType.TEMPERATURE)],
         )
+
+
+def test_adcp_instrument_type(adcp_expedition):
+    """ADCPInstrument returns the correct InstrumentType and if is underway instrument."""
+    adcp_instrument = ADCPInstrument(adcp_expedition, from_data=None)
+    assert adcp_instrument.instrument_type == InstrumentType.ADCP
+    assert adcp_instrument.instrument_type.is_underway

tests/instruments/test_argo_float.py

@@ -10,6 +10,7 @@
 
 from virtualship.instruments.argo_float import ArgoFloat, ArgoFloatInstrument
 from virtualship.instruments.sensors import SensorType
+from virtualship.instruments.types import InstrumentType
 from virtualship.models import Location, Spacetime
 from virtualship.models.expedition import (
     ArgoFloatConfig,
@@ -239,3 +240,16 @@ def test_argo_fieldoutofbounds_error(tmpdir) -> None:
 
     # TODO: capturing the warnings in the tests is complicated by the Parcels C-level print statements; but the logic of not crashing on out-of-bounds is tested if the test simulation runs
     # TODO: when using Parcels v4, this test can become much more robust by capturing the specific warning as well
+
+
+def test_argo_float_instrument_type():
+    """ArgoFloatInstrument returns the correct InstrumentType and if is underway instrument."""
+    sensors = [
+        SensorConfig(sensor_type=SensorType.TEMPERATURE),
+        SensorConfig(sensor_type=SensorType.SALINITY),
+    ]
+    expedition = create_dummy_expedition(sensors)
+
+    argo_instrument = ArgoFloatInstrument(expedition, from_data=None)
+    assert argo_instrument.instrument_type == InstrumentType.ARGO_FLOAT
+    assert not argo_instrument.instrument_type.is_underway

tests/instruments/test_ctd.py

@@ -10,8 +10,8 @@
 import pydantic
 import pytest
 import xarray as xr
-
 from parcels import Field, FieldSet
+
 from virtualship.instruments.ctd import CTD, CTDInstrument
 from virtualship.instruments.sensors import SensorType
 from virtualship.instruments.types import InstrumentType
@@ -23,26 +23,51 @@
     Waypoint,
 )
 
+BASE_TIME = datetime.datetime.strptime("1950-01-01", "%Y-%m-%d")
+MIN_DEPTH = -11
+MAX_DEPTH = -2000
+STATIONKEEPING_TIME = 50
+
+
+def create_dummy_expedition(
+    sensors, lifetime=datetime.timedelta(days=1), location=(1, 2)
+):
+    """Create a DummyExpedition class with specified sensors and parameters."""
+
+    class DummyExpedition:
+        class schedule:
+            waypoints: list[Waypoint] = [  # noqa: RUF012
+                Waypoint(location=Location(*location), time=BASE_TIME)
+            ]
+
+        instruments_config = InstrumentsConfig(
+            ctd_config=CTDConfig(
+                stationkeeping_time_minutes=STATIONKEEPING_TIME,
+                min_depth_meter=MIN_DEPTH,
+                max_depth_meter=MAX_DEPTH,
+                sensors=sensors,
+            )
+        )
+
+    return DummyExpedition()
+
 
 def test_simulate_ctds(tmpdir) -> None:
     """Test that CTDInstrument simulates measurements correctly, incuding sampling physical and bgc variables."""
-    # arbitrary time offset for the dummy fieldset
-    base_time = datetime.datetime.strptime("1950-01-01", "%Y-%m-%d")
-
     # where to cast CTDs
     ctds = [
         CTD(
             spacetime=Spacetime(
                 location=Location(latitude=0, longitude=1),
-                time=base_time + datetime.timedelta(hours=0),
+                time=BASE_TIME + datetime.timedelta(hours=0),
             ),
             min_depth=0,
             max_depth=float("-inf"),
         ),
         CTD(
             spacetime=Spacetime(
                 location=Location(latitude=1, longitude=0),
-                time=base_time,
+                time=BASE_TIME,
             ),
             min_depth=0,
             max_depth=float("-inf"),
@@ -132,8 +157,8 @@ def test_simulate_ctds(tmpdir) -> None:
         {"V": v, "U": u, "T": t, "S": s, "o2": o2, "chl": chl, "no3": no3},
         {
             "time": [
-                np.datetime64(base_time + datetime.timedelta(hours=0)),
-                np.datetime64(base_time + datetime.timedelta(hours=1)),
+                np.datetime64(BASE_TIME + datetime.timedelta(hours=0)),
+                np.datetime64(BASE_TIME + datetime.timedelta(hours=1)),
             ],
             "depth": [-1000, 0],
             "lat": [0, 1],
@@ -142,33 +167,15 @@ def test_simulate_ctds(tmpdir) -> None:
     )
     fieldset.add_field(Field("bathymetry", [-1000], lon=0, lat=0))
 
-    # dummy expedition for CTDInstrument
-    class DummyExpedition:
-        class schedule:
-            # ruff: noqa
-            waypoints = [
-                Waypoint(
-                    location=Location(1, 2),
-                    time=base_time,
-                ),
-            ]
-
-        instruments_config = InstrumentsConfig(
-            ctd_config=CTDConfig(
-                stationkeeping_time_minutes=50,
-                min_depth_meter=-11.0,
-                max_depth_meter=-2000.0,
-                sensors=[
-                    SensorConfig(sensor_type=SensorType.TEMPERATURE),
-                    SensorConfig(sensor_type=SensorType.SALINITY),
-                    SensorConfig(sensor_type=SensorType.OXYGEN),
-                    SensorConfig(sensor_type=SensorType.CHLOROPHYLL),
-                    SensorConfig(sensor_type=SensorType.NITRATE),
-                ],
-            )
-        )
+    sensors = [
+        SensorConfig(sensor_type=SensorType.TEMPERATURE),
+        SensorConfig(sensor_type=SensorType.SALINITY),
+        SensorConfig(sensor_type=SensorType.OXYGEN),
+        SensorConfig(sensor_type=SensorType.CHLOROPHYLL),
+        SensorConfig(sensor_type=SensorType.NITRATE),
+    ]
 
-    expedition = DummyExpedition()
+    expedition = create_dummy_expedition(sensors)
     from_data = None
 
     ctd_instrument = CTDInstrument(expedition, from_data)
@@ -278,22 +285,11 @@ def test_ctd_disabled_sensor_absent(tmpdir) -> None:
     )
     fieldset.add_field(Field("bathymetry", [-1000], lon=0, lat=0))
 
-    class DummyExpedition:
-        class schedule:
-            waypoints = [Waypoint(location=Location(1, 2), time=base_time)]
-
-        instruments_config = InstrumentsConfig(
-            ctd_config=CTDConfig(
-                stationkeeping_time_minutes=50,
-                min_depth_meter=-11.0,
-                max_depth_meter=-2000.0,
-                sensors=[
-                    SensorConfig(sensor_type=SensorType.TEMPERATURE)
-                ],  # SALINITY omitted = disabled
-            )
-        )
+    sensors = (
+        [SensorConfig(sensor_type=SensorType.TEMPERATURE)],
+    )  # SALINITY omitted = disabled
 
-    expedition = DummyExpedition()
+    expedition = create_dummy_expedition(sensors)
     ctd_instrument = CTDInstrument(expedition, None)
     out_path = tmpdir.join("out_disabled.zarr")
     ctd_instrument.load_input_data = lambda: fieldset
@@ -387,23 +383,12 @@ def test_sensor_absent(tmpdir) -> None:
     )
     fieldset.add_field(Field("bathymetry", [-1000], lon=0, lat=0))
 
-    class DummyExpedition:
-        class schedule:
-            waypoints = [Waypoint(location=Location(1, 2), time=base_time)]
-
-        instruments_config = InstrumentsConfig(
-            ctd_config=CTDConfig(
-                stationkeeping_time_minutes=50,
-                min_depth_meter=-11.0,
-                max_depth_meter=-2000.0,
-                sensors=[
-                    SensorConfig(sensor_type=SensorType.OXYGEN),
-                    # CHLOROPHYLL omitted = disabled
-                ],
-            )
-        )
+    sensors = [
+        SensorConfig(sensor_type=SensorType.OXYGEN),
+        # CHLOROPHYLL omitted = disabled
+    ]
 
-    expedition = DummyExpedition()
+    expedition = create_dummy_expedition(sensors)
     ctd_instrument = CTDInstrument(expedition, None)
     out_path = tmpdir.join("out_bgc_disabled.zarr")
     ctd_instrument.load_input_data = lambda: fieldset
@@ -412,3 +397,13 @@ class schedule:
     results = xr.open_zarr(out_path)
     assert "o2" in results, "Enabled BGC sensor variable must be present"
     assert "chl" not in results, "Disabled sensor variable must be absent from output"
+
+
+def test_ctd_instrument_type():
+    """CTDInstrument returns the correct InstrumentType and if is underway instrument."""
+    sensors = [SensorConfig(sensor_type=SensorType.TEMPERATURE)]  # only need one
+    expedition = create_dummy_expedition(sensors)
+
+    ctd_instrument = CTDInstrument(expedition, from_data=None)
+    assert ctd_instrument.instrument_type == InstrumentType.CTD
+    assert not ctd_instrument.instrument_type.is_underway