ecobalyse-data/import_method.py at main · MTES-MCT/ecobalyse-data · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
#!/usr/bin/env python3
import functools
import tempfile
from zipfile import ZipFile

import bw2data
import bw2io
from bw2io.strategies import (
    # drop_unlinked_cfs,
    drop_unspecified_subcategories,
    link_iterable_by_fields,
    match_subcategories,
    normalize_biosphere_categories,
    normalize_biosphere_names,
    normalize_simapro_biosphere_categories,
    normalize_simapro_biosphere_names,
    normalize_units,
    set_biosphere_type,
)
from frozendict import frozendict

from common import brightway_patch as brightway_patch
from common.import_ import setup_project
from config import settings
from ecobalyse_data import s3
from ecobalyse_data.bw.strategy import noLT, uraniumFRU
from ecobalyse_data.logging import logger


def import_method():
    """
    Import file at path `datapath` linked to biosphere named `dbname`
    """

    logger.debug(
        f"{settings.bw.BIOSPHERE} size: {len(bw2data.Database(settings.bw.BIOSPHERE))}"
    )
    logger.info(f"🟢 Importing {settings.dbfiles.METHOD}")
    datapath = s3.get_file(settings.dbfiles.METHOD, settings.dbfiles.METHOD_MD5)
    # unzip
    with tempfile.TemporaryDirectory() as tempdir:
        logger.debug(f"-> Extracting the zip file {datapath}")
        with ZipFile(datapath) as zf:
            extracted_fn = zf.extract(zf.namelist()[0], tempdir)
            logger.debug(f"-> Extracted the zip file as {extracted_fn}")

            ef = bw2io.importers.SimaProLCIACSVImporter(
                extracted_fn, biosphere=settings.bw.BIOSPHERE
            )

            ef.statistics()

            ef.strategies = [
                normalize_units,
                set_biosphere_type,
                drop_unspecified_subcategories,
                functools.partial(normalize_biosphere_categories, lcia=True),
                functools.partial(normalize_biosphere_names, lcia=True),
                normalize_simapro_biosphere_categories,
                normalize_simapro_biosphere_names,
                functools.partial(
                    link_iterable_by_fields,
                    other=(
                        obj
                        for obj in bw2data.Database(ef.biosphere_name)
                        if obj.get("type") == "emission"
                    ),
                    kind="biosphere",
                ),
                functools.partial(
                    match_subcategories, biosphere_db_name=ef.biosphere_name
                ),
            ]
            ef.strategies.append(noLT)
            ef.strategies.append(uraniumFRU)
            ef.apply_strategies()
            logger.debug(f"biosphere3 size: {len(bw2data.Database('biosphere3'))}")
            ef.statistics()

            # ef.write_excel(METHODNAME)
            # drop CFs which are not linked to a biosphere substance
            ef.drop_unlinked()
            # remove duplicates in exchanges
            for m in ef.data:
                m["exchanges"] = [
                    dict(f) for f in list(set([frozendict(d) for d in m["exchanges"]]))
                ]

            ef.write_methods(overwrite=True)
    logger.info(f"🟢 Finished importing {settings.bw.METHOD}")


if __name__ == "__main__":
    setup_project()

    if (
        len([method for method in bw2data.methods if method[0] == settings.bw.METHOD])
        == 0
    ):
        import_method()
    else:
        logger.debug(f"{settings.bw.METHOD} already imported")