Add support for PI-specific non-billed SU types

Introduced a new discount processor, PISUCreditProcessor, to handle the credit for PI-specific non-billed SU types.
The credit amount of 100% of eligible SU costs, with credit code 0005

Updated unit and e2e tests for the new YAML behavior and processor

Author: jimmysway

process_report/loader.py

@@ -112,9 +112,28 @@ def get_alias_map(self) -> dict:
     def load_dataframe(self, filepath: str) -> pandas.DataFrame:
         return pandas.read_csv(filepath)
 
+    @functools.lru_cache
+    def _load_pi_config(self, filepath: str) -> list[dict]:
+        with open(filepath) as file:
+            pi_list = yaml.safe_load(file)
+
+        if not isinstance(pi_list, list):
+            raise ValueError("pi.yaml must contain a YAML list")
+
+        return pi_list
+
     def get_nonbillable_pis(self) -> list[str]:
-        with open(invoice_settings.nonbillable_pis_filepath) as file:
-            return [line.rstrip() for line in file]
+        pi_list = self._load_pi_config(invoice_settings.nonbillable_pis_filepath)
+        return [pi["username"] for pi in pi_list if "non_billed_su_types" not in pi]
+
+    def get_pi_non_billed_su_types(self) -> dict[str, list[str]]:
+        """PI usernames -> list of SU types that receive credit (zeroed out)."""
+        pi_list = self._load_pi_config(invoice_settings.nonbillable_pis_filepath)
+        return {
+            pi["username"]: [su["name"] for su in pi["non_billed_su_types"]]
+            for pi in pi_list
+            if "non_billed_su_types" in pi
+        }
 
     @functools.lru_cache
     def get_nonbillable_projects(self) -> pandas.DataFrame:

process_report/process_report.py

@@ -26,6 +26,7 @@
     add_institution_processor,
     lenovo_processor,
     validate_billable_pi_processor,
+    pi_su_credit_processor,
     new_pi_credit_processor,
     bu_subsidy_processor,
     prepayment_processor,
@@ -74,6 +75,7 @@ def main():
             add_institution_processor.AddInstitutionProcessor,
             lenovo_processor.LenovoProcessor,
             validate_billable_pi_processor.ValidateBillablePIsProcessor,
+            pi_su_credit_processor.PISUCreditProcessor,
             new_pi_credit_processor.NewPICreditProcessor,
             bu_subsidy_processor.BUSubsidyProcessor,
             prepayment_processor.PrepaymentProcessor,

process_report/processors/discount_processor.py

@@ -50,7 +50,12 @@ def apply_flat_discount(
         def apply_discount_on_project(remaining_discount_amount, project_i, project):
             remaining_project_balance = project[pi_balance_field]
             applied_discount = min(remaining_project_balance, remaining_discount_amount)
-            invoice.at[project_i, discount_field] = applied_discount
+
+            if invoice.at[project_i, discount_field] is None:
+                invoice.at[project_i, discount_field] = applied_discount
+            else:
+                invoice.at[project_i, discount_field] += applied_discount
+
             invoice.at[project_i, pi_balance_field] -= applied_discount
             if self.IS_DISCOUNT_BY_NERC:
                 invoice.at[project_i, balance_field] -= applied_discount

process_report/processors/pi_su_credit_processor.py

@@ -0,0 +1,47 @@
+import logging
+
+from dataclasses import dataclass, field
+
+from process_report.loader import loader
+from process_report.invoices import invoice
+from process_report.processors import discount_processor
+
+
+logger = logging.getLogger(__name__)
+logging.basicConfig(level=logging.INFO)
+
+
+@dataclass
+class PISUCreditProcessor(discount_processor.DiscountProcessor):
+    """
+    This processor operates on data processed by these Processors:
+    - ValidateBillablePIsProcessor
+
+    Certain PIs using certain SU types receive a 100% discount on those projects
+    """
+
+    IS_DISCOUNT_BY_NERC = True
+    PI_SU_CREDIT_CODE = "0005"
+
+    pi_su_mapping: dict[str, list[str]] = field(
+        default_factory=loader.get_pi_non_billed_su_types
+    )
+
+    def _process(self):
+        for pi, su_types in self.pi_su_mapping.items():
+            credit_eligible_projects = self.data[
+                self.data[invoice.SU_TYPE_FIELD].isin(su_types)
+                & (self.data[invoice.PI_FIELD] == pi)
+            ]
+            self.apply_flat_discount(
+                invoice=self.data,
+                pi_projects=credit_eligible_projects,
+                pi_balance_field=invoice.PI_BALANCE_FIELD,
+                discount_amount=credit_eligible_projects[
+                    invoice.COST_FIELD
+                ].sum(),  # Discount the entire cost of eligible projects
+                discount_field=invoice.CREDIT_FIELD,
+                balance_field=invoice.BALANCE_FIELD,
+                code_field=invoice.CREDIT_CODE_FIELD,
+                discount_code=self.PI_SU_CREDIT_CODE,
+            )

process_report/settings.py

@@ -24,7 +24,7 @@ class Settings(BaseSettings):
     prepay_debits_remote_filepath: str = "Prepay/prepay_debits.csv"
 
     # Local input files
-    nonbillable_pis_filepath: str = "pi.txt"
+    nonbillable_pis_filepath: str = "pi.yaml"
     nonbillable_projects_filepath: str = "projects.yaml"
     prepay_projects_filepath: str = "prepaid_projects.csv"
     prepay_credits_filepath: str = "prepaid_credits.csv"

process_report/tests/e2e/README.md

@@ -41,7 +41,7 @@ The `test_data/` directory contains the minimal dataset required by the pipeline
 
 - **Invoice Files**: `test_nerc-ocp-test 2025-04.csv`, `test_NERC OpenShift 2025-04.csv`
 - **Configuration Files**:
-  - PI lists: `test_pi.txt`
+  - PI lists: `test_pi.yaml`
   - Project lists: `test_projects.txt`, `test_timed_projects.txt`
   - Prepay configurations: `test_prepay_debits.csv`, `test_prepay_credits.csv`, `test_prepay_projects.csv`, `test_prepay_contacts.csv`
 - **Historical Data**: `test_PI.csv` (old PI file), `test_alias.csv`

process_report/tests/e2e/test_data/test_pi.txt

@@ -0,0 +1,4 @@
+- username: PI9
+- username: PI10
+  non_billed_su_types:
+    - name: SU1

process_report/tests/e2e/test_data/test_pi.yaml

@@ -124,7 +124,7 @@ def _prepare_pipeline_execution(
     env["PREPAY_CREDITS_FILEPATH"] = str(test_files["test_prepay_credits.csv"])
     env["PREPAY_PROJECTS_FILEPATH"] = str(test_files["test_prepay_projects.csv"])
     env["PREPAY_CONTACTS_FILEPATH"] = str(test_files["test_prepay_contacts.csv"])
-    env["nonbillable_pis_filepath"] = str(test_files["test_pi.txt"])
+    env["nonbillable_pis_filepath"] = str(test_files["test_pi.yaml"])
     env["nonbillable_projects_filepath"] = str(test_files["test_projects.yaml"])
 
     # Fallback ensures test works even when CI environment doesn't set Chrome path

process_report/tests/e2e/test_e2e_pipeline.py

@@ -0,0 +1,138 @@
+from unittest import TestCase
+
+import pandas
+
+from process_report.invoices import invoice
+from process_report.processors.pi_su_credit_processor import PISUCreditProcessor
+
+
+class TestDiscountProcessor(TestCase):
+    def _get_test_invoice(
+        self,
+        pi,
+        costs,
+        su_type,
+        credit=None,
+        credit_code=None,
+        pi_balance=None,
+        balance=None,
+    ):
+        if credit is None:
+            credit = [None for _ in range(len(pi))]
+        if credit_code is None:
+            credit_code = [None for _ in range(len(pi))]
+        if pi_balance is None:
+            pi_balance = costs
+        if balance is None:
+            balance = costs
+
+        return pandas.DataFrame(
+            {
+                invoice.PI_FIELD: pi,
+                invoice.SU_TYPE_FIELD: su_type,
+                invoice.COST_FIELD: costs,
+                invoice.CREDIT_FIELD: credit,
+                invoice.CREDIT_CODE_FIELD: credit_code,
+                invoice.PI_BALANCE_FIELD: pi_balance,
+                invoice.BALANCE_FIELD: balance,
+            }
+        )
+
+    def test_preexisting_credit(
+        self,
+    ):
+        """Tests that if there is already a credit in the invoice, the discount is added to it rather than overwriting it"""
+        invoice_data = self._get_test_invoice(
+            pi=["PI"],
+            costs=[100],
+            su_type=["Openstack Storage"],
+            credit=[10],
+            credit_code=["0003"],
+            pi_balance=[90],
+            balance=[90],
+        )
+
+        processor = PISUCreditProcessor(
+            invoice_month="2024-06",
+            data=invoice_data,
+            name="test",
+            pi_su_mapping={"PI": ["Openstack Storage"]},
+        )
+        processor.process()
+        output_invoice = processor.data
+
+        expected_invoice = self._get_test_invoice(
+            pi=["PI"],
+            costs=[100],
+            su_type=["Openstack Storage"],
+            credit=[100],
+            credit_code=["0003,0005"],
+            pi_balance=[0],
+            balance=[0],
+        )
+
+        assert expected_invoice.equals(output_invoice)
+
+    def test_one_eligible_project_only(self):
+        """If PI has multiple projects but only one is eligible, only that project should get the credit"""
+        invoice_data = self._get_test_invoice(
+            pi=["PI", "PI"],
+            costs=[50, 75],
+            su_type=["Openstack Storage", "HPC"],
+        )
+
+        processor = PISUCreditProcessor(
+            invoice_month="2024-06",
+            data=invoice_data,
+            name="test",
+            pi_su_mapping={
+                "PI": ["Openstack Storage"]
+            },  # Only Openstack Storage SU type is eligible for credit, not HPC
+        )
+        processor.process()
+        output_invoice = processor.data
+
+        expected_invoice = self._get_test_invoice(
+            pi=["PI", "PI"],
+            costs=[50, 75],
+            su_type=["Openstack Storage", "HPC"],
+            credit=[50, None],
+            credit_code=["0005", None],
+            pi_balance=[0, 75],
+            balance=[0, 75],
+        )
+
+        expected_invoice = expected_invoice.astype(output_invoice.dtypes)
+        assert expected_invoice.equals(output_invoice)
+
+    def test_all_eligible_projects(self):
+        """PI has multiple projects and all are eligible, so all should get the credit"""
+        invoice_data = self._get_test_invoice(
+            pi=["PI", "PI"],
+            costs=[40, 60],
+            su_type=["Openstack Storage", "Openstack Compute"],
+        )
+
+        processor = PISUCreditProcessor(
+            invoice_month="2024-06",
+            data=invoice_data,
+            name="test",
+            pi_su_mapping={
+                "PI": ["Openstack Storage", "Openstack Compute"]
+            },  # Both SU types are eligible for credit
+        )
+        processor.process()
+        output_invoice = processor.data
+
+        expected_invoice = self._get_test_invoice(
+            pi=["PI", "PI"],
+            costs=[40, 60],
+            su_type=["Openstack Storage", "Openstack Compute"],
+            credit=[40, 60],
+            credit_code=["0005", "0005"],
+            pi_balance=[0, 0],
+            balance=[0, 0],
+        )
+
+        expected_invoice = expected_invoice.astype(output_invoice.dtypes)
+        assert expected_invoice.equals(output_invoice)