[Econ] Anaerobic Digester CHP and RNG

RUFAS/EEE/economics/digester_costs.py

@@ -47,6 +47,41 @@ def annual_operating_costs(self, cows: float) -> Dict[str, float]:
         }
 
 
+@dataclass(frozen=True)
+class BiogasEnergyConversion:
+    """Conversion factors for turning captured biogas into usable energy.
+
+    Parameters
+    ----------
+    methane_content_fraction : float
+        Fraction of the total biogas volume that is methane (unitless).
+    methane_density_kg_per_m3 : float
+        Density used to convert methane volume to mass (kg CH4 / m^3 CH4).
+    methane_energy_content_kwh_per_kg : float
+        Energy content of methane fed into a combined heat and power unit (kWh / kg CH4).
+    chp_electrical_efficiency : float
+        Fraction of the methane energy input converted to electricity by the combined heat and power unit (unitless).
+    chp_thermal_efficiency : float
+        Fraction of the methane energy input recovered as heat by the combined heat and power unit (unitless).
+    rng_leakage_loss_fraction : float
+        Fraction of methane lost to leakage during renewable natural gas upgrading (unitless).
+    rng_process_heat_loss_fraction : float
+        Fraction of methane combusted to supply process heat for renewable natural gas upgrading (unitless).
+    rng_energy_yield_kwh_per_kg : float
+        Energy content of the upgraded renewable natural gas (kWh / kg RNG).
+
+    """
+
+    methane_content_fraction: float = 0.60
+    methane_density_kg_per_m3: float = 0.68
+    methane_energy_content_kwh_per_kg: float = 13.89
+    chp_electrical_efficiency: float = 0.33
+    chp_thermal_efficiency: float = 0.43
+    rng_leakage_loss_fraction: float = 0.02
+    rng_process_heat_loss_fraction: float = 0.03
+    rng_energy_yield_kwh_per_kg: float = 13.10
+
+
 class DigesterCostCalculator:
     """Collection of digester-related economic calculations."""
 
@@ -273,3 +308,78 @@ def estimate_digester_trucking_cost(cls, cows: float) -> float:
         """Estimate the RNG trucking and injection cost for the herd size."""
 
         return cls._TRANSPORT_COST.evaluate(cows)
+
+    @staticmethod
+    def estimate_biogas_electricity(
+        biogas_volume_m3: float,
+        conversion: BiogasEnergyConversion = BiogasEnergyConversion(),
+    ) -> Dict[str, float]:
+        """Estimate electricity and heat produced from biogas by a combined heat and power unit.
+
+        Parameters
+        ----------
+        biogas_volume_m3 : float
+            Volume of captured biogas routed to the combined heat and power unit (m^3).
+        conversion : BiogasEnergyConversion
+            Conversion factors governing the methane content, energy content, and combined heat and power
+            efficiencies.
+
+        Returns
+        -------
+        Dict[str, float]
+            Mapping with the methane volume (m^3), methane mass (kg), methane energy input (kWh), electricity
+            output (kWh), and heat output (kWh).
+
+        """
+
+        if biogas_volume_m3 < 0:
+            raise ValueError("biogas_volume_m3 must be non-negative")
+
+        methane_volume = biogas_volume_m3 * conversion.methane_content_fraction
+        methane_mass = methane_volume * conversion.methane_density_kg_per_m3
+        energy_input = methane_mass * conversion.methane_energy_content_kwh_per_kg
+        return {
+            "methane_volume_m3": methane_volume,
+            "methane_mass_kg": methane_mass,
+            "energy_input_kwh": energy_input,
+            "electricity_output_kwh": energy_input * conversion.chp_electrical_efficiency,
+            "heat_output_kwh": energy_input * conversion.chp_thermal_efficiency,
+        }
+
+    @staticmethod
+    def estimate_biogas_rng(
+        biogas_volume_m3: float,
+        conversion: BiogasEnergyConversion = BiogasEnergyConversion(),
+    ) -> Dict[str, float]:
+        """Estimate renewable natural gas produced from biogas after upgrading.
+
+        Parameters
+        ----------
+        biogas_volume_m3 : float
+            Volume of captured biogas routed to renewable natural gas upgrading (m^3).
+        conversion : BiogasEnergyConversion
+            Conversion factors governing the methane content, upgrading losses, and renewable natural gas energy
+            yield.
+
+        Returns
+        -------
+        Dict[str, float]
+            Mapping with the methane volume before upgrading (m^3), the renewable natural gas volume (m^3), the
+            renewable natural gas mass (kg), and the renewable natural gas energy yield (kWh).
+
+        """
+
+        if biogas_volume_m3 < 0:
+            raise ValueError("biogas_volume_m3 must be non-negative")
+
+        methane_volume = biogas_volume_m3 * conversion.methane_content_fraction
+        methane_mass = methane_volume * conversion.methane_density_kg_per_m3
+        retained_fraction = 1 - conversion.rng_leakage_loss_fraction - conversion.rng_process_heat_loss_fraction
+        rng_volume = methane_volume * retained_fraction
+        rng_mass = methane_mass * retained_fraction
+        return {
+            "methane_volume_m3": methane_volume,
+            "rng_volume_m3": rng_volume,
+            "rng_mass_kg": rng_mass,
+            "rng_energy_kwh": rng_mass * conversion.rng_energy_yield_kwh_per_kg,
+        }

RUFAS/EEE/economics/mapping.py

@@ -469,30 +469,18 @@
                 "outputs",
             },
             "Electricity production from anaerobic digester": {
-                "input_manager": ["economic_inputs.Manure.digester.kwh_per_day_produced"],
+                "biophysical_simulation": [r"Manure\.Digester\.energy\..*\.electricity_produced_kwh"],
                 "economics_files": ["commodity_prices_elec_industrial_dollar_per_kwh"],
-                "future_expansion": "Placeholder "
-                "for "
-                "the "
-                "future, "
-                "scale "
-                "to "
-                "manure "
-                "module "
-                "outputs",
+                "aggregate_by_year": True,
+                "notes": "Daily electricity generated per digester (kWh) is summed by year and priced at the "
+                "industrial electricity rate for that year.",
             },
             "Renewable natural gas (RNG) production": {
-                "input_manager": ["economic_inputs.Manure.digester.digester_rng_produced"],
+                "biophysical_simulation": [r"Manure\.Digester\.energy\..*\.rng_produced_megajoules"],
                 "economics_files": ["commodity_prices_natgas_industrial_dollar_per_megajoule"],
-                "future_expansion": "Placeholder "
-                "for "
-                "the "
-                "future, "
-                "scale "
-                "to "
-                "manure "
-                "module "
-                "outputs",
+                "aggregate_by_year": True,
+                "notes": "Daily RNG generated per digester (MJ) is summed by year and priced at the industrial "
+                "natural gas rate for that year.",
             },
             "Sold manure": {
                 "input_manager": ["economic_inputs.Manure.manure_sales"],

RUFAS/EEE/economics/preprocessing.py

@@ -16,7 +16,9 @@
 
 import math
 import re
+from collections import defaultdict
 from dataclasses import dataclass
+from datetime import datetime, timedelta
 from pathlib import Path
 from typing import Any, Dict, Iterable, List, Set
 
@@ -48,6 +50,7 @@ class EconomicItem:
     match_source: str | None
     wildcard_value_map: Dict[str, str] | None
     preprocessing: str | None
+    aggregate_by_year: bool
 
 
 class EconomicPreprocessor:
@@ -168,6 +171,7 @@ def _build_mapping(self) -> List[EconomicItem]:
                     economics_files = details.get("economics_files")
                     match_source = details.get("match_source")
                     wildcard_value_map = details.get("wildcard_value_map")
+                    aggregate_by_year = bool(details.get("aggregate_by_year", False))
                     if not biophysical_simulation and not input_manager and not economics_files:
                         continue
                     if isinstance(biophysical_simulation, str):
@@ -185,6 +189,7 @@ def _build_mapping(self) -> List[EconomicItem]:
                             match_source=match_source,
                             wildcard_value_map=wildcard_value_map if isinstance(wildcard_value_map, dict) else None,
                             preprocessing=preprocessing,
+                            aggregate_by_year=aggregate_by_year,
                         )
                     )
         return items
@@ -356,6 +361,43 @@ def _expand_input_path_with_wildcards(
 
         return expanded_paths
 
+    def _resolve_input_path(self, path: str) -> Any:
+        """Resolve an InputManager path, expanding a list ancestor into a list of field values.
+
+        A plain scalar or object path resolves exactly as ``InputManager.get_data`` would. When an
+        ancestor along the path is a list (e.g. ``economic_inputs.Manure.digester`` is now a list of
+        digesters), the trailing field is collected from every list element and returned as a list, so
+        downstream aggregation sums the field across all entries.
+        """
+        parts = path.split(".")
+        for split_index in range(len(parts), 0, -1):
+            prefix = ".".join(parts[:split_index])
+            value = self.im.get_data(prefix)
+            if value is None:
+                continue
+
+            remaining = parts[split_index:]
+            if not remaining:
+                return value
+
+            if isinstance(value, list):
+                collected: List[Any] = []
+                for element in value:
+                    current: Any = element
+                    for key in remaining:
+                        if isinstance(current, dict) and key in current:
+                            current = current[key]
+                        else:
+                            current = None
+                            break
+                    if current is not None:
+                        collected.append(current)
+                return collected
+
+            # The prefix resolved to a non-list whose remaining keys did not resolve; treat as missing.
+            return None
+        return None
+
     def _fetch_input_values(
         self,
         input_paths: Iterable[str],
@@ -384,7 +426,7 @@ def _fetch_input_values(
                     continue
 
             for candidate_path in candidate_paths:
-                data = self.im.get_data(candidate_path)
+                data = self._resolve_input_path(candidate_path)
                 if data is None:
                     self.om.add_warning(
                         "MissingEconomicInput",
@@ -638,6 +680,109 @@ def _fetch_prices_with_exact_matches(
                 prices[option] = price_data
         return prices
 
+    def _compute_revenue_by_year(self, item: "EconomicItem") -> Dict[str, Any]:
+        """Compute revenue for a per-digester daily series by summing quantities per year and pricing each year.
+
+        For each biophysical pattern (matching one variable per digester), the daily values are summed into
+        calendar-year buckets using each value's ``simulation_day``. Every year's quantity is multiplied by that
+        year's commodity price (falling back to the average price for years without an explicit entry), and the
+        results are summed into the total revenue line item.
+        """
+        info_map = {"class": self.__class__.__name__, "function": self._compute_revenue_by_year.__name__}
+
+        start_date_str = self.im.get_data("config.start_date")
+        end_date_str = self.im.get_data("config.end_date")
+        start_year: int | None = None
+        end_year: int | None = None
+        start_date: datetime | None = None
+        if start_date_str and end_date_str:
+            try:
+                start_year = int(str(start_date_str).split(":")[0])
+                end_year = int(str(end_date_str).split(":")[0])
+                start_date = datetime.strptime(str(start_date_str), "%Y:%j")
+            except (ValueError, AttributeError):
+                start_year = end_year = None
+                start_date = None
+
+        price_data = self._fetch_prices(item.economics_files)
+        price_values = self._extract_price_values(price_data)
+        price_by_year: dict[int, float] = {}
+        if start_year is not None and end_year is not None:
+            for offset, year in enumerate(range(start_year, end_year + 1)):
+                if offset < len(price_values):
+                    price_by_year[year] = price_values[offset]
+        fallback_price = Aggregator.average(price_values) if price_values else None
+        if fallback_price is None:
+            fallback_price = ECONOMIC_PRICE_FALLBACK.get("revenue", 1.0)
+
+        bio_values_by_digester_by_year: dict[str, dict[int, list[float]]] = {}
+        bio_values_by_digester: dict[str, list[float]] = {}
+        name_pattern = re.compile(r"energy\.(.+?)\.[^.]+$")
+
+        for path in item.biophysical_simulation:
+            filtered_pool = self.om.filter_variables_pool({"filters": [path]})
+            for variable_name, payload in filtered_pool.items():
+                if not isinstance(payload, dict):
+                    continue
+                values = payload.get("values", [])
+                info_maps = payload.get("info_maps", [])
+                name_match = name_pattern.search(variable_name)
+                digester_name = name_match.group(1) if name_match else variable_name
+                if digester_name not in bio_values_by_digester:
+                    bio_values_by_digester[digester_name] = []
+                    bio_values_by_digester_by_year[digester_name] = {year: [] for year in range(start_year, end_year + 1)}
+                for index, value in enumerate(values):
+                    entry_info = (
+                        info_maps[index] if index < len(info_maps) and isinstance(info_maps[index], dict) else {}
+                    )
+                    simulation_day = entry_info.get("simulation_day")
+                    try:
+                        numeric_value = float(value)
+                    except (TypeError, ValueError):
+                        continue
+                    bio_values_by_digester[digester_name].append(numeric_value)
+                    if simulation_day is None or start_date is None:
+                        # Without a day we cannot place the value in a year; fold it into the start year.
+                        year = start_year if start_year is not None else 0
+                    else:
+                        year = (start_date + timedelta(days=int(simulation_day))).year
+                    bio_values_by_digester_by_year[digester_name][year].append(numeric_value)
+
+        revenue_by_year: dict[str, dict[int, list[float]]] = {}
+        price_data_by_day: list[float] = []
+        total_revenue = 0.0
+        for digester_name, year_quantities in bio_values_by_digester_by_year.items():
+            if digester_name not in revenue_by_year:
+                revenue_by_year[digester_name] = {}
+            for year, quantity in year_quantities.items():
+                if year not in revenue_by_year[digester_name]:
+                    revenue_by_year[digester_name][year] = []
+                price = price_by_year.get(year, fallback_price)
+                year_revenue: list[float] = [daily_quantity * price for daily_quantity in quantity]
+                revenue_by_year[digester_name][year] = year_revenue
+                price_data_by_day.extend([price] * len(year_revenue))
+                total_revenue += sum(year_revenue)
+
+        if not bio_values_by_digester_by_year:
+            self.om.add_warning(
+                "MissingDigesterEnergyOutputs",
+                f"No per-digester energy outputs matched patterns {item.biophysical_simulation} for '{item.name}'.",
+                info_map,
+            )
+
+        return {
+            "biophysical_values": bio_values_by_digester,
+            "biophysical_aggregate": {digester_name: sum(bio_values)for digester_name, bio_values in bio_values_by_digester.items()},
+            "biophysical_values_by_scenario": {"baseline": bio_values_by_digester},
+            "biophysical_aggregate_by_scenario": {"baseline": {digester_name: sum(bio_values)for digester_name, bio_values in bio_values_by_digester.items()}},
+            "price_data": price_data,
+            "price_values": price_by_year,
+            "price_aggregate": Aggregator.average(price_data_by_day) if price_data_by_day else None,
+            "line_item_values_by_scenario": {"baseline": total_revenue},
+            "revenue_by_year": revenue_by_year,
+            "flow_type": "revenue",
+        }
+
     def _aggregate(self, values: List[float], desc: str) -> float | None:
         """Aggregate values according to a textual description."""
         if not values:
@@ -670,6 +815,10 @@ def preprocess(self) -> Dict[str, Dict[str, Dict[str, Dict[str, Any]]]]:
             section_data = results.setdefault(item.section, {})
             category_data = section_data.setdefault(item.category, {})
 
+            if item.aggregate_by_year:
+                category_data[item.name] = self._compute_revenue_by_year(item)
+                continue
+
             values_by_scenario = self._fetch_values_by_scenario(item.biophysical_simulation)
             wildcard_values = self._collect_biophysical_wildcards(item.biophysical_simulation)
             input_values, input_match_values = self._fetch_input_values(