refactor: iamc export

remind_mfa/cement/cement_export.py

@@ -1,120 +1,47 @@
-import pyam
-import flodym as fd
-
-from remind_mfa.common.common_export import CommonDataExporter
+from remind_mfa.common.common_export import CommonDataExporter, IamcVariable
 
 
 class CementDataExporter(CommonDataExporter):
-    def write_iamc(self, mfa: fd.MFASystem):
-
-        model = "REMIND 3.0"
-        scenario = "SSP2_NPi"
-        constants = {"model": model, "scenario": scenario}
-
-        reported_dims = ("t", "r", "s")
-
-        # cement production
-        cement_prod_by_stock_type = (
-            mfa.flows["prod_cement => market_cement"] + mfa.flows["prod_cement => sysenv"]
-        )
-        prod_df = self.to_iamc_df(cement_prod_by_stock_type)
-        prod_df["variable"] = "Production|Non-Metallic Minerals|Cement"
-        cement_prod_idf = pyam.IamDataFrame(
-            prod_df,
-            unit="t/yr",
-            **constants,
-        )
-        cement_prod_idf.aggregate(
-            variable="Production|Non-Metallic Minerals|Cement",
-            append=True,
-        )
-
-        # clinker production
-        clinker_prod_by_stock_type = (
-            mfa.flows["prod_clinker => market_clinker"] + mfa.flows["prod_clinker => sysenv"]
-        )
-        clinker_prod_df = self.to_iamc_df(clinker_prod_by_stock_type)
-        clinker_prod_df["variable"] = "Production|Non-Metallic Minerals|Cement Clinker"
-        clinker_prod_idf = pyam.IamDataFrame(
-            clinker_prod_df,
-            unit="t/yr",
-            **constants,
-        )
-        clinker_prod_idf.aggregate(
-            variable="Production|Non-Metallic Minerals|Cement Clinker",
-            append=True,
-        )
-
-        # cement demand
-        cement_demand_by_stock_type = mfa.flows["market_cement => prod_product"].sum_to(
-            reported_dims
-        )
-        demand_df = self.to_iamc_df(cement_demand_by_stock_type)
-        demand_df["variable"] = (
-            "Material Demand|Non-Metallic Minerals|Cement|" + demand_df["Stock Type"]
-        )
-        demand_df = demand_df.drop(columns=["Stock Type"])
-        demand_idf = pyam.IamDataFrame(
-            demand_df,
-            unit="t/yr",
-            **constants,
-        )
-        demand_idf.aggregate(
-            variable="Material Demand|Non-Metallic Minerals|Cement",
-            append=True,
-        )
-
-        # cement stocks
-        cement_stock_by_stock_type = (
-            mfa.stocks["in_use"].stock[{"k": "cement"}].sum_to(reported_dims)
-        )
-        stock_df = self.to_iamc_df(cement_stock_by_stock_type)
-        stock_df["variable"] = (
-            "Material Stock|Non-Metallic Minerals|Cement|" + stock_df["Stock Type"]
-        )
-        stock_df = stock_df.drop(columns=["Stock Type"])
-        stock_idf = pyam.IamDataFrame(
-            stock_df,
-            unit="t",
-            **constants,
-        )
-        stock_idf.aggregate(
-            variable="Material Stock|Non-Metallic Minerals|Cement",
-            append=True,
-        )
-
-        # cement eol
-        cement_scrap_by_stock_type = (
-            mfa.stocks["in_use"].outflow[{"k": "cement"}].sum_to(reported_dims)
-        )
-        scrap_df = self.to_iamc_df(cement_scrap_by_stock_type)
-        scrap_df["variable"] = "Scrap|Non-Metallic Minerals|Cement|" + scrap_df["Stock Type"]
-        scrap_df = scrap_df.drop(columns=["Stock Type"])
-        scrap_idf = pyam.IamDataFrame(
-            scrap_df,
-            unit="t/yr",
-            **constants,
-        )
-        scrap_idf.aggregate(
-            variable="Scrap|Non-Metallic Minerals|Cement",
-            append=True,
-        )
-
-        idf = pyam.concat(
-            [
-                cement_prod_idf,
-                clinker_prod_idf,
-                demand_idf,
-                stock_idf,
-                scrap_idf,
-            ]
-        )
-        idf.aggregate_region(
-            variable=idf.variable,
-            region="World",
-            append=True,
-        )
-        idf.convert_unit(current="t/yr", to="Mt/yr", inplace=True)
-        idf.convert_unit(current="t", to="Mt", inplace=True)
 
-        idf.to_excel(self.export_path("iamc", f"output_iamc.xlsx"))
+    def iamc_variables(self) -> list[IamcVariable]:
+        return [
+            IamcVariable(
+                variable_name="Production|Non-Metallic Minerals|Cement",  # PRISMA nomenclature
+                calculation_function=lambda mfa: (
+                    mfa.flows["prod_cement => market_cement"] + mfa.flows["prod_cement => sysenv"]
+                ).sum_to(("t", "r")),
+                unit="t/yr",
+            ),
+            IamcVariable(
+                variable_name="Production|Non-Metallic Minerals|Cement Clinker",  # PRISMA nomenclature
+                calculation_function=lambda mfa: (
+                    mfa.flows["prod_clinker => market_clinker"]
+                    + mfa.flows["prod_clinker => sysenv"]
+                ).sum_to(("t", "r")),
+                unit="t/yr",
+            ),
+            IamcVariable(
+                variable_name="Material Demand|Non-Metallic Minerals|Cement",  # PRISMA nomenclature
+                calculation_function=lambda mfa: mfa.flows["market_cement => prod_product"].sum_to(
+                    ("t", "r", "s")
+                ),
+                unit="t/yr",
+                split_name="Stock Type",
+            ),
+            IamcVariable(
+                variable_name="Material Stock|Non-Metallic Minerals|Cement",  # PRISMA nomenclature
+                calculation_function=lambda mfa: mfa.stocks["in_use"]
+                .stock[{"k": "cement"}]
+                .sum_to(("t", "r", "s")),
+                unit="t",
+                split_name="Stock Type",
+            ),
+            IamcVariable(
+                variable_name="Scrap|Non-Metallic Minerals|Cement",  # PRISMA nomenclature
+                calculation_function=lambda mfa: (
+                    mfa.stocks["in_use"].outflow[{"k": "cement"}].sum_to(("t", "r", "s"))
+                ),
+                unit="t/yr",
+                split_name="Stock Type",
+            ),
+        ]

remind_mfa/common/common_config.py

@@ -69,6 +69,33 @@ class BaseExportCfg(RemindMFABaseModel):
     """Path to export folder for this entity"""
 
 
+class IamcExportCfg(BaseExportCfg):
+    t: str = "1950:2100"
+    """Inclusive "start:stop" year range to export in IAMC format (e.g. "1950:2100")."""
+
+    @property
+    def time_items(self) -> list[int]:
+        start, stop = self._parse_range()
+        return list(range(start, stop + 1))  # inclusive of stop
+
+    def _parse_range(self) -> tuple[int, int]:
+        parts = str(self.t).split(":")
+        if len(parts) != 2:
+            raise ValueError(f"iamc.t must be a 'start:stop' year range, got {self.t!r}")
+        try:
+            start, stop = int(parts[0]), int(parts[1])
+        except ValueError:
+            raise ValueError(f"iamc.t must be a 'start:stop' year range, got {self.t!r}")
+        if start > stop:
+            raise ValueError(f"iamc.t start must not be after stop, got {self.t!r}")
+        return start, stop
+
+    @model_validator(mode="after")
+    def validate_time_range(self):
+        self._parse_range()
+        return self
+
+
 class ExportCfg(BaseExportCfg):
     csv: BaseExportCfg
     """Configuration of export to CSV files"""
@@ -78,7 +105,7 @@ class ExportCfg(BaseExportCfg):
     """Configuration of export of assumptions to a txt file."""
     docs: BaseExportCfg
     """Configuration of export to documentation files."""
-    iamc: BaseExportCfg
+    iamc: IamcExportCfg
     """Configuration of export of results in IAMC format."""