pip3 install -i https://pypi.tuna.tsinghua.edu.cn/simple paramiko psutil scikit-learn numpy matplotlib

import sys
import io
# 将stdout的编码设置为utf-8
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
import paramiko
import psutil
import time
import numpy as np
import matplotlib.pyplot as plt
from sklearn.linear_model import LinearRegression
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_squared_error

# 连接远程Linux机器的函数
def get_cpu_usage(ip, username, password):
    # 使用 paramiko 库连接到远程 Linux 机器
    ssh_client = paramiko.SSHClient()
    ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh_client.connect(ip, username=username, password=password)

    # 执行命令获取CPU使用率
    stdin, stdout, stderr = ssh_client.exec_command("top -bn1 | grep 'Cpu(s)' | sed 's/.*, *\\([0-9.]*\\)%* id.*/\\1/' | awk '{print 100 - $1}'")
    cpu_usage = float(stdout.read().decode('utf-8').strip())

    ssh_client.close()
    return cpu_usage

# 获取一定时间间隔的CPU使用率数据
def collect_cpu_data(ip, username, password, duration=120, interval=6):
    cpu_data = []
    for _ in range(duration // interval):
        cpu_usage = get_cpu_usage(ip, username, password)
        cpu_data.append(cpu_usage)
        time.sleep(interval)
    return cpu_data

# 训练模型预测CPU使用率
def train_model(cpu_data):
    # 创建特征数据集(时间步长,CPU使用率)
    X = np.array(range(len(cpu_data))).reshape(-1, 1)
    # 时间步长,假设cpu_data有十个数据,那range(10)会返回列表[0,1,2,3,...9]
    #X(n,1)的二维数组,n是数据点的数量
    y = np.array(cpu_data)  # CPU使用率,[10%,16%,18%,...88%]

    # 分割训练集和测试集,20%测试,80%训练
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=100)

    # 创建并训练线性回归模型
    model = LinearRegression()
    model.fit(X_train, y_train)

    # 预测并评估模型
    y_pred = model.predict(X_test)
    mse = mean_squared_error(y_test, y_pred)
    print(f"模型的均方误差 (MSE): {mse:.2f}")

    return model

# 预测未来CPU使用率
def predict_cpu_usage(model, steps=10):
    last_time = model.coef_[0] * (len(cpu_data) - 1) + model.intercept_
    future_steps = np.array(range(len(cpu_data), len(cpu_data) + steps)).reshape(-1, 1)
    predictions = model.predict(future_steps)
    return predictions

# 配置远程Linux机器的连接信息
ip = '192.168.40.183'
username = 'root'
password = 'root'

# 收集数据并训练模型
cpu_data = collect_cpu_data(ip, username, password)
model = train_model(cpu_data)

# 预测未来10个时间步的CPU使用率
future_predictions = predict_cpu_usage(model, steps=10)

# 用实际的时间(秒)作为 X 轴
time_steps = np.arange(0, len(cpu_data) * 5, 5)  # 每个时间步是5秒,所以对应 0, 5, 10, ..., 45
future_time_steps = np.arange(len(cpu_data) * 5, (len(cpu_data) + 10) * 5, 5)  # 预测的时间步

plt.rcParams['font.family'] = ['Microsoft YaHei']  # 使用微软雅黑字体以支持中文显示

# 可视化数据和预测结果
plt.plot(time_steps, cpu_data, label='历史CPU使用率')
plt.plot(future_time_steps, future_predictions, label='预测的CPU使用率', linestyle='--')
plt.xlabel('时间(秒)')
plt.ylabel('CPU使用率 (%)')
plt.legend()
plt.show()

def collect_cpu_data(ip, username, password, duration=300, interval=5):  # 延长采集时间到 300 秒
    cpu_data = []
    for _ in range(duration // interval):
        cpu_usage = get_cpu_usage(ip, username, password)
        cpu_data.append(cpu_usage)
        time.sleep(interval)
return cpu_data

from sklearn.ensemble import RandomForestRegressor

def train_model(cpu_data):
    X = np.array(range(len(cpu_data))).reshape(-1, 1)
    y = np.array(cpu_data)
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

    # 使用随机森林回归模型
    model = RandomForestRegressor(n_estimators=100, random_state=42)
    model.fit(X_train, y_train)

    y_pred = model.predict(X_test)
    mse = mean_squared_error(y_test, y_pred)
    print(f"模型的均方误差 (MSE): {mse:.2f}")
    return model

from sklearn.model_selection import GridSearchCV

def train_model(cpu_data):
    X = np.array(range(len(cpu_data))).reshape(-1, 1)
    y = np.array(cpu_data)
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

    # 使用网格搜索调优随机森林参数
    param_grid = {
        'n_estimators': [50, 100, 200],
        'max_depth': [None, 10, 20]
    }
    model = GridSearchCV(RandomForestRegressor(random_state=42), param_grid, cv=5)
    model.fit(X_train, y_train)

    print(f"最佳参数: {model.best_params_}")
    y_pred = model.predict(X_test)
    mse = mean_squared_error(y_test, y_pred)
    print(f"模型的均方误差 (MSE): {mse:.2f}")
    return model

def create_sliding_window_features(cpu_data, window_size=5):
    X = []
    y = []
    for i in range(len(cpu_data) - window_size):
        X.append(cpu_data[i:i + window_size])
        y.append(cpu_data[i + window_size])
    return np.array(X), np.array(y)

def train_model(cpu_data):
    X, y = create_sliding_window_features(cpu_data, window_size=5)
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

    model = RandomForestRegressor(n_estimators=100, random_state=42)
    model.fit(X_train, y_train)

    y_pred = model.predict(X_test)
    mse = mean_squared_error(y_test, y_pred)
    print(f"模型的均方误差 (MSE): {mse:.2f}")
    return model

from sklearn.metrics import mean_absolute_error, r2_score

def train_model(cpu_data):
    X = np.array(range(len(cpu_data))).reshape(-1, 1)
    y = np.array(cpu_data)
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

    model = RandomForestRegressor(n_estimators=100, random_state=42)
    model.fit(X_train, y_train)

    y_pred = model.predict(X_test)
    mse = mean_squared_error(y_test, y_pred)
    mae = mean_absolute_error(y_test, y_pred)
    r2 = r2_score(y_test, y_pred)
    print(f"模型的均方误差 (MSE): {mse:.2f}")
    print(f"模型的平均绝对误差 (MAE): {mae:.2f}")
    print(f"模型的 R² 分数: {r2:.2f}")
    return model

def plot_results(time_steps, cpu_data, future_time_steps, future_predictions):
    plt.figure(figsize=(10, 6))
    plt.plot(time_steps, cpu_data, label='历史CPU使用率', marker='o')
    plt.plot(future_time_steps, future_predictions, label='预测的CPU使用率', linestyle='--', marker='x')
    plt.fill_between(future_time_steps, future_predictions - 5, future_predictions + 5, alpha=0.2, label='预测区间')
    plt.xlabel('时间(秒)')
    plt.ylabel('CPU使用率 (%)')
    plt.title('CPU使用率预测')
    plt.legend()
    plt.grid(True)
    plt.show()

from sklearn.tree import DecisionTreeClassifier

# 准备数据
X = [[1, 0], [0, 1], [2, 1], [0, 0], [1, 2]]  # 邮件特征("免费"和"优惠"出现次数)
y = [1, 0, 1, 0, 1]  # 标签(1表示垃圾邮件,0表示非垃圾邮件)

# 创建决策树分类器
model = DecisionTreeClassifier()

# 训练模型
model.fit(X, y)

# 输出决策树的规则
print("决策树的规则:", model.tree_)

# 使用模型预测新邮件的类别
new_email = [[1, 1]]  # 新邮件的特征("免费"出现1次,"优惠"出现1次)
prediction = model.predict(new_email)
print("预测结果:", "垃圾邮件" if prediction[0] == 1 else "非垃圾邮件")

import sys
import io
# 将stdout的编码设置为utf-8
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')

import tensorflow as tf
# 导入TensorFlow库
from tensorflow.keras.models import Sequential
# 导入Sequential模型
from tensorflow.keras.layers import Dense, Flatten, Conv2D, MaxPooling2D, Dropout
# 导入Dense、Flatten、Conv2D、MaxPooling2D、Dropout层
from tensorflow.keras.utils import to_categorical
# 导入to_categorical函数,用于将标签转换为one-hot编码
import numpy as np
# 导入numpy库,用于处理数据
import matplotlib.pyplot as plt
# 导入matplotlib.pyplot库,用于绘制图像
import matplotlib.font_manager as fm  # 用于加载中文字体
# 导入matplotlib.font_manager库,用于加载中文字体
import random  # 用于随机选择图片

"""
导入相关的库:
   - tensorflow:用于构建和训练神经网络。
   - keras:这是 TensorFlow 的高级接口,提供更简便的 API 来定义和训练神经网络模型。
   - numpy:用于处理数据和数组操作。
   - matplotlib:用于绘制图像和显示图片。
   - random:用于随机选择测试集中的一张图片。
"""

# 1. 加载 CIFAR-10 数据集
(x_train, y_train), (x_test, y_test) = tf.keras.datasets.cifar10.load_data()

"""
加载 CIFAR-10 数据集:
   CIFAR-10 是一个包含 10 类图像的数据集。每类有 60000 张 32x32 彩色图像。数据集分为训练集和测试集。
   - x_train:训练集图像数据,是训练集中的图像数据,包含 50,000 张 32x32 像素的彩色图像(每张图像有 3 个颜色通道:红色、绿色、蓝色)。
   - y_train:训练集标签数据,形状是 (50000, 1),是训练集的标签,表示每张图像的类别(0 到 9,代表不同的物体类别)。
   - x_test:测试集图像数据,形状是 (10000, 32, 32, 3)。
   - y_test:测试集标签数据,形状是 (10000, 1)。是测试集的标签,表示每张图像的真实类别(0 到 9,表示不同的物体类别)。
"""

# 2. 数据预处理
x_train, x_test = x_train / 255.0, x_test / 255.0  # 归一化到 0-1 范围
y_train, y_test = to_categorical(y_train), to_categorical(y_test)  # one-hot 编码

"""
 数据预处理:
   - 图像数据归一化:图像的像素值范围从 [0, 255],我们将其归一化到 [0, 1]。这样做可以加速模型的训练,使其更稳定。
   - one-hot 编码:标签数据是整数型(0-9),代表类别。为了适应神经网络的输出,我们将它们转换为 one-hot 编码形式。
   例如,类别"0"会变成 [1, 0, 0, 0, 0, 0, 0, 0, 0, 0],类别"1"会变成 [0, 1, 0, 0, 0, 0, 0, 0, 0, 0]。
"""

# 3. 构建卷积神经网络模型
model = Sequential([
    Conv2D(32, (3, 3), activation='relu', input_shape=(32, 32, 3)),
    MaxPooling2D((2, 2)),
    Conv2D(64, (3, 3), activation='relu'),
    MaxPooling2D((2, 2)),
    Conv2D(64, (3, 3), activation='relu'),
    Flatten(),
    Dense(64, activation='relu'),
    Dropout(0.5),  # 添加 Dropout 层,防止过拟合
    Dense(10, activation='softmax')
])

"""
构建卷积神经网络(CNN)模型:
   - Sequential:表示模型是一个层的线性堆叠。
   - Conv2D:卷积层,用于从图像中提取特征。
     - 第一个卷积层使用 32 个 3x3 的卷积核,输入的图像形状是 (32, 32, 3)。
     - 第二、第三个卷积层使用 64 个 3x3 的卷积核。
   - MaxPooling2D:池化层,用于减少数据的尺寸,从而减少计算量和防止过拟合。我们使用 2x2 的池化窗口。
   - Flatten:将多维的输入一维化,通常用于连接到全连接层。
   - Dense:全连接层。第一个全连接层有 64 个神经元,第二个(输出层)有 10 个神经元,分别对应 10 个类别。
   - Dropout:为了防止过拟合,在全连接层后加入 Dropout 层,丢弃 50% 的神经元。
"""

# 4. 编译模型
model.compile(optimizer='adam', loss='categorical_crossentropy', metrics=['accuracy'])

"""
编译模型:
   - optimizer='adam':使用 Adam 优化器,它是常用的自适应学习率优化算法。
   - loss='categorical_crossentropy':选择类别交叉熵损失函数,用于多分类问题。
   - metrics=['accuracy']:在训练和评估过程中我们关注的指标是分类准确率(accuracy)。
"""

# 5. 训练模型
model.fit(x_train, y_train, epochs=10, batch_size=64, validation_data=(x_test, y_test))  # 增加训练轮数,批量大小设置为64

"""
训练模型:
   - x_train 和 y_train:训练数据和标签。
   - epochs=2:训练的轮数(每次使用整个训练集训练一次)。通常需要多个轮次来优化模型,但本例中使用 2 轮以减少计算量。
   - batch_size=10:每次更新模型参数时使用 10 张图片。较大的批量大小可能会提高训练速度,但也需要更多内存。
   - validation_data=(x_test, y_test):使用测试集进行验证,以便跟踪模型在训练过程中的表现。
"""

# 保存模型
model.save("cifar10_model.h5")

# CIFAR-10 标签名称(中文)
labels = ['飞机', '汽车', '鸟', '猫', '鹿', '狗', '青蛙', '马', '船', '卡车']

# 6. 设置中文字体
font_path = "C:/Windows/Fonts/simhei.ttf"  # 适用于Windows的字体路径
my_font = fm.FontProperties(fname=font_path)

# 7. 随机选择测试集中的一张图片进行预测
random_index = random.randint(0, len(x_test) - 1)  # 随机选择一个索引
test_image = x_test[random_index]
test_label = np.argmax(y_test[random_index])  # 真实标签

# 使用模型进行预测
prediction = model.predict(np.expand_dims(test_image, axis=0))
predicted_label = np.argmax(prediction)  # 预测结果

# 显示图片和预测结果
plt.imshow(test_image)
plt.title(f"预测: {labels[predicted_label]}, 实际: {labels[test_label]}", fontproperties=my_font)
plt.axis('off')
# 显示图形
plt.show()

"""
随机选择测试集中的一张图片进行预测:
   - random.randint(0, len(x_test) - 1):生成一个随机整数,作为测试集图片的索引。
   - test_image:获取随机选中的测试图像。
   - test_label:获取该图像的真实标签,使用 `np.argmax` 获取 one-hot 编码的索引值。
   - 使用训练好的模型进行预测,`model.predict` 返回的是一个 10 维的概率分布,我们通过 `np.argmax` 获取最大概率对应的类别。
   - 显示预测结果:使用 Matplotlib 显示图片,并在标题中显示模型的预测类别和真实类别。
"""

#通过加载训练好的 CIFAR-10 分类模型,对上传的图片进行分类,并显示预测结果

import tkinter as tk
from tkinter import filedialog, Label, Button, Canvas
from PIL import Image, ImageTk, ImageOps
import numpy as np
import tensorflow as tf
import matplotlib.font_manager as fm

# 加载已训练的模型
model = tf.keras.models.load_model("C:/代码/图片分类大模型开发/cifar10_model.h5")

# CIFAR-10 标签名称(中文)
labels = ['飞机', '汽车', '鸟', '猫', '鹿', '狗', '青蛙', '马', '船', '卡车']

# 设置字体
font_path = "C:/Windows/Fonts/simhei.ttf"  # 适用于Windows的字体路径
my_font = fm.FontProperties(fname=font_path)

# 创建主窗口
window = tk.Tk()
window.title("CIFAR-10 图像分类器")
window.geometry("700x700")

# 图片显示 Canvas
canvas = Canvas(window, width=500, height=500)
canvas.pack(pady=10)

result_label = Label(window, text="", font=("Arial", 14))
result_label.pack(pady=10)

def preprocess_image(image_path):
    """加载和预处理图像,使其符合模型输入格式"""
    image = Image.open(image_path)
    image = ImageOps.fit(image, (32, 32), Image.LANCZOS)  # 使用 LANCZOS 缩放图像
    image = np.array(image) / 255.0  # 归一化
    return image.reshape((1, 32, 32, 3))  # 添加批次维度

def classify_image():
    """选择图片并进行分类"""
    file_path = filedialog.askopenfilename()
    if file_path:
        # 预处理图像
        img_array = preprocess_image(file_path)

        # 使用模型进行预测
        prediction = model.predict(img_array)
        predicted_label = np.argmax(prediction)  # 获取预测类别

        # 显示预测结果
        result_label.config(text=f"预测结果: {labels[predicted_label]}", font=my_font)

        # 在界面显示选择的图片
        display_image = Image.open(file_path)
        display_image.thumbnail((600, 600))  # 初始缩放图片的最大尺寸
        img = ImageTk.PhotoImage(display_image)

        # 清除旧图像并显示新图像
        canvas.delete("all")
        canvas.create_image(0, 0, anchor="nw", image=img)
        canvas.image = img  # 需要保存引用

# 上传按钮
upload_btn = Button(window, text="上传图片并预测", command=classify_image)
upload_btn.pack(pady=20)

# 运行主循环
window.mainloop()

import sys
import io
# 将stdout的编码设置为utf-8
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')

import os  # 用于操作文件和目录
import pandas as pd  # 用于数据处理和分析
from sklearn.feature_extraction.text import TfidfVectorizer  # 用于文本特征提取,将文本转换为TF-IDF特征
from sklearn.model_selection import train_test_split  # 用于将数据集划分为训练集和测试集
from sklearn.ensemble import RandomForestClassifier  # 随机森林分类器,用于构建分类模型
from sklearn.metrics import classification_report  # 用于评估分类模型的性能,包括精确率、召回率等指标

# Step 1: 数据预处理与加载日志文件
def load_logs(log_file_path):
    if not os.path.exists(log_file_path):
        raise FileNotFoundError(f"Log file '{log_file_path}' not found.")

    # 假设每行是一条日志
    with open(log_file_path, 'r', encoding='utf-8') as file:
        logs = file.readlines()
    return logs

# Step 2: 构建示例数据集并标记数据(真实场景需根据具体情况标注)
def create_dataset():
    data = {
        'log': [
            'User login successful',
            'Disk space running low',
            'Error: Unable to connect to database',
            'Backup completed successfully',
            'Warning: High memory usage detected',
            'User logout',
            'Critical: System crash',
            'Service restarted',
        ],
        'label': [0, 1, 1, 0, 1, 0, 1, 0]  # 0: 正常, 1: 异常
    }
    return pd.DataFrame(data)

# Step 3: TF-IDF 特征提取与模型训练
def train_model(dataset):
    vectorizer = TfidfVectorizer()
    X = vectorizer.fit_transform(dataset['log'])
    y = dataset['label']

    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

    model = RandomForestClassifier(random_state=42)
    model.fit(X_train, y_train)

    y_pred = model.predict(X_test)
    print("Classification Report:\n", classification_report(y_test, y_pred))

    return model, vectorizer

# Step 4: 实时检测日志
def detect_anomalies(logs, model, vectorizer):
    features = vectorizer.transform(logs)
    predictions = model.predict(features)
    for log, label in zip(logs, predictions):
        status = "异常" if label == 1 else "正常"
        print(f"日志: {log.strip()} -> 检测结果: {status}")

# 示例用法
def main():
    # 创建训练数据
    dataset = create_dataset()

    # 训练模型
    model, vectorizer = train_model(dataset)

    # 加载实际日志文件(替换为自己的日志文件路径)
    log_file_path = r'F:\\1-Python 全栈运维开发:从基础到企业级实战\\班级直播主题内容\\第5场直播:Python赋能AI:在DevOps领域的应用与实践\\path_to_your_log_file.log'  # 修改为实际路径
    try:
        logs = load_logs(log_file_path)
    except FileNotFoundError as e:
        print(e)
        return

    # 实时检测
    detect_anomalies(logs, model, vectorizer)

if __name__ == "__main__":
    main()