[WIP] SAM-EM MCY Levelized Metrics Update

docs/SAM-Economic-Models.md

@@ -51,6 +51,7 @@ The following table describes how GEOPHIRES parameters are transformed into SAM
 
 1. Only Electricity end-use is supported
 2. Add-ons with electricity and heat are not currently supported. (Add-ons CAPEX, OPEX, and profit are supported.)
+3. Carbon Revenue is not currently supported, but will be in future releases. See [tracking issue](https://github.com/NatLabRockies/GEOPHIRES-X/issues/458?title=SAM+Economic+Models+do+not+calculate+carbon+revenue) for details.
 
 ## Multiple Construction Years
 

src/geophires_x/EconomicsSam.py

@@ -184,6 +184,9 @@ def _get_row(row_name__: str) -> list[Any]:
         if self._royalties_rate_schedule is not None:
             ret = self._insert_royalties_rate_schedule(ret)
 
+        # FIXME WIP
+        ret = self._insert_calculated_levelized_metrics_line_items(ret)
+
         return ret
 
     def _insert_royalties_rate_schedule(self, cf_ret: list[list[Any]]) -> list[list[Any]]:
@@ -206,6 +209,153 @@ def _get_row_index(row_name_: str) -> list[Any]:
 
         return ret
 
+    # noinspection DuplicatedCode
+    def _insert_calculated_levelized_metrics_line_items(self, cf_ret: list[list[Any]]) -> list[list[Any]]:
+        """
+        FIXME WIP re: https://github.com/NatLabRockies/GEOPHIRES-X/issues/444#issuecomment-3730443078
+        """
+
+        insert_backfilled_rows: bool = False
+
+        ret = cf_ret.copy()
+
+        def _get_row_index(row_name_: str) -> list[Any]:
+            return [it[0] for it in ret].index(row_name_)
+
+        # Backfill annual costs
+        annual_costs_usd_row_name = 'Annual costs ($)'
+        annual_costs = cf_ret[_get_row_index(annual_costs_usd_row_name)].copy()
+        after_tax_net_cash_flow_usd = cf_ret[_get_row_index('After-tax net cash flow ($)')]
+
+        annual_costs_backfilled = [
+            *after_tax_net_cash_flow_usd[1 : (self._pre_revenue_years_count + 1)],
+            *annual_costs[(self._pre_revenue_years_count + 1) :],
+        ]
+
+        if insert_backfilled_rows:
+            ret.insert(
+                _get_row_index(annual_costs_usd_row_name) + 1,
+                [
+                    *['Annual costs [backfilled] ($)'],
+                    *annual_costs_backfilled,
+                ],
+            )
+
+        electricity_to_grid_kwh_row_name = 'Electricity to grid (kWh)'
+        electricity_to_grid = cf_ret[_get_row_index(electricity_to_grid_kwh_row_name)].copy()
+        electricity_to_grid_backfilled = [0 if it == '' else it for it in electricity_to_grid[1:]]
+
+        if insert_backfilled_rows:
+            ret.insert(
+                # _get_row_index(electricity_to_grid_kwh_row_name),  # there are multiple rows with this name
+                _get_row_index(annual_costs_usd_row_name) + 4,
+                [
+                    *['Electricity to grid [backfilled] (kWh)'],
+                    *electricity_to_grid_backfilled,
+                ],
+            )
+        else:
+            pass  # TODO: update existing row (requires finding the right row among duplicates)
+
+        pv_of_annual_costs_backfilled_row_name = 'Present value of annual costs [backfilled] ($)'
+
+        # Backfill PV of annual costs
+        annual_costs_backfilled_pv_processed = annual_costs_backfilled.copy()
+        pv_of_annual_costs_backfilled = []
+        for year in range(self._pre_revenue_years_count):
+            pv_at_year = abs(
+                round(
+                    npf.npv(
+                        self.nominal_discount_rate.quantity().to('dimensionless').magnitude,
+                        annual_costs_backfilled_pv_processed,
+                    )
+                )
+            )
+
+            pv_of_annual_costs_backfilled.append(pv_at_year)
+
+            cost_at_year = annual_costs_backfilled_pv_processed.pop(0)
+            annual_costs_backfilled_pv_processed[0] = annual_costs_backfilled_pv_processed[0] + cost_at_year
+
+        pv_of_annual_costs_backfilled_row = [
+            *[pv_of_annual_costs_backfilled_row_name],
+            *pv_of_annual_costs_backfilled,
+        ]
+
+        pv_of_annual_costs_row_index = _get_row_index('Present value of annual costs ($)')
+        if insert_backfilled_rows:
+            ret.insert(
+                pv_of_annual_costs_row_index + 1,
+                pv_of_annual_costs_backfilled_row,
+            )
+        else:
+            ret[pv_of_annual_costs_row_index][1:] = pv_of_annual_costs_backfilled
+
+        # Backfill PV of electricity to grid
+        pv_of_electricity_to_grid_backfilled_row_name = 'Present value of annual energy nominal [backfilled] (kWh)'
+
+        electricity_to_grid_backfilled_pv_processed = electricity_to_grid_backfilled.copy()
+        pv_of_electricity_to_grid_backfilled = []
+        for year in range(self._pre_revenue_years_count):
+            pv_at_year = abs(
+                round(
+                    npf.npv(
+                        self.nominal_discount_rate.quantity().to('dimensionless').magnitude,
+                        electricity_to_grid_backfilled_pv_processed,
+                    )
+                )
+            )
+
+            pv_of_electricity_to_grid_backfilled.append(pv_at_year)
+
+            electricity_to_grid_at_year = electricity_to_grid_backfilled_pv_processed.pop(0)
+            electricity_to_grid_backfilled_pv_processed[0] = (
+                electricity_to_grid_backfilled_pv_processed[0] + electricity_to_grid_at_year
+            )
+
+        pv_of_annual_energy_row_index = _get_row_index('Present value of annual energy nominal (kWh)')
+        if insert_backfilled_rows:
+            ret.insert(
+                pv_of_annual_energy_row_index + 1,
+                [
+                    *[pv_of_electricity_to_grid_backfilled_row_name],
+                    *pv_of_electricity_to_grid_backfilled,
+                ],
+            )
+        else:
+            ret[pv_of_annual_energy_row_index][1:] = pv_of_electricity_to_grid_backfilled
+
+        def backfill_lcoe_nominal() -> None:
+            # pv_of_annual_costs_backfilled_row = ret[_get_row_index(pv_of_annual_costs_backfilled_row_name)][1:]
+            pv_of_electricity_to_grid_backfilled_row = pv_of_electricity_to_grid_backfilled
+            pv_of_annual_costs_backfilled_row_values = pv_of_annual_costs_backfilled_row[
+                1 if isinstance(pv_of_annual_costs_backfilled_row[0], str) else 0 :
+            ]
+
+            lcoe_nominal_backfilled = []
+            for _year in range(len(pv_of_annual_costs_backfilled_row_values)):
+                lcoe_nominal_backfilled.append(
+                    pv_of_annual_costs_backfilled_row_values[_year]
+                    * 100
+                    / pv_of_electricity_to_grid_backfilled_row[_year]
+                )
+
+            lcoe_nominal_row_index = _get_row_index('LCOE Levelized cost of energy nominal (cents/kWh)')
+            if insert_backfilled_rows:
+                ret.insert(
+                    lcoe_nominal_row_index + 1,
+                    [
+                        *['LCOE Levelized cost of energy nominal [backfilled] (cents/kWh)'],
+                        *lcoe_nominal_backfilled,
+                    ],
+                )
+            else:
+                ret[lcoe_nominal_row_index][1:] = lcoe_nominal_backfilled
+
+        backfill_lcoe_nominal()
+
+        return ret
+
     @property
     def sam_after_tax_net_cash_flow_all_years(self) -> list[float]:
         return _after_tax_net_cash_flow_all_years(self.sam_cash_flow_profile, self._pre_revenue_years_count)

tests/base_test_case.py

@@ -3,6 +3,7 @@
 import inspect
 import numbers
 import os.path
+import sys
 import unittest
 
 from geophires_x.GeoPHIRESUtils import sig_figs
@@ -123,6 +124,17 @@ def assertAlmostEqualWithinSigFigs(self, expected: float | int, actual: float |
     def _is_github_actions(self):
         return 'CI' in os.environ or 'TOXPYTHON' in os.environ
 
+    def _handle_assert_logs_failure(self, ae: AssertionError):
+        if sys.version_info[:2] == (3, 8) and self._is_github_actions():
+            # FIXME - see
+            #  https://github.com/softwareengineerprogrammer/GEOPHIRES/actions/runs/19646240874/job/56262028512#step:5:344
+            print(
+                f'WARNING: Skipping logs assertion in GitHub Actions '
+                f'for Python {sys.version_info.major}.{sys.version_info.minor}'
+            )
+        else:
+            raise ae
+
     @staticmethod
     def get_input_parameter(params: GeophiresInputParameters, param_name: str) -> float | str | None:
         """

tests/geophires_x_tests/test_economics_sam.py

@@ -1137,14 +1137,3 @@ def _new_model(input_file: Path, additional_params: dict[str, Any] | None = None
             m.Calculate()
 
         return m
-
-    def _handle_assert_logs_failure(self, ae: AssertionError):
-        if sys.version_info[:2] == (3, 8) and self._is_github_actions():
-            # FIXME - see
-            #  https://github.com/softwareengineerprogrammer/GEOPHIRES/actions/runs/19646240874/job/56262028512#step:5:344
-            _log.warning(
-                f'WARNING: Skipping logs assertion in GitHub Actions '
-                f'for Python {sys.version_info.major}.{sys.version_info.minor}'
-            )
-        else:
-            raise ae

tests/geophires_x_tests/test_reservoir.py

@@ -323,34 +323,37 @@ def _del_metadata(r: GeophiresXResult) -> GeophiresXResult:
             del r.result['Simulation Metadata']
             return r
 
-        with self.assertLogs(level='INFO') as logs:
-            _del_metadata(
-                GeophiresXClient().get_geophires_result(
-                    GeophiresInputParameters(
-                        from_file_path=self._get_test_file_path('../examples/example5b.txt'),
-                        params={
-                            'Reservoir Output Profile': ','.join(
-                                [str(it) for it in [30 * v for v in [*([10] * 7), 9, 8, 7]]]
-                            )
-                        },
+        try:
+            with self.assertLogs(level='INFO') as logs:
+                _del_metadata(
+                    GeophiresXClient().get_geophires_result(
+                        GeophiresInputParameters(
+                            from_file_path=self._get_test_file_path('../examples/example5b.txt'),
+                            params={
+                                'Reservoir Output Profile': ','.join(
+                                    [str(it) for it in [30 * v for v in [*([10] * 7), 9, 8, 7]]]
+                                )
+                            },
+                        )
                     )
                 )
-            )
 
-            self.assertHasLogRecordWithMessage(
-                logs, 'Reservoir temperature extrapolation result', treat_substring_match_as_match=True
-            )
+                self.assertHasLogRecordWithMessage(
+                    logs, 'Reservoir temperature extrapolation result', treat_substring_match_as_match=True
+                )
 
-            self.assertHasLogRecordWithMessage(
-                logs,
-                # TODO make this less hard-coded
-                '[207.73, 177.48, 147.23, 116.97, 86.72, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, '
-                '80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, '
-                '80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, '
-                '80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, '
-                '80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, '
-                '80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, '
-                '80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, '
-                '80.0, 80.0]',
-                treat_substring_match_as_match=True,
-            )
+                self.assertHasLogRecordWithMessage(
+                    logs,
+                    # TODO make this less hard-coded
+                    '[207.73, 177.48, 147.23, 116.97, 86.72, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, '
+                    '80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, '
+                    '80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, '
+                    '80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, '
+                    '80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, '
+                    '80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, '
+                    '80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, 80.0, '
+                    '80.0, 80.0]',
+                    treat_substring_match_as_match=True,
+                )
+        except AssertionError as ae:
+            self._handle_assert_logs_failure(ae)

tests/test_base_test_case.py

@@ -33,7 +33,7 @@ def test_assertAlmostEqualWithinPercentage_bad_arguments(self):
 
             self.assertIn(str(msg_type_error), '(you may have meant to pass percent=10)')
 
-        with self.assertLogs(level='INFO') as logs:
+        with self.assertLogs('tests.base_test_case', level='INFO') as logs:
             with self.assertRaises(AssertionError):
                 self.assertAlmostEqualWithinPercentage([1, 2, 3], [1.1, 2.2, 3.3], percent=10.5)