0.完整代码
import os
import pandas as pd
import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader, random_split
from torchvision import transforms, models
from PIL import Image
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.metrics import confusion_matrix, mean_squared_error, r2_score
import tkinter as tk
from tkinter import filedialog, ttk, messagebox
from PIL import ImageTk
import datetime
# 设置matplotlib中文字体和清晰度
plt.rcParams["font.sans-serif"] = ["Microsoft YaHei"]
plt.rcParams["figure.dpi"] = 100
plt.rcParams["font.size"] = 10
# 数据集类
class WeatherImageDataset(Dataset):
"""天气图像数据集"""
def __init__(self, data_dir, transform=None):
self.data_dir = data_dir
self.transform = transform
self.weather_classes = ['Cloudy', 'Rain', 'Shine', 'Sunrise']
self.samples = []
self._load_images()
def _load_images(self):
"""加载天气图像"""
for class_idx, class_name in enumerate(self.weather_classes):
class_dir = os.path.join(self.data_dir, class_name)
if os.path.exists(class_dir):
for img_name in os.listdir(class_dir):
if img_name.lower().endswith(('.jpg', '.jpeg', '.png')):
self.samples.append({
'image_path': os.path.join(class_dir, img_name),
'label': class_idx,
'weather_type': class_name
})
print(f"加载了 {len(self.samples)} 张天气图像")
def __len__(self):
return len(self.samples)
def __getitem__(self, idx):
sample = self.samples[idx]
image = Image.open(sample['image_path']).convert('RGB')
if self.transform:
image = self.transform(image)
label = torch.tensor(sample['label'], dtype=torch.long)
return image, label
class TemperatureDataset(Dataset):
"""温度时序数据集"""
def __init__(self, csv_path, sequence_length=7, prediction_horizon=1):
self.sequence_length = sequence_length
self.prediction_horizon = prediction_horizon
self.data = pd.read_csv(csv_path)
self._prepare_data()
def _prepare_data(self):
"""准备温度数据"""
# 使用温度数据（如果没有温度列，使用相近的列）
if 'Temp' in self.data.columns:
temp_data = self.data['Temp'].values
elif 'temperature (C)' in self.data.columns:
temp_data = self.data['temperature (C)'].values
elif 'Temperature (C)' in self.data.columns:
temp_data = self.data['Temperature (C)'].values
else:
# 使用第一列数值数据作为温度
temp_data = self.data.iloc[:, 0].values
# 归一化
self.temp_min = temp_data.min()
self.temp_max = temp_data.max()
temp_data = (temp_data - self.temp_min) / (self.temp_max - self.temp_min)
# 创建序列
self.sequences = []
self.targets = []
for i in range(len(temp_data) - self.sequence_length - self.prediction_horizon):
sequence = temp_data[i:i + self.sequence_length]
target = temp_data[i + self.sequence_length + self.prediction_horizon - 1]
self.sequences.append(sequence)
self.targets.append(target)
def denormalize(self, temp):
"""反归一化温度"""
return temp * (self.temp_max - self.temp_min) + self.temp_min
def __len__(self):
return len(self.sequences)
def __getitem__(self, idx):
sequence = torch.FloatTensor(self.sequences[idx])
target = torch.FloatTensor([self.targets[idx]])
return sequence, target
# 定义CNN模型（天气图像分类）
class WeatherCNN(nn.Module):
"""使用ResNet18进行天气图像分类"""
def __init__(self, num_classes=4):
super(WeatherCNN, self).__init__()
# 加载预训练的ResNet18模型
self.model = models.resnet18(pretrained=True)
# 修改最后一层以适应我们的分类任务
num_features = self.model.fc.in_features
self.model.fc = nn.Linear(num_features, num_classes)
def forward(self, x):
return self.model(x)
# 定义RNN模型（温度预测）
class TemperatureRNN(nn.Module):
"""使用LSTM进行温度预测"""
def __init__(self, input_size=1, hidden_size=64, num_layers=2, output_size=1):
super(TemperatureRNN, self).__init__()
self.hidden_size = hidden_size
self.num_layers = num_layers
# LSTM层
self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
# 全连接层
self.fc = nn.Linear(hidden_size, output_size)
def forward(self, x):
# 输入形状：(batch_size, seq_len)
# 重塑为LSTM输入形状: (batch_size, seq_len, input_size)
x = x.unsqueeze(2)
# 初始化隐藏状态
h0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(x.device)
c0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(x.device)
# LSTM前向传播
out, _ = self.lstm(x, (h0, c0))
# 取最后一个时间步的输出
out = self.fc(out[:, -1, :])
return out
# 训练CNN模型进行天气图像分类
def train_cnn_model(data_dir, batch_size=128, num_epochs=50, learning_rate=0.001):
print("=" * 50)
print("开始CNN模型训练过程...")
print(f"使用设备: {torch.device('cuda' if torch.cuda.is_available() else 'cpu')}")
print(f"批次大小: {batch_size}, 训练轮数: {num_epochs}, 学习率: {learning_rate}")
print("=" * 50)
# 数据预处理和增强
transform_train = transforms.Compose([
transforms.Resize((224, 224)),
transforms.RandomHorizontalFlip(),
transforms.RandomRotation(10),
transforms.ColorJitter(brightness=0.1, contrast=0.1),
transforms.ToTensor(),
transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
])
transform_test = transforms.Compose([
transforms.Resize((224, 224)),
transforms.ToTensor(),
transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
])
# 加载数据集
print("正在加载天气图像数据集...")
full_dataset = WeatherImageDataset(data_dir, transform=None)
# 拆分数据集
train_size = int(0.7 * len(full_dataset))
val_size = int(0.15 * len(full_dataset))
test_size = len(full_dataset) - train_size - val_size
print(f"数据集划分: 训练集 {train_size}张, 验证集 {val_size}张, 测试集 {test_size}张")
# 使用随机种子确保可重复性
torch.manual_seed(42)
train_dataset, val_dataset, test_dataset = random_split(
full_dataset, [train_size, val_size, test_size]
)
# 应用不同的转换
class TransformedSubset(Dataset):
def __init__(self, subset, transform=None):
self.subset = subset
self.transform = transform
def __getitem__(self, idx):
x, y = self.subset[idx]
if self.transform:
x = self.transform(x)
return x, y
def __len__(self):
return len(self.subset)
train_dataset = TransformedSubset(train_dataset, transform_train)
val_dataset = TransformedSubset(val_dataset, transform_test)
test_dataset = TransformedSubset(test_dataset, transform_test)
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=batch_size)
test_loader = DataLoader(test_dataset, batch_size=batch_size)
# 初始化模型
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = WeatherCNN(num_classes=4).to(device)
# 定义损失函数和优化器
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=learning_rate)
scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, 'min', patience=3, factor=0.5)
# 记录训练过程
train_losses = []
train_accs = []
val_losses = []
val_accs = []
# 训练模型
best_val_acc = 0.0
print("开始训练...")
for epoch in range(num_epochs):
# 训练阶段
model.train()
running_loss = 0.0
correct = 0
total = 0
for images, labels in train_loader:
images, labels = images.to(device), labels.to(device)
optimizer.zero_grad()
outputs = model(images)
loss = criterion(outputs, labels)
loss.backward()
optimizer.step()
running_loss += loss.item()
# 计算训练精度
_, predicted = torch.max(outputs.data, 1)
total += labels.size(0)
correct += (predicted == labels).sum().item()
train_loss = running_loss / len(train_loader)
train_acc = correct / total
train_losses.append(train_loss)
train_accs.append(train_acc)
# 验证阶段
model.eval()
val_correct = 0
val_total = 0
val_running_loss = 0.0
with torch.no_grad():
for images, labels in val_loader:
images, labels = images.to(device), labels.to(device)
outputs = model(images)
loss = criterion(outputs, labels)
val_running_loss += loss.item()
_, predicted = torch.max(outputs.data, 1)
val_total += labels.size(0)
val_correct += (predicted == labels).sum().item()
val_loss = val_running_loss / len(val_loader)
val_acc = val_correct / val_total
val_losses.append(val_loss)
val_accs.append(val_acc)
# 学习率调整
scheduler.step(val_loss)
print(f'Epoch [{epoch + 1}/{num_epochs}], Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.4f}, '
f'Val Loss: {val_loss:.4f}, Val Acc: {val_acc:.4f}')
# 保存最佳模型
if val_acc > best_val_acc:
best_val_acc = val_acc
torch.save(model.state_dict(), 'models/best_cnn_model.pth')
print(f"保存最佳模型，验证准确率: {best_val_acc:.4f}")
print(f'训练完成! 最佳验证准确率: {best_val_acc:.4f}')
# 绘制训练曲线
plot_training_curves(train_losses, val_losses, "Loss", "models/cnn_loss_curve.png")
plot_training_curves(train_accs, val_accs, "Accuracy", "models/cnn_accuracy_curve.png")
# 加载最佳模型并在测试集上评估
model.load_state_dict(torch.load('models/best_cnn_model.pth'))
model.eval()
# 评估模型并生成混淆矩阵
evaluate_cnn_model(model, test_loader, device, full_dataset.weather_classes)
return model, full_dataset.weather_classes
# 训练RNN模型进行温度预测
def train_rnn_model(csv_path, batch_size=128, num_epochs=30, learning_rate=0.001):
print("=" * 50)
print("开始RNN模型训练过程...")
print(f"使用设备: {torch.device('cuda' if torch.cuda.is_available() else 'cpu')}")
print(f"批次大小: {batch_size}, 训练轮数: {num_epochs}, 学习率: {learning_rate}")
print("=" * 50)
# 加载数据集
sequence_length = 7 # 使用前7天的数据
prediction_horizon = 1 # 预测未来1天
print("正在加载温度时序数据...")
dataset = TemperatureDataset(csv_path, sequence_length, prediction_horizon)
# 拆分数据集
train_size = int(0.7 * len(dataset))
val_size = int(0.15 * len(dataset))
test_size = len(dataset) - train_size - val_size
print(f"数据集划分: 训练集 {train_size}个序列, 验证集 {val_size}个序列, 测试集 {test_size}个序列")
# 使用随机种子确保可重复性
torch.manual_seed(42)
train_dataset, val_dataset, test_dataset = random_split(
dataset, [train_size, val_size, test_size]
)
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=batch_size)
test_loader = DataLoader(test_dataset, batch_size=batch_size)
# 初始化模型
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = TemperatureRNN(input_size=1, hidden_size=64, num_layers=2, output_size=1).to(device)
# 定义损失函数和优化器
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=learning_rate)
scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, 'min', patience=5, factor=0.5)
# 记录训练过程
train_losses = []
val_losses = []
# 训练模型
best_val_loss = float('inf')
print("开始训练...")
for epoch in range(num_epochs):
# 训练阶段
model.train()
running_loss = 0.0
for sequences, targets in train_loader:
sequences, targets = sequences.to(device), targets.to(device)
optimizer.zero_grad()
outputs = model(sequences)
loss = criterion(outputs, targets)
loss.backward()
optimizer.step()
running_loss += loss.item()
train_loss = running_loss / len(train_loader)
train_losses.append(train_loss)
# 验证阶段
model.eval()
val_running_loss = 0.0
with torch.no_grad():
for sequences, targets in val_loader:
sequences, targets = sequences.to(device), targets.to(device)
outputs = model(sequences)
loss = criterion(outputs, targets)
val_running_loss += loss.item()
val_loss = val_running_loss / len(val_loader)
val_losses.append(val_loss)
# 学习率调整
scheduler.step(val_loss)
print(f'Epoch [{epoch + 1}/{num_epochs}], Train Loss: {train_loss:.6f}, Val Loss: {val_loss:.6f}')
# 保存最佳模型
if val_loss < best_val_loss:
best_val_loss = val_loss
torch.save(model.state_dict(), 'models/best_rnn_model.pth')
print(f"保存最佳模型，验证损失: {best_val_loss:.6f}")
print(f'训练完成! 最佳验证损失: {best_val_loss:.6f}')
# 绘制训练曲线
plot_training_curves(train_losses, val_losses, "Loss", "models/rnn_loss_curve.png")
# 加载最佳模型并在测试集上评估
model.load_state_dict(torch.load('models/best_rnn_model.pth'))
model.eval()
# 评估模型性能
evaluate_rnn_model(model, test_loader, device, dataset)
return model, dataset
def plot_training_curves(train_metric, val_metric, metric_name, save_path):
"""绘制训练和验证指标曲线"""
plt.figure(figsize=(10, 6))
plt.plot(train_metric, label=f'Training {metric_name}')
plt.plot(val_metric, label=f'Validation {metric_name}')
plt.xlabel('Epochs')
plt.ylabel(metric_name)
plt.legend()
plt.title(f'Training and Validation {metric_name}')
plt.grid(True)
plt.savefig(save_path)
plt.close()
print(f"训练曲线已保存至 {save_path}")
def evaluate_cnn_model(model, test_loader, device, class_names):
"""评估CNN模型并生成混淆矩阵"""
model.eval()
# 收集预测和实际标签
all_preds = []
all_labels = []
correct = 0
total = 0
with torch.no_grad():
for images, labels in test_loader:
images, labels = images.to(device), labels.to(device)
outputs = model(images)
_, predicted = torch.max(outputs.data, 1)
all_preds.extend(predicted.cpu().numpy())
all_labels.extend(labels.cpu().numpy())
total += labels.size(0)
correct += (predicted == labels).sum().item()
# 计算准确率
test_acc = correct / total
print(f"测试集准确率: {test_acc:.4f}")
# 生成混淆矩阵
cm = confusion_matrix(all_labels, all_preds)
# 绘制混淆矩阵
plt.figure(figsize=(10, 8))
sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', xticklabels=class_names, yticklabels=class_names)
plt.xlabel('预测标签')
plt.ylabel('真实标签')
plt.title('CNN模型测试集混淆矩阵')
plt.tight_layout()
plt.savefig('models/cnn_confusion_matrix.png')
plt.close()
print("CNN模型混淆矩阵已保存至 models/cnn_confusion_matrix.png")
# 计算每个类别的准确率
print("\n各类别准确率:")
for i, class_name in enumerate(class_names):
class_correct = cm[i, i]
class_total = np.sum(cm[i, :])
class_acc = class_correct / class_total if class_total > 0 else 0
print(f"{class_name}: {class_acc:.4f} ({class_correct}/{class_total})")
def evaluate_rnn_model(model, test_loader, device, dataset):
"""评估RNN模型并生成预测图"""
model.eval()
test_loss = 0.0
actual_temps = []
predicted_temps = []
with torch.no_grad():
for sequences, targets in test_loader:
sequences, targets = sequences.to(device), targets.to(device)
outputs = model(sequences)
loss = nn.MSELoss()(outputs, targets)
test_loss += loss.item()
# 记录实际和预测温度（反归一化）
for i in range(len(outputs)):
actual_temps.append(dataset.denormalize(targets[i].item()))
predicted_temps.append(dataset.denormalize(outputs[i].item()))
test_loss /= len(test_loader)
print(f'测试集平均损失: {test_loss:.6f}')
# 计算评估指标
rmse = np.sqrt(mean_squared_error(actual_temps, predicted_temps))
r2 = r2_score(actual_temps, predicted_temps)
mae = np.mean(np.abs(np.array(actual_temps) - np.array(predicted_temps)))
print(f'RMSE: {rmse:.4f}°C')
print(f'R²: {r2:.4f}')
print(f'MAE: {mae:.4f}°C')
# 绘制实际温度与预测温度对比图
plt.figure(figsize=(12, 6))
# 只绘制前100个点，以便更清晰
n_samples = min(100, len(actual_temps))
plt.plot(range(n_samples), actual_temps[:n_samples], 'b-', label='实际温度')
plt.plot(range(n_samples), predicted_temps[:n_samples], 'r--', label='预测温度')
plt.xlabel('样本')
plt.ylabel('温度 (°C)')
plt.title('RNN模型温度预测结果')
plt.legend()
plt.grid(True)
plt.savefig('models/rnn_prediction_results.png')
plt.close()
# 绘制散点图
plt.figure(figsize=(8, 8))
plt.scatter(actual_temps, predicted_temps, alpha=0.5)
plt.plot([min(actual_temps), max(actual_temps)], [min(actual_temps), max(actual_temps)], 'r--')
plt.xlabel('实际温度 (°C)')
plt.ylabel('预测温度 (°C)')
plt.title(f'RNN模型预测散点图 (R²={r2:.4f}, RMSE={rmse:.4f}°C)')
plt.grid(True)
plt.savefig('models/rnn_scatter_plot.png')
plt.close()
print("RNN模型评估图表已保存")
# 穿衣推荐系统
def get_clothing_recommendation(weather_type, temperature):
"""根据天气和温度给出穿衣建议"""
clothing_options = {
0: "轻薄衣物 (短袖、短裤)",
1: "适中衣物 (长袖、薄外套)",
2: "保暖衣物 (毛衣、厚外套)",
3: "厚重衣物 (羽绒服、围巾)"
}
# 基于规则的建议（可以替换为学习到的模型）
if weather_type == "Rain":
base_score = 2 # 雨天需要更多保暖
elif weather_type == "Sunrise": # 将Sunrise视为特殊情况，通常是早晨较凉
base_score = 1
elif weather_type == "Cloudy":
base_score = 1 # 阴天适中
else: # Shine/晴天
base_score = 0 # 晴天可以穿轻薄衣物
# 根据温度调整
if temperature < 5:
final_score = 3 # 非常冷
elif temperature < 15:
final_score = max(2, base_score + 1) # 冷
elif temperature < 22:
final_score = max(1, base_score) # 凉爽
else:
final_score = min(1, base_score) # 温暖/热
# 雨天特殊处理
additional_tips = ""
if weather_type == "Rain":
additional_tips = "别忘了带雨伞!"
elif weather_type == "Shine" and temperature > 25:
additional_tips = "别忘了涂防晒霜和戴帽子!"
return clothing_options[final_score], final_score, additional_tips
# GUI界面实现
class WeatherClothingApp:
def __init__(self, root):
self.root = root
self.root.title("智能天气穿衣推荐系统")
self.root.geometry("800x600")
# 加载模型
self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# 初始化模型（稍后加载）
self.cnn_model = WeatherCNN(num_classes=4)
self.rnn_model = TemperatureRNN(input_size=1, hidden_size=64, num_layers=2)
self.temp_dataset = None
# 图像转换
self.transform = transforms.Compose([
transforms.Resize((224, 224)),
transforms.ToTensor(),
transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
])
# 类别标签
self.weather_classes = ['Cloudy', 'Rain', 'Shine', 'Sunrise']
# 创建UI
self.create_ui()
# 加载预训练模型
self.load_models()
def create_ui(self):
# 主界面分为左右两部分
main_frame = ttk.Frame(self.root, padding="10")
main_frame.pack(fill=tk.BOTH, expand=True)
# 左侧：图像上传和天气预测
left_frame = ttk.LabelFrame(main_frame, text="天气图像分析", padding="10")
left_frame.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=5, pady=5)
# 图像预览
self.image_frame = ttk.Frame(left_frame, width=300, height=300)
self.image_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
self.image_label = ttk.Label(self.image_frame)
self.image_label.pack(fill=tk.BOTH, expand=True)
# 图像上传按钮
self.upload_button = ttk.Button(left_frame, text="上传天气图像", command=self.upload_image)
self.upload_button.pack(fill=tk.X, padx=5, pady=5)
# 天气预测结果
self.weather_result_frame = ttk.LabelFrame(left_frame, text="天气预测结果")
self.weather_result_frame.pack(fill=tk.X, padx=5, pady=5)
self.weather_result_label = ttk.Label(self.weather_result_frame, text="请上传图像...")
self.weather_result_label.pack(fill=tk.X, padx=5, pady=5)
# 右侧：温度预测和穿衣推荐
right_frame = ttk.LabelFrame(main_frame, text="温度和穿衣推荐", padding="10")
right_frame.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True, padx=5, pady=5)
# 历史温度输入
temp_frame = ttk.Frame(right_frame)
temp_frame.pack(fill=tk.X, padx=5, pady=5)
ttk.Label(temp_frame, text="输入过去7天的温度 (°C)：").pack(anchor=tk.W)
self.temp_entries = []
for i in range(7):
entry_frame = ttk.Frame(temp_frame)
entry_frame.pack(fill=tk.X, pady=2)
ttk.Label(entry_frame, text=f"过去第{7 - i}天: ").pack(side=tk.LEFT)
entry = ttk.Entry(entry_frame, width=10)
entry.pack(side=tk.LEFT, padx=5)
self.temp_entries.append(entry)
# 示例数据按钮
self.example_button = ttk.Button(temp_frame, text="使用示例数据", command=self.fill_example_data)
self.example_button.pack(pady=5)
# 预测按钮
self.predict_button = ttk.Button(right_frame, text="分析并推荐穿衣", command=self.predict_clothing)
self.predict_button.pack(fill=tk.X, padx=5, pady=10)
# 结果显示
self.result_frame = ttk.LabelFrame(right_frame, text="分析结果")
self.result_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
self.prediction_label = ttk.Label(self.result_frame, text="")
self.prediction_label.pack(fill=tk.X, padx=5, pady=5)
self.recommendation_label = ttk.Label(self.result_frame, text="")
self.recommendation_label.pack(fill=tk.X, padx=5, pady=5)
self.tip_label = ttk.Label(self.result_frame, text="", foreground="blue")
self.tip_label.pack(fill=tk.X, padx=5, pady=5)
def load_models(self):
try:
# 检查并创建模型目录
os.makedirs('models', exist_ok=True)
# 加载CNN模型
if os.path.exists('models/best_cnn_model.pth'):
self.cnn_model.load_state_dict(torch.load('models/best_cnn_model.pth', map_location=self.device))
print("CNN模型加载成功")
else:
print("找不到CNN模型，需要先训练")
# 加载RNN模型
if os.path.exists('models/best_rnn_model.pth'):
self.rnn_model.load_state_dict(torch.load('models/best_rnn_model.pth', map_location=self.device))
print("RNN模型加载成功")
else:
print("找不到RNN模型，需要先训练")
# 加载到设备
self.cnn_model = self.cnn_model.to(self.device)
self.rnn_model = self.rnn_model.to(self.device)
# 设置为评估模式
self.cnn_model.eval()
self.rnn_model.eval()
# 加载温度数据集以获取归一化参数
if os.path.exists('data/temperature_data.csv'):
self.temp_dataset = TemperatureDataset('data/temperature_data.csv')
print("温度数据集加载成功")
except Exception as e:
print(f"模型加载错误: {str(e)}")
def upload_image(self):
file_path = filedialog.askopenfilename(
filetypes=[("Image files", "*.jpg *.jpeg *.png")]
)
# 如果没选择图片就直接返回，忽视
if not file_path:
return
try:
# 打开并显示图像
img = Image.open(file_path).convert('RGB')
img_display = img.copy()
# 调整图像大小以适应UI
img_display.thumbnail((300, 300))
photo = ImageTk.PhotoImage(img_display)
self.image_label.config(image=photo)
self.image_label.image = photo # 保持引用
# 预测天气类型
self.predict_weather(img)
except Exception as e:
messagebox.showerror("错误", f"图像加载失败: {str(e)}")
def predict_weather(self, img):
try:
# 预处理图像
img_tensor = self.transform(img).unsqueeze(0).to(self.device)
# 预测
with torch.no_grad():
outputs = self.cnn_model(img_tensor)
_, predicted = torch.max(outputs, 1)
weather_idx = predicted.item()
weather_type = self.weather_classes[weather_idx]
# 显示结果
self.weather_result_label.config(text=f"预测天气: {weather_type}")
return weather_type
except Exception as e:
messagebox.showerror("错误", f"天气预测失败: {str(e)}")
return None
def fill_example_data(self):
# 填充示例温度数据
example_temps = [18, 20, 19, 21, 22, 20, 19] # 过去7天温度示例
for i, temp in enumerate(example_temps):
self.temp_entries[i].delete(0, tk.END)
self.temp_entries[i].insert(0, str(temp))
def predict_clothing(self):
# 检查是否有天气预测
weather_result = self.weather_result_label.cget("text")
if "预测天气:" not in weather_result:
messagebox.showinfo("提示", "请先上传天气图像")
return
# 获取天气类型
weather_type = weather_result.split(": ")[1].strip()
# 检查温度输入
temps = []
for entry in self.temp_entries:
try:
temp = float(entry.get())
temps.append(temp)
except ValueError:
messagebox.showinfo("提示", "请输入有效的温度数据")
return
if len(temps) != 7:
messagebox.showinfo("提示", "请输入7天的温度数据")
return
# 预测明天温度
try:
# 归一化温度
if self.temp_dataset:
norm_temps = [
(t - self.temp_dataset.temp_min) / (self.temp_dataset.temp_max - self.temp_dataset.temp_min)
for t in temps]
else:
# 如果没有加载数据集，简单归一化
min_temp, max_temp = min(temps), max(temps)
norm_temps = [(t - min_temp) / (max_temp - min_temp) for t in temps]
# 转换为张量
temp_tensor = torch.FloatTensor(norm_temps).unsqueeze(0).to(self.device)
# 预测
with torch.no_grad():
output = self.rnn_model(temp_tensor)
# 反归一化
if self.temp_dataset:
predicted_temp = self.temp_dataset.denormalize(output.item())
else:
predicted_temp = output.item() * (max_temp - min_temp) + min_temp
# 四舍五入到一位小数
predicted_temp = round(predicted_temp, 1)
# 获取穿衣建议
recommendation, clothing_level, additional_tip = get_clothing_recommendation(weather_type, predicted_temp)
# 显示结果
self.prediction_label.config(text=f"预测明天温度: {predicted_temp}°C")
self.recommendation_label.config(text=f"穿衣建议: {recommendation}")
self.tip_label.config(text=additional_tip)
except Exception as e:
messagebox.showerror("错误", f"温度预测失败: {str(e)}")
def main():
# 确保模型和数据目录存在，不存在的话就创建
os.makedirs('models', exist_ok=True)
os.makedirs('data', exist_ok=True)
# 训练和评估模式选择
training_mode = input("是否需要训练模型? (y/n): ").lower() == 'y'
if training_mode:
# 检查数据集是否存在
weather_data_exists = os.path.exists('data/weather_images')
temp_data_exists = os.path.exists('data/temperature_data.csv')
if weather_data_exists:
print("开始训练CNN天气分类模型...")
train_cnn_model('data/weather_images')
else:
print(" 找不到天气图像数据集! 请将数据集放在 'data/weather_images' 目录下")
if temp_data_exists:
print("开始训练RNN温度预测模型...")
train_rnn_model('data/temperature_data.csv')
else:
print("找不到温度数据集! 请将数据集保存为 'data/temperature_data.csv'")
# 启动GUI
print("启动智能穿衣推荐系统的GUI...")
root = tk.Tk()
app = WeatherClothingApp(root)
root.mainloop()
if __name__ == "__main__":
main()1.数据集准备
从https://www.kaggle.com/datasets/pratik2901/multiclass-weather-dataset和https://www.kaggle.com/datasets/budincsevity/szeged-weather下载两个数据集。
2.数据加载和预处理
# 数据集类
class WeatherImageDataset(Dataset):
"""天气图像数据集"""
def __init__(self, data_dir, transform=None):
self.data_dir = data_dir
self.transform = transform
self.weather_classes = ['Cloudy', 'Rain', 'Shine', 'Sunrise']
self.samples = []
self._load_images()
def _load_images(self):
"""加载天气图像"""
for class_idx, class_name in enumerate(self.weather_classes):
class_dir = os.path.join(self.data_dir, class_name)
if os.path.exists(class_dir):
for img_name in os.listdir(class_dir):
if img_name.lower().endswith(('.jpg', '.jpeg', '.png')):
self.samples.append({
'image_path': os.path.join(class_dir, img_name),
'label': class_idx,
'weather_type': class_name
})
print(f"加载了 {len(self.samples)} 张天气图像")
def __len__(self):
return len(self.samples)
def __getitem__(self, idx):
sample = self.samples[idx]
image = Image.open(sample['image_path']).convert('RGB')
if self.transform:
image = self.transform(image)
label = torch.tensor(sample['label'], dtype=torch.long)
return image, label
class TemperatureDataset(Dataset):
"""温度时序数据集"""
def __init__(self, csv_path, sequence_length=7, prediction_horizon=1):
self.sequence_length = sequence_length
self.prediction_horizon = prediction_horizon
self.data = pd.read_csv(csv_path)
self._prepare_data()
def _prepare_data(self):
"""准备温度数据"""
# 使用温度数据（如果没有温度列，使用相近的列）
if 'Temp' in self.data.columns:
temp_data = self.data['Temp'].values
elif 'temperature (C)' in self.data.columns:
temp_data = self.data['temperature (C)'].values
elif 'Temperature (C)' in self.data.columns:
temp_data = self.data['Temperature (C)'].values
else:
# 使用第一列数值数据作为温度
temp_data = self.data.iloc[:, 0].values
# 归一化
self.temp_min = temp_data.min()
self.temp_max = temp_data.max()
temp_data = (temp_data - self.temp_min) / (self.temp_max - self.temp_min)
# 创建序列
self.sequences = []
self.targets = []
for i in range(len(temp_data) - self.sequence_length - self.prediction_horizon):
sequence = temp_data[i:i + self.sequence_length]
target = temp_data[i + self.sequence_length + self.prediction_horizon - 1]
self.sequences.append(sequence)
self.targets.append(target)
def denormalize(self, temp):
"""反归一化温度"""
return temp * (self.temp_max - self.temp_min) + self.temp_min
def __len__(self):
return len(self.sequences)
def __getitem__(self, idx):
sequence = torch.FloatTensor(self.sequences[idx])
target = torch.FloatTensor([self.targets[idx]])
return sequence, target
# 数据预处理和增强
transform_train = transforms.Compose([
transforms.Resize((224, 224)),
transforms.RandomHorizontalFlip(),
transforms.RandomRotation(10),
transforms.ColorJitter(brightness=0.1, contrast=0.1),
transforms.ToTensor(),
transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
])
transform_test = transforms.Compose([
transforms.Resize((224, 224)),
transforms.ToTensor(),
transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
])
# 加载数据集
print("正在加载天气图像数据集...")
full_dataset = WeatherImageDataset(data_dir, transform=None)
# 拆分数据集
train_size = int(0.7 * len(full_dataset))
val_size = int(0.15 * len(full_dataset))
test_size = len(full_dataset) - train_size - val_size
print(f"数据集划分: 训练集 {train_size}张, 验证集 {val_size}张, 测试集 {test_size}张")
# 使用随机种子确保可重复性
torch.manual_seed(42)
train_dataset, val_dataset, test_dataset = random_split(
full_dataset, [train_size, val_size, test_size]
)
# 应用不同的转换
class TransformedSubset(Dataset):
def __init__(self, subset, transform=None):
self.subset = subset
self.transform = transform
def __getitem__(self, idx):
x, y = self.subset[idx]
if self.transform:
x = self.transform(x)
return x, y
def __len__(self):
return len(self.subset)
train_dataset = TransformedSubset(train_dataset, transform_train)
val_dataset = TransformedSubset(val_dataset, transform_test)
test_dataset = TransformedSubset(test_dataset, transform_test)
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=batch_size)
test_loader = DataLoader(test_dataset, batch_size=batch_size)
3.模型定义
# 定义CNN模型（天气图像分类）
class WeatherCNN(nn.Module):
"""使用ResNet18进行天气图像分类"""
def __init__(self, num_classes=4):
super(WeatherCNN, self).__init__()
# 加载预训练的ResNet18模型
self.model = models.resnet18(pretrained=True)
# 修改最后一层以适应我们的分类任务
num_features = self.model.fc.in_features
self.model.fc = nn.Linear(num_features, num_classes)
def forward(self, x):
return self.model(x)
# 定义RNN模型（温度预测）
class TemperatureRNN(nn.Module):
"""使用LSTM进行温度预测"""
def __init__(self, input_size=1, hidden_size=64, num_layers=2, output_size=1):
super(TemperatureRNN, self).__init__()
self.hidden_size = hidden_size
self.num_layers = num_layers
# LSTM层
self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
# 全连接层
self.fc = nn.Linear(hidden_size, output_size)
def forward(self, x):
# 输入形状：(batch_size, seq_len)
# 重塑为LSTM输入形状: (batch_size, seq_len, input_size)
x = x.unsqueeze(2)
# 初始化隐藏状态
h0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(x.device)
c0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(x.device)
# LSTM前向传播
out, _ = self.lstm(x, (h0, c0))
# 取最后一个时间步的输出
out = self.fc(out[:, -1, :])
return out4.模型训练
从上面的日志和折线图中可以看出：CNN模型训练后最佳验证准确率达0.9881，测试集准确率0.9765，各类别准确率较高，Sunrise类达到了100%，说明对天气图像的分类效果好；训练过程中损失下降、准确率上升，模型收敛稳定。 RNN模型中温度这一类别预测最佳验证损失低至0.000463，测试集RMSE约1.3969°C、R²达0.9787，表明温度预测误差小、拟合效果好，模型对时序温度的预测能力较强。
5.模型测试和评估
CNN测试
Sunrise类别准确率100%，Shine、Rain、Cloudy均超95%，整体测试集准确率达97.65%，验证了模型对天气图像分类的高可靠性。
对角线元素（正确分类数）占比极高，误分类案例极少。Cloudy仅有2例误判、Rain1例、Shine1例，说明模型对各类天气的区分能力强。
RNN测试
误差指标数值小，结合测试集平均损失极低，表明模型预测精度高，对温度的回归预测效果稳定。
散点图中预测温度与实际温度的散点紧密贴合对角线，R^2=0.9787接近1，说明模型预测值与真实值线性相关性极强，整体预测的趋势高度一致。
温度预测折线图显示实际温度（蓝线）与预测温度（红线）的波动趋势几乎重合，仅在极端值处有微小偏差，体现模型对温度时序变化的捕捉能力好。
6.系统界面制作和测试
# GUI界面实现
class WeatherClothingApp:
def __init__(self, root):
self.root = root
self.root.title("智能天气穿衣推荐系统")
self.root.geometry("800x600")
# 加载模型
self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# 初始化模型（稍后加载）
self.cnn_model = WeatherCNN(num_classes=4)
self.rnn_model = TemperatureRNN(input_size=1, hidden_size=64, num_layers=2)
self.temp_dataset = None
# 图像转换
self.transform = transforms.Compose([
transforms.Resize((224, 224)),
transforms.ToTensor(),
transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
])
# 类别标签
self.weather_classes = ['Cloudy', 'Rain', 'Shine', 'Sunrise']
# 创建UI
self.create_ui()
# 加载预训练模型
self.load_models()
def create_ui(self):
# 主界面分为左右两部分
main_frame = ttk.Frame(self.root, padding="10")
main_frame.pack(fill=tk.BOTH, expand=True)
# 左侧：图像上传和天气预测
left_frame = ttk.LabelFrame(main_frame, text="天气图像分析", padding="10")
left_frame.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=5, pady=5)
# 图像预览
self.image_frame = ttk.Frame(left_frame, width=300, height=300)
self.image_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
self.image_label = ttk.Label(self.image_frame)
self.image_label.pack(fill=tk.BOTH, expand=True)
# 图像上传按钮
self.upload_button = ttk.Button(left_frame, text="上传天气图像", command=self.upload_image)
self.upload_button.pack(fill=tk.X, padx=5, pady=5)
# 天气预测结果
self.weather_result_frame = ttk.LabelFrame(left_frame, text="天气预测结果")
self.weather_result_frame.pack(fill=tk.X, padx=5, pady=5)
self.weather_result_label = ttk.Label(self.weather_result_frame, text="请上传图像...")
self.weather_result_label.pack(fill=tk.X, padx=5, pady=5)
# 右侧：温度预测和穿衣推荐
right_frame = ttk.LabelFrame(main_frame, text="温度和穿衣推荐", padding="10")
right_frame.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True, padx=5, pady=5)
# 历史温度输入
temp_frame = ttk.Frame(right_frame)
temp_frame.pack(fill=tk.X, padx=5, pady=5)
ttk.Label(temp_frame, text="输入过去7天的温度 (°C)：").pack(anchor=tk.W)
self.temp_entries = []
for i in range(7):
entry_frame = ttk.Frame(temp_frame)
entry_frame.pack(fill=tk.X, pady=2)
ttk.Label(entry_frame, text=f"过去第{7 - i}天: ").pack(side=tk.LEFT)
entry = ttk.Entry(entry_frame, width=10)
entry.pack(side=tk.LEFT, padx=5)
self.temp_entries.append(entry)
# 示例数据按钮
self.example_button = ttk.Button(temp_frame, text="使用示例数据", command=self.fill_example_data)
self.example_button.pack(pady=5)
# 预测按钮
self.predict_button = ttk.Button(right_frame, text="分析并推荐穿衣", command=self.predict_clothing)
self.predict_button.pack(fill=tk.X, padx=5, pady=10)
# 结果显示
self.result_frame = ttk.LabelFrame(right_frame, text="分析结果")
self.result_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
self.prediction_label = ttk.Label(self.result_frame, text="")
self.prediction_label.pack(fill=tk.X, padx=5, pady=5)
self.recommendation_label = ttk.Label(self.result_frame, text="")
self.recommendation_label.pack(fill=tk.X, padx=5, pady=5)
self.tip_label = ttk.Label(self.result_frame, text="", foreground="blue")
self.tip_label.pack(fill=tk.X, padx=5, pady=5)
def load_models(self):
try:
# 检查并创建模型目录
os.makedirs('models', exist_ok=True)
# 加载CNN模型
if os.path.exists('models/best_cnn_model.pth'):
self.cnn_model.load_state_dict(torch.load('models/best_cnn_model.pth', map_location=self.device))
print("CNN模型加载成功")
else:
print("找不到CNN模型，需要先训练")
# 加载RNN模型
if os.path.exists('models/best_rnn_model.pth'):
self.rnn_model.load_state_dict(torch.load('models/best_rnn_model.pth', map_location=self.device))
print("RNN模型加载成功")
else:
print("找不到RNN模型，需要先训练")
# 加载到设备
self.cnn_model = self.cnn_model.to(self.device)
self.rnn_model = self.rnn_model.to(self.device)
# 设置为评估模式
self.cnn_model.eval()
self.rnn_model.eval()
# 加载温度数据集以获取归一化参数
if os.path.exists('data/temperature_data.csv'):
self.temp_dataset = TemperatureDataset('data/temperature_data.csv')
print("温度数据集加载成功")
except Exception as e:
print(f"模型加载错误: {str(e)}")
def upload_image(self):
file_path = filedialog.askopenfilename(
filetypes=[("Image files", "*.jpg *.jpeg *.png")]
)
# 如果没选择图片就直接返回，忽视
if not file_path:
return
try:
# 打开并显示图像
img = Image.open(file_path).convert('RGB')
img_display = img.copy()
# 调整图像大小以适应UI
img_display.thumbnail((300, 300))
photo = ImageTk.PhotoImage(img_display)
self.image_label.config(image=photo)
self.image_label.image = photo # 保持引用
# 预测天气类型
self.predict_weather(img)
except Exception as e:
messagebox.showerror("错误", f"图像加载失败: {str(e)}")
def predict_weather(self, img):
try:
# 预处理图像
img_tensor = self.transform(img).unsqueeze(0).to(self.device)
# 预测
with torch.no_grad():
outputs = self.cnn_model(img_tensor)
_, predicted = torch.max(outputs, 1)
weather_idx = predicted.item()
weather_type = self.weather_classes[weather_idx]
# 显示结果
self.weather_result_label.config(text=f"预测天气: {weather_type}")
return weather_type
except Exception as e:
messagebox.showerror("错误", f"天气预测失败: {str(e)}")
return None
def fill_example_data(self):
# 填充示例温度数据
example_temps = [18, 20, 19, 21, 22, 20, 19] # 过去7天温度示例
for i, temp in enumerate(example_temps):
self.temp_entries[i].delete(0, tk.END)
self.temp_entries[i].insert(0, str(temp))
def predict_clothing(self):
# 检查是否有天气预测
weather_result = self.weather_result_label.cget("text")
if "预测天气:" not in weather_result:
messagebox.showinfo("提示", "请先上传天气图像")
return
# 获取天气类型
weather_type = weather_result.split(": ")[1].strip()
# 检查温度输入
temps = []
for entry in self.temp_entries:
try:
temp = float(entry.get())
temps.append(temp)
except ValueError:
messagebox.showinfo("提示", "请输入有效的温度数据")
return
if len(temps) != 7:
messagebox.showinfo("提示", "请输入7天的温度数据")
return
# 预测明天温度
try:
# 归一化温度
if self.temp_dataset:
norm_temps = [
(t - self.temp_dataset.temp_min) / (self.temp_dataset.temp_max - self.temp_dataset.temp_min)
for t in temps]
else:
# 如果没有加载数据集，简单归一化
min_temp, max_temp = min(temps), max(temps)
norm_temps = [(t - min_temp) / (max_temp - min_temp) for t in temps]
# 转换为张量
temp_tensor = torch.FloatTensor(norm_temps).unsqueeze(0).to(self.device)
# 预测
with torch.no_grad():
output = self.rnn_model(temp_tensor)
# 反归一化
if self.temp_dataset:
predicted_temp = self.temp_dataset.denormalize(output.item())
else:
predicted_temp = output.item() * (max_temp - min_temp) + min_temp
# 四舍五入到一位小数
predicted_temp = round(predicted_temp, 1)
# 获取穿衣建议
recommendation, clothing_level, additional_tip = get_clothing_recommendation(weather_type, predicted_temp)
# 显示结果
self.prediction_label.config(text=f"预测明天温度: {predicted_temp}°C")
self.recommendation_label.config(text=f"穿衣建议: {recommendation}")
self.tip_label.config(text=additional_tip)
except Exception as e:
messagebox.showerror("错误", f"温度预测失败: {str(e)}")
从网上另外下载的天气图片，可见预测结果正确，以下是更多测试结果。
以上两张我自己的实拍图预测也正确
![图片[22] - AI科研 编程 读书笔记 - 【人工智能】【Python】深度学习模型在实际问题中的应用实验 - AI科研 编程 读书笔记 - 小竹の笔记本](https://img.smallbamboo.cn/i/2025/11/03/690837a469992.png)
© 版权声明
1. 除特殊说明外，本网站所有原创文章的版权归作者所有，未经授权，禁止以任何形式（包括但不限于转载、摘编、复制、镜像等）发布至任何平台。
2. 论文总结类文章中涉及的图表、数据等素材，版权归原出版商及论文作者所有，仅为学术交流目的引用；若相关权利人认为存在侵权，请联系本网站删除，联系方式：i@smallbamboo.cn。
3. 违反上述声明者，将依法追究其相关法律责任。
2. 论文总结类文章中涉及的图表、数据等素材，版权归原出版商及论文作者所有，仅为学术交流目的引用；若相关权利人认为存在侵权，请联系本网站删除，联系方式：i@smallbamboo.cn。
3. 违反上述声明者，将依法追究其相关法律责任。
THE END
暂无评论内容