Source code for bw2io.importers.json_ld_lcia
from bw2data import Database, databases
from ..extractors.json_ld import JSONLDExtractor
from ..strategies import (
json_ld_lcia_add_method_metadata,
json_ld_lcia_convert_to_list,
json_ld_lcia_reformat_cfs_as_exchanges,
json_ld_lcia_set_method_metadata,
normalize_units,
)
from .base_lcia import LCIAImporter
[docs]
class JSONLDLCIAImporter(LCIAImporter):
"""
Importer for the `OLCD JSON-LD LCIA data format <https://github.com/GreenDelta/olca-schema>`__.
Parameters
----------
dirpath: str
Directory path for the JSON-LD data.
Attributes
----------
format: str
Data format description.
extractor: :class:`JSONLDExtractor`
Extractor class for the JSON-LD data.
data: dict
Extracted LCIA data.
strategies: list
List of strategies to apply to the LCIA data.
Methods
-------
match_biosphere_by_id(database_name)
Matches biosphere flows to a specified database by ID.
"""
def __init__(self, dirpath):
"""
Initialize the JSONLDLCIAImporter object.
Parameters
----------
dirpath: str
Directory path for the JSON-LD data.
Returns
-------
None
"""
[docs]
self.data = self.extractor.extract(dirpath)
KEEP = ("lcia_categories", "lcia_methods", "flows")
for key in list(self.data.keys()):
if key not in KEEP:
del self.data[key]
[docs]
self.strategies = [
json_ld_lcia_add_method_metadata,
json_ld_lcia_convert_to_list,
json_ld_lcia_set_method_metadata,
json_ld_lcia_reformat_cfs_as_exchanges,
normalize_units,
]
[docs]
def match_biosphere_by_id(self, database_name):
"""
Matches biosphere flows to a specified database by ID.
Parameters
----------
database_name: str
Name of the biosphere database.
Returns
-------
None
"""
assert database_name in databases
codes = {o["code"] for o in Database(database_name)}
for method in self.data:
for cf in method["exchanges"]:
if cf["flow"]["@id"] in codes:
cf["input"] = (database_name, cf["flow"]["@id"])
