From fb76c14eccd3033ccf8c87a42d97b0045c08838d Mon Sep 17 00:00:00 2001 From: Niko <70217952+ew3361zh@users.noreply.github.com> Date: Tue, 23 Jun 2026 13:51:38 -0400 Subject: [PATCH] reconfigure digester cost calcs --- RUFAS/EEE/economics/dcfror.py | 38 ++----- RUFAS/EEE/economics/digester_costs.py | 139 ++++---------------------- RUFAS/EEE/economics/preprocessing.py | 2 + 3 files changed, 32 insertions(+), 147 deletions(-) diff --git a/RUFAS/EEE/economics/dcfror.py b/RUFAS/EEE/economics/dcfror.py index fc30d8ddc6..fe47b1c14c 100644 --- a/RUFAS/EEE/economics/dcfror.py +++ b/RUFAS/EEE/economics/dcfror.py @@ -184,36 +184,20 @@ def _to_numpy(values: Any) -> np.ndarray: digester_rows = [row for row in cap_breakdown if "digester" in str(row.get("Item", "")).lower()] if digester_rows: cow_count = float(self.im.get_data("animal_properties.herd_information.cow_num")) + digester_config = self.im.get_data("manure_management_properties.anaerobic_digester") - hydraulic_retention_time_days = float(digester_config[0]["hydraulic_retention_time"]) - digester_volume_proxy = max(1.0, cow_count * hydraulic_retention_time_days) + digester_system = DigesterCostCalculator.get_digester_system_name(digester_config) - annual_fixed_cost = DigesterCostCalculator.calculate_digester_capital_cost( - animal_units=cow_count, - digester_volume=digester_volume_proxy, - ) - discount_rate = float(input_dict["loan_interest_rate"]) - project_life_years = int(input_dict["project_term"]) - crf = DigesterCostCalculator.capital_recovery_factor(discount_rate, project_life_years) - digester_capex = DigesterCostCalculator.calculate_digester_capex(annual_fixed_cost, crf) - # Equation-5 scaling factor (six-tenths rule) for installed cost - # adjusted by effective digester volume proxy. - digester_capex = DigesterCostCalculator.scale_installed_cost( - base_cost=digester_capex, - volume=digester_volume_proxy, - base_volume=max(1.0, cow_count), - install_factor=0.6, - ) - annual_opex_total = DigesterCostCalculator.calculate_digester_operational_cost( - True, - animal_units=cow_count, - farm_type_flag=1.0, - below_ground_flag=1.0, - concrete_flag=1.0, - steel_flag=0.0, - ) + digester_costs = DigesterCostCalculator.estimate_digester_costs(digester_system, cow_count) + + digester_capex = float(digester_costs["capital_expenditure"]) + digester_operating_costs = digester_costs["operating_costs"] + annual_opex_total = sum(float(cost) for cost in digester_operating_costs.values()) + + if "rng" in digester_system.lower(): + annual_opex_total += DigesterCostCalculator.estimate_digester_trucking_cost(cow_count) - prior_digester_capex = float(sum(float(r.get("Cost", 0.0)) for r in digester_rows)) + prior_digester_capex = float(sum(float(row.get("Cost", 0.0)) for row in digester_rows)) base_cost = base_cost - prior_digester_capex + digester_capex operating_costs = operating_costs + annual_opex_total capital_cost = base_cost diff --git a/RUFAS/EEE/economics/digester_costs.py b/RUFAS/EEE/economics/digester_costs.py index 981cd71bec..4fac2f5a8b 100644 --- a/RUFAS/EEE/economics/digester_costs.py +++ b/RUFAS/EEE/economics/digester_costs.py @@ -4,7 +4,7 @@ import math from dataclasses import dataclass -from typing import Dict, Mapping +from typing import Any, Dict, Mapping @dataclass(frozen=True) @@ -50,131 +50,30 @@ def annual_operating_costs(self, cows: float) -> Dict[str, float]: class DigesterCostCalculator: """Collection of digester-related economic calculations.""" - # Equation 1 from the economics documentation - # A_FC_ij = EXP(alpha + R1*ln(A_j) + R2*ln(V_j) + psi1*F_j + psi2*G_j + psi3*C_j + psi4*S_j + epsilon_j) - - @staticmethod - def calculate_digester_capital_cost( - animal_units: float | None = None, - digester_volume: float | None = None, - farm_type_flag: float | None = None, - below_ground_flag: float | None = None, - concrete_flag: float | None = None, - steel_flag: float | None = None, - *, - alpha: float = 10.4, - r1: float = -0.413, - r2: float = -0.147, - psi1: float = -1.085, - psi2: float = -0.582, - psi3: float = 0.519, - psi4: float = -0.007, - epsilon: float = 0.35, - area: float | None = None, - volume: float | None = None, - f_j: float | None = None, - g_j: float | None = None, - c_j: float | None = None, - s_j: float | None = None, - ) -> float: - """Return the estimated digester capital cost using Equation 1.""" - - if animal_units is None: - animal_units = float(area) if area is not None else None - if digester_volume is None: - digester_volume = float(volume) if volume is not None else None - if animal_units is None or digester_volume is None: - raise ValueError("animal_units and digester_volume are required inputs") - if animal_units <= 0 or digester_volume <= 0: - raise ValueError("animal_units and digester_volume must be positive") - - if farm_type_flag is None: - farm_type_flag = float(f_j) if f_j is not None else 1.0 - if below_ground_flag is None: - below_ground_flag = float(g_j) if g_j is not None else 1.0 - if concrete_flag is None: - concrete_flag = float(c_j) if c_j is not None else 1.0 - if steel_flag is None: - steel_flag = float(s_j) if s_j is not None else 0.0 - - return math.exp( - alpha - + r1 * math.log(animal_units) - + r2 * math.log(digester_volume) - + psi1 * farm_type_flag - + psi2 * below_ground_flag - + psi3 * concrete_flag - + psi4 * steel_flag - + epsilon - ) - - # Equation 2 - Capital Recovery Factor - @staticmethod - def capital_recovery_factor(rate: float, periods: int) -> float: - """Compute the capital recovery factor (Equation 2).""" - if periods <= 0: - raise ValueError("periods must be positive") - if rate == 0: - return 1 / periods - numerator = rate * (1 + rate) ** periods - denominator = (1 + rate) ** periods - 1 - return numerator / denominator - - # Equation 5 - Generalized equipment scaling + def get_digester_system_name(digester_config: Any) -> str: + """Return the digester cost-profile name from the manure-management config.""" - @staticmethod - def scale_installed_cost(base_cost: float, volume: float, base_volume: float, install_factor: float) -> float: - """Scale equipment cost to a new volume using Equation 5.""" - if base_volume <= 0: - raise ValueError("base_volume must be positive") - ratio = volume / base_volume - return base_cost * (ratio**install_factor) + config = digester_config[0] if isinstance(digester_config, list) else digester_config - # Equation 3 - Digester CAPEX derived from annual fixed cost and CRF + digester_type = str(config["digester_type"]).lower() + biogas_use = str(config["biogas_use"]).lower() - @staticmethod - def calculate_digester_capex(afc_ij: float, crf: float) -> float: - """Return the digester capital expenditure using Equation 3.""" - if crf == 0: - raise ValueError("capital recovery factor cannot be zero") - return afc_ij / crf + if "lagoon" in digester_type: + system_type = "Covered Lagoon" + elif "plug" in digester_type: + system_type = "Plug Flow" + else: + raise ValueError(f"Unsupported digester type for costing: {digester_type}") - # Equation 4 - Digester variable operational cost + if "rng" in biogas_use: + use_case = "RNG" + elif "chp" in biogas_use: + use_case = "CHP" + else: + raise ValueError(f"Unsupported digester biogas use case for costing: {biogas_use}") - @staticmethod - def calculate_digester_operational_cost( - use_cost_function: bool, - animal_units: float, - farm_type_flag: float, - below_ground_flag: float, - concrete_flag: float, - steel_flag: float, - *, - beta: float = 3.531, - omega1: float = -0.321, - phi1: float = -0.686, - phi2: float = -0.418, - phi3: float = 0.488, - phi4: float = -0.113, - epsilon: float = 0.237, - ) -> float: - """Return the variable operational cost using Equation 4.""" - - if not use_cost_function: - return 0.0 - if animal_units <= 0: - raise ValueError("animal_units must be positive to evaluate the AVC function") - - return math.exp( - beta - + omega1 * math.log(animal_units) - + phi1 * farm_type_flag - + phi2 * below_ground_flag - + phi3 * concrete_flag - + phi4 * steel_flag - + epsilon - ) + return f"{system_type} - {use_case}" @staticmethod def _normalize_system_name(system: str) -> str: diff --git a/RUFAS/EEE/economics/preprocessing.py b/RUFAS/EEE/economics/preprocessing.py index 5b718af977..632efab047 100644 --- a/RUFAS/EEE/economics/preprocessing.py +++ b/RUFAS/EEE/economics/preprocessing.py @@ -15,6 +15,7 @@ from __future__ import annotations import math +from pathlib import Path import re from dataclasses import dataclass from typing import Any, Dict, Iterable, List, Set @@ -765,6 +766,7 @@ def preprocess(self) -> Dict[str, Dict[str, Dict[str, Dict[str, Any]]]]: data=results, properties_blob_key="economic_preprocessing_properties", eager_termination=False, + input_path=Path("") ) self.om.add_log( "Economic preprocessing",