使用 Python 构建实时系统监控仪表板


本文我将分享如何使用 Python 创建一个仪表板,用以实时监控系统性能。该仪表板将显示您计算机的 RAM、CPU 负载和磁盘空间使用情况。

我们将使用 Dash 来构建 Web 界面,使用 Plotly 来绘制交互式图表,使用 psutil 进行系统监控。

该系统可以在 Windows 和 Linux 系统上运行。对于 macOS 用户,可能需要进行一些小的修改。

项目设置

步骤 1:创建项目文件夹

首先,创建一个项目文件夹。打开终端(或命令提示符),并创建一个新目录。

mkdir system_monitor_dashboard
cd system_monitor_dashboard

步骤 2:设置虚拟环境

隔离项目依赖项非常重要。我们将创建一个虚拟环境,以保持我们的环境干净且可预测。

Windows:

python -m venv venv
venv\Scripts\activate

Linux/Mac:

python3 -m venv venv
source venv/bin/activate

步骤 3:安装所需库

我们需要安装以下 Python 包:

  • dash:用于构建仪表板的 Web 框架。
  • plotly:用于创建图形和图表。
  • pandas:用于处理表格格式的数据。
  • psutil:用于监控系统资源。

通过运行以下命令安装它们:

pip install dash plotly pandas psutil

这将安装构建仪表板所需的所有库。

构建仪表板

步骤 4:编写代码

现在,我们将创建一个 Python 脚本,收集系统统计信息并使用动态实时仪表板显示这些信息。

监控 RAM、CPU 和磁盘:psutil 库提供跨平台的系统监控信息访问。它可以在 Windows、Linux 和 macOS 上运行。

创建一个名为 app.py 的新 Python 文件:

touch app.py

打开 app.py ,输入以下代码:

import dash
from dash import dcc, html
from dash.dependencies import Input, Output
import plotly.graph_objs as go
import psutil
import logging
from collections import deque
from datetime import datetime
import sys

# Set up basic logging to debug
logging.basicConfig(level=logging.INFO)

# Initialize the Dash app
app = dash.Dash(__name__)

# Define fixed-size lists (deque) to store the last 20 data points for RAM, CPU, Disk usage, and time
history = {
    'ram': deque(maxlen=20),
    'cpu': deque(maxlen=20),
    'disk': deque(maxlen=20),
    'time': deque(maxlen=20)  # Store timestamps for x-axis
}

# Function to get system statistics (RAM, CPU, and Disk)
def get_system_stats():
    try:
        # Get memory stats
        memory = psutil.virtual_memory()
        ram = memory.percent

        # Get CPU usage
        cpu = psutil.cpu_percent(interval=1)

        # Get Disk usage
        disk = psutil.disk_usage('/').percent

        # Return RAM, CPU, and Disk data
        return {
            'RAM Usage (%)': ram,
            'CPU Usage (%)': cpu,
            'Disk Usage (%)': disk
        }
    except Exception as e:
        logging.error(f"Error fetching system stats: {e}")
        return {}

# Determine whether to run in 'one' or 'multiple' mode based on command-line argument
mode = sys.argv[1] if len(sys.argv) > 1 else 'multiple'

if mode == 'one':
    app.layout = html.Div([
        html.H1('System Monitoring Dashboard (Combined Graph)'),

        # Combined Line Chart for RAM, CPU, and Disk
        dcc.Graph(id='combined-graph'),

        # Interval for updating the dashboard every 5 seconds
        dcc.Interval(
            id='interval-component',
            interval=5*1000,  # 5000 milliseconds (5 seconds)
            n_intervals=0
        )
    ])

    # Update callback to refresh the combined RAM, CPU, and Disk usage graph every interval
    @app.callback(
        Output('combined-graph', 'figure'),
        [Input('interval-component', 'n_intervals')]
    )
    def update_combined_graph(n):
        # Fetch system stats (RAM, CPU, and Disk)
        data = get_system_stats()

        if not data:
            logging.info("No data fetched")
            return {}

        # Log fetched data in the terminal
        logging.info(f"Fetched data: {data}")

        # Append the current time, RAM, CPU, and Disk usage to history
        current_time = datetime.now().strftime('%H:%M:%S')  # Get the current time as a string
        history['ram'].append(data['RAM Usage (%)'])
        history['cpu'].append(data['CPU Usage (%)'])
        history['disk'].append(data['Disk Usage (%)'])
        history['time'].append(current_time)

        # Create Combined Line Chart
        combined_figure = {
            'data': [
                go.Scatter(
                    x=list(history['time']),
                    y=list(history['ram']),
                    mode='lines+markers',
                    name='RAM Usage (%)'
                ),
                go.Scatter(
                    x=list(history['time']),
                    y=list(history['cpu']),
                    mode='lines+markers',
                    name='CPU Usage (%)'
                ),
                go.Scatter(
                    x=list(history['time']),
                    y=list(history['disk']),
                    mode='lines+markers',
                    name='Disk Usage (%)'
                )
            ],
            'layout': go.Layout(
                title='RAM, CPU, and Disk Usage Over Time',
                xaxis=dict(title='Time', tickformat='%H:%M:%S'),  # Format the time
                yaxis=dict(title='Percentage'),
            )
        }

        return combined_figure

else:
    # Layout for multiple graphs (RAM, CPU, Disk each on its own graph)
    app.layout = html.Div([
        html.H1('System Monitoring Dashboard (Separate Graphs)'),

        # RAM Usage Line Chart
        dcc.Graph(id='ram-usage-graph'),

        # CPU Usage Line Chart
        dcc.Graph(id='cpu-usage-graph'),

        # Disk Usage Line Chart
        dcc.Graph(id='disk-usage-graph'),

        # Interval for updating the dashboard every 5 seconds
        dcc.Interval(
            id='interval-component',
            interval=5*1000,  # 5000 milliseconds (5 seconds)
            n_intervals=0
        )
    ])

    # Update callback to refresh the RAM, CPU, and Disk usage graphs every interval
    @app.callback(
        [Output('ram-usage-graph', 'figure'),
         Output('cpu-usage-graph', 'figure'),
         Output('disk-usage-graph', 'figure')],
        [Input('interval-component', 'n_intervals')]
    )
    def update_separate_graphs(n):
        # Fetch system stats (RAM, CPU, and Disk)
        data = get_system_stats()

        if not data:
            logging.info("No data fetched")
            return {}, {}, {}

        # Log fetched data in the terminal
        logging.info(f"Fetched data: {data}")

        # Append the current time, RAM, CPU, and Disk usage to history
        current_time = datetime.now().strftime('%H:%M:%S')  # Get the current time as a string
        history['ram'].append(data['RAM Usage (%)'])
        history['cpu'].append(data['CPU Usage (%)'])
        history['disk'].append(data['Disk Usage (%)'])
        history['time'].append(current_time)

        # Create RAM Usage Line Chart
        ram_figure = {
            'data': [go.Scatter(
                x=list(history['time']),
                y=list(history['ram']),
                mode='lines+markers',
                name='RAM Usage (%)'
            )],
            'layout': go.Layout(
                title='RAM Usage Over Time',
                xaxis=dict(title='Time', tickformat='%H:%M:%S'),  # Format the time
                yaxis=dict(title='Percentage'),
            )
        }

        # Create CPU Usage Line Chart
        cpu_figure = {
            'data': [go.Scatter(
                x=list(history['time']),
                y=list(history['cpu']),
                mode='lines+markers',
                name='CPU Usage (%)'
            )],
            'layout': go.Layout(
                title='CPU Usage Over Time',
                xaxis=dict(title='Time', tickformat='%H:%M:%S'),  # Format the time
                yaxis=dict(title='Percentage'),
            )
        }

        # Create Disk Usage Line Chart
        disk_figure = {
            'data': [go.Scatter(
                x=list(history['time']),
                y=list(history['disk']),
                mode='lines+markers',
                name='Disk Usage (%)'
            )],
            'layout': go.Layout(
                title='Disk Usage Over Time',
                xaxis=dict(title='Time', tickformat='%H:%M:%S'),  # Format the time
                yaxis=dict(title='Percentage'),
            )
        }

        return ram_figure, cpu_figure, disk_figure

# Run the app
if __name__ == '__main__':
    app.run_server(debug=True)

步骤 5:运行仪表板

确保您的虚拟环境已激活。

使用以下命令运行仪表板,以显示包含所有三个指标的一个图表:

python app.py one

使用以下命令运行仪表板,以显示每个指标的三个图表:

python app.py multiple

这将启动一个本地 Web 服务器上的应用程序。打开您的浏览器并访问:

http://127.0.0.1:8050/

您将看到一个每 5 秒更新一次的仪表板,显示:

  • RAM 使用率(百分比)
  • CPU 使用率(百分比)
  • 磁盘使用率(百分比)

步骤 6:附加说明

  • psutil 是跨平台的,因此代码可以在 Windows 和 Linux 上运行而无需修改。
  • 如果您使用的是 macOS 系统,除磁盘使用情况外,所有功能应均可正常工作。如果您的文件系统配置不同,您可能需要调整 psutil.disk_usage('/')。

上述代码的解析:

  1. 设置日志系统

日志配置在脚本早期初始化,使我们能够捕获并将重要的调试信息打印到控制台。

# Set up basic logging to debug
logging.basicConfig(level=logging.INFO)

这一行配置了日志系统,以显示 INFO 级别或更高级别的消息,在程序执行期间提供有用的反馈。

  1. 初始化 Dash 应用程序

Dash 是一个强大的 Python 框架,用于构建交互式 Web 应用程序。应用程序的初始化如下:

# Initialize the Dash app
app = dash.Dash(__name__)

这将创建一个 Dash 应用程序,作为实时仪表板的基础。__name__ 参数帮助 Dash 正确定位资源。

  1. 使用 Deques 设置历史记录

Deques(双端队列)用于存储 RAM、CPU、磁盘使用情况和时间戳的最后 20 个数据点。

# Define fixed-size lists (deque) to store the last 20 data points for RAM, CPU, Disk usage, and time
history = {
    'ram': deque(maxlen=20),
    'cpu': deque(maxlen=20),
    'disk': deque(maxlen=20),
    'time': deque(maxlen=20)  # Store timestamps for x-axis
}

maxlen=20 确保内存中仅保留最后 20 个值,较旧的值会自动被移除。这对于实时图表特别有用,因为它们只需要显示有限数量的最近数据点。

  1. 获取系统统计信息

我们使用 psutil 库收集系统数据,例如 RAM、CPU 和磁盘使用情况。get_system_stats 函数负责此操作:

def get_system_stats():
    try:
        # Get memory stats
        memory = psutil.virtual_memory()
        ram = memory.percent

        # Get CPU usage
        cpu = psutil.cpu_percent(interval=1)

        # Get Disk usage
        disk = psutil.disk_usage('/').percent

        # Return RAM, CPU, and Disk data
        return {
            'RAM Usage (%)': ram,
            'CPU Usage (%)': cpu,
            'Disk Usage (%)': disk
        }
    except Exception as e:
        logging.error(f"Error fetching system stats: {e}")
        return {}

RAM 使用情况:我们调用 psutil.virtual_memory() 来获取内存信息,并提取正在使用的 RAM 百分比。

CPU 使用情况:psutil.cpu_percent() 函数返回当前的 CPU 使用率。interval=1 参数告诉函数在 1 秒的时间段内计算 CPU 使用率。

磁盘使用情况:我们使用 psutil.disk_usage('/') 来获取根目录(/)的磁盘使用百分比。

此函数收集所有这些数据并将其作为字典返回。如果发生错误,它将记录错误并返回一个空字典。

  1. 模式选择:单图或多图

仪表板可以在两种模式下运行:要么将所有数据合并在一个图表中,要么每个指标(RAM、CPU 和磁盘)都有自己的图表。这通过检查命令行参数来确定:

# Determine whether to run in 'one' or 'multiple' mode based on command-line argument
mode = sys.argv[1] if len(sys.argv) > 1 else 'multiple'

如果在运行脚本时提供了参数 one,应用程序将把所有数据合并到一个图表中。如果没有提供参数或提供了 multiple,则会显示单独的图表。

  1. 为合并图表模式创建布局和回调

在 one 模式下运行时,布局包含一个单一的图表,并且数据每 5 秒刷新一次:

if mode == 'one':
    app.layout = html.Div([
        html.H1('System Monitoring Dashboard (Combined Graph)'),

        # Combined Line Chart for RAM, CPU, and Disk
        dcc.Graph(id='combined-graph'),

        # Interval for updating the dashboard every 5 seconds
        dcc.Interval(
            id='interval-component',
            interval=5*1000,  # 5000 milliseconds (5 seconds)
            n_intervals=0
        )
    ])

该布局包括一个标题(html.H1)和一个图表组件(dcc.Graph)。dcc.Interval 组件用于每 5 秒(5000 毫秒)刷新数据。

下面的回调函数通过获取最新的系统统计信息并将其添加到历史记录双端队列中来更新合并图表:

@app.callback(
    Output('combined-graph', 'figure'),
    [Input('interval-component', 'n_intervals')]
)
def update_combined_graph(n):
    # Fetch system stats (RAM, CPU, and Disk)
    data = get_system_stats()

    if not data:
        logging.info("No data fetched")
        return {}

    # Log fetched data in the terminal
    logging.info(f"Fetched data: {data}")

    # Append the current time, RAM, CPU, and Disk usage to history
    current_time = datetime.now().strftime('%H:%M:%S')  # Get the current time as a string
    history['ram'].append(data['RAM Usage (%)'])
    history['cpu'].append(data['CPU Usage (%)'])
    history['disk'].append(data['Disk Usage (%)'])
    history['time'].append(current_time)

    # Create Combined Line Chart
    combined_figure = {
        'data': [
            go.Scatter(
                x=list(history['time']),
                y=list(history['ram']),
                mode='lines+markers',
                name='RAM Usage (%)'
            ),
            go.Scatter(
                x=list(history['time']),
                y=list(history['cpu']),
                mode='lines+markers',
                name='CPU Usage (%)'
            ),
            go.Scatter(
                x=list(history['time']),
                y=list(history['disk']),
                mode='lines+markers',
                name='Disk Usage (%)'
            )
        ],
        'layout': go.Layout(
            title='RAM, CPU, and Disk Usage Over Time',
            xaxis=dict(title='Time', tickformat='%H:%M:%S'),  # Format the time
            yaxis=dict(title='Percentage'),
        )
    }

    return combined_figure

这个回调每 5 秒获取一次数据,将其附加到历史记录双端队列中,然后创建一个合并的折线图,以显示三个指标随时间的变化。

  1. 为多图模式创建布局和回调

在多图模式下,布局由三个单独的图表组成(每个指标一个):

else:
    app.layout = html.Div([
        html.H1('System Monitoring Dashboard (Separate Graphs)'),

        # RAM Usage Line Chart
        dcc.Graph(id='ram-usage-graph'),

        # CPU Usage Line Chart
        dcc.Graph(id='cpu-usage-graph'),

        # Disk Usage Line Chart
        dcc.Graph(id='disk-usage-graph'),

        # Interval for updating the dashboard every 5 seconds
        dcc.Interval(
            id='interval-component',
            interval=5*1000,  # 5000 milliseconds (5 seconds)
            n_intervals=0
        )
    ])

在这种模式下,回调函数单独更新每个图表:

@app.callback(
    [Output('ram-usage-graph', 'figure'),
     Output('cpu-usage-graph', 'figure'),
     Output('disk-usage-graph', 'figure')],
    [Input('interval-component', 'n_intervals')]
)
def update_separate_graphs(n):
    # Fetch system stats (RAM, CPU, and Disk)
    data = get_system_stats()

    if not data:
        logging.info("No data fetched")
        return {}, {}, {}

    # Append the current time, RAM, CPU, and Disk usage to history
    current_time = datetime.now().strftime('%H:%M:%S')
    history['ram'].append(data['RAM Usage (%)'])
    history['cpu'].append(data['CPU Usage (%)'])
    history['disk'].append(data['Disk Usage (%)'])
    history['time'].append(current_time)

    # Create RAM, CPU, and Disk Usage Line Charts
    ram_figure = {
        'data': [go.Scatter(
            x=list(history['time']),
            y=list(history['ram']),
            mode='lines+markers',
            name='RAM Usage (%)'
        )],
        'layout': go.Layout(title='RAM Usage Over Time', xaxis=dict(title='Time'), yaxis=dict(title='Percentage'))
    }

    cpu_figure = {
        'data': [go.Scatter(
            x=list(history['time']),
            y=list(history['cpu']),
            mode='lines+markers',
            name='CPU Usage (%)'
        )],
        'layout': go.Layout(title='CPU Usage Over Time', xaxis=dict(title='Time'), yaxis=dict(title='Percentage'))
    }

    disk_figure = {
        'data': [go.Scatter(
            x=list(history['time']),
            y=list(history['disk']),
            mode='lines+markers',
            name='Disk Usage (%)'
        )],
        'layout': go.Layout(title='Disk Usage Over Time', xaxis=dict(title='Time'), yaxis=dict(title='Percentage'))
    }

    return ram_figure, cpu_figure, disk_figure
  1. 运行应用程序

最后,使用 app.run_server() 函数启动 Dash 应用程序:

if __name__ == '__main__':
    app.run_server(debug=True)

运行应用程序时,使用 python app.py one 将显示一个合并的图表,而使用 python app.py multiple 将为每个指标显示单独的图表。

最终效果:

原文链接:,转发请注明来源!