042 · 软件开发与代码协作

起草测试数据采集与清洗小工具

第 42 / 94 篇免费阅读

这个场景能帮你做什么

测试数据需要重复导入、转换、清洗和画图时，WorkBuddy 可以依据真实样例起草小工具，减少手工搬运。连接设备与执行脚本必须有实际协议、运行环境和授权。

开始前需要准备什么

提供样例文件或设备协议、数据列含义、单位、时间戳、异常码与目标输出。说明只处理导出文件还是允许采集设备，列允许的命令、目录及资源限制。

具体怎么做

第一步：先定义转换规则

说明哪些为有效数据、哪些缺失或异常，是否需要对齐时间和换算单位。异常保留原值与原因，不能默认删掉尖峰和失败样本。

第二步：起草最小工具

从小样本文件解析开始，输出处理明细、异常和核对数。涉及串口或采集接口时，先检查协议与权限，不猜命令控制设备。

第三步：用已知样例验证

准备正常、空值、断行、单位变化和异常码样例，比较预期与实际。允许执行时在副本和测试环境运行；未运行的代码明确为草稿。

第四步：交付使用说明

写清输入格式、运行命令、依赖、输出、失败提示与停止方式。保存原始数据，避免工具直接覆盖采集证据。

用导出的 CSV 做一个最小清洗示例

以下为教学示例，数据与规则均为演示设计，代码未接入设备。先将以下内容保存为 sample.csv。时间戳是带时区的原始记录，本例仅保留，不做重采样或时钟校正。

time,value,unit,status
2026-01-01T10:00:00+08:00,1000,mV,OK
2026-01-01T10:00:01+08:00,1.2,V,OK
2026-01-01T10:00:02+08:00,,V,OK
2026-01-01T10:00:03+08:00,900,mV,E01

规则是：仅将状态为 OK、数值有效且单位为 mV 或 V 的记录换算为伏；mV÷1000，V 原值保留。空值、未知状态和未知单位全部保留原始行并标记，不插值、不当成零、不删除。E01 的具体含义由设备资料解释，本例不猜测。

可让 WorkBuddy 按此规则起草下列 Python 3 小工具，并在有权限的测试目录验证。保存为 clean_test.py，不需要安装额外库：

import csv
import sys
from decimal import Decimal, InvalidOperation
from pathlib import Path

source = Path(sys.argv[1])
target = Path(sys.argv[2])
fields = ["time", "value", "unit", "status"]
rows = []
with source.open(encoding="utf-8-sig", newline="") as handle:
    reader = csv.DictReader(handle)
    if reader.fieldnames != fields:
        raise ValueError("列名或顺序变化，停止处理")
    for number, row in enumerate(reader, start=2):
        if None in row or any(v is None for v in row.values()):
            raise ValueError(f"第 {number} 行列数异常，停止处理")
        reasons = []
        value = None
        if not row["time"].strip():
            reasons.append("缺少时间戳")
        if row["status"] != "OK":
            reasons.append("状态待确认")
        try:
            value = Decimal(row["value"])
            if not value.is_finite():
                reasons.append("数值非有限值")
        except InvalidOperation:
            reasons.append("缺失或非法数值")
        if row["unit"] not in ("V", "mV"):
            reasons.append("未知单位")
        result = ""
        if not reasons:
            result = format(value / 1000 if row["unit"] == "mV" else value, "f")
        rows.append(dict(row, source_line=number, value_V=result,
                         flag="；".join(reasons) or "可用"))

# 输出目录必须尚不存在，避免覆盖原文件或上次结果。
target.mkdir(parents=True, exist_ok=False)
with (target / "processed.csv").open("w", encoding="utf-8-sig", newline="") as handle:
    writer = csv.DictWriter(handle, fieldnames=fields + ["source_line", "value_V", "flag"])
    writer.writeheader()
    writer.writerows(rows)
valid = sum(row["flag"] == "可用" for row in rows)
print(f"输入 {len(rows)} 行；可用 {valid} 行；待确认 {len(rows) - valid} 行")

在两个文件所在目录运行 python3 clean_test.py sample.csv clean_output。如果输出目录已存在，先选一个新的目录名；不要为了重跑而覆盖原始数据。预期控制台显示“输入 4 行；可用 2 行；待确认 2 行”，输出保留全部原始字段，并新增来源行号、换算值和标记：

来源行号	原始数值及单位	value_V	flag
2	1000 mV	1	可用
3	1.2 V	1.2	可用
4	空值，V	空	缺失或非法数值
5	900 mV，E01	空	状态待确认

先手工核对这四行，再换成真实文件副本。该最小版本不校验时间戳格式、不判断重复采样，也不识别业务阈值和离群点；有这些需求时先补规则和测试样例。格式不符会报错停止；它不会连接仪器，也不提供设备通信能力。

可以直接交给 WorkBuddy 的完整要求

根据附件样例与规则起草测试数据处理小工具，原文件只读，输出到独立目录。
保留原值、单位、来源和时间，缺失/异常单列，输出输入行数、成功、失败与去向核对。
用正常及异常样例验证，仅执行获授权测试环境命令。未运行明确说明。
涉及设备协议先列接入条件，不猜控制命令。交付代码、用法、限制和错误处理，不宣称已连接未测试设备。

完成后会得到什么

一份可审阅的小工具代码、样例处理结果或预期，以及独立运行所需说明。

结果应该怎么检查

测试人员手工复算几组值，核对时间、单位、异常与行数去向。检查清洗有没有把真实失效当噪声抹掉。

遇到这些情况怎么处理

数据格式变化时停止对应解析并提示。设备通信不稳定时先排查采集条件，不伪造连续数据。量纲不明时保留原值，不换算。