创建数据/采集数据+从PI数据到PC+实时UI+To PLC

Get_Data

----------

import csv
import os
import random
from datetime import datetime
import logging
import time

# 配置日志记录
logging.basicConfig(filename='D:/_Study/Case/Great_Data/log.txt',
          level=logging.INFO,
          format='%(asctime)s - %(levelname)s - %(message)s')

def 记录日志(消息):
  logging.info(消息)
  print(消息)

def 生成随机记录():
  # 生成随机数据
  aX_U_1 = random.uniform(-0.29969177, 0.29969177)
  aY_U_1 = random.uniform(0.6930143, 2.29969177)
  aZ_U_1 = random.uniform(-0.6866665, 0.69969177)
  timestamp = datetime.now().strftime("%Y/%m/%d %H:%M:%S:%f")[:-3] # 格式化时间戳

  aX_D_1 = random.uniform(-0.38232422, 0.69969177)
  aY_D_1 = random.uniform(0.5595703, 0.79969177)
  aZ_D_1 = random.uniform(0.7294922, 2.7294922)

  # 数据记录列表
  record = [
    aX_U_1,
    aY_U_1,
    aZ_U_1,
    timestamp,
    aX_D_1,
    aY_D_1,
    aZ_D_1
  ]
  return record

def 追加到CSV(文件路径, 记录, 表头=None):
  # 尝试写入CSV文件
  try:
    if not os.path.exists(文件路径):
      # 如果文件不存在，则创建新文件并写入表头
      with open(文件路径, mode='w', newline='', encoding='utf-8') as 文件:
        writer = csv.writer(文件)
        if 表头:
          writer.writerow(表头)
        writer.writerow(记录)
    else:
      # 否则追加记录
      with open(文件路径, mode='a', newline='', encoding='utf-8') as 文件:
        writer = csv.writer(文件)
        writer.writerow(记录)
    记录日志(f"记录已追加到 {文件路径}")
  except PermissionError:
    记录日志("权限拒绝：请确保目录可访问，并尝试以管理员身份运行脚本。")
  except Exception as e:
    记录日志(f"写入CSV时发生错误：{e}")

# 定义文件路径
基础目录 = 'D:/_Study/Case/Great_Data/Data'
os.makedirs(基础目录, exist_ok=True) # 创建目录（如果不存在）

# 文件名称模板
文件名模板 = 'AA@BB@{index}@{time}.csv'
文件索引 = 0

# 获取所有符合条件的文件
文件列表 = []
for 文件名 in os.listdir(基础目录):
  if 文件名.startswith('AA@BB@'):
    文件路径 = os.path.join(基础目录, 文件名)
    文件列表.append(文件路径)

# 按照时间戳排序文件列表
文件列表.sort(key=lambda x: int(os.path.basename(x).split('@')[2].split('.')[0]), reverse=True)

# 只保留最新的两个文件
保留文件列表 = 文件列表[:1]
删除文件列表 = 文件列表[1:]

# 删除多余的文件，并记录到日志
for 文件路径 in 删除文件列表:
  os.remove(文件路径)
  记录日志(f"{文件路径} 已被删除。")

# 如果没有文件，则创建第一个文件
if not 保留文件列表:
  文件索引 += 1
  当前时间 = datetime.now().strftime("%Y%m%d_%H%M%S_%f")[:-3]
  文件路径 = os.path.join(基础目录, 文件名模板.format(index=文件索引, time=当前时间))
  表头 = [ 'aX_U_1', 'aY_U_1', 'aZ_U_1', '时间戳','aX_D_1', 'aY_D_1', 'aZ_D_1']
  追加到CSV(文件路径, 表头, 表头=表头)
  记录日志(f"{文件路径} 已创建并写入表头。")
  保留文件列表.append(文件路径)

# 开始生成记录
记录总数 = 0
while 文件索引 <= 1:
  for 文件路径 in 保留文件列表:
    if 记录总数 >= 10:
      文件索引 += 1
      if 文件索引 > 1:
        记录日志("所有文件均已达到最大行数。")
        break
      当前时间 = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
      文件路径 = os.path.join(基础目录, 文件名模板.format(index=文件索引, time=当前时间))
      表头 = [ 'aX_U_1', 'aY_U_1', 'aZ_U_1', '时间戳','aX_D_1', 'aY_D_1', 'aZ_D_1']
      追加到CSV(文件路径, 表头, 表头=表头)
      记录日志(f"{文件路径} 已创建并写入表头。")
      保留文件列表.append(文件路径)
      记录总数 = 0

    记录 = 生成随机记录()
    追加到CSV(文件路径, 记录)
    记录总数 += 1
    记录日志(f"当前 {文件路径} 总行数为：{记录总数}")

  if 文件索引 > 2:
    break

记录日志("处理完成。")

print("Task completed. Will reboot in 3 minutes.")
time.sleep(180) # 等待180秒
print("Rebooting now...")
os.system('sudo reboot')

'''
[Unit]
Description=Great Data Service
After=network.target

[Service]
User=pi
WorkingDirectory=/home/pi/Great_Data
ExecStart=/usr/bin/python3 /home/pi/Great_Data/Great_data.py

[Install]
WantedBy=multi-user.target

'''

----------

Copy+Del Data

----------

import pyautogui
import time
import os
import shutil
import datetime

#10 秒计时后开始运行
pyautogui.countdown(3)

#到达位置鼠标左键单击
pyautogui.click(1154,1038,button='left')#任务栏图标
time.sleep(0.1)
pyautogui.click(1158,910,button='left')#VNC程序
time.sleep(0.1)
pyautogui.click(659,65,button='left')#PI程序
time.sleep(0.1)##
pyautogui.click(1285,366,button='left')#MENU 菜单
time.sleep(0.1)
pyautogui.click(1170,486,button='left')#File Transfer
time.sleep(0.1)
pyautogui.click(743,677,button='left')#SendFile
time.sleep(0.1)
pyautogui.click(953,498,button='left')#File选择
time.sleep(0.1)
pyautogui.click(1039,615,button='left')#OK
time.sleep(0.1)
pyautogui.click(1039,616,button='left')#桌面路径
time.sleep(0.1)
pyautogui.click(1002,762,button='left')#确定
time.sleep(0.1)
pyautogui.click(1298,300,button='left')#X1
time.sleep(0.1)
pyautogui.click(1190,373,button='left')#X2
time.sleep(0.1)
pyautogui.click(1295,333,button='left')#X3
time.sleep(0.1)

pyautogui.click(1157,1037,button='left')##VNC程序
time.sleep(0.1)
pyautogui.click(1465,874,button='left')#Close File Transfere
time.sleep(0.1)

#到达位置鼠标左键双击
pyautogui.doubleClick(x=1298, y=301, button="left")
time.sleep(1.2)
#pyautogui.click(922,68,button='left')
#time.sleep(1)

#存储截图
im = pyautogui.screenshot()
im.save(r'D:\GGY\屏幕截图.png')

#####获取最新文件，并单独复制至处理文件夹
# 定义源目录和目标目录
src_dir = r'C:\Users\Administrator\Desktop\Great_Data\Data'
dst_dir = r'C:\Users\Administrator\Desktop\Great_Data\Data\Send'

# 如果目标目录不存在，则创建之
if not os.path.exists(dst_dir):
  os.makedirs(dst_dir)

# 获取目录下所有的文件及其修改日期，并找到最新的那个
files = [os.path.join(src_dir, f) for f in os.listdir(src_dir) if os.path.isfile(os.path.join(src_dir, f))]
files.sort(key=lambda x: os.path.getmtime(x), reverse=True)

# 检查是否找到了文件
if files:
  # 复制最新的文件
  latest_file = files[0]
  print(f'Copying the latest file: {latest_file}')
  shutil.copy2(latest_file, dst_dir) # 使用shutil.copy2来保持元数据
 
  # 删除其他的旧文件
  for file in files[1:]:
    print(f'Deleting old file: {file}')
    try:
      os.remove(file)
    except OSError as e:
      print(f"Error: {e.strerror} - {file}")

  # 记录操作日志到目标目录下的 log.txt 文件
  log_file_path = os.path.join(dst_dir, 'log.txt')
  with open(log_file_path, 'a') as log_file:
    log_file.write(f'{datetime.datetime.now()}: Latest file copied and old files deleted.\n')
else:
  print('No files found in the directory.')

#####获取最新文件，并单独复制至处理文件夹  

#填入参数， 第一参数是输入内容，第二个参数是每个字符间的间隔时间；
#pyautogui.moveTo(1241,311)
#count=(r'd:\aa')
#for c in str(count):
#  pyautogui.keyDown(c)

#填入参数， 第一参数是输入内容，第二个参数是每个字符间的间隔时间；
#pyautogui.click(915,87,button='left')
#time.sleep(2)
#pyautogui.click(364,284,button='left')
#count=(2 )
#for cc in str(count):
#  pyautogui.keyDown(cc)
 

#实时显示坐标
#pyautogui.displayMousePosition()
'''
import pyautogui
import time
#10 秒计时后开始运行
pyautogui.countdown(3)

#实时显示坐标d:\aa
while True:
  x,y=pyautogui.position()
  print('Pos:',(x,y))

'''

----------

From data to Chart

----------

##

import os
import pygame
import pandas as pd
import glob
import logging
from pygame.locals import *

# 配置日志记录
log_path = r'D:\Study\Case\Chart_RealTime\log.txt'
logging.basicConfig(filename=log_path, level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')

# 初始化pygame
pygame.init()

# 设置窗口大小和标题
WINDOW_WIDTH = 1920
WINDOW_HEIGHT = 1080
WINDOW_TITLE = "UI"
screen = pygame.display.set_mode((WINDOW_WIDTH, WINDOW_HEIGHT))
pygame.display.set_caption(WINDOW_TITLE)

# 定义颜色
WHITE = (255, 255, 255)
RED = (255, 0, 0)
GREEN = (0, 255, 0)
BLUE = (0, 0, 255)
ORANGE = (255, 165, 0)
PURPLE = (128, 0, 128)
BROWN = (139, 69, 19)  
BLACK = (0, 0, 0)

# 源目录路径
source_dir = r'C:\Users\Administrator\Desktop\Great_Data\Data\Send'

# 查找目录下的所有CSV文件
csv_files = glob.glob(os.path.join(source_dir, '*.csv'))

if not csv_files:
    print("未找到CSV文件，请检查目录路径是否正确。")
else:
    # 用于绘制图表的数据
    data = []

    # 定义处理时间戳的函数
    def process_timestamp(ts):
        try:
            if ':' in ts:
                ts, microseconds = ts.rsplit(':', 1)
                microseconds = microseconds.ljust(3, '0')
                ts += '.' + microseconds
            else:
                ts += '.000'
        except Exception as e:
            logging.error(f"Failed to process timestamp {ts}: {e}")
            ts = ts + '.000'
        return ts

    # 遍历每个CSV文件
    for csv_file in csv_files:
        logging.info("Reading CSV file...")