放电曲线数字化：用 Python 解析电池测试仪数据并计算容量与能量

专栏：《电池上云笔记》第 2 篇 ｜ 分类：物联网
关键词：放电曲线、电池测试仪、CSV 解析、数据清洗、梯形积分、容量 mAh、能量 mWh、pandas、matplotlib

一、从一堆 CSV 说起

在上一篇导读里我们说过：电池最有价值的原始数据，是一条条放电曲线。但现实中，这些曲线通常是这样的：

• 测试仪每个通道导出一个 CSV，文件名是 CH01_20261020.csv，三个月后谁也不记得是哪个型号、什么负载；
• 文件是 GBK 编码，用 UTF-8 一读就乱码；
• 表头前面还有好几行"设备型号、通道、负载模式"之类的信息，pd.read_csv 直接报错；
• 不同品牌的测试仪，列名有的叫"电压(V)“，有的叫"Voltage”，电流单位有的是 mA、有的是 A；
• 数据里夹着重复行、丢点、接触不良造成的电压尖峰；
• 最后，工程师在 Excel 里手动找"电压第一次低于 0.9 V 的那一行"，再用"平均电流 × 时间"估算容量。

这样做一两个文件还行，一旦要对比几十个批次、几种负载，或者要把数据上传到云端做长期分析，就必须把这个过程变成可重复、可验证的代码。

本文的目标：写一个约 130 行的 Python 脚本，把测试仪原始 CSV 变成可计算、可对比、可上云的结构化结果：

原始 CSV ──► ① 读取（编码/表头/列名/单位自适应）
         ──► ② 清洗（去重、去缺失、剔除尖峰）
         ──► ③ 截止判定（连续 N 点确认）
         ──► ④ 积分（梯形法算 mAh、mWh、平均电压）
         ──► ⑤ 输出（汇总表 CSV/JSON + 对比曲线图）

说明：为了让大家都能直接运行，本文使用模拟生成的测试仪数据（文末附生成脚本），它刻意加入了真实数据中常见的"脏数据"。曲线形态参考 AA 碱性（LR6）与碳性（R6P）电池的典型特征，数值仅用于演示方法，不代表任何具体产品的性能。

二、测试数据长什么样

我们模拟了 4 组恒阻放电测试，截止电压设为 0.80 V（比分析用的 0.9 V 更低，留出余量）：

用文本编辑器打开（GBK 编码），前几行是这样的：

设备型号,BT-8CH
通道,CH01
负载模式,恒阻 10 Ω
截止电压,0.80 V

记录序号,测试时间,电压(V),电流(mA),温度(℃),工步
1,2026/10/20 09:00:10,1.5821,158.72,25.0,恒阻放电
2,2026/10/20 09:00:20,1.5794,159.08,25.3,恒阻放电

每个文件里都埋了这些"坑"：20 行重复记录、约 0.5% 的丢点、6 个接触不良造成的电压尖峰（瞬间跌落 0.3~0.9 V）、采样时间抖动、2 mV 级别的电压噪声。

为什么用恒阻放电？因为它最接近真实用电器（手电筒、玩具、遥控器）的负载特性：电压下降时电流也随之下降。IEC 60086-2 中各类应用放电测试也大多以恒阻负载规定。恒流、恒功率测试同样适用本文方法，只是电流列不再变化（或按功率变化）。

三、核心代码逐段讲解

3.1 读取：编码、表头、列名、单位"四自适应"

COLUMN_ALIASES = {
    "time":    ["测试时间", "绝对时间", "时间", "Time", "DateTime"],
    "voltage": ["电压(V)", "电压/V", "电压(mV)", "Voltage(V)", "Voltage"],
    "current": ["电流(mA)", "电流(A)", "电流/mA", "Current(mA)", "Current(A)"],
    "temp":    ["温度(℃)", "温度", "Temperature"],
}

思路是：

1. 编码：依次尝试 utf-8-sig → gbk → utf-16，能解码就用；
2. 表头：逐行扫描，第一个包含"电压列别名"的行就是真正的列名行，之前的行解析成元数据（设备型号、负载模式等，后面会写进结果）；
3. 列名：用别名字典映射到统一的 time / voltage / current / temp；
4. 单位：根据列名里的 mV、(A) 自动换算成 V 和 mA。

以后接入新品牌的测试仪，通常只需要往别名字典里加一个词，不用改逻辑。

3.2 清洗：去重、去缺失、剔除尖峰

med = df["voltage"].rolling(window, center=True, min_periods=1).median()
spike = (df["voltage"] - med).abs() > spike_mv / 1000

尖峰检测用的是滑动中位数：每个点与它前后 9 个点的中位数比较，偏差超过 50 mV 就判定为尖峰。用中位数而不是均值，是因为中位数不会被尖峰本身拉偏。

阈值怎么选？放电末期电压下降很快，但 10 s 采样间隔内的正常变化通常只有几 mV 到十几 mV，50 mV 足以区分正常下降与接触不良。若采样间隔很长（如 5 分钟），需要适当放宽。

3.3 截止判定：连续 N 点确认

def cutoff_index(v, cutoff, n_confirm=3):
    below = (v < cutoff).to_numpy()
    run = 0
    for k, b in enumerate(below):
        run = run + 1 if b else 0
        if run == n_confirm:
            return k - n_confirm + 1
    return len(v) - 1

"电压第一次低于 0.9 V"听起来简单，但只要有一个尖峰，就会把截止点提前好几个小时。后面的实验会看到这个错误有多严重。连续 3 点确认 + 尖峰剔除，是双保险。

3.4 积分：梯形法算容量和能量

容量和能量的定义：

$$Q=\frac{1}{3600}\int_0^{t_c} I(t)\,dt \quad (\text{mAh}) \qquad E=\frac{1}{3600}\int_0^{t_c} V(t)\,I(t)\,dt \quad (\text{mWh})$$

离散数据用梯形法近似：相邻两点的电流取平均，再乘以真实时间间隔。注意这里用的是每个点实际的时间差，而不是假设固定 10 s，这样丢点、时间抖动都不会引入额外误差。

平均放电电压则用 E / Q 计算，它比简单的电压算术平均更有物理意义（按电量加权）。

四、完整代码

依赖：pandas、numpy、matplotlib。把测试数据放在 data/ 目录下，直接运行即可。

"""放电曲线数字化：解析测试仪 CSV → 清洗 → 截止判定 → 容量/能量计算 → 出图"""
import io, json, re
from pathlib import Path
import numpy as np
import pandas as pd

# ---------- 1. 读取：自动识别编码、跳过表头信息区、统一列名与单位 ----------
COLUMN_ALIASES = {
    "time":    ["测试时间", "绝对时间", "时间", "Time", "DateTime"],
    "voltage": ["电压(V)", "电压/V", "电压(mV)", "Voltage(V)", "Voltage"],
    "current": ["电流(mA)", "电流(A)", "电流/mA", "Current(mA)", "Current(A)"],
    "temp":    ["温度(℃)", "温度", "Temperature"],
}

def read_raw(path):
    raw = Path(path).read_bytes()
    for enc in ("utf-8-sig", "gbk", "utf-16"):
        try:
            text = raw.decode(enc); break
        except UnicodeDecodeError:
            continue
    else:
        raise ValueError(f"无法识别编码：{path}")
    lines = text.splitlines()
    # 表头前的"设备信息区"解析成元数据，找到真正的列名行
    meta, header_idx = {}, None
    for idx, line in enumerate(lines):
        cells = [c.strip() for c in line.split(",")]
        if any(a in cells for a in COLUMN_ALIASES["voltage"]):
            header_idx = idx; break
        if len(cells) >= 2 and cells[0]:
            meta[cells[0]] = cells[1]
    if header_idx is None:
        raise ValueError(f"找不到电压列：{path}")
    df = pd.read_csv(io.StringIO("\n".join(lines[header_idx:])))
    return df, meta

def normalize(df):
    out = pd.DataFrame()
    for std, aliases in COLUMN_ALIASES.items():
        col = next((a for a in aliases if a in df.columns), None)
        if col is None:
            continue
        s = df[col]
        if std == "time":
            out["time"] = pd.to_datetime(s)
        elif "mV" in col:                     # 统一：电压 V、电流 mA
            out[std] = s / 1000
        elif col.endswith("(A)"):
            out[std] = s * 1000
        else:
            out[std] = s
    return out

# ---------- 2. 清洗：排序去重、去缺失、剔除尖峰 ----------
def clean(df, spike_mv=50, window=9):
    n0 = len(df)
    df = df.dropna(subset=["time", "voltage", "current"])
    df = df.drop_duplicates(subset="time").sort_values("time").reset_index(drop=True)
    df["t_s"] = (df["time"] - df["time"].iloc[0]).dt.total_seconds()
    # 与滑动中位数偏差超过阈值的点视为接触不良尖峰
    med = df["voltage"].rolling(window, center=True, min_periods=1).median()
    spike = (df["voltage"] - med).abs() > spike_mv / 1000
    df = df[~spike].reset_index(drop=True)
    stats = dict(raw=n0, duplicates_or_nan=n0 - len(df) - int(spike.sum()), spikes=int(spike.sum()), kept=len(df))
    return df, stats

# ---------- 3. 截止判定：连续 N 点低于截止电压才算真正到达 ----------
def cutoff_index(v, cutoff, n_confirm=3):
    below = (v < cutoff).to_numpy()
    run = 0
    for k, b in enumerate(below):
        run = run + 1 if b else 0
        if run == n_confirm:
            return k - n_confirm + 1
    return len(v) - 1                           # 未到截止：返回最后一点

# ---------- 4. 积分：梯形法计算容量与能量 ----------
def integrate(df):
    t, i, v = df["t_s"].to_numpy(), df["current"].to_numpy(), df["voltage"].to_numpy()
    dt = np.diff(t)
    q = np.concatenate([[0], np.cumsum((i[1:] + i[:-1]) / 2 * dt)]) / 3600              # mAh
    e = np.concatenate([[0], np.cumsum((v[1:]*i[1:] + v[:-1]*i[:-1]) / 2 * dt)]) / 3600  # mWh
    return q, e

def analyze(path, cutoffs=(1.1, 1.0, 0.9)):
    raw, meta = read_raw(path)
    df, stats = clean(normalize(raw))
    q, e = integrate(df)
    df["capacity_mAh"], df["energy_mWh"] = q, e
    result = {"file": Path(path).name, "load": meta.get("负载模式", ""), **stats,
              "v_start_V": round(df["voltage"].iloc[:3].mean(), 3)}
    for vc in cutoffs:
        k = cutoff_index(df["voltage"], vc)
        result[f"t_{vc}V_h"] = round(df["t_s"].iloc[k] / 3600, 2)
        result[f"Q_{vc}V_mAh"] = round(q[k], 0)
        result[f"E_{vc}V_mWh"] = round(e[k], 0)
    k = cutoff_index(df["voltage"], 0.9)
    result["Vavg_0.9V"] = round(e[k] / q[k], 3)       # 平均放电电压 = 能量 / 容量
    return df, result

if __name__ == "__main__":
    import matplotlib
    matplotlib.use("Agg")
    import matplotlib.pyplot as plt
    plt.rcParams["font.sans-serif"] = ["Noto Sans CJK SC", "Noto Sans SC", "SimHei"]
    plt.rcParams["axes.unicode_minus"] = False

    files = sorted(Path("data").glob("*.csv"))
    curves, results = {}, []
    for f in files:
        df, res = analyze(f)
        curves[f.stem] = df
        results.append(res)

    summary = pd.DataFrame(results).set_index("file")
    summary.to_csv("summary.csv", encoding="utf-8-sig")
    Path("summary.json").write_text(json.dumps(results, ensure_ascii=False, indent=2), encoding="utf-8")
    print(summary.T.to_string())

    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(13, 4.8))
    for name, df in curves.items():
        chem, load = name.split("_")
        label = f"{'碱性 LR6' if chem == 'LR6' else '碳性 R6P'} · {load.replace('ohm', ' Ω')}"
        ax1.plot(df["t_s"] / 3600, df["voltage"], label=label, lw=1.4)
        ax2.plot(df["capacity_mAh"], df["voltage"], label=label, lw=1.4)
    for ax, xl, tt in ((ax1, "放电时间 (h)", "电压 – 时间"), (ax2, "放电容量 (mAh)", "电压 – 容量")):
        ax.axhline(0.9, ls="--", c="grey", lw=1); ax.annotate("截止 0.9 V", (0.02, 0.91), xycoords=("axes fraction", "data"), fontsize=9, color="grey")
        ax.set_xlabel(xl); ax.set_ylabel("电压 (V)"); ax.set_title(tt); ax.grid(alpha=.3); ax.set_ylim(0.75, 1.65)
    ax1.set_xscale("log"); ax1.set_xlim(0.1, 120); ax1.set_xlabel("放电时间 (h，对数坐标)")
    ax2.legend(fontsize=9, loc="upper right")
    fig.tight_layout(); fig.savefig("discharge_curves.png", dpi=150)

五、运行结果

5.1 数据清洗统计

埋进去的 20 行重复和 6 个尖峰，全部被准确识别。

5.2 容量与能量汇总

[放电曲线对比图]

左图横轴是时间（对数坐标），右图横轴是放电容量。做对比分析时，"电压–容量"图通常比"电压–时间"图更有用：它消除了负载大小对时间轴的影响，可以直接看出"同样放出 1000 mAh 时，谁的电压更高"。

5.3 从数据中读出的三个结论

1. 放电倍率对容量影响巨大：同一款碱性电池，43 Ω 轻载能放出 2651 mAh，3.9 Ω 重载只有 1568 mAh，相差约 41%。所以只写"容量 xxxx mAh"而不写负载条件，是没有意义的。
2. 截止电压同样关键：10 Ω 负载下，截止电压从 1.1 V 降到 0.9 V，容量从 1963 mAh 增加到 2138 mAh（约 +9%）。这正是续航设计那篇文章里强调的"系统最低工作电压就是免费续航"。
3. 能量比容量更能反映"做功能力"：碳性电池平均电压更低（1.282 V），同样的容量能做的功更少。比较不同化学体系，或者比较 1.5 V 恒压锂电池与碱性电池时，应该看 mWh，而不只是 mAh。

六、两个小实验：数据处理细节到底有多重要？

实验①：不清洗尖峰，会怎样？

对 LR6_10ohm.csv，用三种方式判定 0.9 V 截止点：

一个接触不良的尖峰，就能让容量结果从 2138 mAh 变成 271 mAh，误差接近 90%。如果这是来料抽检，一批合格品就可能被误判为不合格。这也解释了为什么"Excel 里找第一次低于 0.9 V 的行"这种做法风险很大。

实验②：采样间隔和积分方法对结果的影响

对 LR6_3.9ohm.csv 降采样，比较梯形法与左矩形法（"电流 × 时间间隔"直接累加）：

两点启示：

• 梯形法的误差约为矩形法的一半，代码复杂度几乎一样，没理由不用；
• 误差主要来自截止点"过冲"：采样越稀，判定到截止时实际已经多放了一段。重载测试建议采样间隔不超过 60 s；如果采样间隔受限，可以在截止点前后做线性插值，进一步减小误差。

七、工程化建议：让结果"可上云"

脚本最后输出了 summary.csv 和 summary.json，这是为后续上云准备的。实际落地时建议：

1. 文件命名遵循"一物一码"：比如 R03P-261102-L03_CR10_CH01.csv（批次 + 负载 + 通道），和导读篇里的编码体系对齐，后续追溯才能自动关联；
2. 原始文件和结果分开存：原始 CSV 原样上传对象存储（OBS），按"型号/日期/产线"分目录；汇总结果写入关系数据库，便于查询和统计；
3. 把处理参数记录下来：尖峰阈值、确认点数、截止电压都应写进结果，否则半年后没人能复现当时的数字；
4. 加一道合理性校验：例如容量超过该型号历史均值 ±3σ 时标记为"待复核"，而不是直接入库。

下一周我们会把这些结果接入华为云 IoTDA 和 OBS，形成完整的数据链路。

八、总结

放电曲线是电池的"体检报告"。把它从 Excel 里解放出来、变成可计算的数据，是电池数字化最小、也最值得先做的一步。

下一篇预告：有了放电曲线，能不能反推出电池内部的"参数"？11/03 我们用 Python 实现电池一阶 RC 等效电路模型，做参数辨识与仿真。

如果你手里有不同品牌测试仪导出的数据格式，欢迎在评论区贴出表头（脱敏即可），我会把列名补充进别名字典。

附录：模拟数据生成脚本

想复现本文结果，先运行此脚本生成 data/ 目录下的 4 个 CSV 文件（随机种子固定为 42）：

"""生成模拟的电池恒阻放电测试仪导出数据（GBK 编码、中文表头、含噪声/缺失/重复/尖峰）"""
import numpy as np, pandas as pd
from pathlib import Path

rng = np.random.default_rng(42)

CHEM = {
    # 开路电压曲线 E(dod)、内阻 R(dod)、低倍率容量 Q0、倍率衰减系数 I_ref
    "LR6": dict(E=lambda d: 1.60 - 0.32*d - 0.22*d**4 - 0.15*np.exp((d-1)/0.04),
                R=lambda d: 0.12 + 0.30*d**2, Q0=2900, I_ref=420, k=0.9),
    "R6P": dict(E=lambda d: 1.62 - 0.50*d - 0.15*d**3 - 0.15*np.exp((d-1)/0.05),
                R=lambda d: 0.35 + 0.90*d**2, Q0=1500, I_ref=110, k=0.8),
}

def simulate(chem, r_load, dt=10, cutoff=0.80, t0="2026-10-20 09:00:00"):
    c = CHEM[chem]
    i_est = 1.3 / r_load * 1000                              # 估计平均电流 mA
    q_eff = c["Q0"] / (1 + (i_est / c["I_ref"]) ** c["k"])   # 倍率相关有效容量
    q, t, rows = 0.0, 0, []
    while True:
        d = min(q / q_eff, 1.08)
        e = c["E"](d); r = c["R"](d)
        v = e * r_load / (r_load + r)
        i = v / r_load * 1000
        rows.append((t, v, i))
        if v < cutoff - 0.02:
            break
        q += i * dt / 3600; t += dt
    df = pd.DataFrame(rows, columns=["t", "v", "i"])
    # —— 加入真实测试仪常见的"脏数据" ——
    df["t"] = df["t"] + rng.normal(0, 0.3, len(df)).round(1)          # 采样时间抖动
    df["v"] = df["v"] + rng.normal(0, 0.002, len(df))                 # 电压噪声 2 mV
    df["i"] = df["i"] * (1 + rng.normal(0, 0.003, len(df)))           # 电流噪声
    spikes = rng.choice(len(df), 6, replace=False)                    # 接触不良尖峰
    df.loc[spikes, "v"] -= rng.uniform(0.3, 0.9, 6)
    df = df.drop(rng.choice(df.index, len(df)//200, replace=False))   # 0.5% 丢点
    df = pd.concat([df, df.sample(20, random_state=1)]).sort_index()  # 重复行
    temp = 25 + rng.normal(0, 0.3, len(df))
    ts = pd.Timestamp(t0) + pd.to_timedelta(df["t"], unit="s")
    out = pd.DataFrame({
        "记录序号": range(1, len(df) + 1),
        "测试时间": ts.dt.strftime("%Y/%m/%d %H:%M:%S"),
        "电压(V)": df["v"].round(4),
        "电流(mA)": df["i"].round(2),
        "温度(℃)": temp.round(1),
        "工步": "恒阻放电",
    })
    return out

if __name__ == "__main__":
    for chem, r in [("LR6", 3.9), ("LR6", 10), ("LR6", 43), ("R6P", 10)]:
        df = simulate(chem, r)
        name = f"data/{chem}_{r}ohm.csv"
        with open(name, "w", encoding="gbk", newline="") as f:
            f.write(f"设备型号,BT-8CH\n通道,CH01\n负载模式,恒阻 {r} Ω\n截止电压,0.80 V\n\n")
            df.to_csv(f, index=False)
        print(name, len(df), "行")

另附实验脚本 experiments.py 的核心逻辑：实验①分别对未清洗数据用 n_confirm=1 和 n_confirm=3 调用 cutoff_index；实验②对清洗后的数据按 iloc[::step] 降采样，再分别用梯形法和 np.sum(i[:k] * np.diff(t[:k+1])) 的左矩形法计算容量并与 10 s 基准对比。

参考资料

1. IEC 60086-1 / IEC 60086-2：原电池通用要求与物理、电气规格（含应用放电测试条件）
2. pandas 官方文档：read_csv、rolling、to_datetime
3. 《电池上云笔记》第 1 篇：电池行业数字化全景——从电芯到云端的数据链路一图读懂