Modbus-rtu/HostComputer/PythonTest/parse_logic_csv.py

#!/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())