diff --git a/RUFAS/EEE/economics/framework.py b/RUFAS/EEE/economics/framework.py index c76bce8264..eeaa07a5cf 100644 --- a/RUFAS/EEE/economics/framework.py +++ b/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) diff --git a/RUFAS/EEE/economics/partial_budget.py b/RUFAS/EEE/economics/partial_budget.py index b62bba8b2e..18d3062c74 100644 --- a/RUFAS/EEE/economics/partial_budget.py +++ b/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.""" diff --git a/changelog.md b/changelog.md index fcfb45a25b..bb8caceb8d 100644 --- a/changelog.md +++ b/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. diff --git a/tests/test_EEE/test_economic_breakdown_output.py b/tests/test_EEE/test_economic_breakdown_output.py index e73f1b5c75..375b295ef4 100644 --- a/tests/test_EEE/test_economic_breakdown_output.py +++ b/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] diff --git a/tests/test_EEE/test_partial_budget_outputs.py b/tests/test_EEE/test_partial_budget_outputs.py index 7a252d5083..9705c411fb 100644 --- a/tests/test_EEE/test_partial_budget_outputs.py +++ b/tests/test_EEE/test_partial_budget_outputs.py @@ -1,6 +1,8 @@ import pytest from RUFAS.EEE.economics import partial_budget +from RUFAS.output_manager import OutputManager +from RUFAS.units import MeasurementUnits class DummyOutputManager: @@ -103,3 +105,183 @@ def test_partial_budget_exports_net_annual_cash_flow_for_single_scenario( assert exported["econ_pba_additional_costs"] == [0.0] assert exported["econ_pba_reduced_revenue"] == [0.0] assert "econ_pba_summary" in exported + + +BREAKDOWN_PREPROCESSED = { + "Animal": { + "Revenue": { + "Milk": { + "flow_type": "revenue", + "biophysical_aggregate": 30.0, + "biophysical_aggregate_by_scenario": {"baseline": 10.0, "alternative": 20.0}, + "price_aggregate": 2.0, + "line_item_values_by_scenario": {"baseline": 20.0, "alternative": 40.0}, + } + }, + "Costs": { + "Feed": { + "flow_type": "cost", + "biophysical_aggregate": 5.0, + "price_aggregate": None, + "line_item_values_by_scenario": {"baseline": 5.0, "alternative": float("nan")}, + }, + "Unpriced": {"flow_type": "cost", "biophysical_values": [1.0]}, + }, + }, + "Manure": { + "Costs": { + "Labor": { + "biophysical_aggregate": "3", + "price_aggregate": "1.5", + "line_item_values_by_scenario": {"baseline": "4.5", "alternative": 6.0}, + } + } + }, +} + +EXPECTED_BREAKDOWN_ROWS = [ + { + "module": "Animal", + "flow_type": "revenue", + "item": "Milk", + "scenario": "baseline", + "biophysical_aggregate": 10.0, + "price_aggregate": 2.0, + "line_item_value": 20.0, + }, + { + "module": "Animal", + "flow_type": "revenue", + "item": "Milk", + "scenario": "alternative", + "biophysical_aggregate": 20.0, + "price_aggregate": 2.0, + "line_item_value": 40.0, + }, + { + "module": "Animal", + "flow_type": "cost", + "item": "Feed", + "scenario": "baseline", + "biophysical_aggregate": 5.0, + "price_aggregate": None, + "line_item_value": 5.0, + }, + { + "module": "Animal", + "flow_type": "cost", + "item": "Feed", + "scenario": "alternative", + "biophysical_aggregate": 5.0, + "price_aggregate": None, + "line_item_value": 0.0, + }, + { + "module": "Manure", + "flow_type": "cost", + "item": "Labor", + "scenario": "baseline", + "biophysical_aggregate": 3.0, + "price_aggregate": 1.5, + "line_item_value": 4.5, + }, + { + "module": "Manure", + "flow_type": "cost", + "item": "Labor", + "scenario": "alternative", + "biophysical_aggregate": 3.0, + "price_aggregate": 1.5, + "line_item_value": 6.0, + }, +] + + +def _make_partial_budget(monkeypatch: pytest.MonkeyPatch) -> tuple[partial_budget.PartialBudget, DummyOutputManager]: + dummy_om = DummyOutputManager() + monkeypatch.setattr(partial_budget, "InputManager", lambda: object()) + monkeypatch.setattr(partial_budget, "OutputManager", lambda: dummy_om) + return partial_budget.PartialBudget(), dummy_om + + +def test_line_item_breakdown_lists_one_row_per_item_and_scenario(monkeypatch: pytest.MonkeyPatch) -> None: + pb, _ = _make_partial_budget(monkeypatch) + + rows = pb.build_line_item_breakdown(BREAKDOWN_PREPROCESSED) + + assert rows == EXPECTED_BREAKDOWN_ROWS + + +def test_line_item_breakdown_is_empty_without_preprocessed_data(monkeypatch: pytest.MonkeyPatch) -> None: + pb, _ = _make_partial_budget(monkeypatch) + + assert pb.build_line_item_breakdown(None) == [] + assert pb.build_line_item_breakdown({}) == [] + + +def test_line_item_breakdown_reconciles_with_single_scenario_totals(monkeypatch: pytest.MonkeyPatch) -> None: + preprocessed = { + "Animal": { + "Revenue": { + "Milk": {"flow_type": "revenue", "line_item_values_by_scenario": {"baseline": 120.0}}, + "Calves": {"flow_type": "revenue", "line_item_values_by_scenario": {"baseline": "30"}}, + }, + "Costs": { + "Feed": {"flow_type": "cost", "line_item_values_by_scenario": {"baseline": 80.0}}, + "Bedding": {"flow_type": "cost", "line_item_values_by_scenario": {"baseline": float("nan")}}, + }, + } + } + pb, dummy_om = _make_partial_budget(monkeypatch) + + pb.calculate_partial_budget(preprocessed) + rows = pb.build_line_item_breakdown(preprocessed) + + exported = {name: value for name, value, _ in dummy_om.variables} + revenue_rows = sum(row["line_item_value"] for row in rows if row["flow_type"] == "revenue") + cost_rows = sum(row["line_item_value"] for row in rows if row["flow_type"] == "cost") + assert revenue_rows == pytest.approx(exported["econ_pba_revenue_total"][0]) == pytest.approx(150.0) + assert cost_rows == pytest.approx(exported["econ_pba_cost_total"][0]) == pytest.approx(80.0) + assert [row["item"] for row in rows] == ["Milk", "Calves", "Feed", "Bedding"] + + +def test_export_line_item_breakdown_logs_one_entry_per_row(monkeypatch: pytest.MonkeyPatch) -> None: + pb, dummy_om = _make_partial_budget(monkeypatch) + + rows = pb.export_line_item_breakdown(BREAKDOWN_PREPROCESSED) + + logged = [(value, info) for name, value, info in dummy_om.variables if name == "econ_pba_breakdown"] + assert [value for value, _ in logged] == rows == EXPECTED_BREAKDOWN_ROWS + units = logged[0][1]["units"] + assert set(units) == set(EXPECTED_BREAKDOWN_ROWS[0]) + assert units["line_item_value"] is MeasurementUnits.DOLLARS + assert all(info["function"] == "export_line_item_breakdown" for _, info in logged) + + +def test_export_line_item_breakdown_renders_as_csv_columns(monkeypatch: pytest.MonkeyPatch) -> None: + om = OutputManager() + om.flush_pools() + monkeypatch.setattr(partial_budget, "InputManager", lambda: object()) + pb = partial_budget.PartialBudget() + + rows = pb.export_line_item_breakdown(BREAKDOWN_PREPROCESSED) + + filtered = om.filter_variables_pool({"filters": ["econ_pba_breakdown$"]}) + ((key, data),) = filtered.items() + assert key.endswith("partial_budget.export_line_item_breakdown.econ_pba_breakdown") + assert data["values"] == rows + + columns = om._dict_to_csv_column_list(key, data) + assert [column.name for column in columns] == [ + f"{key}.module (unitless)", + f"{key}.flow_type (unitless)", + f"{key}.item (unitless)", + f"{key}.scenario (unitless)", + f"{key}.biophysical_aggregate (unitless)", + f"{key}.price_aggregate (unitless)", + f"{key}.line_item_value ($)", + ] + assert columns[2].tolist() == ["Milk", "Milk", "Feed", "Feed", "Labor", "Labor"] + assert columns[-1].tolist() == [row["line_item_value"] for row in rows] + + om.flush_pools()