启程研习社

Appearance

第 12 章 · 文件读写与 csv 模块 ​

这一章要解决什么：第一次读"真实存在的数据"—— 把 cycle_raw.csv 的 13,967 行读进程序并统计出"一共 12 支电芯"。 四个关键坑：相对路径是相对"当前工作目录"而不是脚本文件、忘了 encoding="utf-8"、 读进来全是字符串、写 CSV 忘了 newline="" 会每行多一个空行。

本章你需要完成的代码：demos/ch12-csv.py　前置：数据已在下载包里（data/battery/cycle_raw.csv），无需生成。

12.1 路径是第一大坑：相对路径相对于谁？ ​

python
with open("data/battery/cycle_raw.csv") as f:      # ❌ 只在 code 目录下运行才对
    ...

它能跑通只是因为"运气好" —— 你恰好在 code 目录下运行。换台机器、换个终端目录、 或者从 IDE 里点"运行"，就会得到 FileNotFoundError: [Errno 2] No such file or directory: 'data/battery/cycle_raw.csv'。

根本原因：相对路径是相对当前工作目录（CWD），不是相对脚本文件所在目录， 而这两个目录大多数时候不一样。正确做法是用 __file__ 反推：

python
from pathlib import Path

CODE_DIR = Path(__file__).resolve().parent.parent     # demos/ch12-csv.py → code/
DATA_DIR = CODE_DIR / "data" / "battery"
CSV_PATH = DATA_DIR / "cycle_raw.csv"
print(CSV_PATH.exists())        # True
写法从哪算起换目录运行
"data/battery/cycle_raw.csv"当前工作目录❌ 找不到
Path(__file__).resolve().parent.parent / "data" / ...脚本自己✅ 永远对

记忆法：__file__ = "我（这个脚本）在哪"。只要文件在项目里，就用它定位，别写裸的相对路径。

12.2 pathlib.Path：比字符串拼路径好用的多 ​

python
p = Path("data/battery/cycle_raw.csv")
p.name          # 'cycle_raw.csv'      文件名
p.suffix        # '.csv'               后缀
p.parent        # 'data/battery'       父目录
p.exists()      # True / False
p.stat().st_size / 1024                 # 738 KB   文件大小
想做的事写法别写
拼路径CODE_DIR / "data" / "battery"❌ CODE_DIR + "/data" + "/battery"
取父目录 / 后缀p.parent / p.suffix❌ 字符串切片
判断存在p.exists()❌ os.path.exists（能用，但风格混杂）
建目录 / 列目录OUT_DIR.mkdir(exist_ok=True)❌ 建多层目录要加 parents=True

为什么用 Path：/ 运算符自动处理分隔符（Windows 上是 \），属性名不用背 os.path 那一串函数 —— 本书统一用 pathlib。注意写文件前别忘了 mkdir，否则 open(OUT_DIR / "x.csv", "w") 直接 FileNotFoundError。

12.3 with open(...)：出了代码块自动关文件 ​

python
with open(CSV_PATH, encoding="utf-8") as f:
    lines = [line.rstrip("\n") for line in f]
# 出了缩进块，文件自动关闭 —— 不用手写 f.close()
参数作用常用值
mode怎么开"r" 读（默认）、"w" 覆写、"a" 追加
encoding文字编码永远写 "utf-8"
newline换行处理写 CSV 时写 ""

为什么一定要写 encoding="utf-8"：不写就用操作系统的默认编码 —— Linux/macOS 上是 UTF-8（所以你不会发现问题），Windows 上是 GBK， 于是同一份含中文的 CSV 在同事的 Windows 上直接 UnicodeDecodeError: 'gbk' codec can't decode byte ...。 写不写编码是"能不能跨平台"的分界线。 另外注意 w 模式会清空文件重写（最容易造成数据丢失的一行），想追加用 "a"。

12.4 csv.DictReader：每行一个字典（和第 09 章接上了） ​

python
import csv

with open(CSV_PATH, encoding="utf-8") as f:
    reader = csv.DictReader(f)
    rows = list(reader)

len(rows)              # 13967
rows[0]                # {'cell_id': 'C001', 'batch': 'A', 'soh': '0.9999', ...}
reader.fieldnames      # 表头：['cell_id', 'batch', 'cycle', ..., 'coulombic_eff']

"Dict"就是第 09 章的字典 —— 表头自动变成键，每一行的内容变成值。

csv.DictReadercsv.reader
每行拿到字典（按列名取值）列表（按位置取值）
取值row["soh"]row[5]
加一列会不会错位不会会（位置全变）
什么时候用默认选它表头是数据、或要按位置处理
点开：为什么"按列名取值"更安全？

真实场景：上游系统在中间插了一列 device_id。用位置取值的 row[5] 不会报错， 只会悄悄把别的字段当成 soh 算 —— 最典型的一类静默 bug。按名字取值，改名时会直接 KeyError（该崩就崩），插列时毫无影响。 另外 fieldnames 还能检查表头有没有被改 （第 04 章"把期望写进代码"）：assert reader.fieldnames == EXPECT, f"表头变了：{reader.fieldnames}"。

12.5 读进来全是字符串（为第 23 章埋伏笔） ​

python
v = rows[0]["soh"]
type(v)          # str
v                # '0.9999'    ← 带引号！    v * 2 = '0.99990.9999' ← 重复，不是乘法！
v - 0.01         # TypeError: unsupported operand type(s) for -: 'str' and 'float'
float(v) - 0.01  # 0.9899    ← 想算就得先转类型

CSV 文件里没有"类型"这个概念，一切都是文本。 所以读出来全是 str， 想要数字必须自己转：soh = float(row["soh"])，整列转就 sohs = [float(r["soh"]) for r in rows]（推导式，第 11 章）。 注意 v * 2 不报错 —— 字符串的 * 是"重复"，这和第 03 章的坑一模一样： 它不会崩，只会给你一串长度翻倍的乱码，所以比 TypeError 更危险。

第 23 章会有转机：Pandas 的 pd.read_csv() 会自动推断类型，数值列直接变成 float64， 不用手写 float()。这就是本书从 csv 模块转向 Pandas 的最大动机。

12.6 用 set 统计"一共多少支电芯" ​

python
ids_set = {r["cell_id"] for r in rows}    # 集合版：一行，O(n)   → len(ids_set) == 12

ids_loop = []                             # 列表版：O(n²)，数据一大就慢
for r in rows:
    if r["cell_id"] not in ids_loop:
        ids_loop.append(r["cell_id"])     # 结果一样是 12，但慢得多

13,967 行数据，只要 12 支不同电芯 —— 这个"记录数 ≠ 实体数"的区别， 是第 09 章讲过的口径问题，也是写报告时最常被问到的数。集合版快在 O(n) vs O(n²)： 数据量再大十倍，差距就从"感觉不到"变成"要等一分钟"。

12.7 写 CSV：newline="" 不能忘 ​

python
OUT_DIR = CODE_DIR / "outputs"
OUT_DIR.mkdir(exist_ok=True)               # 目录不存在时，写文件会失败

with open(OUT_DIR / "ch12_per_cell.csv", "w", encoding="utf-8", newline="") as f:
    writer = csv.writer(f)
    writer.writerow(["cell_id", "n_cycles", "soh_min", "soh_max", "soh_mean"])
    for cid in sorted(per_cell):
        writer.writerow([cid, len(per_cell[cid]), min(per_cell[cid]), ...])

三个参数一个都不能少：mode="w" 是清空重写（要追加用 "a"）、encoding="utf-8"、 newline=""。newline="" 的作用：csv 模块要自己负责换行，如果 open 又把 \n 翻译成 \r\n，就会变成 \r\r\n —— 在 Excel 里表现为"每隔一行一个空行"。 记住结论：写 CSV 时 open(..., newline="")，一个字母都不能少。 另外注意 sorted(per_cell)：字典保留插入顺序，但"读进来的顺序"取决于文件， 要让输出稳定（可对比、可 diff），就显式排序（第 09 章那条纪律）。

12.8 写 / 读 JSON：结构化数据的通用交换格式 ​

python
import json

report = {"source": "cycle_raw.csv", "n_rows": len(rows),
          "n_cells": len(ids_set), "batches": sorted({r["batch"] for r in rows})}

json_path = OUT_DIR / "ch12_report.json"
json_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")

data = json.loads(json_path.read_text(encoding="utf-8"))    # 读回来还是字典

CSV 是表格（行 × 列，全是文本），JSON 是嵌套结构（字典/列表随便套，有类型）—— 所以配置、接口响应、结果报告都用 JSON。 两个必写参数：ensure_ascii=False（不然中文变成 \u7535\u82af）、 indent=2（不然挤成一行，没法 diff）。JSON 天然是"字典 + 列表"的组合， 第 09 章学的 .get() 在这里全部派上用场（data.get("n_cells", 0)）。

12.9 其他常用文件操作 ​

python
[p.name for p in DATA_DIR.iterdir()]        # ['README.md', 'dirty.csv', 'cells.csv', 'cycle_raw.csv']
CSV_PATH.stat().st_size / 1024              # 738.0   KB
CSV_PATH.name, CSV_PATH.suffix              # 'cycle_raw.csv', '.csv'
OUT_DIR.mkdir(exist_ok=True)                # 已存在时不报错

iterdir() 的顺序取决于操作系统（不保证字母序），所以输出前要 sorted()， 否则你的脚本在不同机器上会给出不同的输出顺序。

12.10 完整示例与真实输出 ​

运行 cd code && python demos/ch12-csv.py，真实输出（节选，绝对路径前缀已省略为 …/code）：

text
=== ② with open 读文本 ===
   cell_id,batch,cycle,timestamp,capacity_ah,soh,resistance_mohm,temp_rise_c,coulombic_eff
   C001,A,1,2023-01-01,2.5092,0.9999,40.33,6.0,0.998
  ↑ encoding='utf-8' 必须写：Windows 默认 GBK

=== ③ csv.DictReader：每行一个字典 ===
  总行数: 13,967    第一行: {'cell_id': 'C001', 'batch': 'A', 'soh': '0.9999', ...}

=== ④ 读进来全是字符串 ===
  type(r['soh']) = str   值 = '0.9999'
  字符串 * 2 = '0.99990.9999'  ← 是「重复」，不是乘法！
  字符串 - 数字 → TypeError: unsupported operand type(s) for -: 'str' and 'float'

=== ⑥ 统计不同电芯数 ===
  循环版: 12 支     集合版: 12 支  ['C001', ..., 'C012']

注意第 ④ 块的两个输出：'0.9999' 带引号说明它是字符串； '0.99990.9999' 是字符串乘法（重复），这两行是本章最该记住的证据。

常见坑 ​

报错 / 现象原因怎么改
FileNotFoundError: ... 'data/battery/cycle_raw.csv'相对路径相对当前工作目录用 Path(__file__).resolve().parent 反推
UnicodeDecodeError: 'gbk' codec can't decode ...漏了 encoding，Windows 默认 GBKopen(..., encoding="utf-8")（写也要写）
Windows 上 CSV 每行之间多一个空行写文件漏了 newline=""open(..., "w", encoding="utf-8", newline="")
TypeError: ... for -: 'str' and 'float'读进来的数字是字符串入口处统一 float() / int()
'0.99' * 2 得到 '0.990.99'字符串的 * 是重复先 float() 再算，它不报错，最危险
写文件时 FileNotFoundError（目录不存在）输出目录还没建先 OUT_DIR.mkdir(exist_ok=True)
每次运行输出顺序不同字典 / iterdir() 顺序不稳定输出前 sorted(...)

本章练习 ​

练习 1：统计有多少支不同的电芯

读 data/battery/cycle_raw.csv（用 __file__ 反推路径 + with open + csv.DictReader）， 打印：① 读的是哪个文件；② 总行数；③ 不同电芯数和排序后的电芯列表；④ 前 3 行（格式 C001 cycle=1 soh=0.9999）。

期望输出：

text
读的文件: .../code/data/battery/cycle_raw.csv
总行数: 13967
不同电芯数: 12
电芯列表: ['C001', 'C002', 'C003', ..., 'C012']
前 3 行:
  C001  cycle=1  soh=0.9999
参考答案
python
import csv
from pathlib import Path

CODE_DIR = Path(__file__).resolve().parent.parent.parent    # ① 往上三层才是 code/
CSV_PATH = CODE_DIR / "data" / "battery" / "cycle_raw.csv"
print(f"读的文件: {CSV_PATH}")
if not CSV_PATH.exists():
    raise SystemExit(f"数据文件不存在：{CSV_PATH}\n请确认下载包已完整解压。")

with open(CSV_PATH, encoding="utf-8") as f:                 # ② 每行变成一个字典
    rows = list(csv.DictReader(f))

print(f"总行数: {len(rows)}")                                # ③ 行数 + 去重计数
ids_set = {r["cell_id"] for r in rows}                      # 集合推导式：一行、O(n)
print(f"不同电芯数: {len(ids_set)}")
print(f"电芯列表: {sorted(ids_set)}")

print("前 3 行:")                                            # ④ 切片
for r in rows[:3]:
    print(f"  {r['cell_id']}  cycle={r['cycle']}  soh={r['soh']}")

四个要点：① parent.parent.parent 数了几层 —— 文件在 exercises/ex04-12-csv/ 下， 往上三层才是 code/，层数写错是最常见的低级错误，先 print(CODE_DIR) 确认； ② {r["cell_id"] for r in rows} 是集合推导式（第 11 章），一行代替 "建空列表 + 循环 + if not in"，从 O(n²) 降到 O(n)；③ sorted(ids_set) 不能省， set 顺序每次运行都可能不同（第 09 章）； ④ rows[:3] 打印出的 '0.9999' 是字符串 —— 要算就得先 float()。

完整练习在 exercises/ex04-12-csv —— 含骨架 exercise.py 和参考答案 solution.py。

练习 2：算出每支电芯的平均 SOH，并按批次统计

在练习 1 的基础上：① 按 cell_id 分组，算出每支电芯的循环次数和平均 SOH（保留 4 位）； ② 写出 outputs/ch12_exercise.csv（写完打印前 3 行确认）；③ 顺便打印每个批次的记录数。

参考答案
python
import csv
from collections import Counter
from pathlib import Path

CODE_DIR = Path(__file__).resolve().parent.parent.parent
CSV_PATH = CODE_DIR / "data" / "battery" / "cycle_raw.csv"
with open(CSV_PATH, encoding="utf-8") as f:
    rows = list(csv.DictReader(f))

# ① 分组：读进来的全是字符串，所以这里必须 float()
per_cell = {}
for r in rows:
    per_cell.setdefault(r["cell_id"], []).append(float(r["soh"]))   # 第 09 章的 setdefault

summary = {cid: (len(sohs), round(sum(sohs) / len(sohs), 4))
           for cid, sohs in per_cell.items()}                        # 字典推导式（第 11 章）

# ② 写 CSV：mkdir + newline="" 一个都不能少
OUT_DIR = CODE_DIR / "outputs"
OUT_DIR.mkdir(exist_ok=True)
out_path = OUT_DIR / "ch12_exercise.csv"
with open(out_path, "w", encoding="utf-8", newline="") as f:
    writer = csv.writer(f)
    writer.writerow(["cell_id", "n_cycles", "soh_mean"])
    for cid in sorted(summary):
        writer.writerow([cid, *summary[cid]])                        # 星号解包（第 11 章）

print(f"已写出 {out_path}")
print(" ", out_path.read_text(encoding="utf-8").splitlines()[1])   # C001,862,0.9069

# ③ 按批次统计
print("按批次记录数:", dict(sorted(Counter(r["batch"] for r in rows).items())))

四个要点：① 入口处一次性把类型转干净（float(r["soh"])），别在后面的计算里零散地转， 漏一个就是 TypeError（第 03 章的纪律）；② setdefault(cid, []).append(...) 是"分组"最省事的写法， 比 if cid not in d: d[cid] = [] 少两行；③ newline="" + encoding="utf-8" + mkdir(exist_ok=True) 是写 CSV 的固定三件套；④ 写完读回来检查前几行是个好习惯 —— 文件真写对了、顺序是不是预期的，一眼就知道，比等别人打开 Excel 才发现问题便宜得多。

加分：把 soh_mean 和 n_cycles 画成散点图（第 36 章）会看到明显的负相关 —— 循环越多的电芯平均 SOH 越低，这正是这批数据的物理规律。

练习 3：找出这段代码的四处问题

下面这段代码有四处问题（有的一运行就报错，有的在 Windows 上才出问题，有的不报错但结果错）。 先自己找，再对照答案：

python
import csv

with open("data/battery/cycle_raw.csv") as f:
    rows = list(csv.DictReader(f))

with open("outputs/report.csv", "w") as f:
    writer = csv.writer(f)
    writer.writerow(["cell_id", "soh"])
    for r in rows:
        writer.writerow([r["cell_id"], r["soh"] * 2])
参考答案

问题 ①：相对路径。 open("data/battery/cycle_raw.csv") 相对的是当前工作目录， 在别的目录运行时直接 FileNotFoundError。改成用 __file__ 反推： CSV_PATH = Path(__file__).resolve().parent.parent / "data" / "battery" / "cycle_raw.csv"。

问题 ②：漏了 encoding="utf-8"。 Linux/macOS 上碰巧是对的（默认 UTF-8）， Windows 上会乱码或 UnicodeDecodeError。读写两侧都要写 encoding="utf-8"。

问题 ③：写文件缺 newline=""，而且目录没建。open("outputs/report.csv", "w") 在目录不存在时直接 FileNotFoundError； 即使目录在，缺 newline="" 也会让 Windows 上每行多一个空行。 改：OUT_DIR.mkdir(exist_ok=True) + open(out, "w", encoding="utf-8", newline="")。

问题 ④：字符串乘 2 —— 不报错，但结果全错。 r["soh"] * 2 得到 '0.99990.9999'： soh 是字符串，* 2 是"重复"而不是乘法。这一行不会崩溃， 写进 CSV 的就是一列长度翻倍的乱码。改：float(r["soh"]) * 2。

修好后的完整代码：

python
CODE_DIR = Path(__file__).resolve().parent.parent               # ①
CSV_PATH = CODE_DIR / "data" / "battery" / "cycle_raw.csv"
OUT_DIR = CODE_DIR / "outputs"
OUT_DIR.mkdir(exist_ok=True)                                    # ③

with open(CSV_PATH, encoding="utf-8") as f:                     # ②
    rows = list(csv.DictReader(f))

with open(OUT_DIR / "report.csv", "w", encoding="utf-8", newline="") as f:    # ② ③
    writer = csv.writer(f)
    writer.writerow(["cell_id", "soh"])
    for r in rows:
        writer.writerow([r["cell_id"], float(r["soh"]) * 2])    # ④

三个要点：① "不报错但结果错"的问题（④）最值得警惕 —— 前三个都会崩、都有清晰报错，只有它悄悄给你一份错数据； ② ②③ 是"跨平台"问题，在你机器上完全看不出来，只能靠纪律防 （写 open 就把参数补齐：路径、模式、encoding="utf-8"、写 CSV 加 newline=""）； ③ 路径问题（①）只在换个目录运行时暴露，这就是本书所有示例都用 __file__ 的原因。

面向 AI 开发：数据入口决定一切 ​

AI 项目里 80% 的时间花在"把数据正确地读进来"，而 80% 的诡异 bug 也出在这一步。

① 数据入口是唯一该做类型转换的地方。

python
NUM_COLS = ["capacity_ah", "soh", "resistance_mohm", "temp_rise_c", "coulombic_eff"]

def load_rows(path):
    with open(path, encoding="utf-8") as f:
        rows = list(csv.DictReader(f))
    for r in rows:
        for c in NUM_COLS:
            r[c] = float(r[c])          # 就地转成数字，后面所有函数拿到的都是数字
    return rows

为什么必须集中在入口：散落在各处的 float(...) 只要漏一个就是 TypeError； 更糟的是漏掉的 * 变成字符串重复 —— 不报错，数据全错。

② 路径纪律：永远用 __file__ 反推。 ROOT = Path(__file__).resolve().parent.parent。 为什么在 AI 项目里更重要：训练脚本常在三种环境下跑 —— 你的 IDE、 服务器上的 python train.py、容器里被 subprocess 拉起。 只有 __file__ 在这三种情况下都正确，os.getcwd() 和裸相对路径都不是。

③ 编码问题：统一 UTF-8，写死在代码里。 读 CSV / JSON / 文本都用 encoding="utf-8"； 写文件同理，CSV 再加 newline=""；写 JSON 加 ensure_ascii=False。 中文数据踩坑的三种典型形态：UnicodeDecodeError（读）、Excel 打开全是乱码（写）、 \u7535\u82af（JSON 没关 ensure_ascii）—— 三行代码就能全避开。

④ CSV 是最常见的数据交换格式，但只在"表格"这件事上：实验导出、Excel 交换、日志表用 CSV； 配置、接口响应、训练结果用 JSON（能嵌套、有类型）；大模型语料用 JSONL / Parquet； 上百万行反复查询用 Parquet / 数据库（CSV 每次都要重新解析，第 31 章）。

⑤ 这一步到 Pandas 只有一步之遥（第 23 章）：本章是"手工打开文件 → 手工 float()"， 下一部分只要 df = pd.read_csv(CSV_PATH) 一行，类型还会自动推断。 本章学的每一样东西都不会白费：pd.read_csv 内部做的事， 就是"打开文件 → 按分隔符切 → 处理表头 → 推断每列类型 → 存成列式结构" —— 你亲手写过 DictReader + float()，才知道 read_csv 的 dtype 参数在管什么。 而当数据大到一次读不进来时，你也会知道该用 chunksize（第 24 章）， 因为"流式读取"的思想，你在第 11 章的生成器里已经学过一遍了。

下一章：第 13 章 · 异常处理与调试