Standardize how we check for a run being completed

flood_adapt/dbs_classes/dbs_scenario.py

@@ -40,7 +40,7 @@ def list_objects(self) -> dict[str, Any]:
         scenarios["Event"] = [obj.attrs.event for obj in objects]
         scenarios["Strategy"] = [obj.attrs.strategy for obj in objects]
         scenarios["finished"] = [
-            obj.init_object_model().direct_impacts.has_run for obj in objects
+            obj.init_object_model().has_run_check() for obj in objects
         ]
 
         return scenarios

flood_adapt/integrator/fiat_adapter.py

@@ -60,7 +60,7 @@ def __del__(self) -> None:
         gc.collect()
 
     def set_hazard(self, hazard: Hazard) -> None:
-        map_fn = hazard.flood_map_path
+        map_fn = hazard._get_flood_map_path()
         map_type = hazard.site.attrs.fiat.floodmap_type
         var = "zsmax" if hazard.event_mode == Mode.risk else "risk_maps"
         is_risk = hazard.event_mode == Mode.risk

flood_adapt/object_model/hazard/hazard.py

@@ -727,6 +727,7 @@ def _get_flood_map_path(self):
                 map_fn.append(results_path.joinpath(f"RP_{rp:04d}_maps.nc"))
 
         self.flood_map_path = map_fn
+        return map_fn
 
     def write_water_level_map(self):
         """Read simulation results from SFINCS and saves a netcdf with the maximum water levels."""

flood_adapt/object_model/scenario.py

@@ -10,6 +10,7 @@
 from flood_adapt.object_model.direct_impacts import DirectImpacts
 from flood_adapt.object_model.hazard.hazard import ScenarioModel
 from flood_adapt.object_model.interface.scenarios import IScenario
+from flood_adapt.object_model.utils import finished_file_exists, write_finished_file
 
 
 class Scenario(IScenario):
@@ -94,6 +95,13 @@ def run(self):
                 f"Finished evaluation of {self.attrs.name} for {self.site_info.attrs.name}"
             )
 
+        # write finished file to indicate that the scenario has been run
+        write_finished_file(self.results_path)
+
+    def has_run_check(self):
+        """Check if the scenario has been run."""
+        return finished_file_exists(self.results_path)
+
     def __eq__(self, other):
         if not isinstance(other, Scenario):
             # don't attempt to compare against unrelated types

flood_adapt/object_model/utils.py

@@ -11,3 +11,14 @@ def cd(newdir: Path):
         yield
     finally:
         os.chdir(prevdir)
+
+
+def write_finished_file(path: Path):
+    if not path.exists():
+        path.mkdir(parents=True)
+    with open(Path(path) / "finished.txt", "w") as f:
+        f.write("run finished")
+
+
+def finished_file_exists(path: Path):
+    return (Path(path) / "finished.txt").exists()