如何用Python Pandas接收持续追加更新的CSV文件中的信号？

针对你需要持续监测signal.csv最后一行、并将信号写入仅保留一行的out.csv的需求，以下是基于Python的实现方案：

核心逻辑说明 #

1. 读取signal.csv的最后一行数据（跳过表头）
2. 按优先级判断信号：
   • 若Answers列非0，直接使用该值作为信号
   • 若Answers为0但step列非0，使用step的值作为信号
   • 两者都为0时，信号为0
3. 将最终信号覆盖写入out.csv，确保文件始终只有一行内容

单次执行代码（手动触发更新） #

import csv
import os

def get_last_line(csv_path):
    """获取CSV文件的最后一行数据（跳过表头）"""
    with open(csv_path, 'r', newline='', encoding='utf-8') as f:
        reader = csv.reader(f)
        next(reader)  # 跳过表头行
        last_line = None
        for line in reader:
            last_line = line
        return last_line

def update_out_signal(signal_csv, out_csv):
    """根据signal.csv最后一行更新out.csv的信号"""
    last_line = get_last_line(signal_csv)
    
    # 处理空文件情况
    if not last_line:
        with open(out_csv, 'w', newline='', encoding='utf-8') as f:
            csv.writer(f).writerow([0])
        return
    
    # 解析列值
    answers = int(last_line[0])
    step = int(last_line[2])
    
    # 确定最终信号
    if answers != 0:
        signal = answers
    elif step != 0:
        signal = step
    else:
        signal = 0
    
    # 覆盖写入out.csv
    with open(out_csv, 'w', newline='', encoding='utf-8') as f:
        csv.writer(f).writerow([signal])

# 执行示例
if __name__ == "__main__":
    update_out_signal("signal.csv", "out.csv")

实时监测版本（自动响应文件追加） #

如果需要持续监听signal.csv的追加更新，可使用watchdog库实现自动触发：

1. 先安装依赖：

pip install watchdog

2. 监听代码：

import csv
import os
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

def get_last_line(csv_path):
    with open(csv_path, 'r', newline='', encoding='utf-8') as f:
        reader = csv.reader(f)
        next(reader)
        last_line = None
        for line in reader:
            last_line = line
        return last_line

def update_out_signal(signal_csv, out_csv):
    last_line = get_last_line(signal_csv)
    if not last_line:
        with open(out_csv, 'w', newline='', encoding='utf-8') as f:
            csv.writer(f).writerow([0])
        return
    
    answers = int(last_line[0])
    step = int(last_line[2])
    
    signal = answers if answers != 0 else (step if step != 0 else 0)
    
    with open(out_csv, 'w', newline='', encoding='utf-8') as f:
        csv.writer(f).writerow([signal])

class SignalFileHandler(FileSystemEventHandler):
    def __init__(self, signal_csv, out_csv):
        self.signal_csv = signal_csv
        self.out_csv = out_csv
    
    def on_modified(self, event):
        # 仅处理目标CSV文件的修改事件
        if event.src_path == os.path.abspath(self.signal_csv):
            try:
                update_out_signal(self.signal_csv, self.out_csv)
            except Exception as e:
                # 捕获文件占用等异常，避免程序崩溃
                print(f"更新信号时出错: {e}")

if __name__ == "__main__":
    SIGNAL_CSV = "signal.csv"
    OUT_CSV = "out.csv"
    
    # 初始化信号文件
    update_out_signal(SIGNAL_CSV, OUT_CSV)
    
    # 启动文件监听
    event_handler = SignalFileHandler(SIGNAL_CSV, OUT_CSV)
    observer = Observer()
    observer.schedule(event_handler, path=os.path.dirname(os.path.abspath(SIGNAL_CSV)), recursive=False)
    observer.start()
    
    print(f"开始监测 {SIGNAL_CSV}... 按Ctrl+C停止")
    try:
        while True:
            pass
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

注意事项 #

• 如果你的CSV文件使用了非UTF-8编码，需要在open函数中指定对应的encoding参数（如gbk）
• 实时监听版本会响应文件的所有修改事件，包括其他程序的追加写入
• 若存在文件占用问题，可在update_out_signal函数中增加重试逻辑或异常捕获

内容的提问来源于stack exchange，提问作者Genry