99 lines
3.3 KiB
Python
99 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
直接测试训练器的日志输出
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import time
|
||
|
||
# 设置编码环境
|
||
os.environ['PYTHONIOENCODING'] = 'utf-8'
|
||
if os.name == 'nt':
|
||
try:
|
||
os.system('chcp 65001 >nul 2>&1')
|
||
if hasattr(sys.stdout, 'reconfigure'):
|
||
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
|
||
sys.stderr.reconfigure(encoding='utf-8', errors='replace')
|
||
except Exception as e:
|
||
print(f"Warning: Failed to set UTF-8 encoding: {e}")
|
||
|
||
def test_direct_training():
|
||
"""直接测试训练器功能和日志输出"""
|
||
print("🧪 直接测试训练器日志输出")
|
||
print("=" * 50)
|
||
|
||
try:
|
||
# 添加server目录到路径
|
||
server_dir = os.path.join(os.getcwd(), 'server')
|
||
if server_dir not in sys.path:
|
||
sys.path.insert(0, server_dir)
|
||
|
||
print("1️⃣ 导入模块...")
|
||
from trainers.transformer_trainer import train_product_model_with_transformer
|
||
from utils.training_progress import TrainingProgressManager
|
||
print("✅ 模块导入成功")
|
||
|
||
print("\n2️⃣ 创建进度管理器...")
|
||
def progress_callback(message):
|
||
print(f"📊 [进度回调] {message}")
|
||
|
||
progress_manager = TrainingProgressManager(websocket_callback=progress_callback)
|
||
print("✅ 进度管理器创建成功")
|
||
|
||
print("\n3️⃣ 检查数据文件...")
|
||
if not os.path.exists('pharmacy_sales_multi_store.csv'):
|
||
print("❌ 数据文件不存在，请先运行生成数据脚本")
|
||
return False
|
||
print("✅ 数据文件存在")
|
||
|
||
print("\n4️⃣ 开始训练测试...")
|
||
print("-" * 30)
|
||
|
||
# 直接调用训练器
|
||
result = train_product_model_with_transformer(
|
||
product_id='P001',
|
||
epochs=2, # 只训练2轮用于测试
|
||
store_id=None,
|
||
training_mode='product',
|
||
aggregation_method='sum',
|
||
model_dir='saved_models',
|
||
version=None,
|
||
socketio=None, # 不使用socketio
|
||
task_id='direct_test_001',
|
||
continue_training=False
|
||
)
|
||
|
||
print("-" * 30)
|
||
print(f"4️⃣ 训练完成，结果: {type(result)}")
|
||
|
||
if result and len(result) >= 2:
|
||
model, metrics = result[0], result[1]
|
||
print(f"✅ 训练成功！")
|
||
print(f"📊 返回指标: {metrics}")
|
||
else:
|
||
print(f"⚠️ 训练结果异常: {result}")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 测试失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
if __name__ == "__main__":
|
||
print("🔧 直接测试训练器日志功能")
|
||
print("这将直接调用训练器，观察日志输出\n")
|
||
|
||
success = test_direct_training()
|
||
|
||
print("\n" + "=" * 50)
|
||
if success:
|
||
print("🎉 测试完成!")
|
||
print("\n📋 观察上面的输出:")
|
||
print("- 如果看到详细的训练进度信息，说明训练器日志正常")
|
||
print("- 如果看到进度回调信息，说明进度管理器工作正常")
|
||
else:
|
||
print("❌ 测试失败") |