Black reformating

pathways/lca.py

@@ -1,14 +1,14 @@
+import csv
 import sys
 import warnings
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple
 
 import numpy as np
 import scipy
-import csv
-import xarray as xr
-from typing import List, Tuple, Dict, Optional
 import scipy.sparse
+import xarray as xr
 from scipy import sparse
-from pathlib import Path
 from scipy.sparse import csr_matrix
 
 # Attempt to import pypardiso's spsolve function. If it isn't available, fall back on scipy's spsolve.
@@ -18,7 +18,12 @@
     from scipy.sparse.linalg import spsolve
 
 
-def create_demand_vector(activities_idx: List[int], A: scipy.sparse, demand: xr.DataArray, unit_conversion: np.ndarray) -> np.ndarray:
+def create_demand_vector(
+    activities_idx: List[int],
+    A: scipy.sparse,
+    demand: xr.DataArray,
+    unit_conversion: np.ndarray,
+) -> np.ndarray:
     """
     Create a demand vector with the given activities indices, sparse matrix A, demand values, and unit conversion factors.
 
@@ -73,11 +78,11 @@ def read_indices_csv(file_path: Path) -> Dict[Tuple[str, str, str, str], str]:
 
 
 def load_matrix_and_index(
-        file_path: Path,
-        num_indices: int,
-        sign: int = 1,
-        transpose: bool = False,
-        extra_indices: Optional[int] = None
+    file_path: Path,
+    num_indices: int,
+    sign: int = 1,
+    transpose: bool = False,
+    extra_indices: Optional[int] = None,
 ) -> csr_matrix:
     """
     Reads a CSV file and returns its contents as a CSR sparse matrix.
@@ -107,7 +112,10 @@ def load_matrix_and_index(
         shape = (num_indices, I.max() + 1)
         idx = (J, I)
     else:
-        shape = (extra_indices if extra_indices is not None else I.max() + 1, num_indices)
+        shape = (
+            extra_indices if extra_indices is not None else I.max() + 1,
+            num_indices,
+        )
         idx = (I, J)
 
     # Create and return the CSR matrix
@@ -116,10 +124,7 @@ def load_matrix_and_index(
 
 
 def get_lca_matrices(
-    datapackage: str,
-    model: str,
-    scenario: str,
-    year: int
+    datapackage: str, model: str, scenario: str, year: int
 ) -> Tuple[sparse.csr_matrix, sparse.csr_matrix, Dict, Dict]:
     """
     Retrieve Life Cycle Assessment (LCA) matrices from disk.
@@ -138,7 +143,9 @@ def get_lca_matrices(
     B_inds = read_indices_csv(dirpath / "B_matrix_index.csv")
 
     A = load_matrix_and_index(dirpath / "A_matrix.csv", len(A_inds), transpose=True)
-    B = load_matrix_and_index(dirpath / "B_matrix.csv", len(B_inds), sign = -1, extra_indices=len(A_inds))
+    B = load_matrix_and_index(
+        dirpath / "B_matrix.csv", len(B_inds), sign=-1, extra_indices=len(A_inds)
+    )
 
     return A, B, A_inds, B_inds
 
@@ -159,7 +166,7 @@ def remove_double_counting(A: csr_matrix, vars_info: dict) -> csr_matrix:
 
     for region in vars_info:
         for variable in vars_info[region]:
-            idx = vars_info[region][variable]['idx']
+            idx = vars_info[region][variable]["idx"]
             row_mask = np.isin(A_coo.row, idx)
             col_mask = np.isin(A_coo.col, idx)
             A_coo.data[row_mask & ~col_mask] = 0  # zero out rows
@@ -203,8 +210,6 @@ def solve_inventory(A: csr_matrix, B: np.ndarray, f: np.ndarray) -> np.ndarray:
     if B.shape[0] != A.shape[0]:
         raise ValueError("Incompatible dimensions between A and B")
 
-
-
     with warnings.catch_warnings():
         warnings.simplefilter("ignore")
         # Solve the system Ax = f for x using sparse solver

pathways/pathways.py

@@ -4,38 +4,39 @@
 LCA datasets, and LCA matrices.
 """
 
+import csv
 import json
 import sys
+from collections import defaultdict
+from csv import reader
+from multiprocessing import Pool, cpu_count
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple, Union
 
+import numpy as np
 import pandas as pd
+import xarray as xr
 import yaml
 from datapackage import DataPackage
-from pathlib import Path
-from csv import reader
-import csv
-import numpy as np
+from premise.geomap import Geomap
 from scipy import sparse
-from .lcia import fill_characterization_factors_matrix, get_lcia_method_names
-from .utils import (
-    load_classifications,
-    load_units_conversion,
-    display_results,
-    create_lca_results_array,
-    _get_activity_indices,
-    get_unit_conversion_factors,
-)
+
 from .lca import (
-    solve_inventory,
     characterize_inventory,
     create_demand_vector,
     get_lca_matrices,
     remove_double_counting,
+    solve_inventory,
+)
+from .lcia import fill_characterization_factors_matrix, get_lcia_method_names
+from .utils import (
+    _get_activity_indices,
+    create_lca_results_array,
+    display_results,
+    get_unit_conversion_factors,
+    load_classifications,
+    load_units_conversion,
 )
-import xarray as xr
-from collections import defaultdict
-from premise.geomap import Geomap
-from multiprocessing import Pool, cpu_count
-from typing import Any, Dict, List, Tuple, Union, Optional
 
 # if pypardiso is installed, use it
 try:
@@ -123,7 +124,6 @@ def fetch_indices(mapping, regions, variables, A_index, geo):
     vars_idx = {}
 
     for region in regions:
-
         activities = [
             (
                 mapping[x]["dataset"]["name"],
@@ -149,6 +149,7 @@ def fetch_indices(mapping, regions, variables, A_index, geo):
 
     return vars_idx
 
+
 def generate_A_indices(A_index, reverse_classifications, lca_results_coords):
     # Generate a list of activity indices for each activity category
     acts_idx = []
@@ -192,10 +193,7 @@ def process_region(data: Tuple) -> Union[None, Dict[str, Any]]:
     category_idx = []
     for cat in lca_results_coords["act_category"].values:
         category_idx.append(
-            [
-                int(A_index[a])
-                for a in reverse_classifications[cat]
-                if a in A_index]
+            [int(A_index[a]) for a in reverse_classifications[cat] if a in A_index]
         )
 
     act_categories = lca_results_coords["act_category"].values
@@ -209,7 +207,6 @@ def process_region(data: Tuple) -> Union[None, Dict[str, Any]]:
         target = np.zeros((len(act_categories), len(list(vars_idx)), len(B_index)))
 
     for v, variable in enumerate(variables):
-
         idx, dataset = vars_idx[variable]["idx"], vars_idx[variable]["dataset"]
 
         # Compute the unit conversion vector for the given activities
@@ -249,21 +246,31 @@ def process_region(data: Tuple) -> Union[None, Dict[str, Any]]:
             # Sum along the first axis of D to get final result
             D = D.sum(axis=1)
 
-            acts_idx = generate_A_indices(A_index, reverse_classifications, lca_results_coords,)
+            acts_idx = generate_A_indices(
+                A_index,
+                reverse_classifications,
+                lca_results_coords,
+            )
 
             # Sum over the first axis of D, using acts_idx for advanced indexing
             target[:, v] = D[acts_idx, ...].sum(axis=0)
 
         else:
             # else, just sum the results of the inventory
-            acts_idx = generate_A_indices(A_index, reverse_classifications, lca_results_coords, )
+            acts_idx = generate_A_indices(
+                A_index,
+                reverse_classifications,
+                lca_results_coords,
+            )
             target[:, v] = C[acts_idx, ...].sum(axis=0)
 
     # Return a dictionary containing the processed LCA data for the given region
     return {
         "act_category": act_categories,
         "variable": list(vars_idx.keys()),
-        "impact_category": impact_categories if lcia_matrix is not None else [" - ".join(a) for a in list(B_index.keys())],
+        "impact_category": impact_categories
+        if lcia_matrix is not None
+        else [" - ".join(a) for a in list(B_index.keys())],
         "year": year,
         "region": region,
         "D": target,
@@ -347,8 +354,7 @@ def get_scenarios(self):
 
         scenario_data = self.data.get_resource("scenarios").read()
         scenario_data = pd.DataFrame(
-            scenario_data,
-            columns=self.data.get_resource("scenarios").headers
+            scenario_data, columns=self.data.get_resource("scenarios").headers
         )
 
         # remove rows which do not have a value under the `variable`
@@ -560,7 +566,6 @@ def characterize_planetary_boundaries(
         years: Optional[List[int]] = None,
         variables: Optional[List[str]] = None,
     ):
-
         self.calculate(
             models=models,
             scenarios=scenarios,
@@ -571,5 +576,4 @@ def characterize_planetary_boundaries(
         )
 
     def display_results(self, cutoff: float = 0.01) -> xr.DataArray:
-
         return display_results(self.lca_results, cutoff=cutoff)

pathways/utils.py

@@ -179,11 +179,7 @@ def create_lca_results_array(
 
     # Create the xarray DataArray with the defined coordinates and dimensions.
     # The array is initialized with zeros.
-    return xr.DataArray(
-        np.zeros(dims),
-        coords=coords,
-        dims=list(coords.keys())
-    )
+    return xr.DataArray(np.zeros(dims), coords=coords, dims=list(coords.keys()))
 
 
 def display_results(