diff --git a/pathways/lca.py b/pathways/lca.py index c6fc545..d8037cd 100644 --- a/pathways/lca.py +++ b/pathways/lca.py @@ -14,7 +14,9 @@ import bw_processing as bwp import numpy as np import pyprind -from bw2calc.monte_carlo import MonteCarloLCA ### Dev version coming: removed `MonteCarloLCA` (normal LCA class can do Monte Carlo) and added `IterativeLCA` (different solving strategy) +from bw2calc.monte_carlo import ( # ## Dev version coming: removed `MonteCarloLCA` (normal LCA class can do Monte Carlo) and added `IterativeLCA` (different solving strategy) + MonteCarloLCA, +) from bw2calc.utils import get_datapackage from bw_processing import Datapackage from numpy import dtype, ndarray @@ -24,6 +26,13 @@ from .filesystem_constants import DIR_CACHED_DB from .lcia import fill_characterization_factors_matrices +from .stats import ( + create_mapping_sheet, + log_intensities_to_excel, + log_results_to_excel, + log_subshares_to_excel, + run_stats_analysis, +) from .subshares import ( adjust_matrix_based_on_shares, find_technology_indices, @@ -37,14 +46,6 @@ read_indices_csv, ) -from .stats import ( - log_subshares_to_excel, - log_intensities_to_excel, - log_results_to_excel, - create_mapping_sheet, - run_stats_analysis, -) - # disable warnings warnings.filterwarnings("ignore") @@ -172,14 +173,16 @@ def select_filepath(keyword: str, fps): return dp, technosphere_inds, biosphere_inds, uncertain_parameters -def find_uncertain_parameters(distributions_array: np.ndarray, indices_array: np.ndarray) -> list[tuple[int, int]]: +def find_uncertain_parameters( + distributions_array: np.ndarray, indices_array: np.ndarray +) -> list[tuple[int, int]]: """ Find the uncertain parameters in the distributions array. They will be used for the stats report :param distributions_array: :param indices_array: :return: """ - uncertain_indices = np.where(distributions_array['uncertainty_type'] != 0)[0] + uncertain_indices = np.where(distributions_array["uncertainty_type"] != 0)[0] uncertain_parameters = [tuple(indices_array[idx]) for idx in uncertain_indices] return uncertain_parameters @@ -295,12 +298,16 @@ def process_region(data: Tuple) -> dict[str, ndarray[Any, dtype[Any]] | list[int matrix_result = (characterization_matrix @ lca.inventory).toarray() temp_results.append(matrix_result) for i in range(len(uncertain_parameters)): - param_key = f'{uncertain_parameters[i][0]}_to_{uncertain_parameters[i][1]}' - param_keys.add(param_key) - if param_key not in params: - params[param_key] = [] - value = - lca.technosphere_matrix[uncertain_parameters[i][0], uncertain_parameters[i][1]] - params[param_key].append(value) + param_key = ( + f"{uncertain_parameters[i][0]}_to_{uncertain_parameters[i][1]}" + ) + param_keys.add(param_key) + if param_key not in params: + params[param_key] = [] + value = -lca.technosphere_matrix[ + uncertain_parameters[i][0], uncertain_parameters[i][1] + ] + params[param_key].append(value) results = np.array(temp_results) for idx, method in enumerate(methods): @@ -313,7 +320,6 @@ def process_region(data: Tuple) -> dict[str, ndarray[Any, dtype[Any]] | list[int log_intensities_to_excel(model, scenario, year, params) - d.append(characterized_inventory) if debug: @@ -387,9 +393,12 @@ def _calculate_year(args: tuple): # Try to load LCA matrices for the given model, scenario, and year try: - bw_datapackage, technosphere_indices, biosphere_indices, uncertain_parameters = get_lca_matrices( - filepaths, model, scenario, year - ) + ( + bw_datapackage, + technosphere_indices, + biosphere_indices, + uncertain_parameters, + ) = get_lca_matrices(filepaths, model, scenario, year) except FileNotFoundError: # If LCA matrices can't be loaded, skip to the next iteration @@ -461,7 +470,14 @@ def _calculate_year(args: tuple): logging.info("Calculating LCA results with subshares.") shares_indices = find_technology_indices(regions, technosphere_indices, geo) correlated_arrays = adjust_matrix_based_on_shares( - filepaths, lca, shares_indices, shares, use_distributions, model, scenario, year + filepaths, + lca, + shares_indices, + shares, + use_distributions, + model, + scenario, + year, ) bw_correlated = get_subshares_matrix(correlated_arrays) diff --git a/pathways/lcia.py b/pathways/lcia.py index ed7cc5b..7e02a6e 100644 --- a/pathways/lcia.py +++ b/pathways/lcia.py @@ -24,11 +24,11 @@ def get_lcia_method_names(): def format_lcia_method_exchanges(method): """ - Format LCIA method data to fit such structure: - (name, unit, type, category, subcategory, amount, uncertainty type, uncertainty amount) -- - :param method: LCIA method - :return: list of tuples + Format LCIA method data to fit such structure: + (name, unit, type, category, subcategory, amount, uncertainty type, uncertainty amount) + - + :param method: LCIA method + :return: list of tuples """ return { diff --git a/pathways/pathways.py b/pathways/pathways.py index 301e2b2..d959138 100644 --- a/pathways/pathways.py +++ b/pathways/pathways.py @@ -431,7 +431,6 @@ def calculate( self._fill_in_result_array(results) - def _fill_in_result_array(self, results: dict): # Assuming DIR_CACHED_DB, results, and self.lca_results are already defined diff --git a/pathways/subshares.py b/pathways/subshares.py index fc564c6..0a50cef 100644 --- a/pathways/subshares.py +++ b/pathways/subshares.py @@ -4,8 +4,8 @@ import bw2calc import bw_processing import bw_processing as bwp -import pandas as pd import numpy as np +import pandas as pd import yaml from bw_processing import Datapackage from premise.geomap import Geomap @@ -15,7 +15,6 @@ from pathways.filesystem_constants import DATA_DIR from pathways.utils import get_activity_indices - SUBSHARES = DATA_DIR / "technologies_shares.yaml" logging.basicConfig( diff --git a/pathways/utils.py b/pathways/utils.py index 42e4d5d..f46a3ea 100644 --- a/pathways/utils.py +++ b/pathways/utils.py @@ -102,6 +102,7 @@ def load_units_conversion() -> dict: return data + def read_indices_csv(file_path: Path) -> dict[tuple[str, str, str, str], int]: """ Reads a CSV file and returns its contents as a dictionary. @@ -328,6 +329,7 @@ def clean_cache_directory(): # # return scenario_data + def resize_scenario_data( scenario_data: xr.DataArray, model: List[str],