[Econ] Update ROI calculations

RUFAS/EEE/economics/dcfror.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Dict, Iterable, Tuple
+from typing import Any, Iterable
 
 import numpy as np
 import pandas as pd
@@ -22,7 +22,7 @@ def __init__(self) -> None:
         self.om = OutputManager()
         self.inputs = self._load_inputs()
 
-    def _load_inputs(self) -> Dict[str, Any]:
+    def _load_inputs(self) -> dict[str, Any]:
         info_map = {"class": self.__class__.__name__, "function": self._load_inputs.__name__}
         try:
 
@@ -50,7 +50,7 @@ def _get_input(path: str) -> Any:
 
             cost_capital_multiple = _get_input("economic_inputs.capital_costs.capital_cost_breakdown")
 
-            inputs: Dict[str, Any] = {
+            inputs: dict[str, Any] = {
                 "cost_capital_multiple": cost_capital_multiple,
                 "cost_operational_units": _get_input("economic_inputs.cashflow_inputs.operating_units"),
                 "cost_operational_unit_cost": _get_input("economic_inputs.cashflow_inputs.operating_unit_costs"),
@@ -85,7 +85,7 @@ def _get_input(path: str) -> Any:
             self.om.add_error("MissingInputKey", f"Missing input key: {str(e)}", info_map)
             raise
 
-    def calculate(self, override_inputs: Dict[str, Any] | None = None) -> None:
+    def calculate(self, override_inputs: dict[str, Any] | None = None) -> None:
         """Run the DCFROR cash-flow model and export summary outputs.
 
         Parameters
@@ -131,7 +131,7 @@ def calculate(self, override_inputs: Dict[str, Any] | None = None) -> None:
             self.om.add_error("DCFRORCalculationFailed", f"DCFROR calculation failed: {exc}", info_map)
             raise
 
-    def _prepare_costs(self, input_dict: Dict[str, Any]) -> Dict[str, Any]:
+    def _prepare_costs(self, input_dict: dict[str, Any]) -> dict[str, Any]:
         """Build capital, operating-cost, and revenue schedules from inputs."""
 
         info_map = {"class": self.__class__.__name__, "function": self._prepare_costs.__name__}
@@ -267,7 +267,7 @@ def _get_digester_config(self) -> list:
         except (KeyError, TypeError):
             return []
 
-    def _prepare_financing(self, input_dict: Dict[str, Any], project_term: int) -> Dict[str, Any]:
+    def _prepare_financing(self, input_dict: dict[str, Any], project_term: int) -> dict[str, Any]:
         """Normalise financing inputs and enforce valid borrowing shares."""
         depreciation_rate = input_dict["depreciation_rate"]
         depreciation_rate = depreciation_rate[~np.isnan(depreciation_rate)]
@@ -320,7 +320,7 @@ def _compute_cash_flows(
         operating_costs: np.ndarray,
         tax_rate: float,
         internal_rate_of_return: float,
-    ) -> Tuple[np.ndarray, pd.DataFrame]:
+    ) -> tuple[np.ndarray, pd.DataFrame]:
         """Build the annual cash flow table using Equations 7–26."""
 
         years = EconomicEquations.construct_timeline(construction_term, loan_term, project_term)
@@ -497,15 +497,11 @@ def _export_results(
             cash_flow_df.loc[construction_mask, "NPVCapitalPlusInterest"].to_numpy(),
         )
 
-        positive_benefits = float(CF[CF > 0].sum())
-        investment_costs = float(-CF[CF < 0].sum())
-        roi = EconomicMetrics.calculate_roi(positive_benefits, investment_costs)
         payback = EconomicMetrics.calculate_payback_period(CF)
         net_cash_flow = EconomicMetrics.calculate_net_annual_cash_flow(revenue, operating_costs)
         mpsp = EconomicMetrics.calculate_mpsp(capital_cost + operating_costs.sum(), revenue.sum())
 
         self.om.add_variable("econ_dcfror_npv", npv, {**info_map, "units": MeasurementUnits.DOLLARS})
-        self.om.add_variable("econ_dcfror_roi", roi, {**info_map, "units": MeasurementUnits.UNITLESS})
         self.om.add_variable(
             "econ_dcfror_payback_period",
             payback,
@@ -532,7 +528,7 @@ def goal_seek(
         self,
         variable_name: str,
         target_npv: float = 0.0,
-        bounds: Tuple[float, float] = (0.01, 100.0),
+        bounds: tuple[float, float] = (0.01, 100.0),
         tol: float = 1e-6,
         max_iter: int = 100,
     ) -> float:

RUFAS/EEE/economics/partial_budget.py

@@ -2,12 +2,16 @@
 
 from __future__ import annotations
 
+import ast
+from pathlib import Path
+import re
 from typing import Any, Dict
 
 import numpy as np
 import pandas as pd
 import math
 
+from RUFAS.EEE.economics.metrics import EconomicMetrics
 from RUFAS.input_manager import InputManager
 from RUFAS.output_manager import OutputManager
 from RUFAS.units import MeasurementUnits
@@ -244,6 +248,17 @@ def calculate_partial_budget(self, preprocessed_data: Dict[str, Dict[str, Dict[s
             self.om.add_variable("econ_pba_net_annual_cash_flow", net_annual_cash_flow.tolist(), info_map)
             self.om.add_variable("econ_pba_summary", result_df.to_dict(orient="list"), info_map)
             self.om.add_log("PartialBudget", "Partial budget analysis completed.", info_map)
+
+            should_run_roi_comparison: bool = self.im.get_data("economic_inputs.roi.compare_roi")
+            if should_run_roi_comparison:
+                revenue = float(revenue_total.item())
+                costs = float(cost_total.item())
+                current_simulation_roi = EconomicMetrics.calculate_roi(benefits=revenue, costs=costs)
+                comparison_roi_path_data: list[dict[str, str]] = \
+                    self.im.get_data("economic_inputs.roi.roi_comparison_paths")
+                for comparison in comparison_roi_path_data:
+                    self._run_roi_comparison(comparison, current_simulation_roi)
+
             return
 
         else:
@@ -280,6 +295,139 @@ def calculate_partial_budget(self, preprocessed_data: Dict[str, Dict[str, Dict[s
         self.om.add_variable("econ_pba_summary", result_df.to_dict(orient="list"), info_map)
         self.om.add_log("PartialBudget", "Partial budget analysis completed.", info_map)
 
+    def _run_roi_comparison(
+        self,
+        comparison_roi_data: dict[str, str],
+        current_simulation_roi: float,
+    ) -> None:
+        """
+        Compare a previous simulation ROI with the current simulation ROI.
+
+        Parameters
+        ----------
+        comparison_roi_data : dict[str, str]
+            A dictionary containing the user specified locations and names for comparison roi data.
+        current_simulation_roi : float
+            The roi calculated for the current simulation.
+
+        Raises
+        ------
+        ValueError
+            If the comparison data has a different number of revenues and costs.
+
+        """
+        info_map = {
+            "class": self.__class__.__name__,
+            "function": self._run_roi_comparison.__name__,
+            "units": MeasurementUnits.DOLLARS
+        }
+
+        comparison_path = Path(comparison_roi_data["address"])
+        comparison_pool = self.im.load_data_from_csv(comparison_path)
+
+        comparison_revenues = self._extract_numeric_values(
+            comparison_pool,
+            r"\.RevenueTotal$",
+        )
+        comparison_costs = self._extract_numeric_values(
+            comparison_pool,
+            r"\.CostTotal$",
+        )
+
+        if len(comparison_revenues) != len(comparison_costs):
+            error_message = "Comparison revenue and cost data must contain the same number of values."
+            self.om.add_error(
+                "ROI comparison error",
+                error_message,
+                info_map
+            )
+            raise ValueError(error_message)
+
+        comparison_roi_name = comparison_roi_data["name"]
+
+        for index, (comparison_revenue, comparison_cost) in enumerate(
+            zip(comparison_revenues, comparison_costs, strict=True)
+        ):
+            comparison_roi = EconomicMetrics.calculate_roi(
+                benefits=comparison_revenue,
+                costs=comparison_cost,
+            )
+            roi_delta = current_simulation_roi - comparison_roi
+
+            output_name = f"roi_delta_for_{comparison_roi_name}"
+            if len(comparison_revenues) > 1:
+                output_name = f"{output_name}_{index}"
+
+            self.om.add_variable(
+                output_name,
+                roi_delta,
+                info_map,
+            )
+
+    def _extract_numeric_values(
+        self,
+        data: dict[str, Any],
+        column_pattern: str,
+    ) -> list[float]:
+        """
+        Helper function for _run_roi_comparison().
+        Extract numeric values from the single column matching a pattern.
+
+        Parameters
+        ----------
+        data : dict[str, Any]
+            The data structure from which the numeric values are extracted.
+        column_pattern : str
+            The regex pattern used to search the column name for the desired variable.
+
+        Returns
+        -------
+        list[float]
+            A list of floats extracted from the pattern-matched column of data in the data structure.
+
+        Raises
+        ------
+        ValueError
+            If there are multiple columns of matched data where we're only expecting one.
+
+        """
+        matching_columns = [
+            column_name
+            for column_name in data
+            if re.search(column_pattern, column_name)
+        ]
+
+        if len(matching_columns) != 1:
+            error_message = (
+                f"In prepping ROI data, expected exactly one column matching {column_pattern!r}, "
+                f"but found {matching_columns}."
+            )
+            self.om.add_error(
+                "ROI comparison error",
+                error_message,
+                {
+                    "class": self.__class__.__name__,
+                    "function": self._extract_numeric_values.__name__
+                }
+            )
+            raise ValueError(error_message)
+
+        column_values = data[matching_columns[0]]
+        extracted_values: list[float] = []
+
+        for value in column_values:
+            if isinstance(value, str):
+                parsed_value = ast.literal_eval(value)
+            else:
+                parsed_value = value
+
+            if isinstance(parsed_value, (list, tuple)):
+                extracted_values.extend(float(item) for item in parsed_value)
+            else:
+                extracted_values.append(float(parsed_value))
+
+        return extracted_values
+
     def has_partial_budget_activity(
         self, preprocessed_data: Dict[str, Dict[str, Dict[str, Any]]] | None = None
     ) -> bool:

RUFAS/input_manager.py

@@ -545,7 +545,7 @@ def _runtime_data_loader_map(self) -> dict[str, Callable[[Path], dict[str, Any]]
         """Helper function for runtime data mapping."""
         return {
             "json": self._load_data_from_json,
-            "csv": self._load_data_from_csv,
+            "csv": self.load_data_from_csv,
         }
 
     def _process_runtime_file(
@@ -889,7 +889,7 @@ def _load_data_from_json(self, file_path: Path) -> dict[str, Any]:
             self.om.add_error(f"Unexpected error when loading file at path {file_path}: {e}", str(e), info_map)
             raise
 
-    def _load_data_from_csv(self, file_path: Path) -> dict[str, Any]:
+    def load_data_from_csv(self, file_path: Path) -> dict[str, Any]:
         """
         Loads data from input csv file.
 
@@ -917,7 +917,7 @@ def _load_data_from_csv(self, file_path: Path) -> dict[str, Any]:
         """
         info_map = {
             "class": self.__class__.__name__,
-            "function": self._load_data_from_csv.__name__,
+            "function": self.load_data_from_csv.__name__,
         }
         self.om.add_log("open_csv_file", f"Attempting to open {file_path}.", info_map)
         try:
@@ -972,7 +972,7 @@ def _populate_pool(self, input_root: Path, eager_termination: bool) -> bool:
         self.input_root = input_root
         data_type_to_loader_map: dict[str, Callable[[Path], dict[str, Any]]] = {
             "json": self._load_data_from_json,
-            "csv": self._load_data_from_csv,
+            "csv": self.load_data_from_csv,
         }
         valid_data = True
         for file_blob_key, file_details in self.__metadata["files"].items():

input/data/EEE/economic_inputs.json

@@ -189,5 +189,14 @@
         "Cost": 15000.0
       }
     ]
+  },
+  "roi": {
+    "compare_roi": false,
+    "roi_comparison_paths": [
+       {
+        "name": "base_freestall_costs_and_revenue",
+        "address": "output/reports/freestall_report_report_econ.json_04-Aug-2026_Tue_09-01-28.csv"
+       }
+    ]
   }
 }

input/metadata/properties/default.json

@@ -3535,6 +3535,30 @@
           }
         }
       }
+    },
+    "roi": {
+      "type": "object",
+      "description": "Inputs related to return on investment calculations.", 
+      "compare_roi": {
+        "type": "bool",
+        "description": "If true, triggers loading previous simulation roi results for comparison to current simulation roi.",
+        "default": false
+      },
+      "roi_comparison_paths": {
+        "type": "array",
+        "description": "The list of saved roi output addresses the user wants to compare to the roi for the current simulation.",
+        "properties": {
+          "type": "object",
+          "name": {
+            "type": "string",
+            "description": "The user-specified name for the roi being used for comparison to the current simulation roi."
+          },
+          "address": {
+            "type": "string",
+            "description": "The user-specified directory location of the saved roi comparison result."
+          }
+        }
+      }
     }
   },
   "emissions_properties": {

tests/test_EEE/test_partial_budget_outputs.py

@@ -62,44 +62,3 @@ def test_partial_budget_exports_all_series(monkeypatch: pytest.MonkeyPatch) -> N
     assert exported["econ_pba_net_change"] == pytest.approx([7.0])
     assert exported["econ_pba_cumulative_net_change"] == pytest.approx([7.0])
     assert "econ_pba_summary" in exported
-
-
-def test_partial_budget_exports_net_annual_cash_flow_for_single_scenario(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    preprocessed = {
-        "Section": {
-            "Revenue": {
-                "Milk": {
-                    "flow_type": "revenue",
-                    "line_item_values_by_scenario": {"baseline": 120.0},
-                }
-            },
-            "Costs": {
-                "Feed": {
-                    "flow_type": "cost",
-                    "line_item_values_by_scenario": {"baseline": 80.0},
-                }
-            },
-        }
-    }
-
-    dummy_im = object()
-    dummy_om = DummyOutputManager()
-
-    monkeypatch.setattr(partial_budget, "InputManager", lambda: dummy_im)
-    monkeypatch.setattr(partial_budget, "OutputManager", lambda: dummy_om)
-
-    pb = partial_budget.PartialBudget()
-    pb.calculate_partial_budget(preprocessed)
-
-    exported = {name: value for name, value, _ in dummy_om.variables}
-
-    assert exported["econ_pba_net_annual_cash_flow"] == [40.0]
-    assert exported["econ_pba_revenue_total"] == [120.0]
-    assert exported["econ_pba_cost_total"] == [80.0]
-    assert exported["econ_pba_additional_revenue"] == [0.0]
-    assert exported["econ_pba_reduced_costs"] == [0.0]
-    assert exported["econ_pba_additional_costs"] == [0.0]
-    assert exported["econ_pba_reduced_revenue"] == [0.0]
-    assert "econ_pba_summary" in exported