性能监视服务端-客户端构建

毛毛睡似了吗?看了这个项目，突发奇想，打算也做个视监自己电脑的服务端出来。

于是，昨天晚上，费了九牛二虎以及AI之力，终于把这个东西给实现出来了。

项目已开源，仓库位置：YMDG-BM/device-monitoring-server-and-client

项目结构

TREE
├── server
│   ├── static
│   │   ├── css
│   │   │   └── style.css				# 样式文件
│   │   ├── img
│   │   │   └── background.jpg			# 背景图片
│   │   └── js
│   │       └── script.js				# JavaScript 文件
│   ├── templates
│   │   └── display.html				# HTML 模板文件
│   ├── server.py					# 服务端入口点，设置HTTP服务器并处理请求
│   └── requirements.txt				# 项目依赖项
├── client
│   ├── utils
│   │   └── __init__.py				# 工具函数
│   └──client.py					# 客户端入口点，发送请求并处理响应
├── dockerfile					# Docker 配置文件
├── README.md						# 项目文档
└── requirements.txt					# 项目依赖项

客户端

客户端的目的有以下几个：

	1. 监视设备是否在线
	2. 监视设备各项性能信息
	3. 监视当前激活的窗口
	4. 发送json信息给服务器用于解析

于是，围绕这些需求开始构建代码。

监视设备是否在线

这个非常简单，在服务端设置一个heartbeat，在一定时间内如果没有发包就判定客户端离线。

部分实现如下：

client_last_seen = time.time()
client_timeout = 30  # 超时时间，单位为秒

def monitor_client_status():
    global client_last_seen
    while True:
        if time.time() - client_last_seen > client_timeout:
            client_data.clear()  # 清空客户端数据，表示客户端下线
        time.sleep(5)
        
@app.route('/status', methods=['GET'])
def get_status():
    global client_data
    if not client_data:
        return jsonify({"error": "No data received from client"}), 404
    return jsonify(client_data)

@app.route('/client_data', methods=['GET'])
def client_data_endpoint():
    global client_data
    if not client_data:
        return jsonify({"error": "No data received from client"}), 404
    return jsonify(client_data)

前端js实现

const container = document.getElementById('progressBars');
	if (data.error) {
		container.innerHTML = '<div class="Offline">计算机已离线</div>';
}        

监视设备各项性能信息

这部分在客户端实现，主要在utils中实现。

import psutil
import GPUtil
import subprocess

一些必要的模块，用于获取各项信息，比如CPU名、各项占用等。

def get_cpu_name():
    try:
        # 执行wmic命令获取CPU名称
        command = 'wmic cpu get name'
        cpu_name = subprocess.check_output(command, shell=True).decode('utf-8').split('\n')[1].strip()
        return cpu_name
    except Exception as e:
        print("无法获取CPU名称:", e)
        return None

获取CPU名。

def get_performance():
    cpu_usage = psutil.cpu_percent(interval=1)
    memory_info = psutil.virtual_memory()
    disk_info = psutil.disk_usage('/')
    gpus = GPUtil.getGPUs()
    gpu_info = [{"name": gpu.name, "load": gpu.load * 100, "memory_total": gpu.memoryTotal, "memory_used": gpu.memoryUsed, "memory_percent": gpu.memoryUtil * 100} for gpu in gpus]

    performance_data = {
        "cpu": {
            "name": get_cpu_name(),
            "usage": round(cpu_usage, 1)
        },
        "memory": {
            "percent": memory_info.percent
        },
        "disk": {
            "percent": disk_info.percent
        },
        "gpu": {
            "name": gpu_info[0]["name"] if gpu_info else "No GPU",
            "load": round(gpu_info[0]["load"], 1) if gpu_info else 0.0,
            "memory_total": int(gpu_info[0]["memory_total"]) if gpu_info else 0,
            "memory_percent": round(gpu_info[0]["memory_percent"], 1) if gpu_info else 0.0
        },
        "active_window": {
            "title": get_active_window()["title"],
        }
    }
    return performance_data

获取各项性能信息。

监视当前激活的窗口

import pygetwindow as gw

导入pygetwindow用来监视当前激活窗口

def get_active_window():
    try:
        window = gw.getActiveWindow()
        if window is not None:
            return {
                'title': window.title,
                'hwnd': window._hWnd
            }
        else:
            return None
    except Exception as e:
        print(f"An error occurred: {e}")
        return None

获取当前激活窗口信息

发送json信息给服务器用于解析

首先准备要发送的data

def send_data(self, data):
        headers = {'Content-Type': 'application/json'}
        payload = {
            "device_key": self.device_key, #设备认证信息，后面要考
            "performance_data": data
        }
        try:
            response = requests.post(self.server_url, data=json.dumps(payload), headers=headers)
            response.raise_for_status()  # 如果响应状态码不是200，抛出HTTPError
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"Error sending data: {e}")
            return None
 def get_performance_data(self):
        return utils.get_performance()

死循环防止炸客户端

while True:
        # retry_duration = 3  # 重试间隔时间
        performance_data = client.get_performance_data()
        response = client.send_data(performance_data)
        if response:
            print("Response from server:", response)
        else:
            print(f"Failed to send data, will retry in {retry_duration} seconds.")
        time.sleep(retry_duration)  # 每隔retry_duration秒重试一次

服务端

服务端要实现的内容也很简单，有以下几项：

1.获取客户端发送的信息
2.将客户端发送的信息进行解析，并呈现在网页上
3.认证客户端信息

获取客户端发送的信息

用flask实现

from flask import Flask, jsonify, request, render_template

@app.route('/monitor', methods=['POST'])
def monitor():
    global client_data, client_last_seen
    data = request.json
    if data.get("device_key") != device_key:
        return jsonify({"error": "Unauthorized"}), 401
    client_data = data.get("performance_data")
    client_last_seen = time.time()
    return jsonify({"status": "success"})

@app.route('/client_data', methods=['GET'])
def client_data_endpoint():
    global client_data
    if not client_data:
        return jsonify({"error": "No data received from client"}), 404
    return jsonify(client_data)

将客户端发送的信息进行解析，并呈现在网页上

python部分

@app.route('/display', methods=['GET'])
def display():
    return render_template('display.html')

@app.route('/client_data', methods=['GET'])
def client_data_endpoint():
    global client_data
    if not client_data:
        return jsonify({"error": "No data received from client"}), 404
    return jsonify(client_data)

json解析部分用js实现

function createProgressBar(category, label, percent) {
    return `
        <div class="progress-item ${category}">
            <div class="progress-title">
                <span>${label}</span>
                <span>${percent.toFixed(1)}%</span>
            </div>
            <div class="progress-bar">
                <div class="progress-fill" style="width: ${percent}%"></div>
            </div>
        </div>
    `;
}
function renderProgressBars() {
    fetch('/client_data')
        .then(response => response.json())
        .then(data => {
            const container = document.getElementById('progressBars');
            if (data.error) {
                container.innerHTML = '<div class="Offline">计算机已离线</div>';
            }
          else{
    container.innerHTML = [
        createProgressBar('cpu', 'CPU 使用率', data.cpu.usage),
        createProgressBar('memory', '内存使用率', data.memory.percent),
        createProgressBar('disk', '硬盘使用率', data.disk.percent),
        createProgressBar('gpu-load', 'GPU 使用率', data.gpu.load),
        createProgressBar('gpu-mem', '显存使用率', data.gpu.memory_percent),
        `<div class="active_window">当前激活的窗口为：${data.active_window.title}</div>`
    ].join('');}
    });
}
setInterval(renderProgressBars, 5000); // 每5秒更新一次数据
renderProgressBars(); // 页面加载时立即获取一次数据

认证客户端信息

客户端与服务端协同实现

这里我使用客户端的mac地址作为device_key认证。

import getmac

def send_data(self, data):
        headers = {'Content-Type': 'application/json'}
        payload = {
            "device_key": self.device_key,
            "performance_data": data
        }
        try:
            response = requests.post(self.server_url, data=json.dumps(payload), headers=headers)
            response.raise_for_status()  # 如果响应状态码不是200，抛出HTTPError
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"Error sending data: {e}")
            return None

服务端存放device_key信息

with open("./device_key.txt", "r") as f:
    device_key = f.read().strip()

monitor方法附带认证

def monitor():
    global client_data, client_last_seen
    data = request.json
    if data.get("device_key") != device_key:
        return jsonify({"error": "Unauthorized"}), 401
    client_data = data.get("performance_data")
    client_last_seen = time.time()
    return jsonify({"status": "success"})