Add missings

pathways/lca.py

@@ -136,6 +136,7 @@ def select_filepath(keyword: str, fps):
         return matches[0]
 
     fps = filter_filepaths(".csv", [model, scenario, str(year)])
+
     if len(fps) != 4:
         raise ValueError(f"Expected 4 filepaths, got {len(fps)}")
 

pathways/lcia.py

@@ -8,6 +8,7 @@
 import numpy as np
 from scipy import sparse
 from scipy.sparse import csr_matrix
+import bw_processing as bwp
 
 from .filesystem_constants import DATA_DIR
 
@@ -41,16 +42,34 @@ def format_lcia_method_exchanges(method):
     }
 
 
-def get_lcia_methods(methods: list = None):
+def get_lcia_methods(methods: list = None) -> dict:
     """Get a list of available LCIA methods."""
     with open(LCIA_METHODS, "r") as f:
         data = json.load(f)
 
     if methods:
         data = [x for x in data if " - ".join(x["name"]) in methods]
 
-    return {" - ".join(x["name"]): format_lcia_method_exchanges(x) for x in data}
+    cfs = {" - ".join(x["name"]): format_lcia_method_exchanges(x) for x in data}
+    units = {" - ".join(x["name"]): x["unit"] for x in data}
 
+    return cfs, units
+
+
+def build_characterization_matrix_for_sankey(method: str, biosphere_dict: dict):
+
+    lcia_data, unit = get_lcia_methods(methods=[method])
+    method_data = lcia_data[method]
+
+
+    indices, data = [], []
+
+    for flow in biosphere_dict:
+        if flow in method_data:
+            indices.append((biosphere_dict[flow], biosphere_dict[flow]))
+            data.append(method_data[flow])
+
+    return np.array(data), np.array(indices, dtype=bwp.INDICES_DTYPE), unit[method]
 
 def fill_characterization_factors_matrices(
     methods: list, biosphere_matrix_dict: dict, biosphere_dict: dict, debug=False
@@ -64,7 +83,7 @@ def fill_characterization_factors_matrices(
     :return: a sparse matrix with the characterization factors
     """
 
-    lcia_data = get_lcia_methods(methods=methods)
+    lcia_data, _ = get_lcia_methods(methods=methods)
 
     # Prepare data for efficient creation of the sparse matrix
     data = []

pathways/pathways.py

@@ -33,6 +33,7 @@
     load_units_conversion,
     resize_scenario_data,
 )
+from .sankey import Sankey
 
 
 def _get_mapping(data) -> dict:
@@ -238,6 +239,41 @@ def _get_scenarios(self, scenario_data: pd.DataFrame) -> xr.DataArray:
 
         return data
 
+    def sankey(
+            self,
+            method: str,
+            model: str,
+            scenario: str,
+            region: str,
+            year: int,
+            variable: str,
+            cutoff: float = 1e-3,
+    ):
+        """
+        Display a Sankey diagram of the scenario data.
+        """
+
+        dp, technosphere_inds, biosphere_inds, uncertain_parameters = get_lca_matrices(
+            self.filepaths, model, scenario, year
+        )
+
+        sankey = Sankey(
+            method,
+            model,
+            scenario,
+            region,
+            year,
+            variable,
+            dp,
+            biosphere_inds,
+            technosphere_inds,
+            self.mapping,
+            self.scenarios,
+            cutoff,
+        )
+
+        return sankey
+
     def calculate(
         self,
         methods: Optional[List[str]] = None,

pathways/sankey.py

@@ -0,0 +1,73 @@
+import bw_processing as bwp
+import bw2calc as bc
+import bw2data
+import xarray as xr
+
+from .lcia import build_characterization_matrix_for_sankey
+from .utils import get_unit_conversion_factors, load_units_conversion
+
+class Sankey:
+    """
+    A class to represent a Sankey diagram.
+    """
+
+    def __init__(self,
+                 method: str,
+                 model: str,
+                 scenario: str,
+                 region: str,
+                 year: int,
+                 variable: str,
+                 dp: bwp,
+                 biosphere_dict: dict,
+                 activity_dict: dict,
+                 mapping: dict,
+                 scenario_data: xr.DataArray,
+                 cutoff: float = 1e-3,
+                 ):
+        """
+        Initialize the Sankey object.
+        """
+
+        self.method = method
+        self.model = model
+        self.scenario = scenario
+        self.region = region
+        self.year = year
+        self.variable = variable
+        self.dp = dp
+        self.biosphere_dict = biosphere_dict
+        self.activity_dict = activity_dict
+        self.cutoff = cutoff
+        self.scenario_data = scenario_data
+
+        c_data, c_indices, c_unit = build_characterization_matrix_for_sankey(
+            method=self.method, biosphere_dict=self.biosphere_dict
+        )
+
+        # fetch dataset name of variables
+        dataset_name = tuple(list(mapping[self.variable]["dataset"][0].values()))
+        # add region to tuple
+        dataset_name = dataset_name + (self.region,)
+        # getch index of dataset
+        dataset_index = self.activity_dict[dataset_name]
+        self.dataset_index = dataset_index
+        # fetch demand
+        demand = self.scenario_data.sel(model=self.model, pathway=self.scenario, region=self.region, year=self.year, variables=self.variable).values
+        demand_unit = self.scenario_data.attrs["units"][self.variable]
+        demand = float(demand) * float(get_unit_conversion_factors(demand_unit, dataset_name[2], load_units_conversion()))
+
+        self.lcia_unit = c_unit
+
+        self.dp.add_persistent_vector(
+            matrix='characterization_matrix',
+            indices_array=c_indices,
+            data_array=c_data,
+        )
+
+        self.lca = bc.LCA(
+            demand={dataset_index: demand},
+            data_objs=[
+                self.dp,
+            ],
+        )