动态LCA：多活动多交换量高效修改方法咨询（电网小时级模型）

问题背景 #

我正在构建电网小时级消耗组合LCA模型，使用Activity-Browser完成太阳能光伏、风电等技术的建模与更新，各技术对消耗组合的贡献以交换量表示。计划借助Brightway 2.5基于小时级数据动态修改各技术的交换量，示例小时级消耗组合如下：

Hour   Biomass  Natural Gas     Hydro    Solar       Wind   Imports
0     0  0.044560     0.296973  0.193629  0.118828  0.256997  0.089014
1     1  0.043051     0.205872  0.205036  0.136175  0.237763  0.172102
2     2  0.046465     0.126948  0.210244  0.151587  0.243857  0.220899
3     3  0.045191     0.118660  0.214444  0.151929  0.230907  0.238869

当前通过遍历每小时数据更新exchange['amount']并调用exchange.save()，同时执行LCIA计算，但数千次迭代耗时过长，需要更高效的动态修改方法。

高效优化方案 #

1. 取消不必要的数据库持久化 #

exchange.save()会将修改写入Brightway的数据库，属于IO密集型操作，数千次迭代的开销极大。如果仅需计算小时级LCA结果（无需永久保存每小时的交换量），完全可以跳过save()步骤，直接在内存中完成修改和计算。

2. 直接操作LCA技术矩阵 #

Brightway的LCA计算核心是技术矩阵（technosphere matrix），直接修改矩阵元素比逐个修改交换对象效率高得多，具体实现如下：

import brightway2 as bw
import pandas as pd

# 加载项目与数据库
bw.projects.set_current("your_project_name")
db = bw.Database("your_database_name")

# 定位核心活动
grid_activity = db.get("grid_activity_code")  # 电网消耗组合活动
tech_activity_map = {
    "Biomass": db.get("biomass_tech_code"),
    "Natural Gas": db.get("natural_gas_tech_code"),
    "Hydro": db.get("hydro_tech_code"),
    "Solar": db.get("solar_tech_code"),
    "Wind": db.get("wind_tech_code"),
    "Imports": db.get("imports_tech_code"),
}

# 初始化LCA对象
lca = bw.LCA({grid_activity: 1}, method=("your_lcia_method", "category", "indicator"))
lca.lci()
lca.lcia()

# 获取活动在矩阵中的索引
grid_row_idx = lca.activity_dict[grid_activity.key]
tech_col_indices = {
    name: lca.activity_dict[act.key] for name, act in tech_activity_map.items()
}

# 加载小时级数据
hourly_data = pd.read_csv("your_hourly_data.csv")

# 遍历计算每小时LCIA结果
hourly_scores = []
for _, row in hourly_data.iterrows():
    # 更新技术矩阵对应元素
    for tech_name, amount in row.items():
        if tech_name == "Hour":
            continue
        lca.technosphere_matrix[grid_row_idx, tech_col_indices[tech_name]] = amount
    # 重新计算LCI与LCIA
    lca.redo_lci()
    lca.redo_lcia()
    hourly_scores.append({"Hour": row["Hour"], "LCIA_Score": lca.score})

# 整理输出结果
result_df = pd.DataFrame(hourly_scores)
print(result_df)

3. 批量预加载数据 #

提前将所有需要修改的活动、交换对象加载到内存，避免每次迭代从数据库重复查询，减少IO开销。

4. 并行计算加速 #

针对数千小时的计算任务，可使用多进程并行处理，每个进程独立初始化LCA对象并计算单小时结果：

from multiprocessing import Pool
import brightway2 as bw
import pandas as pd

def compute_single_hour(args):
    row, grid_code, tech_code_map, lcia_method = args
    # 进程内初始化Brightway环境
    bw.projects.set_current("your_project_name")
    db = bw.Database("your_database_name")
    grid_act = db.get(grid_code)
    tech_acts = {name: db.get(code) for name, code in tech_code_map.items()}
    
    # 初始化LCA
    lca = bw.LCA({grid_act: 1}, method=lcia_method)
    lca.lci()
    lca.lcia()
    
    # 获取矩阵索引
    grid_row = lca.activity_dict[grid_act.key]
    tech_cols = {name: lca.activity_dict[act.key] for name, act in tech_acts.items()}
    
    # 更新矩阵并计算
    for tech_name, amount in row.items():
        if tech_name == "Hour":
            continue
        lca.technosphere_matrix[grid_row, tech_cols[tech_name]] = amount
    lca.redo_lci()
    lca.redo_lcia()
    
    return row["Hour"], lca.score

# 准备计算参数
grid_code = "grid_activity_code"
tech_code_map = {
    "Biomass": "biomass_tech_code",
    "Natural Gas": "natural_gas_tech_code",
    "Hydro": "hydro_tech_code",
    "Solar": "solar_tech_code",
    "Wind": "wind_tech_code",
    "Imports": "imports_tech_code",
}
lcia_method = ("your_lcia_method", "category", "indicator")
hourly_data = pd.read_csv("your_hourly_data.csv")
args_list = [(row, grid_code, tech_code_map, lcia_method) for _, row in hourly_data.iterrows()]

# 并行计算
with Pool(processes=4) as pool:
    results = pool.map(compute_single_hour, args_list)

# 整理并排序结果
result_df = pd.DataFrame(results, columns=["Hour", "LCIA_Score"]).sort_values("Hour")
print(result_df)

内容的提问来源于stack exchange，提问作者Carlos Hernandez