[Econ] Purchased feed cost preprocessing

RUFAS/EEE/economics/fallback_values.py

@@ -10,7 +10,7 @@
     "Economic_preprocessing.Animal.FPCM": [250.0],
     "AnimalModuleReporter.report_life_cycle_manager_data.sold_heiferII_num": [2.0],
     "AnimalModuleReporter.report_life_cycle_manager_data.sold_heiferIII_oversupply_num": [1.0],
-    "FeedManager.purchase_feed.ration_interval_*_cost": [100.0],
+    "FeedManager.purchase_feed.ration_interval_.*_cost": [100.0],
     "SEE NOTES": [10.0],
     "ManureManager._record_manure_request_results.off_farm_manure.total_manure_mass": [50.0],
     "see future_expansion": [5.0],

RUFAS/EEE/economics/mapping.py

@@ -129,6 +129,8 @@
             },
             "Purchased feed costs": {
                 "biophysical_simulation": ["FeedManager.purchase_feed.ration_interval_.*_cost"],
+                "values_are_costs": True,
+                "expand_interval_to_daily": True,
                 "economics_files": [
                     "commodity_prices_alfalfa_hay_dollar_per_kilogram",
                     "commodity_prices_alfalfa_silage_dollar_per_kilogram",

RUFAS/EEE/economics/preprocessing.py

@@ -48,6 +48,8 @@ class EconomicItem:
     match_source: str | None
     wildcard_value_map: Dict[str, str] | None
     preprocessing: str | None
+    values_are_costs: bool = False
+    expand_interval_to_daily: bool = False
 
 
 class EconomicPreprocessor:
@@ -185,6 +187,8 @@ 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,
+                            values_are_costs=bool(details.get("values_are_costs", False)),
+                            expand_interval_to_daily=bool(details.get("expand_interval_to_daily", False)),
                         )
                     )
         return items
@@ -236,11 +240,35 @@ def _fetch_values(self, sim_paths: Iterable[str]) -> List[float]:
                     )
         return values
 
-    def _fetch_values_by_scenario(self, sim_paths: Iterable[str]) -> Dict[str, List[float]]:
+    def _build_filter_content(self, path: str, expand_interval_to_daily: bool) -> Dict[str, Any]:
+        """Build OutputManager filter options, requesting daily expansion when applicable."""
+
+        filter_content: Dict[str, Any] = {"filters": [path]}
+        if not expand_interval_to_daily:
+            return filter_content
+
+        if getattr(self.om, "time", None) is None:
+            self.om.add_warning(
+                "MissingTimeForIntervalExpansion",
+                f"Cannot expand interval data to daily for '{path}' because the OutputManager time is not initialized",
+                {"class": self.__class__.__name__, "function": self._build_filter_content.__name__},
+            )
+            return filter_content
+
+        # Interval-reported variables (e.g. ration interval feed purchases) are only recorded on the days
+        # they occur; pad the gap days with zeros so the series aligns with daily-reported data.
+        filter_content["expand_data"] = True
+        filter_content["fill_value"] = 0.0
+        return filter_content
+
+    def _fetch_values_by_scenario(
+        self, sim_paths: Iterable[str], expand_interval_to_daily: bool = False
+    ) -> Dict[str, List[float]]:
         """Collect values per scenario from the OutputManager."""
 
         filtered_by_path: Dict[str, Dict[str, Any]] = {
-            path: self.om.filter_variables_pool({"filters": [path]}) for path in sim_paths
+            path: self.om.filter_variables_pool(self._build_filter_content(path, expand_interval_to_daily))
+            for path in sim_paths
         }
         if not any(filtered_by_path.values()):
             fallback_values = self._fallback_values_by_scenario(sim_paths)
@@ -490,95 +518,30 @@ def _infer_flow_type(self, item: EconomicItem) -> str | None:
     def _fetch_prices(self, economics_files: Any) -> Dict[str, Any]:
         """Collect commodity pricing using the InputManager."""
 
-        prices: Dict[str, Any] = {}
         info_map = {"class": self.__class__.__name__, "function": self._fetch_prices.__name__}
 
-        if economics_files is None:
-            return prices
-
         if isinstance(economics_files, list):
-            for file_key in economics_files:
-                price_data = self._get_data_with_handling(file_key, info_map)
-                if price_data is None:
-                    self.om.add_warning(
-                        "MissingEconomicsFile",
-                        f"Commodity pricing '{file_key}' not found in InputManager",
-                        info_map,
-                    )
-                    continue
-                prices[file_key] = price_data
-            return prices
+            return self._fetch_labeled_prices([(file_key, file_key) for file_key in economics_files], info_map)
 
         if not isinstance(economics_files, dict):
-            return prices
+            return {}
 
         selector_path = economics_files.get("input_manager_location")
         if selector_path:
-            selection = self._get_data_with_handling(selector_path, info_map)
-            if selection is None:
-                self.om.add_warning(
-                    "MissingSelection",
-                    f"Selector value not found at '{selector_path}'",
-                    info_map,
-                )
-                for option, file_key in economics_files.items():
-                    if option == "input_manager_location":
-                        continue
-                    if not isinstance(file_key, str):
-                        continue
-                    price_data = self._get_data_with_handling(file_key, info_map)
-                    if price_data is not None:
-                        prices[file_key] = price_data
-                if prices:
-                    self.om.add_warning(
-                        "MissingSelectionFallback",
-                        f"No selector match; using all available pricing options for '{selector_path}'.",
-                        info_map,
-                    )
-                return prices
-            selection_key = str(selection).lower()
-            selected_file = None
-            for option, file_key in economics_files.items():
-                if option == "input_manager_location":
-                    continue
-                if option.lower() == selection_key:
-                    selected_file = file_key
-                    break
-            if selected_file is None:
-                self.om.add_warning(
-                    "UnknownSelection",
-                    f"No price file matched selection '{selection}' at '{selector_path}'",
-                    info_map,
-                )
-                for option, file_key in economics_files.items():
-                    if option == "input_manager_location":
-                        continue
-                    if not isinstance(file_key, str):
-                        continue
-                    price_data = self._get_data_with_handling(file_key, info_map)
-                    if price_data is not None:
-                        prices[file_key] = price_data
-                if prices:
-                    self.om.add_warning(
-                        "UnknownSelectionFallback",
-                        f"No matching selection; using all available pricing options for '{selector_path}'.",
-                        info_map,
-                    )
-                return prices
-            price_data = self._get_data_with_handling(selected_file, info_map)
-            if price_data is None:
-                self.om.add_warning(
-                    "MissingEconomicsFile",
-                    f"Commodity pricing '{selected_file}' not found in InputManager",
-                    info_map,
-                )
-                return prices
-            prices[selected_file] = price_data
-            return prices
+            return self._fetch_prices_by_selector(economics_files, selector_path, info_map)
 
-        for label, file_key in economics_files.items():
-            if not isinstance(file_key, str):
-                continue
+        return self._fetch_labeled_prices(
+            [(label, file_key) for label, file_key in economics_files.items() if isinstance(file_key, str)],
+            info_map,
+        )
+
+    def _fetch_labeled_prices(
+        self, labeled_files: Iterable[tuple[str, Any]], info_map: Dict[str, str]
+    ) -> Dict[str, Any]:
+        """Fetch pricing for ``(label, file_key)`` pairs, warning on missing files."""
+
+        prices: Dict[str, Any] = {}
+        for label, file_key in labeled_files:
             price_data = self._get_data_with_handling(file_key, info_map)
             if price_data is None:
                 self.om.add_warning(
@@ -590,6 +553,79 @@ def _fetch_prices(self, economics_files: Any) -> Dict[str, Any]:
             prices[label] = price_data
         return prices
 
+    def _fetch_prices_by_selector(
+        self, economics_files: Dict[str, Any], selector_path: str, info_map: Dict[str, str]
+    ) -> Dict[str, Any]:
+        """Collect the pricing option chosen by a selector value from the InputManager."""
+
+        selection = self._get_data_with_handling(selector_path, info_map)
+        if selection is None:
+            self.om.add_warning(
+                "MissingSelection",
+                f"Selector value not found at '{selector_path}'",
+                info_map,
+            )
+            return self._fetch_all_price_options(
+                economics_files, selector_path, info_map, "MissingSelectionFallback", "No selector match"
+            )
+
+        selected_file = self._find_selected_price_file(economics_files, selection)
+        if selected_file is None:
+            self.om.add_warning(
+                "UnknownSelection",
+                f"No price file matched selection '{selection}' at '{selector_path}'",
+                info_map,
+            )
+            return self._fetch_all_price_options(
+                economics_files, selector_path, info_map, "UnknownSelectionFallback", "No matching selection"
+            )
+
+        price_data = self._get_data_with_handling(selected_file, info_map)
+        if price_data is None:
+            self.om.add_warning(
+                "MissingEconomicsFile",
+                f"Commodity pricing '{selected_file}' not found in InputManager",
+                info_map,
+            )
+            return {}
+        return {selected_file: price_data}
+
+    def _find_selected_price_file(self, economics_files: Dict[str, Any], selection: Any) -> Any | None:
+        """Find the pricing file whose option label matches the selector value."""
+
+        selection_key = str(selection).lower()
+        for option, file_key in economics_files.items():
+            if option == "input_manager_location":
+                continue
+            if option.lower() == selection_key:
+                return file_key
+        return None
+
+    def _fetch_all_price_options(
+        self,
+        economics_files: Dict[str, Any],
+        selector_path: str,
+        info_map: Dict[str, str],
+        fallback_warning: str,
+        fallback_reason: str,
+    ) -> Dict[str, Any]:
+        """Fetch every available pricing option as a fallback when no selection matched."""
+
+        prices: Dict[str, Any] = {}
+        for option, file_key in economics_files.items():
+            if option == "input_manager_location" or not isinstance(file_key, str):
+                continue
+            price_data = self._get_data_with_handling(file_key, info_map)
+            if price_data is not None:
+                prices[file_key] = price_data
+        if prices:
+            self.om.add_warning(
+                fallback_warning,
+                f"{fallback_reason}; using all available pricing options for '{selector_path}'.",
+                info_map,
+            )
+        return prices
+
     def _extract_selector_values(self, selection: Any) -> List[str]:
         """Normalize selector values into lowercase keys."""
 
@@ -660,6 +696,59 @@ def _aggregate(self, values: List[float], desc: str) -> float | None:
         # Default aggregation is sum
         return Aggregator.sum(values)
 
+    def _resolve_price_aggregate(self, item: EconomicItem, price_values: List[float]) -> float | None:
+        """Average commodity prices, applying flow-type fallbacks for quantity-based items."""
+
+        price_aggregate = self._aggregate(price_values, "average")
+        if price_aggregate is None and not item.values_are_costs:
+            flow_type = self._infer_flow_type(item) or "cost"
+            if flow_type in ECONOMIC_PRICE_FALLBACK:
+                price_aggregate = ECONOMIC_PRICE_FALLBACK[flow_type]
+        return price_aggregate
+
+    def _compute_line_item_values(
+        self,
+        item: EconomicItem,
+        aggregates_by_scenario: Dict[str, float | None],
+        aggregated_value: float | None,
+        price_aggregate: float | None,
+        info_map: Dict[str, str],
+    ) -> Dict[str, float]:
+        """Convert scenario aggregates into line item totals, scaling quantities by price."""
+
+        scenario_aggregates = {
+            scenario: aggregate_value
+            for scenario, aggregate_value in aggregates_by_scenario.items()
+            if aggregate_value is not None
+        }
+
+        if item.values_are_costs:
+            # The biophysical values are already dollar amounts (e.g. ration interval feed
+            # purchase costs priced from the feed input file), so commodity pricing is
+            # reference-only and must not scale the aggregate.
+            line_item_values = scenario_aggregates
+        elif price_aggregate is not None:
+            line_item_values = {
+                scenario: aggregate_value * price_aggregate for scenario, aggregate_value in scenario_aggregates.items()
+            }
+        else:
+            line_item_values = scenario_aggregates
+            if line_item_values:
+                self.om.add_warning(
+                    "MissingPriceForLineItem",
+                    f"No price found for '{item.name}'. Using aggregated values as totals.",
+                    info_map,
+                )
+
+        if not line_item_values and aggregated_value is not None:
+            if item.values_are_costs:
+                line_item_values["baseline"] = aggregated_value
+            else:
+                fallback_flow_type = self._infer_flow_type(item) or "cost"
+                fallback_price = ECONOMIC_PRICE_FALLBACK.get(fallback_flow_type, 1.0)
+                line_item_values["baseline"] = aggregated_value * fallback_price
+        return line_item_values
+
     def preprocess(self) -> Dict[str, Dict[str, Dict[str, Dict[str, Any]]]]:
         """Run preprocessing and store results in the InputManager."""
 
@@ -670,7 +759,9 @@ 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, {})
 
-            values_by_scenario = self._fetch_values_by_scenario(item.biophysical_simulation)
+            values_by_scenario = self._fetch_values_by_scenario(
+                item.biophysical_simulation, item.expand_interval_to_daily
+            )
             wildcard_values = self._collect_biophysical_wildcards(item.biophysical_simulation)
             input_values, input_match_values = self._fetch_input_values(
                 item.input_manager,
@@ -718,38 +809,16 @@ def preprocess(self) -> Dict[str, Dict[str, Dict[str, Dict[str, Any]]]]:
                 )
 
             price_values = self._extract_price_values(price_data)
-            price_aggregate = self._aggregate(price_values, "average")
-            if price_aggregate is None:
-                flow_type = self._infer_flow_type(item) or "cost"
-                if flow_type in ECONOMIC_PRICE_FALLBACK:
-                    price_aggregate = ECONOMIC_PRICE_FALLBACK[flow_type]
+            price_aggregate = self._resolve_price_aggregate(item, price_values)
             aggregates_by_scenario: Dict[str, float | None] = {}
             for scenario, scenario_values in values_by_scenario.items():
                 scenario_aggregate = self._aggregate(scenario_values, item.preprocessing or "")
                 if scenario_aggregate is None and scenario_values:
                     scenario_aggregate = Aggregator.sum(scenario_values)
                 aggregates_by_scenario[scenario] = scenario_aggregate
-            line_item_values_by_scenario: Dict[str, float] = {}
-            if price_aggregate is not None:
-                for scenario, aggregate_value in aggregates_by_scenario.items():
-                    if aggregate_value is None:
-                        continue
-                    line_item_values_by_scenario[scenario] = aggregate_value * price_aggregate
-            else:
-                for scenario, aggregate_value in aggregates_by_scenario.items():
-                    if aggregate_value is None:
-                        continue
-                    line_item_values_by_scenario[scenario] = aggregate_value
-                if line_item_values_by_scenario:
-                    self.om.add_warning(
-                        "MissingPriceForLineItem",
-                        f"No price found for '{item.name}'. Using aggregated values as totals.",
-                        info_map,
-                    )
-            if not line_item_values_by_scenario and aggregated_value is not None:
-                fallback_flow_type = self._infer_flow_type(item) or "cost"
-                fallback_price = ECONOMIC_PRICE_FALLBACK.get(fallback_flow_type, 1.0)
-                line_item_values_by_scenario["baseline"] = aggregated_value * fallback_price
+            line_item_values_by_scenario = self._compute_line_item_values(
+                item, aggregates_by_scenario, aggregated_value, price_aggregate, info_map
+            )
 
             flow_type = self._infer_flow_type(item) or "cost"
             category_data[item.name] = {