Generate Dataframe of multiple LCIA results using Brightway

simple_multilca_dataframe.py

from typing import Any

from tqdm import tqdm
import bw2calc as bc
import bw2data as bd
import pandas as pd

PRODUCT_NODE_TYPES = set(bd.labels.product_node_types + [bd.labels.chimaera_node_default])


def multiple_lca_scores(
    products: list[bd.Node],
    impact_categories: list[tuple],
    product_attributes: dict[str, str] = {"fu": "name"},
    lcia_attributes: dict[str, str] = {"unit": "unit"},
    lcia_label_offset: int = 0,
) -> pd.DataFrame:
    """
    Create a pandas DataFrame with product and impact category metadata and LCIA scores.

    Can add additional attributes from either the products or the impact categories. To do so,
    specify the desired attributes using a dictionary like `{"<dataframe label>": "<data label>"}`.
    Note that product attributes will override lcia attributes if they both have the same dataframe
    label.

    If the impact categories have the form `("ecoinvent-something", "general", "foo", "bar")`, and
    you are only interested in `("foo", "bar")`, you can use `lcia_label_offset` - in this case it
    would be 1, as we will take elements `lcia_label_offset + 1` and `lcia_label_offset + 2`.
    """
    def check_product(obj: Any) -> bd.Node:
        if isinstance(obj, int):
            obj = bd.get_node(id=obj)

        if isinstance(obj, bd.Node) and obj["type"] not in PRODUCT_NODE_TYPES:
            raise TypeError(f"LCA calculations must be on product nodes, but got {obj['type']}")
        elif isinstance(obj, bd.Node):
            return obj
        else:
            raise ValueError(f"LCA calculations must be on product nodes, but got {type(obj)}")

    products = [check_product(obj) for obj in products]

    for ic in impact_categories:
        if ic not in bd.methods:
            raise ValueError(f"Can't find impact category `{ic}` in the installed LCIA data")

    lca = bc.LCA({prod: 1 for prod in products}, method=impact_categories[0])
    lca.lci()
    lca.lcia()

    lcia_matrices = [lca.characterization_matrix.copy()]

    for ic in impact_categories[1:]:
        lca.switch_method(ic)
        lcia_matrices.append(lca.characterization_matrix.copy())

    results = []

    for product in tqdm(products):
        lca.lci({product.id: 1})

        for matrix, ic in zip(lcia_matrices, impact_categories):
            metadata = bd.methods[ic]
            results.append({
                "score": (matrix @ lca.inventory).sum(),
                "category": ic[lcia_label_offset + 1],
                "subcategory": ic[lcia_label_offset + 2],
            } | {
                key: metadata.get(value) for key, value in lcia_attributes.items()
            } | {
                key: product.get(value) for key, value in product_attributes.items()
            })

    return pd.DataFrame(results)