多线程显示 CSV 2 PNG 倒计时循环播放

import argparse
import logging
import sys
from pathlib import Path
import time
import os
import pandas as pd
import matplotlib.pyplot as plt
from concurrent.futures import ProcessPoolExecutor, as_completed

# 初始化日志记录
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s - %(levelname)s - %(message)s',
                    handlers=[logging.FileHandler('script.log'), logging.StreamHandler(sys.stdout)])

def process_file(file_path, input_directory, output_directory):
    try:
        # 使用 Pandas 读取 CSV 文件
        data1 = pd.read_csv(file_path)
    except Exception as e:
        logging.error(f"读取文件 {file_path} 时发生错误: {e}")
        return

    # 获取数据的行数和列数
    nRow, nCol = data1.shape

    # 绘制图表
    plt.figure(figsize=(10, 8))
    index = 0
    for i in range(nCol):
        if data1.columns[i] == "datetime":
            continue

        plt.subplot(nCol - 1, 1, index + 1)
        index += 1
        plt.plot(data1.iloc[:, i])
        plt.title(data1.columns[i])

    # 获取当前时间并格式化
    formatted_time = time.strftime('%Y_%m_%d_%H_%M', time.localtime())

    # 修改文件名以包含当前时间，并替换特殊字符
    file_stem = file_path.stem.replace('@', '_').replace(':', '_')
    file_name = formatted_time + '_' + file_stem + '.png'

    # 生成输出路径
    output_path = Path(output_directory) / file_name

    # 保存图像到 Fig 目录
    try:
        plt.savefig(str(output_path))
        logging.info(f"存储图片中... {output_path}")
        plt.close()
    except Exception as e:
        logging.error(f"保存图像 {output_path} 时发生错误: {e}")
        return
    return output_path

def animate_charts(output_directory):
    # 获取最新生成的两个图表文件
    paths = sorted(Path(output_directory).glob('*.png'), key=lambda p: p.stat().st_mtime, reverse=True)
    latest_paths = paths[:2]

    # 设置初始状态
    paused = False

    def on_key_press(event):
        nonlocal paused
        if event.key == ' ':  # 空格键
            paused = not paused
            if paused:
                print("Paused. Press space to continue.")
            else:
                print("Continuing...")

    # 主循环
    while True:
        for path in latest_paths:
            # 显示文件名
            print(f"正在显示文件: {path}")

            if paused:
                plt.connect('key_press_event', on_key_press)
                img = plt.imread(str(path))
                fig, ax = plt.subplots()
                ax.imshow(img)
                ax.axis('off')  # 关闭坐标轴

                creation_time = time.strftime("%Y-%m-%d %H:%M", time.localtime(path.stat().st_ctime))
                ax.set_title(f'File Created at: {creation_time}')

                plt.show(block=False)

                # 显示倒计时
                duration = 55
                text_obj = None
                for remaining in range(duration, 0, -1):
                    if not paused:
                        break
                    time.sleep(1)
                    if text_obj is not None:
                        text_obj.remove()  # 直接移除文本对象
                    text_obj = ax.text(0.5, 0.15, f"Timing: {remaining}s", transform=ax.transAxes, ha='center', va='center', color='red', fontsize=8)
                    fig.canvas.draw_idle()
                    plt.pause(0.001)  # 更新图形界面

                plt.close()

                while paused:
                    time.sleep(0.1)  # 等待恢复
            else:
                img = plt.imread(str(path))
                fig, ax = plt.subplots()
                ax.imshow(img)
                ax.axis('off')  # 关闭坐标轴

                creation_time = time.strftime("%Y-%m-%d %H:%M", time.localtime(path.stat().st_ctime))
                ax.set_title(f'File Created at: {creation_time}')

                plt.show(block=False)

                # 显示倒计时
                duration = 60
                text_obj = None
                for remaining in range(duration, 0, -1):
                    time.sleep(1)
                    if text_obj is not None:
                        text_obj.remove()  # 直接移除文本对象
                    text_obj = ax.text(0.5, 0.15, f"Timing: {remaining}s", transform=ax.transAxes, ha='center', va='center', color='red', fontsize=8)
                    fig.canvas.draw_idle()
                    plt.pause(0.001)  # 更新图形界面

                plt.close()

def main(input_directory, output_directory):
    # 开始运行提示
    logging.info("-----开始运行-----")

    # 定义传感器数据所在的目录
    sensor_dir = Path(input_directory)

    # 获取传感器目录下的所有文件列表
    files = list(sensor_dir.glob('*.csv'))

    # 输出文件列表
    logging.info("即将进行文件读取...")
    for i, f in enumerate(files, start=1):
        logging.info(f"文件读取中 {i}/{len(files)}")
        logging.info(f.name)

    # 定义输出图像的目录
    fig_dir = Path(output_directory)

    # 检查是否存在 Fig 目录，如果不存在则创建
    if not fig_dir.exists():
        try:
            fig_dir.mkdir(parents=True, exist_ok=True)
        except FileExistsError:
            logging.info("目录已存在。")

    # 打印当前时间
    current_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())
    logging.info(f"读取完毕，读取时间： {current_time}  ^_^ 存储图片进行中... ")

    # 使用并行处理
    with ProcessPoolExecutor() as executor:
        futures = []
        for file_path in files:
            future = executor.submit(process_file, file_path, input_directory, output_directory)
            futures.append(future)

        # 等待所有任务完成
        for future in as_completed(futures):
            future.result()

    # 结束提示
    logging.info("存储完毕")

    # 显示最近两张图片
    animate_charts(output_directory)

if __name__ == "__main__":
    # 定义默认配置值
    input_directory = 'C:\\Users\\Administrator\\Desktop\\Sensor'
    output_directory = 'C:\\Users\\Administrator\\Desktop\\Fig'

    # 解析命令行参数
    parser = argparse.ArgumentParser(description="Process CSV files and generate charts.")
    parser.add_argument('--input_directory', type=str, default=input_directory, help='Input directory containing CSV files.')
    parser.add_argument('--output_directory', type=str, default=output_directory, help='Output directory to save generated charts.')
    args = parser.parse_args()

    # 主函数调用
    main(args.input_directory, args.output_directory)