EG/plugins/user/weather_season_system/weather/weather_system.py
2025-10-30 11:46:41 +08:00

697 lines
25 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
天气系统
负责管理天气现象、天气变化和天气物理效果
"""
import time
import random
from typing import Dict, Any, List, Optional, Tuple
import math
class WeatherSystem:
"""
天气系统
负责管理天气现象、天气变化和天气物理效果
"""
def __init__(self, plugin):
"""
初始化天气系统
Args:
plugin: 天气和季节系统插件实例
"""
self.plugin = plugin
self.enabled = False
self.initialized = False
# 天气现象配置
self.weather_phenomena = {
'precipitation': {
'name': '降水',
'types': ['rain', 'snow', 'hail'],
'intensity_range': (0.0, 1.0),
'accumulation_rate': 0.1, # 每秒积累量
'melt_rate': 0.05, # 雪融化速率
'splash_effect': True
},
'wind': {
'name': '',
'speed_range': (0.0, 50.0), # 风速范围(m/s)
'direction_variability': 0.1, # 方向变化性
'gust_factor': 2.0, # 阵风倍数
'turbulence': 0.2
},
'temperature': {
'name': '温度',
'range': (-30.0, 50.0), # 温度范围(°C)
'change_rate': 0.1, # 温度变化速率
'daily_variation': 10.0, # 日变化幅度
'seasonal_variation': 20.0 # 季节变化幅度
},
'humidity': {
'name': '湿度',
'range': (0.0, 100.0), # 湿度范围(%)
'change_rate': 0.5, # 湿度变化速率
'precipitation_effect': 10.0 # 降水对湿度的影响
},
'pressure': {
'name': '气压',
'range': (950.0, 1050.0), # 气压范围(hPa)
'change_rate': 0.1, # 气压变化速率
'weather_indicator': True # 是否指示天气变化
},
'visibility': {
'name': '能见度',
'range': (10.0, 10000.0), # 能见度范围(米)
'fog_effect': 0.8, # 雾对能见度的影响
'precipitation_effect': 0.6 # 降水对能见度的影响
}
}
# 当前天气状态
self.current_conditions = {
'temperature': 20.0, # 温度(°C)
'humidity': 50.0, # 湿度(%)
'pressure': 1013.25, # 气压(hPa)
'wind_speed': 3.0, # 风速(m/s)
'wind_direction': 0.0, # 风向(度)
'precipitation': 0.0, # 降水强度(0-1)
'precipitation_type': 'none', # 降水类型
'visibility': 10000.0, # 能见度(米)
'cloud_cover': 0.1, # 云层覆盖率(0-1)
'uv_index': 2.0 # UV指数
}
# 天气物理效果
self.physical_effects = {
'friction_modifier': 1.0, # 摩擦系数修改器
'movement_speed_modifier': 1.0, # 移动速度修改器
'sound_attenuation': 1.0, # 声音衰减
'light_scattering': 0.0, # 光线散射
'surface_wetness': 0.0, # 表面湿润度
'surface_ice': 0.0 # 表面结冰
}
# 天气更新配置
self.update_interval = 1.0 # 更新间隔(秒)
self.last_update_time = 0.0
# 天气预测
self.weather_forecast = []
self.forecast_duration = 24 # 预测时长(小时)
# 天气区域配置
self.weather_zones = {}
self.zone_influence_radius = 100.0
# 统计信息
self.stats = {
'conditions_updated': 0,
'weather_changes': 0,
'phenomena_occurred': 0,
'total_weather_time': 0.0
}
print("✓ 天气系统已创建")
def initialize(self) -> bool:
"""
初始化天气系统
Returns:
是否初始化成功
"""
try:
# 生成初始天气预报
self._generate_weather_forecast()
self.initialized = True
print("✓ 天气系统初始化完成")
return True
except Exception as e:
print(f"✗ 天气系统初始化失败: {e}")
import traceback
traceback.print_exc()
return False
def enable(self) -> bool:
"""
启用天气系统
Returns:
是否启用成功
"""
try:
if not self.initialized:
print("✗ 天气系统未初始化")
return False
self.enabled = True
print("✓ 天气系统已启用")
return True
except Exception as e:
print(f"✗ 天气系统启用失败: {e}")
import traceback
traceback.print_exc()
return False
def disable(self):
"""禁用天气系统"""
try:
self.enabled = False
print("✓ 天气系统已禁用")
except Exception as e:
print(f"✗ 天气系统禁用失败: {e}")
import traceback
traceback.print_exc()
def finalize(self):
"""清理天气系统资源"""
try:
self.disable()
self.initialized = False
print("✓ 天气系统资源已清理")
except Exception as e:
print(f"✗ 天气系统资源清理失败: {e}")
import traceback
traceback.print_exc()
def update(self, dt: float):
"""
更新天气系统状态
Args:
dt: 时间增量
"""
try:
if not self.enabled:
return
# 更新计时器
self.last_update_time += dt
if self.last_update_time < self.update_interval:
return
# 重置更新计时器
self.last_update_time = 0.0
# 更新天气条件
self._update_weather_conditions(dt)
# 更新天气物理效果
self._update_physical_effects()
# 更新天气预报
self._update_weather_forecast()
# 更新统计信息
self.stats['total_weather_time'] += dt
except Exception as e:
print(f"✗ 天气系统更新失败: {e}")
import traceback
traceback.print_exc()
def _update_weather_conditions(self, dt: float):
"""
更新天气条件
Args:
dt: 时间增量
"""
try:
# 获取当前季节和天气参数
if self.plugin.season_manager:
season_params = self.plugin.season_manager.get_interpolated_season_params()
else:
season_params = {}
if self.plugin.weather_manager:
weather_params = self.plugin.weather_manager.get_interpolated_weather_params()
else:
weather_params = {}
# 更新温度
self._update_temperature(season_params, weather_params, dt)
# 更新湿度
self._update_humidity(weather_params, dt)
# 更新风
self._update_wind(weather_params, dt)
# 更新降水
self._update_precipitation(weather_params, dt)
# 更新气压
self._update_pressure(weather_params, dt)
# 更新能见度
self._update_visibility(weather_params)
# 更新云层覆盖率
self.current_conditions['cloud_cover'] = weather_params.get('cloud_cover', 0.1)
# 更新UV指数
self._update_uv_index()
# 更新统计信息
self.stats['conditions_updated'] += 1
except Exception as e:
print(f"✗ 天气条件更新失败: {e}")
def _update_temperature(self, season_params: Dict, weather_params: Dict, dt: float):
"""
更新温度
Args:
season_params: 季节参数
weather_params: 天气参数
dt: 时间增量
"""
try:
# 基础温度受季节影响
base_temp = season_params.get('temperature', 20.0)
# 天气影响
temp_modifier = weather_params.get('temperature_modifier', 0.0)
# 日变化(简化模型)
if self.plugin.environment_effects:
time_config = getattr(self.plugin.environment_effects, 'time_config', {'day_time': 12.0})
current_time = time_config.get('day_time', 12.0)
# 正弦变化12点最高0点最低
daily_variation = math.sin((current_time - 6) * math.pi / 12) * 5.0
else:
daily_variation = 0.0
# 计算最终温度
target_temp = base_temp + temp_modifier + daily_variation
# 平滑过渡到目标温度
current_temp = self.current_conditions['temperature']
change_rate = self.weather_phenomena['temperature']['change_rate']
self.current_conditions['temperature'] = current_temp + (target_temp - current_temp) * change_rate * dt
except Exception as e:
print(f"✗ 温度更新失败: {e}")
def _update_humidity(self, weather_params: Dict, dt: float):
"""
更新湿度
Args:
weather_params: 天气参数
dt: 时间增量
"""
try:
# 基础湿度受季节影响
if self.plugin.season_manager:
season_params = self.plugin.season_manager.get_interpolated_season_params()
base_humidity = season_params.get('humidity', 50.0)
else:
base_humidity = 50.0
# 天气影响
humidity_modifier = weather_params.get('humidity_modifier', 0.0)
# 降水影响
precipitation = self.current_conditions['precipitation']
precipitation_effect = self.weather_phenomena['humidity']['precipitation_effect']
# 计算目标湿度
target_humidity = base_humidity + humidity_modifier + precipitation * precipitation_effect
target_humidity = max(0.0, min(100.0, target_humidity))
# 平滑过渡到目标湿度
current_humidity = self.current_conditions['humidity']
change_rate = self.weather_phenomena['humidity']['change_rate']
self.current_conditions['humidity'] = current_humidity + (target_humidity - current_humidity) * change_rate * dt
except Exception as e:
print(f"✗ 湿度更新失败: {e}")
def _update_wind(self, weather_params: Dict, dt: float):
"""
更新风
Args:
weather_params: 天气参数
dt: 时间增量
"""
try:
# 基础风速受天气影响
base_wind_speed = weather_params.get('wind_speed', 3.0)
# 添加随机扰动
turbulence = self.weather_phenomena['wind']['turbulence']
wind_variation = random.uniform(-turbulence, turbulence) * base_wind_speed
# 更新风速
self.current_conditions['wind_speed'] = max(0.0, base_wind_speed + wind_variation)
# 更新风向(缓慢变化)
direction_variability = self.weather_phenomena['wind']['direction_variability']
direction_change = random.uniform(-direction_variability, direction_variability) * 180.0 * dt
self.current_conditions['wind_direction'] = (self.current_conditions['wind_direction'] + direction_change) % 360.0
except Exception as e:
print(f"✗ 风更新失败: {e}")
def _update_precipitation(self, weather_params: Dict, dt: float):
"""
更新降水
Args:
weather_params: 天气参数
dt: 时间增量
"""
try:
# 获取降水强度
precipitation = weather_params.get('precipitation', 0.0)
self.current_conditions['precipitation'] = precipitation
# 根据天气类型设置降水类型
current_weather = self.plugin.weather_manager.get_current_weather() if self.plugin.weather_manager else 'clear'
if precipitation > 0:
if current_weather == 'snow':
self.current_conditions['precipitation_type'] = 'snow'
elif current_weather == 'hail':
self.current_conditions['precipitation_type'] = 'hail'
else:
self.current_conditions['precipitation_type'] = 'rain'
else:
self.current_conditions['precipitation_type'] = 'none'
except Exception as e:
print(f"✗ 降水更新失败: {e}")
def _update_pressure(self, weather_params: Dict, dt: float):
"""
更新气压
Args:
weather_params: 天气参数
dt: 时间增量
"""
try:
# 基础气压
base_pressure = self.current_conditions['pressure']
# 添加随机变化
pressure_change = random.uniform(-0.5, 0.5)
self.current_conditions['pressure'] = max(950.0, min(1050.0, base_pressure + pressure_change))
except Exception as e:
print(f"✗ 气压更新失败: {e}")
def _update_visibility(self, weather_params: Dict):
"""
更新能见度
Args:
weather_params: 天气参数
"""
try:
# 基础能见度受天气影响
visibility_factor = weather_params.get('visibility', 1.0)
base_visibility = self.weather_phenomena['visibility']['range'][1] # 最大能见度
# 计算最终能见度
final_visibility = base_visibility * visibility_factor
# 降水影响
precipitation = self.current_conditions['precipitation']
precipitation_effect = self.weather_phenomena['visibility']['precipitation_effect']
final_visibility *= (1.0 - precipitation * precipitation_effect)
self.current_conditions['visibility'] = max(10.0, final_visibility)
except Exception as e:
print(f"✗ 能见度更新失败: {e}")
def _update_uv_index(self):
"""更新UV指数"""
try:
# 简化的UV指数计算
if self.plugin.environment_effects:
time_config = getattr(self.plugin.environment_effects, 'time_config', {'day_time': 12.0})
current_time = time_config.get('day_time', 12.0)
# 正午最高早晚为0
if 8 <= current_time <= 16:
hour_factor = 1.0 - abs(current_time - 12.0) / 4.0
else:
hour_factor = 0.0
# 晴天时UV指数最高
current_weather = self.plugin.weather_manager.get_current_weather() if self.plugin.weather_manager else 'clear'
weather_factor = 1.0 if current_weather == 'clear' else 0.5
self.current_conditions['uv_index'] = max(0.0, 10.0 * hour_factor * weather_factor)
else:
self.current_conditions['uv_index'] = 2.0
except Exception as e:
print(f"✗ UV指数更新失败: {e}")
def _update_physical_effects(self):
"""更新天气物理效果"""
try:
precipitation = self.current_conditions['precipitation']
precipitation_type = self.current_conditions['precipitation_type']
temperature = self.current_conditions['temperature']
wind_speed = self.current_conditions['wind_speed']
# 更新表面湿润度
if precipitation_type in ['rain', 'hail']:
self.physical_effects['surface_wetness'] = min(1.0, precipitation * 2.0)
elif precipitation_type == 'snow':
self.physical_effects['surface_wetness'] = min(0.3, precipitation)
else:
# 无降水时逐渐干燥
self.physical_effects['surface_wetness'] = max(0.0, self.physical_effects['surface_wetness'] - 0.01)
# 更新表面结冰
if temperature < 0 and self.physical_effects['surface_wetness'] > 0.1:
self.physical_effects['surface_ice'] = min(1.0, self.physical_effects['surface_wetness'] * 0.5)
else:
self.physical_effects['surface_ice'] = max(0.0, self.physical_effects['surface_ice'] - 0.02)
# 更新摩擦系数
wetness = self.physical_effects['surface_wetness']
ice = self.physical_effects['surface_ice']
self.physical_effects['friction_modifier'] = max(0.1, 1.0 - wetness * 0.5 - ice * 0.8)
# 更新移动速度
self.physical_effects['movement_speed_modifier'] = self.physical_effects['friction_modifier'] * 0.8 + 0.2
# 更新声音衰减
self.physical_effects['sound_attenuation'] = 1.0 - precipitation * 0.3
# 更新光线散射
self.physical_effects['light_scattering'] = precipitation * 0.5 + (1.0 - self.current_conditions['visibility'] / 10000.0) * 0.5
except Exception as e:
print(f"✗ 物理效果更新失败: {e}")
def _generate_weather_forecast(self):
"""生成天气预报"""
try:
self.weather_forecast = []
# 获取当前天气
current_weather = self.plugin.weather_manager.get_current_weather() if self.plugin.weather_manager else 'clear'
# 生成未来几小时的预报
for i in range(self.forecast_duration):
# 简化的预报生成
forecast_entry = {
'hour': i,
'weather': current_weather,
'temperature': self.current_conditions['temperature'] + random.uniform(-2, 2),
'precipitation': self.current_conditions['precipitation']
}
self.weather_forecast.append(forecast_entry)
except Exception as e:
print(f"✗ 天气预报生成失败: {e}")
def _update_weather_forecast(self):
"""更新天气预报"""
try:
# 移除过期的预报
if self.weather_forecast:
self.weather_forecast.pop(0)
# 添加新的预报
current_weather = self.plugin.weather_manager.get_current_weather() if self.plugin.weather_manager else 'clear'
new_entry = {
'hour': len(self.weather_forecast),
'weather': current_weather,
'temperature': self.current_conditions['temperature'] + random.uniform(-2, 2),
'precipitation': self.current_conditions['precipitation']
}
self.weather_forecast.append(new_entry)
except Exception as e:
print(f"✗ 天气预报更新失败: {e}")
def get_current_conditions(self) -> Dict[str, Any]:
"""
获取当前天气条件
Returns:
当前天气条件字典
"""
return self.current_conditions.copy()
def get_physical_effects(self) -> Dict[str, float]:
"""
获取天气物理效果
Returns:
天气物理效果字典
"""
return self.physical_effects.copy()
def get_weather_phenomena_list(self) -> List[str]:
"""
获取天气现象列表
Returns:
天气现象名称列表
"""
return list(self.weather_phenomena.keys())
def get_weather_forecast(self, hours: int = 24) -> List[Dict[str, Any]]:
"""
获取天气预报
Args:
hours: 预报时长(小时)
Returns:
天气预报列表
"""
return self.weather_forecast[:hours]
def get_weather_zone(self, zone_id: str) -> Optional[Dict[str, Any]]:
"""
获取天气区域信息
Args:
zone_id: 区域ID
Returns:
天气区域信息或None
"""
return self.weather_zones.get(zone_id)
def set_weather_zone(self, zone_id: str, zone_data: Dict[str, Any]):
"""
设置天气区域
Args:
zone_id: 区域ID
zone_data: 区域数据
"""
self.weather_zones[zone_id] = zone_data
print(f"✓ 天气区域已设置: {zone_id}")
def remove_weather_zone(self, zone_id: str):
"""
移除天气区域
Args:
zone_id: 区域ID
"""
if zone_id in self.weather_zones:
del self.weather_zones[zone_id]
print(f"✓ 天气区域已移除: {zone_id}")
def get_weather_impact_on_object(self, object_position: Tuple[float, float, float]) -> Dict[str, Any]:
"""
获取天气对特定位置物体的影响
Args:
object_position: 物体位置(x, y, z)
Returns:
天气影响字典
"""
try:
# 简化实现,实际中需要考虑物体与天气区域的关系
impact = {
'wetness': self.physical_effects['surface_wetness'],
'icing': self.physical_effects['surface_ice'],
'wind_effect': self.current_conditions['wind_speed'] / 50.0,
'visibility_factor': self.current_conditions['visibility'] / 10000.0,
'temperature': self.current_conditions['temperature']
}
return impact
except Exception as e:
print(f"✗ 物体天气影响计算失败: {e}")
return {}
def set_update_interval(self, interval: float):
"""
设置更新间隔
Args:
interval: 更新间隔(秒)
"""
self.update_interval = max(0.1, interval)
print(f"✓ 天气更新间隔设置为: {interval}")
def get_stats(self) -> Dict[str, Any]:
"""
获取统计信息
Returns:
统计信息字典
"""
return self.stats.copy()
def reset_stats(self):
"""重置统计信息"""
self.stats = {
'conditions_updated': 0,
'weather_changes': 0,
'phenomena_occurred': 0,
'total_weather_time': 0.0
}
print("✓ 天气系统统计信息已重置")
def is_weather_phenomenon_active(self, phenomenon: str) -> bool:
"""
检查天气现象是否活跃
Args:
phenomenon: 天气现象名称
Returns:
天气现象是否活跃
"""
try:
if phenomenon == 'precipitation':
return self.current_conditions['precipitation'] > 0
elif phenomenon == 'wind':
return self.current_conditions['wind_speed'] > 1.0
elif phenomenon == 'extreme_temperature':
temp = self.current_conditions['temperature']
return temp < 0 or temp > 35
return False
except Exception as e:
print(f"✗ 天气现象检查失败: {e}")
return False