Python方案--交互式VR教育应用开发

前言

虚拟现实(VR)技术正在改变教育领域，为学习者提供沉浸式、互动性强的学习体验。本文将详细介绍如何使用Python开发一个交互式VR教育应用，涵盖从项目规划到部署的完整开发流程。

为什么选择Python开发VR教育应用？

Python的优势：

简洁语法：易于学习和维护
丰富生态：强大的科学计算和数据处理库
良好集成：与Unity、Blender等工具无缝对接
快速原型：适合教育应用的迭代开发
AI支持：内置机器学习能力，可实现智能化教学

技术栈架构

核心技术组件

# 主要依赖库
DEPENDENCIES = {'backend': ['Flask', 'FastAPI', 'SQLAlchemy', 'Redis'],'vr_engine': ['OpenXR', 'SteamVR', 'Panda3D'],'ai_ml': ['TensorFlow', 'PyTorch', 'OpenCV'],'data': ['NumPy', 'Pandas', 'Matplotlib'],'networking': ['WebSocket', 'Socket.io'],'testing': ['pytest', 'unittest']
}

系统架构设计

┌─────────────────┐    ┌──────────────────┐    ┌─────────────────┐
│   VR头显设备     │    │   Unity前端      │    │  Python后端     │
│   (Oculus/HTC)  │◄──►│   (VR界面)       │◄──►│   (业务逻辑)     │
└─────────────────┘    └──────────────────┘    └─────────────────┘│                         │▼                         ▼┌──────────────────┐    ┌─────────────────┐│   3D内容管理     │    │   数据库        ││   (教学资源)     │    │   (学习数据)    │└──────────────────┘    └─────────────────┘

项目结构搭建

目录结构

vr_education_app/
├── backend/                 # Python后端
│   ├── app/
│   │   ├── models/         # 数据模型
│   │   ├── services/       # 业务服务
│   │   ├── api/           # API接口
│   │   └── utils/         # 工具函数
│   ├── config/            # 配置文件
│   ├── tests/             # 测试文件
│   └── requirements.txt   # 依赖列表
├── frontend/              # Unity VR前端
│   ├── Assets/
│   ├── Scripts/
│   └── Scenes/
├── content/               # 教学内容
│   ├── 3d_models/
│   ├── textures/
│   └── animations/
└── docs/                  # 项目文档

环境配置

# config/settings.py
import os
from dataclasses import dataclass
@dataclass
class VREducationConfig:"""VR教育应用配置类"""# 数据库配置DATABASE_URL: str = os.getenv('DATABASE_URL', 'postgresql://localhost/vr_edu')# VR设备配置SUPPORTED_VR_DEVICES = ['oculus_quest', 'htc_vive', 'valve_index']DEFAULT_ROOM_SCALE = (4.0, 3.0)  # 4m x 3m# 教学内容配置CONTENT_STORAGE_PATH = './content/'MAX_SCENE_COMPLEXITY = 100000  # 最大面数# AI配置AI_MODEL_PATH = './models/'ENABLE_ADAPTIVE_LEARNING = True# 网络配置WEBSOCKET_PORT = 8765API_PORT = 5000

核心功能开发

1. VR场景管理器

# backend/app/services/scene_manager.py
import json
import asyncio
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
@dataclass
class VRScene:"""VR场景数据类"""scene_id: strtitle: strsubject: strdifficulty_level: intobjects: List[Dict]interactions: List[Dict]learning_objectives: List[str]
class SceneManager:"""VR场景管理器"""def __init__(self):self.scenes: Dict[str, VRScene] = {}self.active_sessions = {}async def load_scene(self, scene_id: str) -> Optional[VRScene]:"""加载VR场景"""try:scene_path = f"./content/scenes/{scene_id}.json"with open(scene_path, 'r', encoding='utf-8') as f:scene_data = json.load(f)scene = VRScene(**scene_data)self.scenes[scene_id] = sceneprint(f"✅ 场景 {scene.title} 加载成功")return sceneexcept Exception as e:print(f"❌ 场景加载失败: {e}")return Noneasync def create_learning_session(self, user_id: str, scene_id: str):"""创建学习会话"""scene = await self.load_scene(scene_id)if not scene:return Nonesession = {'session_id': f"{user_id}_{scene_id}_{int(asyncio.get_event_loop().time())}",'user_id': user_id,'scene': scene,'start_time': asyncio.get_event_loop().time(),'interactions': [],'progress': 0}self.active_sessions[session['session_id']] = sessionreturn sessiondef get_session_progress(self, session_id: str) -> Dict:"""获取学习进度"""if session_id not in self.active_sessions:return {'error': 'Session not found'}session = self.active_sessions[session_id]total_objectives = len(session['scene'].learning_objectives)completed = len([i for i in session['interactions'] if i.get('completed', False)])return {'progress': (completed / total_objectives) * 100 if total_objectives > 0 else 0,'completed_objectives': completed,'total_objectives': total_objectives,'time_spent': asyncio.get_event_loop().time() - session['start_time']}
# 使用示例
scene_manager = SceneManager()

2. 智能学习分析系统

# backend/app/services/learning_analytics.py
import numpy as np
import pandas as pd
from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler
import matplotlib.pyplot as plt
from typing import Dict, List, Tuple
class LearningAnalytics:"""智能学习分析系统"""def __init__(self):self.scaler = StandardScaler()self.student_clusters = Nonedef analyze_learning_behavior(self, interaction_data: List[Dict]) -> Dict:"""分析学习行为模式"""if not interaction_data:return {'error': 'No data available'}df = pd.DataFrame(interaction_data)# 计算关键指标metrics = {'avg_session_duration': df['session_duration'].mean(),'interaction_frequency': len(df) / df['session_duration'].sum(),'mistake_rate': df['mistakes'].sum() / len(df),'help_requests': df['help_requests'].sum(),'completion_rate': (df['completed'] == True).mean()}return metricsdef predict_learning_difficulty(self, student_profile: Dict, content_features: Dict) -> float:"""预测学习难度"""# 简化的难度预测模型student_ability = student_profile.get('ability_score', 50)content_complexity = content_features.get('complexity_score', 50)# 基于能力和内容复杂度计算预测难度difficulty_score = max(0, min(100, content_complexity - student_ability + 50))return difficulty_score / 100.0def generate_adaptive_recommendations(self, user_id: str) -> List[Dict]:"""生成自适应学习推荐"""# 模拟推荐生成recommendations = [{'content_id': 'geometry_basics','reason': '基于你在空间几何方面的表现，推荐加强基础练习','difficulty_adjustment': -0.2},{'content_id': 'physics_simulation','reason': '你在物理概念理解上表现优秀，可以尝试更高级的内容','difficulty_adjustment': 0.3}]return recommendationsdef visualize_learning_progress(self, user_data: Dict) -> str:"""可视化学习进度"""sessions = user_data.get('sessions', [])if not sessions:return "No session data available"# 生成学习进度图表dates = [s['date'] for s in sessions]scores = [s['score'] for s in sessions]plt.figure(figsize=(10, 6))plt.plot(dates, scores, marker='o')plt.title('Learning Progress Over Time')plt.xlabel('Date')plt.ylabel('Score')plt.grid(True)# 保存图表chart_path = f"./charts/progress_{user_data['user_id']}.png"plt.savefig(chart_path)plt.close()return chart_path
# 使用示例
analytics = LearningAnalytics()

3. WebSocket通信模块

# backend/app/services/websocket_server.py
import asyncio
import websockets
import json
from typing import Set, Dict, Any
class VRWebSocketServer:"""VR应用WebSocket服务器"""def __init__(self):self.clients: Set[websockets.WebSocketServerProtocol] = set()self.client_sessions: Dict[str, Dict] = {}async def register_client(self, websocket: websockets.WebSocketServerProtocol):"""注册客户端连接"""self.clients.add(websocket)client_id = f"client_{len(self.clients)}"self.client_sessions[client_id] = {'websocket': websocket,'user_id': None,'current_scene': None}print(f"✅ 客户端 {client_id} 已连接")return client_idasync def unregister_client(self, websocket: websockets.WebSocketServerProtocol):"""注销客户端连接"""self.clients.discard(websocket)# 清理会话数据to_remove = Nonefor client_id, session in self.client_sessions.items():if session['websocket'] == websocket:to_remove = client_idbreakif to_remove:del self.client_sessions[to_remove]print(f"❌ 客户端 {to_remove} 已断开连接")async def handle_message(self, websocket: websockets.WebSocketServerProtocol, message: str):"""处理客户端消息"""try:data = json.loads(message)message_type = data.get('type')if message_type == 'interaction':await self.handle_interaction(websocket, data)elif message_type == 'scene_request':await self.handle_scene_request(websocket, data)elif message_type == 'progress_update':await self.handle_progress_update(websocket, data)else:await self.send_error(websocket, f"Unknown message type: {message_type}")except json.JSONDecodeError:await self.send_error(websocket, "Invalid JSON message")except Exception as e:await self.send_error(websocket, f"Error processing message: {str(e)}")async def handle_interaction(self, websocket: websockets.WebSocketServerProtocol, data: Dict):"""处理VR交互事件"""interaction_data = {'type': 'interaction_response','object_id': data.get('object_id'),'action': data.get('action'),'result': 'success','feedback': self.generate_interaction_feedback(data)}await websocket.send(json.dumps(interaction_data))async def handle_scene_request(self, websocket: websockets.WebSocketServerProtocol, data: Dict):"""处理场景请求"""scene_id = data.get('scene_id')# 这里应该调用SceneManager加载场景scene_data = {'type': 'scene_data','scene_id': scene_id,'objects': [],  # 实际场景对象数据'lighting': {},  # 光照设置'physics': {}    # 物理设置}await websocket.send(json.dumps(scene_data))def generate_interaction_feedback(self, interaction_data: Dict) -> str:"""生成交互反馈"""action = interaction_data.get('action', '')if action == 'grab':return "很好！你成功抓取了这个物体。"elif action == 'examine':return "仔细观察物体的细节，注意它的特征。"else:return "继续探索，发现更多有趣的内容！"async def send_error(self, websocket: websockets.WebSocketServerProtocol, error_message: str):"""发送错误消息"""error_data = {'type': 'error','message': error_message}await websocket.send(json.dumps(error_data))async def broadcast_to_all(self, message: Dict):"""向所有客户端广播消息"""if self.clients:message_str = json.dumps(message)await asyncio.gather(*[client.send(message_str) for client in self.clients],return_exceptions=True)async def start_server(self, host='localhost', port=8765):"""启动WebSocket服务器"""async def handle_client(websocket, path):client_id = await self.register_client(websocket)try:async for message in websocket:await self.handle_message(websocket, message)except websockets.exceptions.ConnectionClosed:passfinally:await self.unregister_client(websocket)print(f" WebSocket服务器启动在 ws://{host}:{port}")await websockets.serve(handle_client, host, port)
# 启动服务器
async def main():vr_server = VRWebSocketServer()await vr_server.start_server()await asyncio.Future()  # 保持运行
if __name__ == "__main__":asyncio.run(main())

4. 数据模型设计

# backend/app/models/database.py
from sqlalchemy import create_engine, Column, Integer, String, DateTime, Boolean, Text, Float
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.sql import func
from datetime import datetime
Base = declarative_base()
class User(Base):"""用户模型"""__tablename__ = 'users'id = Column(Integer, primary_key=True)username = Column(String(50), unique=True, nullable=False)email = Column(String(100), unique=True, nullable=False)created_at = Column(DateTime, default=func.now())last_login = Column(DateTime)# 学习偏好preferred_learning_style = Column(String(20))  # visual, auditory, kinestheticdifficulty_preference = Column(Float, default=0.5)  # 0.0-1.0# 统计数据total_study_time = Column(Integer, default=0)  # 分钟sessions_completed = Column(Integer, default=0)average_score = Column(Float, default=0.0)
class VRSession(Base):"""VR学习会话模型"""__tablename__ = 'vr_sessions'id = Column(Integer, primary_key=True)user_id = Column(Integer, nullable=False)scene_id = Column(String(50), nullable=False)start_time = Column(DateTime, default=func.now())end_time = Column(DateTime)duration_minutes = Column(Integer)# 学习数据interactions_count = Column(Integer, default=0)mistakes_count = Column(Integer, default=0)help_requests = Column(Integer, default=0)completion_percentage = Column(Float, default=0.0)final_score = Column(Float)# VR特定数据head_movement_data = Column(Text)  # JSON格式存储controller_data = Column(Text)     # JSON格式存储eye_tracking_data = Column(Text)   # JSON格式存储（如果支持）
class LearningContent(Base):"""学习内容模型"""__tablename__ = 'learning_content'id = Column(Integer, primary_key=True)content_id = Column(String(50), unique=True, nullable=False)title = Column(String(200), nullable=False)subject = Column(String(50), nullable=False)# 内容属性difficulty_level = Column(Float, nullable=False)  # 0.0-1.0estimated_duration = Column(Integer)  # 分钟learning_objectives = Column(Text)    # JSON格式# 文件路径scene_file_path = Column(String(500))thumbnail_path = Column(String(500))# 统计信息usage_count = Column(Integer, default=0)average_completion_time = Column(Float)average_score = Column(Float)created_at = Column(DateTime, default=func.now())updated_at = Column(DateTime, default=func.now(), onupdate=func.now())
class InteractionEvent(Base):"""交互事件模型"""__tablename__ = 'interaction_events'id = Column(Integer, primary_key=True)session_id = Column(Integer, nullable=False)timestamp = Column(DateTime, default=func.now())event_type = Column(String(50), nullable=False)  # grab, release, examine, etc.object_id = Column(String(100))# 位置数据position_x = Column(Float)position_y = Column(Float)position_z = Column(Float)# 旋转数据rotation_x = Column(Float)rotation_y = Column(Float)rotation_z = Column(Float)rotation_w = Column(Float)# 结果数据success = Column(Boolean, default=True)response_time_ms = Column(Integer)additional_data = Column(Text)  # JSON格式存储额外数据
# 数据库连接管理
class DatabaseManager:"""数据库管理器"""def __init__(self, database_url: str):self.engine = create_engine(database_url)self.SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=self.engine)def create_tables(self):"""创建所有表"""Base.metadata.create_all(bind=self.engine)print("✅ 数据库表创建成功")def get_session(self):"""获取数据库会话"""return self.SessionLocal()async def save_vr_session(self, session_data: dict):"""保存VR会话数据"""db = self.get_session()try:vr_session = VRSession(**session_data)db.add(vr_session)db.commit()db.refresh(vr_session)return vr_session.idexcept Exception as e:db.rollback()raise efinally:db.close()async def get_user_progress(self, user_id: int) -> dict:"""获取用户学习进度"""db = self.get_session()try:user = db.query(User).filter(User.id == user_id).first()if not user:return Nonerecent_sessions = db.query(VRSession).filter(VRSession.user_id == user_id).order_by(VRSession.start_time.desc()).limit(10).all()return {'user_info': {'username': user.username,'total_study_time': user.total_study_time,'sessions_completed': user.sessions_completed,'average_score': user.average_score},'recent_sessions': [{'scene_id': s.scene_id,'duration': s.duration_minutes,'score': s.final_score,'completion': s.completion_percentage} for s in recent_sessions]}finally:db.close()
# 使用示例
db_manager = DatabaseManager('postgresql://localhost/vr_education')

API接口开发

RESTful API设计

# backend/app/api/routes.py
from flask import Flask, request, jsonify
from flask_cors import CORS
import asyncio
from datetime import datetime
app = Flask(__name__)
CORS(app)
# 全局变量
scene_manager = None
analytics = None
db_manager = None
@app.route('/api/scenes', methods=['GET'])
def get_available_scenes():"""获取可用的VR场景列表"""try:# 模拟场景数据scenes = [{'scene_id': 'chemistry_lab','title': '化学实验室','subject': 'chemistry','difficulty': 0.6,'duration': 25,'description': '在虚拟实验室中进行安全的化学实验'},{'scene_id': 'solar_system','title': '太阳系探索','subject': 'astronomy','difficulty': 0.4,'duration': 30,'description': '近距离观察行星和恒星'},{'scene_id': 'human_anatomy','title': '人体解剖','subject': 'biology','difficulty': 0.8,'duration': 40,'description': '3D人体结构学习'}]return jsonify({'success': True,'scenes': scenes,'total': len(scenes)})except Exception as e:return jsonify({'success': False,'error': str(e)}), 500
@app.route('/api/session/start', methods=['POST'])
def start_learning_session():"""开始学习会话"""try:data = request.get_json()user_id = data.get('user_id')scene_id = data.get('scene_id')if not user_id or not scene_id:return jsonify({'success': False,'error': 'Missing user_id or scene_id'}), 400# 创建会话session_id = f"session_{user_id}_{scene_id}_{int(datetime.now().timestamp())}"session_data = {'session_id': session_id,'user_id': user_id,'scene_id': scene_id,'start_time': datetime.now(),'status': 'active'}return jsonify({'success': True,'session_id': session_id,'websocket_url': f'ws://localhost:8765','message': '会话创建成功'})except Exception as e:return jsonify({'success': False,'error': str(e)}), 500
@app.route('/api/session//progress', methods=['GET'])
def get_session_progress(session_id):"""获取会话进度"""try:# 模拟进度数据progress_data = {'session_id': session_id,'progress_percentage': 65.0,'time_spent': 15.5,'interactions_count': 23,'current_objective': '观察细胞结构','completed_objectives': 2,'total_objectives': 5}return jsonify({'success': True,'progress': progress_data})except Exception as e:return jsonify({'success': False,'error': str(e)}), 500
@app.route('/api/analytics/user/', methods=['GET'])
def get_user_analytics(user_id):"""获取用户学习分析数据"""try:# 模拟分析数据analytics_data = {'user_id': user_id,'total_sessions': 15,'total_study_time': 450,  # 分钟'average_session_score': 78.5,'learning_progress': {'chemistry': 85,'biology': 72,'physics': 90},'strengths': ['空间想象能力', '实验操作技能'],'improvement_areas': ['理论知识记忆', '公式应用'],'recommended_content': [{'content_id': 'organic_chemistry','reason': '基于你的化学实验表现，推荐有机化学进阶内容'}]}return jsonify({'success': True,'analytics': analytics_data})except Exception as e:return jsonify({'success': False,'error': str(e)}), 500
@app.route('/api/content/upload', methods=['POST'])
def upload_learning_content():"""上传学习内容"""try:if 'file' not in request.files:return jsonify({'success': False,'error': 'No file uploaded'}), 400file = request.files['file']content_info = request.form.to_dict()# 这里应该处理文件上传和内容解析# 简化实现return jsonify({'success': True,'content_id': f"content_{int(datetime.now().timestamp())}",'message': '内容上传成功'})except Exception as e:return jsonify({'success': False,'error': str(e)}), 500
if __name__ == '__main__':app.run(debug=True, host='0.0.0.0', port=5000)

部署和优化

Docker容器化部署

# Dockerfile
FROM python:3.9-slim
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y \gcc \g++ \libpq-dev \&& rm -rf /var/lib/apt/lists/*
# 复制依赖文件
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 复制应用代码
COPY . .
# 暴露端口
EXPOSE 5000 8765
# 启动脚本
CMD ["python", "run_server.py"]

# docker-compose.yml
version: '3.8'
services:vr-education-backend:build: .ports:- "5000:5000"- "8765:8765"environment:- DATABASE_URL=postgresql://postgres:password@db:5432/vr_education- REDIS_URL=redis://redis:6379depends_on:- db- redisvolumes:- ./content:/app/content- ./logs:/app/logsdb:image: postgres:13environment:- POSTGRES_DB=vr_education- POSTGRES_USER=postgres- POSTGRES_PASSWORD=passwordvolumes:- postgres_data:/var/lib/postgresql/dataports:- "5432:5432"redis:image: redis:6-alpineports:- "6379:6379"nginx:image: nginx:alpineports:- "80:80"- "443:443"volumes:- ./nginx.conf:/etc/nginx/nginx.conf- ./ssl:/etc/nginx/ssldepends_on:- vr-education-backend
volumes:postgres_data:

性能优化配置

# backend/app/utils/performance.py
import asyncio
import aioredis
from functools import wraps
import time
import cProfile
import pstats
from io import StringIO
class PerformanceOptimizer:"""性能优化工具类"""def __init__(self):self.redis_pool = Noneself.cache_enabled = Trueasync def init_redis(self):"""初始化Redis连接池"""self.redis_pool = aioredis.ConnectionPool.from_url("redis://localhost:6379",max_connections=20)def cache_result(self, expire_time=300):"""结果缓存装饰器"""def decorator(func):@wraps(func)async def wrapper(*args, **kwargs):if not self.cache_enabled:return await func(*args, **kwargs)# 生成缓存键cache_key = f"{func.__name__}:{hash(str(args) + str(kwargs))}"try:redis = aioredis.Redis(connection_pool=self.redis_pool)cached_result = await redis.get(cache_key)if cached_result:return eval(cached_result.decode())# 执行函数并缓存结果result = await func(*args, **kwargs)await redis.setex(cache_key, expire_time, str(result))return resultexcept Exception as e:print(f"缓存操作失败: {e}")return await func(*args, **kwargs)return wrapperreturn decoratordef profile_performance(self, func):"""性能分析装饰器"""@wraps(func)def wrapper(*args, **kwargs):profiler = cProfile.Profile()profiler.enable()start_time = time.time()result = func(*args, **kwargs)execution_time = time.time() - start_timeprofiler.disable()# 输出性能报告s = StringIO()ps = pstats.Stats(profiler, stream=s).sort_stats('cumulative')ps.print_stats()print(f" {func.__name__} 执行时间: {execution_time:.4f}s")print(" 详细性能报告:")print(s.getvalue())return resultreturn wrapper
# 全局性能优化实例
perf_optimizer = PerformanceOptimizer()
# 使用示例
@perf_optimizer.cache_result(expire_time=600)
async def get_scene_data(scene_id: str):"""获取场景数据（带缓存）"""# 模拟耗时操作await asyncio.sleep(0.1)return {'scene_id': scene_id, 'data': 'scene_content'}
@perf_optimizer.profile_performance
def complex_calculation():"""复杂计算（性能分析）"""result = sum(i**2 for i in range(10000))return result

测试和质量保证

单元测试

# tests/test_scene_manager.py
import pytest
import asyncio
from unittest.mock import Mock, patch
import sys
import os
# 添加项目路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from backend.app.services.scene_manager import SceneManager, VRScene
class TestSceneManager:"""场景管理器测试类"""@pytest.fixturedef scene_manager(self):"""测试用场景管理器实例"""return SceneManager()@pytest.fixturedef sample_scene_data(self):"""示例场景数据"""return {'scene_id': 'test_scene','title': '测试场景','subject': 'test','difficulty_level': 0.5,'objects': [{'id': 'obj1', 'type': 'cube'}],'interactions': [{'type': 'grab', 'target': 'obj1'}],'learning_objectives': ['目标1', '目标2']}@pytest.mark.asyncioasync def test_load_scene_success(self, scene_manager, sample_scene_data):"""测试成功加载场景"""with patch('builtins.open'), \patch('json.load', return_value=sample_scene_data):scene = await scene_manager.load_scene('test_scene')assert scene is not Noneassert scene.scene_id == 'test_scene'assert scene.title == '测试场景'assert len(scene.learning_objectives) == 2@pytest.mark.asyncioasync def test_load_scene_file_not_found(self, scene_manager):"""测试场景文件不存在"""with patch('builtins.open', side_effect=FileNotFoundError):scene = await scene_manager.load_scene('nonexistent_scene')assert scene is None@pytest.mark.asyncioasync def test_create_learning_session(self, scene_manager, sample_scene_data):"""测试创建学习会话"""with patch('builtins.open'), \patch('json.load', return_value=sample_scene_data):session = await scene_manager.create_learning_session('user123', 'test_scene')assert session is not Noneassert session['user_id'] == 'user123'assert 'session_id' in sessionassert session['progress'] == 0def test_get_session_progress(self, scene_manager):"""测试获取会话进度"""# 模拟会话数据session_id = 'test_session'scene_manager.active_sessions[session_id] = {'scene': VRScene(scene_id='test', title='test', subject='test',difficulty_level=0.5, objects=[], interactions=[],learning_objectives=['obj1', 'obj2']),'interactions': [{'completed': True},{'completed': False}]}progress = scene_manager.get_session_progress(session_id)assert 'progress' in progressassert progress['progress'] == 50.0  # 1/2 completedassert progress['completed_objectives'] == 1assert progress['total_objectives'] == 2
# 运行测试
if __name__ == '__main__':pytest.main([__file__, '-v'])

集成测试

# tests/test_integration.py
import pytest
import asyncio
import json
from unittest.mock import patch
import websockets
from backend.app.services.websocket_server import VRWebSocketServer
class TestVRIntegration:"""VR应用集成测试"""@pytest.fixtureasync def websocket_server(self):"""WebSocket服务器实例"""server = VRWebSocketServer()return server@pytest.mark.asyncioasync def test_websocket_connection_flow(self, websocket_server):"""测试WebSocket连接流程"""# 启动服务器server_task = asyncio.create_task(websocket_server.start_server('localhost', 8766))# 等待服务器启动await asyncio.sleep(0.1)try:# 测试客户端连接uri = "ws://localhost:8766"async with websockets.connect(uri) as websocket:# 发送测试消息test_message = {'type': 'scene_request','scene_id': 'test_scene'}await websocket.send(json.dumps(test_message))# 接收响应response = await websocket.recv()response_data = json.loads(response)assert response_data['type'] == 'scene_data'assert response_data['scene_id'] == 'test_scene'finally:server_task.cancel()@pytest.mark.asyncioasync def test_api_and_websocket_integration(self):"""测试API和WebSocket集成"""# 这里应该测试API创建会话后WebSocket能正确处理pass
# 性能测试
class TestPerformance:"""性能测试类"""@pytest.mark.asyncioasync def test_concurrent_sessions(self):"""测试并发会话处理能力"""scene_manager = SceneManager()# 模拟并发创建多个会话tasks = []for i in range(100):task = scene_manager.create_learning_session(f'user_{i}', 'test_scene')tasks.append(task)with patch('builtins.open'), \patch('json.load', return_value={'scene_id': 'test', 'title': 'test', 'subject': 'test', 'difficulty_level': 0.5, 'objects': [], 'interactions': [], 'learning_objectives': []}):start_time = asyncio.get_event_loop().time()results = await asyncio.gather(*tasks)end_time = asyncio.get_event_loop().time()# 验证所有会话都成功创建successful_sessions = [r for r in results if r is not None]assert len(successful_sessions) == 100# 验证性能要求（例如100个会话在1秒内创建）assert (end_time - start_time) < 1.0def test_memory_usage(self):"""测试内存使用情况"""import psutilimport osprocess = psutil.Process(os.getpid())initial_memory = process.memory_info().rss# 执行一些操作scene_manager = SceneManager()for i in range(1000):scene_manager.scenes[f'scene_{i}'] = VRScene(scene_id=f'scene_{i}', title='test', subject='test',difficulty_level=0.5, objects=[], interactions=[],learning_objectives=[])final_memory = process.memory_info().rssmemory_increase = final_memory - initial_memory# 验证内存增长在合理范围内（例如小于100MB）assert memory_increase < 100 * 1024 * 1024

项目总结与展望

技术亮点

模块化设计：清晰的分层架构，便于维护和扩展
异步处理：使用asyncio提高并发性能
智能分析：集成机器学习进行学习行为分析
实时通信：WebSocket支持VR设备实时交互
性能优化：Redis缓存和性能监控

优化建议

# 未来优化方向
FUTURE_OPTIMIZATIONS = {'performance': ['实现GPU加速的3D渲染','优化WebSocket消息传输协议','添加CDN支持静态资源','实现数据库读写分离'],'features': ['增加眼动追踪支持','实现多用户协同学习','添加语音识别和合成','支持手势识别交互'],'ai_enhancement': ['个性化学习路径推荐','情感计算和反馈','智能内容生成','预测性学习分析']
}