[Econ] Add partial budget breakdown output

RUFAS/EEE/economics/framework.py

@@ -149,6 +149,7 @@ def run_economic_analysis(self) -> None:
             self._build_line_item_breakdown(preprocessed_results),
             info_map=info_map,
         )
+        self.partial_budget.export_line_item_breakdown(preprocessed_results)
         capital_present = self._capital_cost_present()
         partial_budget_requested = self.partial_budget.has_partial_budget_activity(preprocessed_results)
 

RUFAS/EEE/economics/partial_budget.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import Any, Dict
+from typing import Any, Dict, Iterator
 
 import numpy as np
 import pandas as pd
@@ -12,6 +12,16 @@
 from RUFAS.output_manager import OutputManager
 from RUFAS.units import MeasurementUnits
 
+LINE_ITEM_BREAKDOWN_UNITS: dict[str, MeasurementUnits] = {
+    "module": MeasurementUnits.UNITLESS,
+    "flow_type": MeasurementUnits.UNITLESS,
+    "item": MeasurementUnits.UNITLESS,
+    "scenario": MeasurementUnits.UNITLESS,
+    "biophysical_aggregate": MeasurementUnits.UNITLESS,
+    "price_aggregate": MeasurementUnits.UNITLESS,
+    "line_item_value": MeasurementUnits.DOLLARS,
+}
+
 
 class PartialBudget:
     """Container for partial budget inputs and analysis."""
@@ -47,34 +57,66 @@ def _load_inputs(self) -> Dict[str, np.ndarray]:
             "reduced_revenue": zero.copy(),
         }
 
-    # Supporting multi-year scenarios will require accumulating results across
-    # scenarios as outlined in `Documentation of Economic Data and Analytical
-    # Methods (2).pdf`.
+    @staticmethod
+    def _iter_line_items(
+        preprocessed_data: Dict[str, Dict[str, Dict[str, Any]]] | None,
+    ) -> Iterator[tuple[str, str, str, Dict[str, Any], Dict[str, Any]]]:
+        """Yield ``(section, item_name, flow_type, item, line_items)`` for each priced line item.
 
-    def _calculate_from_preprocessed(
-        self, preprocessed_data: Dict[str, Dict[str, Dict[str, Any]]] | None
-    ) -> Dict[str, Any] | None:
-        """Compute partial budget inputs from preprocessed scenario data."""
+        Items without a ``line_item_values_by_scenario`` dictionary are skipped and a missing
+        ``flow_type`` defaults to ``"cost"``, matching the partial budget aggregation rules.
+        """
 
         if not preprocessed_data:
-            return None
-
-        scenario_names: set[str] = set()
-        items: list[tuple[str | None, Dict[str, Any], Dict[str, float]]] = []
-        for section_data in preprocessed_data.values():
+            return
+        for section, section_data in preprocessed_data.items():
             if not isinstance(section_data, dict):
                 continue
             for category_data in section_data.values():
                 if not isinstance(category_data, dict):
                     continue
-                for _, item in category_data.items():
+                for item_name, item in category_data.items():
                     if not isinstance(item, dict):
                         continue
                     line_items = item.get("line_item_values_by_scenario")
                     if not isinstance(line_items, dict):
                         continue
-                    scenario_names.update(line_items.keys())
-                    items.append((item.get("flow_type"), item, line_items))
+                    yield section, item_name, item.get("flow_type") or "cost", item, line_items
+
+    @staticmethod
+    def _to_finite_float(value: Any) -> float:
+        """Coerce a line item value to ``float``, treating unparsable or non-finite values as zero."""
+
+        try:
+            result = float(value)
+        except (TypeError, ValueError):
+            return 0.0
+        return result if math.isfinite(result) else 0.0
+
+    @staticmethod
+    def _to_optional_float(value: Any) -> float | None:
+        """Coerce an aggregate to ``float`` when possible, otherwise return ``None``."""
+
+        if value is None:
+            return None
+        try:
+            return float(value)
+        except (TypeError, ValueError):
+            return None
+
+    def _calculate_from_preprocessed(
+        self, preprocessed_data: Dict[str, Dict[str, Dict[str, Any]]] | None
+    ) -> Dict[str, Any] | None:
+        """Compute partial budget inputs from preprocessed scenario data."""
+
+        if not preprocessed_data:
+            return None
+
+        scenario_names: set[str] = set()
+        items: list[tuple[str, Dict[str, Any], Dict[str, float]]] = []
+        for _section, _item_name, flow_type, item, line_items in self._iter_line_items(preprocessed_data):
+            scenario_names.update(line_items.keys())
+            items.append((flow_type, item, line_items))
 
         if len(scenario_names) == 1:
             scenario = next(iter(scenario_names))
@@ -83,19 +125,10 @@ def _calculate_from_preprocessed(
             cost_total = 0.0
 
             for flow_type, item, line_items in items:
-                flow_type = flow_type or "cost"
                 if flow_type not in {"revenue", "cost"}:
                     continue
 
-                raw = line_items.get(scenario, 0.0)
-
-                try:
-                    value = float(raw)
-                except (TypeError, ValueError):
-                    value = 0.0
-
-                if not math.isfinite(value):
-                    value = 0.0
+                value = self._to_finite_float(line_items.get(scenario, 0.0))
 
                 if flow_type == "revenue":
                     revenue_total += value
@@ -137,7 +170,6 @@ def _pick_scenario(candidates: list[str]) -> str | None:
         reduced_costs = 0.0
 
         for flow_type, item, line_items in items:
-            flow_type = flow_type or "cost"
             if flow_type not in {"revenue", "cost"}:
                 continue
             if baseline not in line_items or alternative not in line_items:
@@ -174,6 +206,59 @@ def _pick_scenario(candidates: list[str]) -> str | None:
             "reduced_costs": reduced_costs,
         }
 
+    def build_line_item_breakdown(
+        self, preprocessed_data: Dict[str, Dict[str, Dict[str, Any]]] | None
+    ) -> list[Dict[str, Any]]:
+        """Flatten the preprocessed economics data into one row per line item and scenario.
+
+        Each row reports the biophysical module (the ``ECONOMIC_MAP`` section), whether the item is
+        a cost or a revenue, the item name, the scenario, the aggregated biophysical quantity, the
+        aggregated price, and the resulting line item value that feeds the partial budget totals.
+        Rows follow the same rules as the totals, so summing ``line_item_value`` by ``flow_type``
+        and ``scenario`` reproduces ``econ_pba_cost_total`` and ``econ_pba_revenue_total``.
+        """
+
+        rows: list[Dict[str, Any]] = []
+        for section, item_name, flow_type, item, line_items in self._iter_line_items(preprocessed_data):
+            if flow_type not in {"revenue", "cost"}:
+                continue
+            aggregates_by_scenario = item.get("biophysical_aggregate_by_scenario")
+            if not isinstance(aggregates_by_scenario, dict):
+                aggregates_by_scenario = {}
+            for scenario, raw_value in line_items.items():
+                biophysical_aggregate = aggregates_by_scenario.get(scenario, item.get("biophysical_aggregate"))
+                rows.append(
+                    {
+                        "module": section,
+                        "flow_type": flow_type,
+                        "item": item_name,
+                        "scenario": scenario,
+                        "biophysical_aggregate": self._to_optional_float(biophysical_aggregate),
+                        "price_aggregate": self._to_optional_float(item.get("price_aggregate")),
+                        "line_item_value": self._to_finite_float(raw_value),
+                    }
+                )
+        return rows
+
+    def export_line_item_breakdown(
+        self, preprocessed_data: Dict[str, Dict[str, Dict[str, Any]]] | None
+    ) -> list[Dict[str, Any]]:
+        """Record the line item breakdown in the OutputManager as ``econ_pba_breakdown``.
+
+        One entry is added per row so the variable renders as a table: a JSON output lists the row
+        dictionaries and a CSV output gets one column per row field.
+        """
+
+        info_map = {
+            "class": __name__,
+            "function": self.export_line_item_breakdown.__name__,
+            "units": LINE_ITEM_BREAKDOWN_UNITS,
+        }
+        rows = self.build_line_item_breakdown(preprocessed_data)
+        for row in rows:
+            self.om.add_variable("econ_pba_breakdown", row, info_map)
+        return rows
+
     def calculate_partial_budget(self, preprocessed_data: Dict[str, Dict[str, Dict[str, Any]]] | None = None) -> None:
         """Perform a partial budget analysis and export multi-year net changes."""
 

changelog.md

@@ -24,6 +24,7 @@ A **changelog** is a structured record of changes made to the codebase over time
 v1.0.0
 
 ### Next Version Updates
+- [3252](https://github.com/RuminantFarmSystems/RuFaS/pull/3252) - [minor change] [Economics] [NoInputChange] [OutputChange] Adds the `econ_pba_breakdown` partial budget line item output, one row per line item and scenario with the biophysical module, cost/revenue flow type, item name, biophysical aggregate, price aggregate, and line item value, so economics preprocessing can be audited in a single CSV or JSON output.
 - [3219](https://github.com/RuminantFarmSystems/RuFaS/pull/3219) - [minor change] [Economics] [NoInputChange] [OutputChange] Port the issue #3088 bedding cost preprocessing onto the special-case handler architecture as BeddingRequirementsHandler, keeping per-pen pairing, lactating-cow-only billing, and identical output.
 - [2793](https://github.com/RuminantFarmSystems/RuFaS/pull/2793) - [minor change] [Animal] [OutputManager] [NoInputChange] [OutputChange] Track and summarize when effective DMI falls below the empirical domain of manure equations for lactating and dry cows at end of simulation.
 - [2865](https://github.com/RuminantFarmSystems/RuFaS/pull/2865) - [minor change] [NoInputChange] [NoOutputChange] Update to tables and formatting in scientific documentation.

tests/test_EEE/test_economic_breakdown_output.py

@@ -18,12 +18,19 @@ def add_error(self, *args, **kwargs):
 
 
 class DummyPartialBudget:
+    def __init__(self) -> None:
+        self.exported = []
+
     def has_partial_budget_activity(self, _preprocessed):
         return False
 
     def calculate_partial_budget(self, _preprocessed):
         return None
 
+    def export_line_item_breakdown(self, preprocessed):
+        self.exported.append(preprocessed)
+        return []
+
 
 def test_framework_exports_line_item_breakdown(monkeypatch):
     preprocessed = {
@@ -48,6 +55,7 @@ def test_framework_exports_line_item_breakdown(monkeypatch):
     }
 
     dummy_om = DummyOutputManager()
+    dummy_pb = DummyPartialBudget()
 
     monkeypatch.setattr(
         framework, "InputManager", lambda: type("IM", (), {"get_data": lambda *_: pd.DataFrame({"Cost": []})})()
@@ -56,7 +64,7 @@ def test_framework_exports_line_item_breakdown(monkeypatch):
     monkeypatch.setattr(
         framework, "EconomicPreprocessor", lambda: type("Pre", (), {"preprocess": lambda *_: preprocessed})()
     )
-    monkeypatch.setattr(framework, "PartialBudget", lambda: DummyPartialBudget())
+    monkeypatch.setattr(framework, "PartialBudget", lambda: dummy_pb)
 
     ef = framework.EconomicFramework()
     ef.run_economic_analysis()
@@ -72,3 +80,5 @@ def test_framework_exports_line_item_breakdown(monkeypatch):
 
     milk = breakdown["Animal"]["revenues"]["milk"]
     assert milk["total"] == 32.0
+
+    assert dummy_pb.exported == [preprocessed]