Feat/validation between classes

feo/osemosys/schemas/commodity.py

@@ -80,11 +80,11 @@ def from_otoole_csv(cls, root_dir):
         if len(df_commodity) != len(df_commodity["VALUE"].unique()):
             raise ValueError("FUEL.csv must not contain duplicate values")
 
-        # Check impact names are consistent with those in FUEL.csv
+        # Check commodity names are consistent with those in FUEL.csv
         for df in dfs.keys():
-            for impact in dfs[df]["FUEL"].unique():
-                if impact not in list(df_commodity["VALUE"]):
-                    raise ValueError(f"{impact} given in {df}.csv but not in FUEL.csv")
+            for commodity in dfs[df]["FUEL"].unique():
+                if commodity not in list(df_commodity["VALUE"]):
+                    raise ValueError(f"{commodity} given in {df}.csv but not in FUEL.csv")
 
         # ########################
         # Define class instances #

feo/osemosys/schemas/model.py

@@ -4,6 +4,7 @@
 
 import xarray as xr
 import yaml
+from pydantic import model_validator
 
 from feo.osemosys.defaults import DefaultsOtoole, defaults
 from feo.osemosys.schemas.base import OSeMOSYSBase
@@ -12,6 +13,12 @@
 from feo.osemosys.schemas.region import Region
 from feo.osemosys.schemas.technology import Technology, TechnologyStorage
 from feo.osemosys.schemas.time_definition import TimeDefinition
+from feo.osemosys.schemas.validation.model_composition import (
+    check_tech_consuming_commodity,
+    check_tech_producing_commodity,
+    check_tech_producing_impact,
+)
+from feo.osemosys.schemas.validation.model_presolve import check_able_to_meet_demands
 from feo.osemosys.utils import to_df_helper
 
 # filter this pandas-3 dep warning for now
@@ -42,6 +49,15 @@ class RunSpec(OSeMOSYSBase):
     # Default values
     defaults_otoole: Optional[DefaultsOtoole] = None
 
+    @model_validator(mode="after")
+    def validation(cls, values):
+        values = check_tech_producing_commodity(values)
+        values = check_tech_producing_impact(values)
+        values = check_tech_consuming_commodity(values)
+        values = check_able_to_meet_demands(values)
+
+        return values
+
     def to_xr_ds(self):
         """
         Return the current RunSpec as an xarray dataset

feo/osemosys/schemas/validation/model_composition.py

@@ -0,0 +1,53 @@
+def check_tech_producing_commodity(values):
+    """
+    For each commodity, check there is a technology which produces it
+    """
+    for commodity in values.commodities:
+        technology_missing = True
+        for technology in values.technologies:
+            if technology_missing and technology.output_activity_ratio is not None:
+                for region in technology.output_activity_ratio.data.keys():
+                    if commodity.id in technology.output_activity_ratio.data[region].keys():
+                        technology_missing = False
+        if technology_missing:
+            raise ValueError(f"Commodity '{commodity.id}' is not an output of any technology")
+
+    return values
+
+
+def check_tech_producing_impact(values):
+    """
+    For each impact, check there is a technology which produces it
+    """
+    for impact in values.impacts:
+        technology_missing = True
+        for technology in values.technologies:
+            if technology_missing and technology.emission_activity_ratio is not None:
+                for region in technology.emission_activity_ratio.data.keys():
+                    if impact.id in technology.emission_activity_ratio.data[region].keys():
+                        technology_missing = False
+        if technology_missing:
+            raise ValueError(f"Impact '{impact.id}' is not an output of any technology")
+
+    return values
+
+
+def check_tech_consuming_commodity(values):
+    """
+    For each commodity which isn't a final demand, check it is the input of a technology
+    """
+    for commodity in values.commodities:
+        if commodity.demand_annual is None and commodity.accumulated_demand is None:
+            technology_missing = True
+            for technology in values.technologies:
+                if technology_missing and technology.input_activity_ratio is not None:
+                    for region in technology.input_activity_ratio.data.keys():
+                        if commodity.id in technology.input_activity_ratio.data[region].keys():
+                            technology_missing = False
+            if technology_missing:
+                raise ValueError(
+                    f"Commodity '{commodity.id}' is neither a final demand nor "
+                    f"an input of any technology"
+                )
+
+    return values