#!/usr/bin/env python3 """解析正点原子逻辑分析仪导出的 CSV 文件。 文件开头以分号开头的说明行会被忽略。输出文件默认不写表头， 每行格式为：时间,电平 示例： python parse_logic_csv.py input.csv output.csv --channel CH1 --mode changes python parse_logic_csv.py input.csv output.csv --channel 2 --mode all """ from __future__ import annotations import argparse import csv import io import sys from pathlib import Path from typing import Iterable, Sequence TIME_NAMES = {"time", "timestamp", "时间", "时刻", "t"} LEVEL_NAMES = { "0": "0", "1": "1", "low": "0", "l": "0", "false": "0", "off": "0", "低": "0", "低电平": "0", "high": "1", "h": "1", "true": "1", "on": "1", "高": "1", "高电平": "1", } class CsvParseError(ValueError): """CSV 格式或参数不符合预期。""" def read_text(path: Path, encoding: str) -> str: """读取文本，utf-8 失败时自动尝试常见的 GBK 编码。""" if encoding != "auto": return path.read_text(encoding=encoding) data = path.read_bytes() for candidate in ("utf-8-sig", "gbk", "utf-16"): try: return data.decode(candidate) except UnicodeDecodeError: continue raise CsvParseError(f"无法识别文件编码：{path}") def remove_comment_lines(text: str) -> str: """删除空行和以分号开头的逻辑分析仪说明行。""" lines = [] for line in text.splitlines(): if not line.strip() or line.lstrip().startswith(";"): continue lines.append(line) return "\n".join(lines) def detect_delimiter(text: str) -> str: """识别逗号、制表符或分号分隔格式。""" sample_lines = text.splitlines()[:20] sample = "\n".join(sample_lines) try: dialect = csv.Sniffer().sniff(sample, delimiters=",\t;") return dialect.delimiter except csv.Error: counts = {delimiter: sample.count(delimiter) for delimiter in (",", "\t", ";")} delimiter = max(counts, key=counts.get) if counts[delimiter] == 0: raise CsvParseError("无法识别 CSV 分隔符，请使用逗号、制表符或分号") return delimiter def parse_level(value: str) -> str: """把常见的高低电平写法统一为 0 或 1。""" text = value.strip().lower() if text in LEVEL_NAMES: return LEVEL_NAMES[text] try: number = float(text) except ValueError as exc: raise CsvParseError(f"无法识别电平值：{value!r}") from exc if number == 0: return "0" if number == 1: return "1" raise CsvParseError(f"电平必须为0或1，实际得到：{value!r}") def find_column(header: Sequence[str], name: str, description: str) -> int: """按列名或列序号查找列，列序号从0开始。""" name = name.strip() if name.isdigit(): index = int(name) if 0 <= index < len(header): return index raise CsvParseError(f"{description}列序号超出范围：{index}") lowered = [item.strip().lower() for item in header] target = name.lower() if target in lowered: return lowered.index(target) choices = ", ".join(header) raise CsvParseError(f"找不到{description}列 {name!r}，可用列：{choices}") def find_time_column(header: Sequence[str]) -> int: """优先查找常见时间列，找不到时默认使用第0列。""" for index, name in enumerate(header): if name.strip().lower() in TIME_NAMES: return index return 0 def parse_rows(text: str, channel: str, time_column: str | None) -> list[tuple[str, str]]: """读取 CSV 并返回有效的（时间，电平）采样点。""" clean_text = remove_comment_lines(text) if not clean_text: raise CsvParseError("文件中没有有效数据") delimiter = detect_delimiter(clean_text) rows = list(csv.reader(io.StringIO(clean_text), delimiter=delimiter)) if not rows or len(rows[0]) < 2: raise CsvParseError("CSV 至少需要时间列和一个通道列") header = [item.strip() for item in rows[0]] channel_column = find_column(header, channel, "通道") time_index = ( find_column(header, time_column, "时间") if time_column is not None else find_time_column(header) ) points: list[tuple[str, str]] = [] for line_number, row in enumerate(rows[1:], start=2): if len(row) <= max(channel_column, time_index): raise CsvParseError(f"第{line_number}行列数不足") time_value = row[time_index].strip() if not time_value: raise CsvParseError(f"第{line_number}行时间为空") points.append((time_value, parse_level(row[channel_column]))) if not points: raise CsvParseError("CSV 表头后没有采样数据") return points def compress_level_changes(points: Iterable[tuple[str, str]]) -> list[tuple[str, str]]: """只保留首个采样点和电平发生变化的采样点。""" output: list[tuple[str, str]] = [] previous_level: str | None = None for point in points: if previous_level is None or point[1] != previous_level: output.append(point) previous_level = point[1] return output def write_points(path: Path, points: Iterable[tuple[str, str]]) -> int: """写出时间、电平两列，不写表头。""" count = 0 with path.open("w", encoding="utf-8", newline="") as file: writer = csv.writer(file, lineterminator="\n") for time_value, level in points: writer.writerow((time_value, level)) count += 1 return count def build_argument_parser() -> argparse.ArgumentParser: """创建命令行参数。""" parser = argparse.ArgumentParser( description="解析正点原子逻辑分析仪 CSV，输出时间和电平两列" ) parser.add_argument("input", type=Path, help="逻辑分析仪导出的 CSV 文件") parser.add_argument("output", type=Path, help="输出 CSV 文件") parser.add_argument( "--channel", required=True, help="通道名称，例如 CH1、D0；也可以填写列序号，序号从0开始", ) parser.add_argument( "--mode", choices=("all", "changes"), default="changes", help="all输出全部采样点，changes只输出电平跳变坐标点（默认）", ) parser.add_argument( "--time-column", help="时间列名称或列序号，默认自动识别，找不到时使用第0列", ) parser.add_argument( "--encoding", default="auto", choices=("auto", "utf-8-sig", "gbk", "utf-16"), help="输入文件编码，默认自动识别", ) return parser def main(argv: Sequence[str] | None = None) -> int: """命令行入口。""" parser = build_argument_parser() args = parser.parse_args(argv) try: points = parse_rows(read_text(args.input, args.encoding), args.channel, args.time_column) output_points = points if args.mode == "all" else compress_level_changes(points) output_count = write_points(args.output, output_points) except (OSError, CsvParseError) as exc: parser.error(str(exc)) return 2 compression = 0.0 if not points else (1.0 - output_count / len(points)) * 100.0 print(f"输入采样点数：{len(points)}") print(f"输出点数：{output_count}") print(f"点数压缩率：{compression:.2f}%") print(f"输出文件：{args.output}") return 0 if __name__ == "__main__": sys.exit(main())