sam cash flow profile csv support

Author: softwareengineerprogrammer

src/geophires_x_client/geophires_x_result.py

@@ -458,7 +458,7 @@ def __init__(self, v_u):
                         v_u = ValueUnit(value_unit)
                         csv_entries.append([category, field_display, '', v_u.value_display, v_u.unit_display])
             else:
-                if category not in (
+                if category in (
                     'POWER GENERATION PROFILE',
                     'HEAT AND/OR ELECTRICITY EXTRACTION AND GENERATION PROFILE',
                     'EXTENDED ECONOMIC PROFILE',
@@ -467,26 +467,50 @@ def __init__(self, v_u):
                     GeophiresXResult.CCUS_PROFILE_LEGACY_NAME,
                     'S-DAC-GT PROFILE',
                 ):
-                    raise RuntimeError('unexpected category')
-
-                for i in range(len(fields[0][1:])):
-                    field_profile = fields[0][i + 1]
-                    unit_split = field_profile.split(' (')
-                    field_display = unit_split[0]
-                    unit_display = ''
-                    if len(unit_split) > 1:
-                        unit_display = unit_split[1].replace(')', '')
-                    for j in range(len(fields[1:])):
-                        year_entry = fields[j + 1]
-                        year = year_entry[0]
-                        profile_year_val = year_entry[i + 1]
-                        csv_entries.append([category, field_display, year, profile_year_val, unit_display])
+
+                    for i in range(len(fields[0][1:])):
+                        field_profile = fields[0][i + 1]
+                        name_unit_split = field_profile.split(' (')
+                        field_display = name_unit_split[0]
+                        unit_display = ''
+                        if len(name_unit_split) > 1:
+                            unit_display = name_unit_split[1].replace(')', '')
+                        for j in range(len(fields[1:])):
+                            year_entry = fields[j + 1]
+                            year = year_entry[0]
+                            profile_year_val = year_entry[i + 1]
+                            csv_entries.append([category, field_display, year, profile_year_val, unit_display])
+                elif category == 'SAM CASH FLOW PROFILE':
+                    for row_num in range(len(fields) - 1):
+                        row_idx = row_num + 1
+                        row_data = fields[row_idx]
+                        if len(row_data) > 1:
+                            name_unit_split = GeophiresXResult._get_sam_cash_flow_row_name_unit_split(
+                                fields[row_idx][0]
+                            )
+                            row_name = name_unit_split[0]
+                            unit_display = name_unit_split[1]
+                            for year in range(len(row_data) - 1):
+                                row_datapoint = row_data[year + 1]
+                                csv_entries.append([category, row_name, year, row_datapoint, unit_display])
+
+                else:
+                    raise RuntimeError(f'Unexpected category: {category}')
 
         for csv_entry in csv_entries:
             w.writerow(csv_entry)
 
         return f.getvalue()
 
+    @staticmethod
+    def _get_sam_cash_flow_row_name_unit_split(first_entry_in_row) -> list[str]:
+        name_unit_split = [it[::-1] for it in first_entry_in_row[::-1].split('( ', maxsplit=1)][::-1]
+        if len(name_unit_split) < 2:
+            name_unit_split.append('')
+
+        name_unit_split[1] = name_unit_split[1].replace(')', '')
+        return name_unit_split
+
     @property
     def json_output_file_path(self) -> Path:
         return Path(self.output_file_path).with_suffix('.json')

tests/geophires_x_client_tests/test_geophires_x_result.py

@@ -0,0 +1,17 @@
+from geophires_x_client import GeophiresXResult
+from tests.base_test_case import BaseTestCase
+
+
+class GeophiresXResultTestCase(BaseTestCase):
+
+    def test_get_sam_cash_flow_row_name_unit_split(self) -> None:
+        cases = [
+            ('Electricity to grid (kWh)', ['Electricity to grid', 'kWh']),
+            ('Federal tax benefit (liability) ($)', ['Federal tax benefit (liability)', '$']),
+            ('Underwater baskets', ['Underwater baskets', '']),
+        ]
+
+        for case in cases:
+            with self.subTest(msg=case[0]):
+                actual = GeophiresXResult._get_sam_cash_flow_row_name_unit_split(case[0])
+                self.assertListEqual(actual, case[1])

tests/test_geophires_x_client.py

@@ -487,37 +487,6 @@ def test_csv(self):
          3. Copy-paste the value of `as_csv` (in Threads & Variables tab in PyCharm) to example1_addons.csv
         """
 
-        def assertFileContentsEqual(expected_file_path, actual_file_path, tol=0.01):
-            with open(expected_file_path, encoding='utf-8') as ef:
-                expected_lines = ef.readlines()
-            with open(actual_file_path, encoding='utf-8') as af:
-                actual_lines = af.readlines()
-
-            self.assertEqual(len(expected_lines), len(actual_lines), 'The number of lines in the files do not match.')
-
-            for line_index, (expected_line, actual_line) in enumerate(zip(expected_lines, actual_lines), start=1):
-                expected_parts = expected_line.strip().split(',')
-                actual_parts = actual_line.strip().split(',')
-                self.assertEqual(
-                    len(expected_parts),
-                    len(actual_parts),
-                    f'The number of columns in line {line_index} does not match.',
-                )
-                for col_index, (expected, actual) in enumerate(zip(expected_parts, actual_parts), start=1):
-                    try:
-                        expected_float = float(expected)
-                        actual_float = float(actual)
-                        self.assertTrue(
-                            abs(expected_float - actual_float) < tol,
-                            f'Float values differ at line {line_index}, column {col_index}: {expected} != {actual}',
-                        )
-                    except ValueError:
-                        self.assertEqual(
-                            expected,
-                            actual,
-                            f'String values differ at line {line_index}, column {col_index}: {expected} != {actual}',
-                        )
-
         def assert_csv_equal(case_report_file_path, expected_csv_file_path):
             test_result_path = self._get_test_file_path(case_report_file_path)
             result = GeophiresXResult(test_result_path)
@@ -527,7 +496,7 @@ def assert_csv_equal(case_report_file_path, expected_csv_file_path):
             result_file = Path(tempfile.gettempdir(), f'test_csv-result_{uuid.uuid1()!s}.csv')
             with open(result_file, 'w', newline='', encoding='utf-8') as rf:
                 rf.write(as_csv)
-            assertFileContentsEqual(self._get_test_file_path(expected_csv_file_path), result_file)
+            self.assertCsvFileContentsEqual(self._get_test_file_path(expected_csv_file_path), result_file)
 
         for case in [
             ('examples/example1_addons.out', 'example1_addons.csv'),
@@ -536,6 +505,54 @@ def assert_csv_equal(case_report_file_path, expected_csv_file_path):
             with self.subTest(msg=case[0]):
                 assert_csv_equal(case[0], case[1])
 
+        op_example_file = 'examples/example_overpressure.out'
+        with self.subTest(msg=op_example_file):
+            # Ensure overpressure-specific RESERVOIR POWER REQUIRED PROFILES doesn't cause issues
+            op_result = GeophiresXResult(self._get_test_file_path(op_example_file))
+            op_csv = op_result.as_csv()
+            self.assertIsNotNone(op_csv)
+
+        sam_example_file = 'examples/example_SAM-single-owner-PPA.out'
+        with self.subTest(msg=sam_example_file):
+            sam_result = GeophiresXResult(self._get_test_file_path(sam_example_file))
+            sam_csv = sam_result.as_csv()
+            self.assertIsNotNone(sam_csv)
+            sam_cf_lines = [line.split(',') for line in sam_csv.split('\n') if line.startswith('SAM CASH FLOW PROFILE')]
+            self.assertGreater(len(sam_cf_lines), 250)
+            # TODO test more of the content (but not full result given how big/complex it is, which would add undue
+            #  maintenance overhead)
+
+    def assertCsvFileContentsEqual(self, expected_file_path, actual_file_path, tol=0.01):
+        with open(expected_file_path, encoding='utf-8') as ef:
+            expected_lines = ef.readlines()
+        with open(actual_file_path, encoding='utf-8') as af:
+            actual_lines = af.readlines()
+
+        self.assertEqual(len(expected_lines), len(actual_lines), 'The number of lines in the files do not match.')
+
+        for line_index, (expected_line, actual_line) in enumerate(zip(expected_lines, actual_lines), start=1):
+            expected_parts = expected_line.strip().split(',')
+            actual_parts = actual_line.strip().split(',')
+            self.assertEqual(
+                len(expected_parts),
+                len(actual_parts),
+                f'The number of columns in line {line_index} does not match.',
+            )
+            for col_index, (expected, actual) in enumerate(zip(expected_parts, actual_parts), start=1):
+                try:
+                    expected_float = float(expected)
+                    actual_float = float(actual)
+                    self.assertTrue(
+                        abs(expected_float - actual_float) < tol,
+                        f'Float values differ at line {line_index}, column {col_index}: {expected} != {actual}',
+                    )
+                except ValueError:
+                    self.assertEqual(
+                        expected,
+                        actual,
+                        f'String values differ at line {line_index}, column {col_index}: {expected} != {actual}',
+                    )
+
     def test_parse_chp_percent_cost_allocation(self):
         result = GeophiresXResult(self._get_test_file_path('examples/example3.out'))
         self.assertEqual(