install_iwplus.py

#!/usr/bin/env python
from collections import defaultdict
from pathlib import Path
import bw2data as bd
import bw2io as bi
import functools
import pandas as pd
import requests
import tempfile


COLUMN_NAME_MAPPING = {
    'Impact category': 'cat name',
    'CF unit': 'cat unit',
    'Compartment': 'c',
    'Sub-compartment': 'sc',
    'Elem flow name': 'name',
    'CF value': 'amount',
    'Elem flow unit': 'unit',
    'MP or Damage': 'midpoint?',
    'Native geographical resolution scale': 'scale'
}
URL = "https://zenodo.org/record/7348580/files/impact_world_plus_2.0_ecoinvent_v39.xlsx?download=1"
FILE = "impact_world_plus_2.0_ecoinvent_v39.xlsx"


def nan(value):
    return value if isinstance(value, str) else None


def install_iwplus(project_name="IW+ 2.0 import"):
    bd.projects.set_current(project_name)

    if 'biosphere3' not in bd.databases:
        bi.bw2setup()

    bd.utils.download_file(FILE, )

    with tempfile.TemporaryDirectory() as td:
        filepath = Path(td) / FILE

        request = requests.get(URL, stream=True)
        if request.status_code != 200:
            raise ValueError
        download = request.raw
        chunk = 128 * 1024
        with open(filepath, "wb") as f:
            while True:
                segment = download.read(chunk)
                if not segment:
                    break
                f.write(segment)

        df = pd.read_excel(filepath)

        column_mapping = {
            COLUMN_NAME_MAPPING[value]: index
            for index, value in enumerate(df.columns)
            if value in COLUMN_NAME_MAPPING
        }

        data = defaultdict(dict)

        for _, row in df.iterrows():
            if _ == 0:
                continue
            obj = data[row[column_mapping['cat name']]]
            obj['name'] = (
                tuple([elem.strip().title() for elem in row[column_mapping['cat name']].split(",")]) +
                ('Midpoint' if row[column_mapping['midpoint?']] == 'Midpoint' else 'Endpoint',)
            )
            obj['unit'] = row[column_mapping['cat unit']]
            obj['filename'] = FILE
            obj['description'] = """
            IMPACT World+ is a life cycle impact assessment methodology which characterizes thousands of
            substances spanning across many compartments and sub-compartments of the environment. It
            differentiates 45 impact categories, both at midpoint and damage levels.

            For more information on IW+, refer to our website (https://www.impactworldplus.org/en/) and
            scientific article (https://doi.org/10.1007/s11367-019-01583-0).
            """.strip()
            obj['native spatial scale'] = row[column_mapping['scale']]
            if 'exchanges' not in obj:
                obj['exchanges'] = []
            obj['exchanges'].append({
                'name': row[column_mapping['name']],
                'amount': row[column_mapping['amount']],
                'unit': row[column_mapping['unit']],
                'categories': (
                    row[column_mapping['c']],
                    nan(row[column_mapping['sc']])
                ),
            })

        importer = bi.importers.base_lcia.LCIAImporter(str(filepath))
        importer.data = list(data.values())
        all_flows = functools.partial(
            bi.strategies.link_iterable_by_fields,
            other=(
                obj
                for obj in bd.Database(importer.biosphere_name)
            ),
            kind="biosphere",
        )

        assert len(importer.strategies) == 7
        importer.strategies[5] = all_flows

        importer.apply_strategies()
        importer.drop_unlinked()
        importer.write_methods()


if __name__ == "__main__":
    install_iwplus()