227 lines
6.9 KiB
Python
227 lines
6.9 KiB
Python
|
import random
|
|||
|
|
import shelve
|
|||
|
|
import json
|
|||
|
|
from datetime import datetime
|
|||
|
|
|
|||
|
|
# 定义数据对象
|
|||
|
|
class Computer_status:
|
|||
|
|
def __init__(self, time, ram_all, ram_use, cpu_use, cpu_c, disk, web, gpu):
|
|||
|
|
self.time = time # 时间戳，单位：秒
|
|||
|
|
self.ram_all = ram_all # 总内存
|
|||
|
|
self.ram_use = ram_use # 已占用内存
|
|||
|
|
self.cpu_use = cpu_use # CPU 占用百分比
|
|||
|
|
self.cpu_c = cpu_c # CPU 温度
|
|||
|
|
self.disk = disk # 磁盘信息
|
|||
|
|
self.web = web # 网络流量信息
|
|||
|
|
self.gpu = gpu # GPU 信息
|
|||
|
|
|
|||
|
|
def to_dict(self):
|
|||
|
|
"""将对象转换为字典"""
|
|||
|
|
return {
|
|||
|
|
"time": self.time,
|
|||
|
|
"ram_all": self.ram_all,
|
|||
|
|
"ram_use": self.ram_use,
|
|||
|
|
"cpu_use": self.cpu_use,
|
|||
|
|
"cpu_c": self.cpu_c,
|
|||
|
|
"disk": self.disk,
|
|||
|
|
"web": self.web,
|
|||
|
|
"gpu": self.gpu,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def from_dict(data):
|
|||
|
|
"""从字典创建 Computer_status 对象"""
|
|||
|
|
return Computer_status(
|
|||
|
|
time=data["time"],
|
|||
|
|
ram_all=data["ram_all"],
|
|||
|
|
ram_use=data["ram_use"],
|
|||
|
|
cpu_use=data["cpu_use"],
|
|||
|
|
cpu_c=data["cpu_c"],
|
|||
|
|
disk=data["disk"],
|
|||
|
|
web=data["web"],
|
|||
|
|
gpu=data["gpu"],
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 数据操作方法
|
|||
|
|
DATABASE_FILE = "data/computer_status.db" # 数据库存储文件
|
|||
|
|
|
|||
|
|
|
|||
|
|
def write_data(computer_status):
|
|||
|
|
"""
|
|||
|
|
写入一条数据到对象数据库
|
|||
|
|
参数：
|
|||
|
|
computer_status: Computer_status 对象
|
|||
|
|
"""
|
|||
|
|
if not isinstance(computer_status, Computer_status):
|
|||
|
|
raise ValueError("Input must be a Computer_status object")
|
|||
|
|
with shelve.open(DATABASE_FILE) as db:
|
|||
|
|
db[str(computer_status.time)] = computer_status
|
|||
|
|
|
|||
|
|
|
|||
|
|
# def get_data_30s():
|
|||
|
|
# """
|
|||
|
|
# 获取最新30秒的数据
|
|||
|
|
# 返回：
|
|||
|
|
# List[Computer_status]，按时间升序排列
|
|||
|
|
# """
|
|||
|
|
# current_time = int(datetime.now().timestamp())
|
|||
|
|
# with shelve.open(DATABASE_FILE) as db:
|
|||
|
|
# data = [
|
|||
|
|
# db[key]
|
|||
|
|
# for key in db
|
|||
|
|
# if current_time - int(key) <= 30 # 过滤最近30秒数据
|
|||
|
|
# ]
|
|||
|
|
# return sorted(data, key=lambda x: x.time)
|
|||
|
|
#
|
|||
|
|
#
|
|||
|
|
# def get_data(time_start, time_end):
|
|||
|
|
# """
|
|||
|
|
# 获取指定时间戳到指定时间戳之间的数据
|
|||
|
|
# 参数：
|
|||
|
|
# time_start: 开始时间戳（int）
|
|||
|
|
# time_end: 结束时间戳（int）
|
|||
|
|
# 返回：
|
|||
|
|
# List[Computer_status]，按时间升序排列
|
|||
|
|
# """
|
|||
|
|
# with shelve.open(DATABASE_FILE) as db:
|
|||
|
|
# data = [
|
|||
|
|
# db[key]
|
|||
|
|
# for key in db
|
|||
|
|
# if time_start <= int(key) <= time_end
|
|||
|
|
# ]
|
|||
|
|
# return sorted(data, key=lambda x: x.time)
|
|||
|
|
#
|
|||
|
|
#
|
|||
|
|
# def get_data_json_30s():
|
|||
|
|
# """
|
|||
|
|
# 获取最新30秒数据并转为 JSON
|
|||
|
|
# 返回：
|
|||
|
|
# JSON 数据（List[Dict]）
|
|||
|
|
# """
|
|||
|
|
# data = get_data_30s()
|
|||
|
|
# return json.dumps([cs.to_dict() for cs in data], indent=4)
|
|||
|
|
#
|
|||
|
|
#
|
|||
|
|
# def get_data_json(time_start, time_end):
|
|||
|
|
# """
|
|||
|
|
# 获取指定时间范围的数据并转为 JSON
|
|||
|
|
# 参数：
|
|||
|
|
# time_start: 开始时间戳（int）
|
|||
|
|
# time_end: 结束时间戳（int）
|
|||
|
|
# 返回：
|
|||
|
|
# JSON 数据（List[Dict]）
|
|||
|
|
# """
|
|||
|
|
# data = get_data(time_start, time_end)
|
|||
|
|
# return json.dumps([cs.to_dict() for cs in data], indent=4)
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
def delete_data(time_start):
|
|||
|
|
"""
|
|||
|
|
删除 time_start 之前的数据
|
|||
|
|
参数：
|
|||
|
|
time_start: 开始时间戳（int），保留此时间戳之后的数据
|
|||
|
|
"""
|
|||
|
|
with shelve.open(DATABASE_FILE, writeback=True) as db:
|
|||
|
|
keys_to_delete = [key for key in db if int(key) < time_start]
|
|||
|
|
for key in keys_to_delete:
|
|||
|
|
del db[key]
|
|||
|
|
print(f"已删除 {len(keys_to_delete)} 条记录")
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 实时数据生成函数
|
|||
|
|
def generate_data(timestamp):
|
|||
|
|
"""生成一个 Computer_status 对象"""
|
|||
|
|
return Computer_status(
|
|||
|
|
time=timestamp,
|
|||
|
|
ram_all=16000,
|
|||
|
|
ram_use=random.randint(8000, 16000),
|
|||
|
|
cpu_use=random.randint(10, 90),
|
|||
|
|
cpu_c=random.randint(40, 70),
|
|||
|
|
disk={"disk1": [100000, random.randint(50000, 90000), random.randint(100, 500), random.randint(200, 600)]},
|
|||
|
|
web={"eth0": [random.randint(500, 1500), random.randint(1000, 2500)]},
|
|||
|
|
gpu={"GPU1": {"gpu_ram_all": 8, "gpu_ram_use": random.uniform(2, 8), "gpu_use": random.randint(10, 90), "gpu_fan": random.randint(20, 80)}}
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# 数据操作方法
|
|||
|
|
def get_data_30s():
|
|||
|
|
"""
|
|||
|
|
实时生成最新30秒的数据
|
|||
|
|
返回：
|
|||
|
|
List[Computer_status]，按时间升序排列
|
|||
|
|
"""
|
|||
|
|
current_time = int(datetime.now().timestamp())
|
|||
|
|
data = [generate_data(current_time - i) for i in range(30)] # 生成过去30秒的数据
|
|||
|
|
return sorted(data, key=lambda x: x.time)
|
|||
|
|
|
|||
|
|
def get_data(time_start, time_end):
|
|||
|
|
"""
|
|||
|
|
实时生成指定时间范围的数据
|
|||
|
|
参数：
|
|||
|
|
time_start: 开始时间戳（int）
|
|||
|
|
time_end: 结束时间戳（int）
|
|||
|
|
返回：
|
|||
|
|
List[Computer_status]，按时间升序排列
|
|||
|
|
"""
|
|||
|
|
data = [generate_data(timestamp) for timestamp in range(time_start, time_end + 1)]
|
|||
|
|
return sorted(data, key=lambda x: x.time)
|
|||
|
|
|
|||
|
|
def get_data_json_30s():
|
|||
|
|
"""
|
|||
|
|
实时生成最新30秒数据并转为 JSON
|
|||
|
|
返回：
|
|||
|
|
JSON 数据（List[Dict]）
|
|||
|
|
"""
|
|||
|
|
data = get_data_30s()
|
|||
|
|
return json.dumps([cs.to_dict() for cs in data], indent=4)
|
|||
|
|
|
|||
|
|
def get_data_json(time_start, time_end):
|
|||
|
|
"""
|
|||
|
|
实时生成指定时间范围的数据并转为 JSON
|
|||
|
|
参数：
|
|||
|
|
time_start: 开始时间戳（int）
|
|||
|
|
time_end: 结束时间戳（int）
|
|||
|
|
返回：
|
|||
|
|
JSON 数据（List[Dict]）
|
|||
|
|
"""
|
|||
|
|
data = get_data(time_start, time_end)
|
|||
|
|
return json.dumps([cs.to_dict() for cs in data], indent=4)
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 示例测试
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
# # 创建一些测试数据
|
|||
|
|
# for i in range(40):
|
|||
|
|
# cs = Computer_status(
|
|||
|
|
# time=int(datetime.now().timestamp()) - i,
|
|||
|
|
# ram_all=16000,
|
|||
|
|
# ram_use=8000 + i * 100,
|
|||
|
|
# cpu_use=50 + i,
|
|||
|
|
# cpu_c=60 - i,
|
|||
|
|
# disk={"disk1": [100000, 50000 + i * 100, 100, 200]},
|
|||
|
|
# web={"eth0": [1000 + i * 10, 2000 + i * 20]},
|
|||
|
|
# gpu={"GPU1": {"gpu_ram_all": 8, "gpu_ram_use": 4 + i / 10, "gpu_use": 30 + i, "gpu_fan": 50 + i}},
|
|||
|
|
# )
|
|||
|
|
# write_data(cs)
|
|||
|
|
#
|
|||
|
|
# # 获取最新30秒数据
|
|||
|
|
# print("最新30秒数据（JSON）:")
|
|||
|
|
# print(get_data_json_30s())
|
|||
|
|
#
|
|||
|
|
# # 获取指定时间段的数据
|
|||
|
|
# time_now = int(datetime.now().timestamp())
|
|||
|
|
# print(f"指定时间段数据（{time_now - 20} 到 {time_now - 10}）:")
|
|||
|
|
# print(get_data_json(time_now - 20, time_now - 10))
|
|||
|
|
# 获取最新30秒的实时数据
|
|||
|
|
print("最新30秒数据（JSON）:")
|
|||
|
|
print(get_data_json_30s())
|
|||
|
|
|
|||
|
|
# 获取指定时间范围的实时数据
|
|||
|
|
time_now = int(datetime.now().timestamp())
|
|||
|
|
print(f"指定时间段数据（{time_now - 20} 到 {time_now - 10}）:")
|
|||
|
|
print(get_data_json(time_now - 20, time_now - 10))
|