Generate post-expedition problems report

src/virtualship/cli/_run.py

@@ -18,8 +18,10 @@
 from virtualship.utils import (
     CHECKPOINT,
     EXPEDITION,
+    LOG_DIR,
     PROBLEMS_ENCOUNTERED_DIR,
     PROJECTION,
+    REPORT,
     SELECTED_PROBLEMS,
     _get_expedition,
     _save_checkpoint,
@@ -90,7 +92,7 @@ def _run(
         checkpoint = Checkpoint(past_schedule=Schedule(waypoints=[]))
 
     # verify that schedule and checkpoint match, and that problems have been resolved
-    checkpoint.verify(expedition, problems_dir)
+    checkpoint.verify(expedition, problems_dir.joinpath(LOG_DIR))
 
     print("\n---- WAYPOINT VERIFICATION ----")
 
@@ -138,16 +140,16 @@ def _run(
     problem_simulator = ProblemSimulator(expedition, expedition_dir)
 
     # re-load previously encountered (same expedition as previously) problems if they exist, else select new problems and cache them
-    if os.path.exists(problems_dir / SELECTED_PROBLEMS):
+    if os.path.exists(problems_dir / LOG_DIR / SELECTED_PROBLEMS):
         problems = problem_simulator.load_selected_problems(
-            problems_dir / SELECTED_PROBLEMS
+            problems_dir / LOG_DIR / SELECTED_PROBLEMS
         )
     else:
         problems = problem_simulator.select_problems(
             instruments_in_expedition, prob_level
         )
         problem_simulator.cache_selected_problems(
-            problems, problems_dir / SELECTED_PROBLEMS
+            problems, problems_dir / LOG_DIR / SELECTED_PROBLEMS
         ) if problems else None
 
     # simulate instrument measurements
@@ -174,7 +176,7 @@ def _run(
                 problem_simulator.execute(
                     problems,
                     instrument_type_validation=itype,
-                    problems_dir=problems_dir,
+                    log_dir=problems_dir.joinpath(LOG_DIR),
                 )
 
             # get measurements to simulate
@@ -215,9 +217,10 @@ def _run(
     )
 
     if problems:
+        ProblemSimulator.post_expedition_report(problems, problems_dir.joinpath(REPORT))
         print("\n----- RECORD OF PROBLEMS ENCOUNTERED ------")
         print(
-            f"\nA record of problems encountered during the expedition is saved in: {problems_dir}"
+            f"\nA post-expedition report of problems encountered during the expedition is saved in: {problems_dir.joinpath(REPORT)}"
         )
 
     # delete checkpoint file (in case it interferes with any future re-runs)

src/virtualship/make_realistic/problems/simulator.py

@@ -190,7 +190,7 @@ def execute(
         self,
         problems: dict[str, list[GeneralProblem | InstrumentProblem] | None],
         instrument_type_validation: InstrumentType | None,
-        problems_dir: Path,
+        log_dir: Path,
         log_delay: float = 4.0,
     ):
         """
@@ -209,7 +209,7 @@ def execute(
                 continue
 
             problem_hash = _make_hash(problem.message + str(problem_waypoint_i), 8)
-            hash_fpath = problems_dir.joinpath(f"problem_{problem_hash}.json")
+            hash_fpath = log_dir.joinpath(f"problem_{problem_hash}.json")
             if hash_fpath.exists():
                 continue  # problem * waypoint combination has already occurred; don't repeat
 
@@ -232,7 +232,7 @@ def execute(
             )
 
             # cache original schedule for reference and/or restoring later if needed (checkpoint.yaml [written in _log_problem] can be overwritten if multiple problems occur so is not a persistent record of original schedule)
-            schedule_original_fpath = problems_dir / SCHEDULE_ORIGINAL
+            schedule_original_fpath = log_dir / SCHEDULE_ORIGINAL
             if not os.path.exists(schedule_original_fpath):
                 self._cache_original_schedule(
                     self.expedition.schedule, schedule_original_fpath
@@ -263,6 +263,25 @@ def cache_selected_problems(
                 indent=4,
             )
 
+    @staticmethod
+    def post_expedition_report(
+        problems: dict[str, list[GeneralProblem | InstrumentProblem] | None],
+        report_fpath: str | Path,
+    ) -> None:
+        """Produce human-readable post-expedition report (.txt), including problems that occured (their full messages), the waypoint and what delay they caused."""
+        for problem, problem_waypoint_i in zip(
+            problems["problem_class"], problems["waypoint_i"], strict=True
+        ):
+            affected_wp = (
+                "in-port" if problem_waypoint_i is None else f"{problem_waypoint_i + 1}"
+            )
+            delay_hours = problem.delay_duration.total_seconds() / 3600.0
+            with open(report_fpath, "a", encoding="utf-8") as f:
+                f.write("---\n")
+                f.write(f"Waypoint: {affected_wp}\n")
+                f.write(f"Problem: {problem.message}\n")
+                f.write(f"Delay caused: {delay_hours} hours\n\n")
+
     @staticmethod
     def load_selected_problems(
         selected_problems_fpath: str,

src/virtualship/utils.py

@@ -37,10 +37,12 @@
 
 EXPEDITION = "expedition.yaml"
 CHECKPOINT = "checkpoint.yaml"
-SCHEDULE_ORIGINAL = "schedule_original.yaml"
 
 PROBLEMS_ENCOUNTERED_DIR = "problems_encountered_" + "{expedition_identifier}"
+LOG_DIR = "log"
+SCHEDULE_ORIGINAL = "schedule_original.yaml"
 SELECTED_PROBLEMS = "selected_problems.json"
+REPORT = "post_expedition_report.txt"
 
 # projection used to sail between waypoints
 PROJECTION = pyproj.Geod(ellps="WGS84")

tests/make_realistic/problems/test_simulator.py

@@ -1,7 +1,12 @@
 import json
+import random
 from datetime import datetime, timedelta
 
-from virtualship.make_realistic.problems.scenarios import GeneralProblem
+from virtualship.instruments.types import InstrumentType
+from virtualship.make_realistic.problems.scenarios import (
+    GeneralProblem,
+    InstrumentProblem,
+)
 from virtualship.make_realistic.problems.simulator import ProblemSimulator
 from virtualship.models.expedition import (
     Expedition,
@@ -11,21 +16,26 @@
     Waypoint,
 )
 from virtualship.models.location import Location
-from virtualship.utils import GENERAL_PROBLEM_REG
+from virtualship.utils import GENERAL_PROBLEM_REG, REPORT
 
 
 def _make_simple_expedition(
-    num_waypoints: int = 2, distance_scale: float = 1.0
+    num_waypoints: int = 2, distance_scale: float = 1.0, no_instruments: bool = False
 ) -> Expedition:
+    """Func. rather than fixture to allow for configurability in different tests."""
     sample_datetime = datetime(2024, 1, 1, 0, 0, 0)
+    instruments_non_underway = [inst for inst in InstrumentType if not inst.is_underway]
+
     waypoints = []
     for i in range(num_waypoints):
         wp = Waypoint(
             location=Location(
                 latitude=0.0 + i * distance_scale, longitude=0.0 + i * distance_scale
             ),
             time=sample_datetime + timedelta(days=i),
-            instrument=[],  # ensure is list, not None
+            instrument=[]
+            if no_instruments
+            else random.sample(instruments_non_underway, 3),
         )
         waypoints.append(wp)
 
@@ -39,8 +49,9 @@ def _make_simple_expedition(
 
 def test_select_problems_single_waypoint_returns_pre_departure(tmp_path):
     expedition = _make_simple_expedition(num_waypoints=1)
+    instruments_in_expedition = expedition.get_instruments()
     simulator = ProblemSimulator(expedition, str(tmp_path))
-    problems = simulator.select_problems(set(), prob_level=2)
+    problems = simulator.select_problems(instruments_in_expedition, prob_level=2)
 
     assert isinstance(problems, dict)
     assert len(problems["problem_class"]) == 1
@@ -51,11 +62,28 @@ def test_select_problems_single_waypoint_returns_pre_departure(tmp_path):
     assert getattr(problem_cls, "pre_departure", False) is True
 
 
+def test_no_instruments_no_instruments_problems(tmp_path):
+    expedition = _make_simple_expedition(num_waypoints=2, no_instruments=True)
+    instruments_in_expedition = expedition.get_instruments()
+    assert len(instruments_in_expedition) == 0, "Expedition should have no instruments"
+
+    simulator = ProblemSimulator(expedition, str(tmp_path))
+    problems = simulator.select_problems(instruments_in_expedition, prob_level=2)
+
+    has_instrument_problems = any(
+        issubclass(cls, InstrumentProblem) for cls in problems["problem_class"]
+    )
+    assert not has_instrument_problems, (
+        "Should not select instrument problems when no instruments are present"
+    )
+
+
 def test_select_problems_prob_level_zero():
     expedition = _make_simple_expedition(num_waypoints=2)
+    instruments_in_expedition = expedition.get_instruments()
     simulator = ProblemSimulator(expedition, ".")
 
-    problems = simulator.select_problems(set(), prob_level=0)
+    problems = simulator.select_problems(instruments_in_expedition, prob_level=0)
     assert problems is None
 
 
@@ -64,10 +92,10 @@ def test_cache_and_load_selected_problems_roundtrip(tmp_path):
     simulator = ProblemSimulator(expedition, str(tmp_path))
 
     # pick two general problems (registry should contain entries)
-    cls1 = GENERAL_PROBLEM_REG[0]
-    cls2 = GENERAL_PROBLEM_REG[1] if len(GENERAL_PROBLEM_REG) > 1 else cls1
+    problem1 = GENERAL_PROBLEM_REG[0]
+    problem2 = GENERAL_PROBLEM_REG[1] if len(GENERAL_PROBLEM_REG) > 1 else problem1
 
-    problems = {"problem_class": [cls1, cls2], "waypoint_i": [None, 0]}
+    problems = {"problem_class": [problem1, problem2], "waypoint_i": [None, 0]}
 
     sel_fpath = tmp_path / "subdir" / "selected_problems.json"
     simulator.cache_selected_problems(problems, str(sel_fpath))
@@ -89,11 +117,11 @@ def test_hash_to_json(tmp_path):
     expedition = _make_simple_expedition(num_waypoints=2)
     simulator = ProblemSimulator(expedition, str(tmp_path))
 
-    cls = GENERAL_PROBLEM_REG[0]
+    any_problem = GENERAL_PROBLEM_REG[0]
 
     hash_path = tmp_path / "problem_hash.json"
     simulator._hash_to_json(
-        cls, "deadbeef", None, hash_path
+        any_problem, "deadbeef", None, hash_path
     )  # "deadbeef" as sub for hex in test
 
     assert hash_path.exists()
@@ -107,18 +135,27 @@ def test_hash_to_json(tmp_path):
 def test_has_contingency_pre_departure(tmp_path):
     expedition = _make_simple_expedition(num_waypoints=2)
     simulator = ProblemSimulator(expedition, str(tmp_path))
-    cls = GENERAL_PROBLEM_REG[0]
 
-    # _has_contingency should return False for pre-departure (None)
-    assert simulator._has_contingency(cls, None) is False
+    pre_departure_problem = next(
+        gp for gp in GENERAL_PROBLEM_REG if getattr(gp, "pre_departure", False)
+    )
+    assert pre_departure_problem is not None, (
+        "Need at least one pre-departure problem class in the general problem registry"
+    )
+
+    # _has_contingency should return False for pre-departure (waypoint = None)
+    assert simulator._has_contingency(pre_departure_problem, None) is False
 
 
 def test_select_problems_prob_levels(tmp_path):
     expedition = _make_simple_expedition(num_waypoints=3)
+    instruments_in_expedition = expedition.get_instruments()
     simulator = ProblemSimulator(expedition, str(tmp_path))
 
     for level in range(3):  # prob levels 0, 1, 2
-        problems = simulator.select_problems(set(), prob_level=level)
+        problems = simulator.select_problems(
+            instruments_in_expedition, prob_level=level
+        )
         if level == 0:
             assert problems is None
         else:
@@ -132,13 +169,22 @@ def test_select_problems_prob_levels(tmp_path):
 def test_prob_level_two_more_problems(tmp_path):
     prob_level = 2
 
-    short_expedition = _make_simple_expedition(num_waypoints=2)
+    short_expedition = _make_simple_expedition(
+        num_waypoints=2
+    )  # short in terms of number of waypoints
+    instruments_in_short_expedition = short_expedition.get_instruments()
     simulator_short = ProblemSimulator(short_expedition, str(tmp_path))
+
     long_expedition = _make_simple_expedition(num_waypoints=12)
+    instruments_in_long_expedition = long_expedition.get_instruments()
     simulator_long = ProblemSimulator(long_expedition, str(tmp_path))
 
-    problems_short = simulator_short.select_problems(set(), prob_level=prob_level)
-    problems_long = simulator_long.select_problems(set(), prob_level=prob_level)
+    problems_short = simulator_short.select_problems(
+        instruments_in_short_expedition, prob_level=prob_level
+    )
+    problems_long = simulator_long.select_problems(
+        instruments_in_long_expedition, prob_level=prob_level
+    )
 
     assert len(problems_long["problem_class"]) >= len(
         problems_short["problem_class"]
@@ -165,3 +211,31 @@ def test_has_contingency_during_expedition(tmp_path):
     # short distance expedition should have contingency, long distance should not (given time between waypoints and ship speed is constant)
     assert short_simulator._has_contingency(problem_cls, problem_waypoint_i=0) is True
     assert long_simulator._has_contingency(problem_cls, problem_waypoint_i=0) is False
+
+
+def test_post_expedition_report(tmp_path):
+    expedition = _make_simple_expedition(
+        num_waypoints=12
+    )  # longer expedition to increase likelihood of multiple problems at prob_level=2
+    instruments_in_expedition = expedition.get_instruments()
+
+    simulator = ProblemSimulator(expedition, str(tmp_path))
+    problems = simulator.select_problems(instruments_in_expedition, prob_level=2)
+
+    report_path = tmp_path / REPORT
+    simulator.post_expedition_report(problems, report_path)
+
+    assert report_path.exists()
+    with open(report_path, encoding="utf-8") as f:
+        content = f.read()
+
+    assert content.count("Problem:") == len(problems["problem_class"]), (
+        "Number of reported problems should match number of selected problems."
+    )
+    assert content.count("Delay caused:") == len(problems["problem_class"]), (
+        "Number of reported delay durations should match number of selected problems."
+    )
+    for problem in problems["problem_class"]:
+        assert problem.message in content, (
+            "Problem messages in report should match those of selected problems."
+        )