Add Global Sensitivity Analysis

activity_browser/app/bwutils/__init__.py

@@ -5,7 +5,6 @@
 """
 import brightway2 as bw
 from .metadata import AB_metadata
-from .montecarlo import CSMonteCarloLCA
 from .multilca import MLCA, Contributions
 from .pedigree import PedigreeMatrix
 from .presamples import PresamplesContributions, PresamplesMLCA
@@ -14,7 +13,8 @@
     CFUncertaintyInterface, ExchangeUncertaintyInterface,
     ParameterUncertaintyInterface, get_uncertainty_interface
 )
-
+from .montecarlo import MonteCarloLCA
+from .sensitivity_analysis import GlobalSensitivityAnalysis
 
 def cleanup():
     n_dir = bw.projects.purge_deleted_directories()

activity_browser/app/bwutils/commontasks.py

@@ -1,12 +1,13 @@
 # -*- coding: utf-8 -*-
 import hashlib
-
+import os
 import textwrap
 import arrow
 import brightway2 as bw
 from bw2data import databases
 from bw2data.proxies import ActivityProxyBase
 from bw2data.utils import natural_sort
+from bw2data.project import ProjectDataset, SubstitutableDatabase
 
 """
 bwutils is a collection of methods that build upon brightway2 and are generic enough to provide here so that we avoid 
@@ -56,6 +57,29 @@ def format_activity_label(key, style='pnl', max_length=40):
             return wrap_text(str(key))
     return wrap_text(label, max_length=max_length)
 
+# Switch brightway directory
+def switch_brightway2_dir(dirpath):
+    if dirpath == bw.projects._base_data_dir:
+        print('dirpath already loaded')
+        return False
+    try:
+        assert os.path.isdir(dirpath)
+        bw.projects._base_data_dir = dirpath
+        bw.projects._base_logs_dir = os.path.join(dirpath, "logs")
+        # create folder if it does not yet exist
+        if not os.path.isdir(bw.projects._base_logs_dir):
+            os.mkdir(bw.projects._base_logs_dir)
+        # load new brightway directory
+        bw.projects.db = SubstitutableDatabase(
+            os.path.join(bw.projects._base_data_dir, "projects.db"),
+            [ProjectDataset]
+        )
+        print('Loaded brightway2 data directory: {}'.format(bw.projects._base_data_dir))
+        return True
+
+    except AssertionError:
+        print('Could not access BW_DIR as specified in settings.py')
+        return False
 
 # Database
 def get_database_metadata(name):

activity_browser/app/bwutils/manager.py

@@ -1,10 +1,12 @@
 # -*- coding: utf-8 -*-
 from abc import abstractmethod
+from collections import defaultdict
 from collections.abc import Iterator
 import itertools
 from typing import Iterable, List, Optional, Tuple
 
 from asteval import Interpreter
+from bw2calc import LCA
 from bw2data.backends.peewee import ExchangeDataset
 from bw2data.parameters import (
     ProjectParameter, DatabaseParameter, ActivityParameter,
@@ -166,6 +168,69 @@ def has_parameterized_exchanges() -> bool:
         """
         return ParameterizedExchange.select().exists()
 
+    def parameter_exchange_dependencies(self) -> dict:
+        """
+
+        Schema: {param1: List[tuple], param2: List[tuple]}
+        """
+        parameters = defaultdict(list)
+        for act in self.initial.act_by_group_db:
+            exchanges = self.initial.exc_by_group(act.group)
+            for exc, formula in exchanges.items():
+                params = get_new_symbols([formula])
+                # Convert exchange from int to Index
+                exc = Index.build_from_exchange(ExchangeDataset.get_by_id(exc))
+                for param in params:
+                    parameters[param].append(exc)
+        return parameters
+
+    def extract_active_parameters(self, lca: LCA) -> dict:
+        """Given an LCA object, extract the used exchanges and build a
+        dictionary of parameters with the exchanges that use these parameters.
+
+        Schema: {param1: {"name":  str, "act": Optional[tuple],
+                          "group": str, "exchanges": List[tuple]}}
+        """
+        params = self.parameter_exchange_dependencies()
+        used_exchanges = set(lca.biosphere_dict.keys())
+        used_exchanges = used_exchanges.union(lca.activity_dict.keys())
+
+        # Make a pass through the 'params' dictionary and remove exchanges
+        # that don't exist in the 'used_exchanges' set.
+        for p in params:
+            keep = [
+                i for i, exc in enumerate(params[p])
+                if (exc.input in used_exchanges and exc.output in used_exchanges)
+            ]
+            params[p] = [params[p][i] for i in keep]
+        # Now only keep parameters with exchanges.
+        keep = [p for p, excs in params.items() if len(excs) > 0]
+        schema = {
+            p: {"name": p, "act": None, "group": None, "exchanges": params[p]}
+            for p in keep
+        }
+        # Now start filling in the parameters that remain with information.
+        for group in self.initial.groups:
+            for name, data in self.initial.act_by_group(group).items():
+                key = (data.get("database"), data.get("code"))
+                if name in schema:
+                    # Make sure that the activity parameter matches the
+                    # output of the exchange.
+                    exc = next(e.output for e in schema[name]["exchanges"])
+                    if key == exc:
+                        schema[name]["name"] = name
+                        schema[name]["group"] = group
+                        schema[name]["act"] = key
+        # Lastly, determine for the remaining parameters if they are database
+        # or project parameters.
+        for db in self.initial.databases:
+            for name, data in self.initial.by_database(db).items():
+                if name in schema and schema[name]["group"] is None:
+                    schema[name]["group"] = db
+        for name in (n for n in schema if schema[n]["group"] is None):
+            schema[name]["group"] = "project"
+        return schema
+
 
 class MonteCarloParameterManager(ParameterManager, Iterator):
     """Use to sample the uncertainty of parameter values, mostly for use in
@@ -226,3 +291,16 @@ def next(self) -> np.ndarray:
         self.parameters.update(values)
         data = self.calculate()
         return self.indices.mock_params(data)
+
+    def retrieve_sampled_values(self, data: dict):
+        """Enters the sampled values into the 'exchanges' list in the 'data'
+        dictionary.
+        """
+        for name, vals in data.items():
+            param = next((p for p in self.parameters
+                         if p.name == vals.get("name")
+                         and p.group == vals.get("group")),
+                         None)
+            if param is None:
+                continue
+            data[name]["values"].append(param.amount)