Great_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_ICM_1 = random.uniform(-0.29969177, 0.29969177)
aY_ICM_1 = random.uniform(0.6930143, 2.29969177)
aZ_ICM_1 = random.uniform(-0.6866665, 0.69969177)
timestamp = datetime.now().strftime("%Y/%m/%d %H:%M:%S:%f")[:-3] # 格式化时间戳
aX_MPU_1 = random.uniform(-0.38232422, 0.69969177)
aY_MPU_1 = random.uniform(0.5595703, 0.79969177)
aZ_MPU_1 = random.uniform(0.7294922, 2.7294922)
# 数据记录列表
record = [
aX_ICM_1,
aY_ICM_1,
aZ_ICM_1,
timestamp,
aX_MPU_1,
aY_MPU_1,
aZ_MPU_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_ICM_1', 'aY_ICM_1', 'aZ_ICM_1', '时间戳','aX_MPU_1', 'aY_MPU_1', 'aZ_MPU_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_ICM_1', 'aY_ICM_1', 'aZ_ICM_1', '时间戳','aX_MPU_1', 'aY_MPU_1', 'aZ_MPU_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_Data
-------------------
import pyautogui
import time
import os
import shutil
#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来保持元数据
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))
'''
-------------------
Chart_RealTime
-------------------
import os
import pygame
import pandas as pd
import glob
import logging
from pygame.locals import *
# 配置日志记录
log_path = r'D:\Tim_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 = "Pump PHM"
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)
