LCA Processor¶

def clear_lca_caches(include_disk: bool = False, cache_dir=None) -> None:
    """
    Clear the module-level background inventory and metadata caches.

    Parameters
    ----------
    include_disk : bool, optional
        Also delete the on-disk inventory cache of the current project.
    cache_dir : str or Path, optional
        Directory of the on-disk cache, if it is not in the default location.
    """
    _BACKGROUND_INVENTORY_CACHE.clear()
    _BIOSPHERE_METADATA_CACHE.clear()
    _NODE_INDEX_CACHE.clear()
    _CHARACTERIZATION_FUNCTION_CACHE.clear()

    if include_disk:
        directory = Path(
            cache_dir
            if cache_dir is not None
            else Path(bd.projects.dir) / "optimex-inventory-cache"
        )
        for path in directory.glob("*.pickle"):
            path.unlink(missing_ok=True)

def compute_db_inventory_entries(
    db_name: str,
    intermediate_flows: dict,
    cutoff: Optional[float] = None,
    biosphere_db_name: Optional[str] = None,
    project: Optional[str] = None,
    base_dirs: Optional[Tuple[str, str]] = None,
) -> dict:
    """
    Compute aggregated background inventories for the given intermediate flows.

    All flows are solved against one technosphere matrix, factorized once when
    there are enough of them to amortize it. For an intermediate flow :math:`j` with unit demand, the
    aggregated elementary flow vector is :math:`g_j = B x_j`, i.e. the column of
    :math:`B A^{-1}` belonging to that flow. The per-background-process breakdown
    that `LCA.lci()` builds (B times diag(x_j)) is never needed here and is skipped,
    since only the aggregate enters the optimization.

    This is a module-level function so that it can also run in a worker process.

    Parameters
    ----------
    db_name : str
        Name of the background database to analyze.
    intermediate_flows : dict
        Dictionary mapping intermediate flow codes (foreground reference codes) to
        identity metadata dicts with keys "name", "reference product", and
        "location".
    cutoff : float, optional
        If given, keep only the ``cutoff`` largest elementary flows (by absolute
        amount) per intermediate flow. Default ``None`` keeps every non-zero flow,
        since a small flow can still carry a large characterized impact.
    biosphere_db_name : str, optional
        Biosphere database to read flow codes and names from. Defaults to the
        project's configured biosphere database.
    project : str, optional
        Brightway project to activate first. Needed when running in a worker
        process, which starts without an active project.
    base_dirs : tuple of str, optional
        ``(data directory, logs directory)`` of the Brightway installation, for
        worker processes that would otherwise fall back to the default location.

    Returns
    -------
    dict
        ``{flow identity: {elementary flow code: (name, amount)}}``.
    """
    if base_dirs is not None and str(bd.projects._base_data_dir) != base_dirs[0]:
        bd.projects.change_base_directories(
            Path(base_dirs[0]), Path(base_dirs[1]), project_name=project
        )
    elif project is not None and bd.projects.current != project:
        bd.projects.set_current(project)
    if biosphere_db_name is None:
        biosphere_db_name = bd.config.biosphere

    logger.info(f"Calculating inventory for database: {db_name}")
    db = bd.Database(name=db_name)

    activities = {}
    for key, meta in intermediate_flows.items():
        try:
            if isinstance(meta, dict):
                activities[key] = _resolve_node(db_name, meta)
            else:
                activities[key] = db.get(code=key)
        except Exception as e:  # Catch exceptions (e.g., if activity not found)
            logger.warning(
                f"Failed to resolve intermediate flow {meta!r} (code '{key}') "
                f"in '{db_name}': {e}"
            )

    if not activities:
        return {}

    # No LCIA method is needed: the inventory does not depend on it, and the
    # characterization factors are applied later, per system year.
    lca = bc.LCA({activity: 1 for activity in activities.values()})
    factorize = len(activities) > _FACTORIZE_MIN_FLOWS
    lca.lci(factorize=factorize)
    logger.info(
        f"Built {'and factorized ' if factorize else ''}technosphere matrix "
        f"for: {db_name}"
    )

    bio_meta = _biosphere_metadata(biosphere_db_name)
    reversed_biosphere = lca.dicts.biosphere.reversed
    row_codes = []
    row_names = []
    for row in range(lca.biosphere_matrix.shape[0]):
        flow_id = reversed_biosphere[row]
        if flow_id in bio_meta:
            code, name = bio_meta[flow_id]
        else:
            node = bd.get_node(id=flow_id)
            code, name = node["code"], node["name"]
        row_codes.append(code)
        row_names.append(name)

    entries = {}
    for key, activity in tqdm(activities.items()):
        # `lci()` is bypassed on purpose: it would build the full
        # (elementary flow x background process) inventory matrix, of which only
        # the row sums are used below.
        lca.build_demand_array({activity.id: 1})
        # `reshape(-1)` guards against pypardiso: its `spsolve` squeezes the
        # result, so a single-process background database yields a 0-d array,
        # which sparse `@` rejects as a scalar operand.
        supply = np.asarray(lca.solve_linear_system()).reshape(-1)
        aggregated = lca.biosphere_matrix @ supply

        rows = np.flatnonzero(aggregated)
        if cutoff is not None and len(rows) > int(cutoff):
            largest = np.argpartition(np.abs(aggregated[rows]), -int(cutoff))
            rows = rows[largest[-int(cutoff) :]]

        if not len(rows):
            logger.warning(
                f"Activity {activity} has no non-zero inventory in '{db_name}'."
            )

        entries[_flow_identity(key, intermediate_flows[key])] = {
            row_codes[row]: (row_names[row], float(aggregated[row])) for row in rows
        }

    logger.info(f"Finished calculating inventory for database: {db_name}")
    return entries