Refactor: move verification methods up to Expedition class

src/virtualship/cli/_plan.py

@@ -1124,20 +1124,24 @@ def save_pressed(self) -> None:
                     "Some waypoint entries are invalid or incomplete:\n\n"
                     + "\n".join(waypoint_errors)
                 )
+
             ship_speed = self.get_ship_speed(editor)
-            schedule = editor.expedition.schedule
-            schedule.verify(
-                ship_speed, editor.expedition.instruments_config, ignore_land_test=True
+            editor.expedition.verify_schedule(
+                ignore_land_test=True, ship_speed=ship_speed
             )
+
             if editor.save_changes():
                 self.notify(
                     "Changes saved successfully", severity="information", timeout=20
                 )
+
             # check for incomplete ports and warn the user, but allow save to continue
+            schedule = editor.expedition.schedule
             if not (
                 schedule.departure_port.is_in_use and schedule.arrival_port.is_in_use
             ):
                 self.notify(INCOMPLETE_PORT_MSG, severity="warning", timeout=20)
+
         except Exception as e:
             self.notify(
                 f"*** Error saving changes ***:\n\n{e}\n",

src/virtualship/cli/_run.py

@@ -92,7 +92,7 @@ def _run(
     )
 
     # verify instruments_config file is consistent with schedule
-    expedition.instruments_config.verify(expedition)
+    expedition.verify_instruments_config()
 
     # load last checkpoint
     checkpoint = _load_checkpoint(expedition_dir)
@@ -103,9 +103,7 @@ def _run(
 
     print("\n---- WAYPOINT VERIFICATION ----")
 
-    schedule.verify(
-        expedition.ship_config.ship_speed_knots,
-        expedition.instruments_config,
+    expedition.verify_schedule(
         from_data=Path(from_data) if from_data else None,
     )
 

src/virtualship/models/expedition.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import itertools
+from collections.abc import Collection
 from datetime import datetime, timedelta
 from pathlib import Path
 from typing import ClassVar, cast
@@ -51,7 +52,7 @@ def from_yaml(cls, file_path: str) -> Expedition:
             data = yaml.safe_load(file)
         return Expedition(**data)
 
-    def get_instruments(self) -> set[InstrumentType]:
+    def get_instruments(self) -> list[InstrumentType]:
         """Return a set of unique InstrumentType enums used in the expedition."""
         instruments_in_expedition = []
 
@@ -76,6 +77,99 @@ def get_instruments(self) -> set[InstrumentType]:
                 "Underway instrument config attribute(s) are missing from YAML. Must be <Instrument>Config object or None."
             ) from e
 
+    def verify_instruments_config(self) -> None:
+        """Check the instrument configurations cover exactly the instruments this expedition deploys."""
+        self.instruments_config.verify(self.get_instruments())
+
+    def verify_schedule(
+        self,
+        ignore_land_test: bool = False,
+        ship_speed: float | None = None,
+        from_data: Path | None = None,
+    ) -> None:
+        """
+        Verify the feasibility and correctness of the expedition's schedule.
+
+        `ship_speed` (also in knots) overrides the configured speed, for checking a prospective speed that has not been saved yet (e.g. in `plan` app).
+        """
+        print("\nVerifying expedition... ")
+
+        if ship_speed is None:
+            ship_speed = self.ship_config.ship_speed_knots
+
+        # waypoints excluding any inactive placeholder departure/arrival ports
+        wps_in_use = self.schedule._get_wps_in_use()
+
+        self.schedule._check_structure(wps_in_use)
+
+        if not ignore_land_test:
+            self.schedule._check_in_water(from_data)
+
+        # check that ship will arrive on time at each waypoint (in case no unexpected event happen)
+        overrun = self._first_overrun(wps_in_use, ship_speed)
+
+        if overrun is not None:
+            wp_i, wp_next, arrival_time = overrun
+
+            # TODO: add messaging of stationkeeping time to the error message, e.g. how much each instrument is taking...
+            raise ScheduleError(
+                f"Waypoint planning is not valid: would arrive too late at {self._describe_waypoint(wp_i, wp_next)}. "
+                f"Location: {wp_next.location} Time: {wp_next.time}. "
+                f"Currently projected to arrive at: {arrival_time}."
+                "\n\nHint: adding instruments may increase the amount of time spent stationary at a waypoints. "
+                "Have you ensured that your schedule includes sufficient time for taking measurements, e.g. CTD casts (in addition to the time it takes to sail between waypoints)?\n"
+            )
+
+        # finally, mark this schedule as verified (so that subsequent stages of the workflow can proceed without re-verifying)
+        self.schedule._verified = True
+
+        print("... All good to go!")
+
+    def _first_overrun(
+        self,
+        wps_in_use: list[Port | Waypoint],
+        ship_speed: float,
+    ) -> tuple[int, Port | Waypoint, datetime] | None:
+        """Project the ship along the schedule and return the first waypoint it reaches late."""
+        time = wps_in_use[0].time
+
+        # offset from wps_in_use indices to self.schedule.waypoints indices (a placeholder departure port is excluded from wps_in_use)
+        wps_in_use_offset = 0 if self.schedule.departure_port.is_in_use else 1
+
+        for wp_i, (wp, wp_next) in enumerate(
+            itertools.pairwise(wps_in_use), start=wps_in_use_offset
+        ):
+            stationkeeping_time = (
+                wp.stationkeeping_time(self.instruments_config)
+                if isinstance(wp, Waypoint)
+                else timedelta()
+            )
+
+            time_to_reach = _calc_sail_time(
+                wp.location,
+                wp_next.location,
+                ship_speed,
+                projection,
+            )[0]
+
+            arrival_time = time + time_to_reach + stationkeeping_time
+
+            if wp_next.time is None:
+                time = arrival_time
+            elif arrival_time > wp_next.time:
+                return wp_i, wp_next, arrival_time
+            else:
+                time = wp_next.time
+
+        return None
+
+    def _describe_waypoint(self, wp_i: int, wp_next: Port | Waypoint) -> str:
+        """Name the waypoint reached after index `wp_i`, for use in user-facing messages."""
+        if isinstance(wp_next, Port):
+            return "the final port of arrival"
+        # +1 to get next
+        return f"waypoint {_get_public_wp(wp_i + 1, self.schedule.waypoints)}"
+
     def _annotate(self):
         """
         Add port/waypoint comments/annotations to the expedition.yaml file.
@@ -159,28 +253,16 @@ def arrival_port(self) -> Port:
         """Arrival port (always the last waypoint)."""
         return cast(Port, self.waypoints[-1])
 
-    def verify(
-        self,
-        ship_speed: float,
-        instruments_config: InstrumentsConfig,
-        ignore_land_test: bool = False,
-        *,
-        from_data: Path | None = None,
-    ) -> None:
-        """Verify the feasibility and correctness of the schedule's waypoints."""
-        print("\nVerifying route... ")
-
-        # waypoints excluding any inactive placeholder departure/arrival ports
-        wps_in_use = self._get_wps_in_use()
-
+    def _check_structure(self, wps_in_use: list[Port | Waypoint]) -> None:
+        """Check the schedule is structured properly: has waypoint(s) and times in order."""
+        # has wps in use
         if not wps_in_use:
             raise ScheduleError(
                 "Schedule has no active waypoints: at least one waypoint, or an "
                 "in-use departure/arrival port (with both a location and a time), "
                 "must be provided."
             )
 
-        # is the departure port in use or a placeholder (i.e. all None)?
         wp_str = "Departure port" if self.departure_port.is_in_use else "Waypoint 1"
 
         if wps_in_use[0].time is None:
@@ -207,88 +289,40 @@ def verify(
                 f"Waypoint(s) {', '.join(invalid_labels)}: each waypoint should be timed after all previous waypoints",
             )
 
-        # check if all non-port waypoints are in water using bathymetry data
+    def _check_in_water(self, from_data: Path | None = None) -> None:
+        """Check non-port waypoints sit in water rather than on land."""
+        try:
+            bathymetry_field = _get_bathy_data(from_data=from_data).bathymetry
+        except Exception as e:
+            raise ScheduleError(
+                f"Problem loading bathymetry data (used to verify waypoints are in water) directly via copernicusmarine. \n\n original message: {e}"
+            ) from e
+
         land_waypoints = []
-        if not ignore_land_test:
+
+        for wp_i, wp in enumerate(self.waypoints):
+            if isinstance(wp, Port):
+                continue  # ports are in harbour; skip bathymetry land check
+            public_wp = _get_public_wp(wp_i, self.waypoints)
             try:
-                bathymetry_field = _get_bathy_data(from_data=from_data).bathymetry
+                value = bathymetry_field.eval(
+                    0,  # time
+                    0,  # depth (surface)
+                    wp.location.lat,
+                    wp.location.lon,
+                )
+                if value == 0.0 or (isinstance(value, float) and np.isnan(value)):
+                    land_waypoints.append((public_wp, wp))
             except Exception as e:
                 raise ScheduleError(
-                    f"Problem loading bathymetry data (used to verify waypoints are in water) directly via copernicusmarine. \n\n original message: {e}"
+                    f"Waypoint #{public_wp} at location {wp.location} could not be evaluated against bathymetry data. \n\n Original error: {e}"
                 ) from e
 
-            for wp_i, wp in enumerate(self.waypoints):
-                if isinstance(wp, Port):
-                    continue  # ports are in harbour; skip bathymetry land check
-                public_wp = _get_public_wp(wp_i, self.waypoints)
-                try:
-                    value = bathymetry_field.eval(
-                        0,  # time
-                        0,  # depth (surface)
-                        wp.location.lat,
-                        wp.location.lon,
-                    )
-                    if value == 0.0 or (isinstance(value, float) and np.isnan(value)):
-                        land_waypoints.append((public_wp, wp))
-                except Exception as e:
-                    raise ScheduleError(
-                        f"Waypoint #{public_wp} at location {wp.location} could not be evaluated against bathymetry data. \n\n Original error: {e}"
-                    ) from e
-
-            if len(land_waypoints) > 0:
-                raise ScheduleError(
-                    f"The following waypoint(s) throw(s) error(s): {['#' + str(public_wp) + ' ' + str(wp) for (public_wp, wp) in land_waypoints]}\n\nINFO: They are likely on land (bathymetry data cannot be interpolated to their location(s)).\n"
-                )
-
-        # check that ship will arrive on time at each waypoint (in case no unexpected event happen)
-        time = wps_in_use[0].time
-
-        # offset from wps_in_use indices to self.waypoints indices (a placeholder departure port is excluded from wps_in_use)
-        wps_in_use_offset = 0 if self.departure_port.is_in_use else 1
-
-        for wp_i, (wp, wp_next) in enumerate(
-            itertools.pairwise(wps_in_use), start=wps_in_use_offset
-        ):
-            stationkeeping_time = (
-                wp.stationkeeping_time(instruments_config)
-                if isinstance(wp, Waypoint)
-                else timedelta()
+        if len(land_waypoints) > 0:
+            raise ScheduleError(
+                f"The following waypoint(s) throw(s) error(s): {['#' + str(public_wp) + ' ' + str(wp) for (public_wp, wp) in land_waypoints]}\n\nINFO: They are likely on land (bathymetry data cannot be interpolated to their location(s)).\n"
             )
 
-            time_to_reach = _calc_sail_time(
-                wp.location,
-                wp_next.location,
-                ship_speed,
-                projection,
-            )[0]
-
-            arrival_time = time + time_to_reach + stationkeeping_time
-
-            if wp_next.time is None:
-                time = arrival_time
-            elif arrival_time > wp_next.time:
-                affected = (
-                    f"waypoint {_get_public_wp(wp_i + 1, self.waypoints)}"  # +1 to get next
-                    if not isinstance(wp_next, Port)
-                    else "the final port of arrival"
-                )
-
-                # TODO: add messaging of stationkeeping time to the error message, e.g. how much each instrument is taking...
-                raise ScheduleError(
-                    f"Waypoint planning is not valid: would arrive too late at {affected}. "
-                    f"Location: {wp_next.location} Time: {wp_next.time}. "
-                    f"Currently projected to arrive at: {arrival_time}."
-                    "\n\nHint: adding instruments may increase the amount of time spent stationary at a waypoints. "
-                    "Have you ensured that your schedule includes sufficient time for taking measurements, e.g. CTD casts (in addition to the time it takes to sail between waypoints)?\n"
-                )
-            else:
-                time = wp_next.time
-
-        # finally, mark this schedule as verified (so that subsequent stages of the workflow can proceed without re-verifying)
-        self._verified = True
-
-        print("... All good to go!")
-
     def _get_wps_in_use(self) -> list[Port | Waypoint]:
         """Return waypoints that are in use (i.e., have a specified time and location), i.e. excluding placeholder departure/arrival ports."""
         start_slice = 0 if self.departure_port.is_in_use else 1
@@ -651,14 +685,13 @@ class InstrumentsConfig(pydantic.BaseModel):
 
     model_config = pydantic.ConfigDict(extra="forbid")
 
-    def verify(self, expedition: Expedition) -> None:
+    def verify(self, instruments_in_expedition: Collection[InstrumentType]) -> None:
         """
-        Verify instrument configurations against the schedule.
+        Verify instrument configurations against the instruments actually being deployed.
 
         Removes instrument configs not present in the schedule and checks that all scheduled instruments are configured.
         Raises ConfigError if any scheduled instrument is missing a config.
         """
-        instruments_in_expedition = expedition.get_instruments()
         instrument_config_map = {
             InstrumentType.ARGO_FLOAT: "argo_float_config",
             InstrumentType.DRIFTER: "drifter_config",

tests/cli/test_run.py

@@ -52,10 +52,11 @@ def test_run(tmp_path, monkeypatch):
     monkeypatch.setattr("virtualship.cli._run.simulate_schedule", _simulate_schedule)
 
     monkeypatch.setattr(
-        "virtualship.models.InstrumentsConfig.verify", lambda self, expedition: None
+        "virtualship.models.InstrumentsConfig.verify", lambda self, instruments: None
     )
     monkeypatch.setattr(
-        "virtualship.models.Schedule.verify", lambda self, *args, **kwargs: None
+        "virtualship.models.Expedition.verify_schedule",
+        lambda self, *args, **kwargs: None,
     )
 
     monkeypatch.setattr(