import json import numpy as np from datetime import datetime, timedelta import random import math class RunoffSimulator: """模拟河流流量数据生成器""" def __init__(self, base_flow=50, seasonal_amp=20, daily_amp=10, random_factor=5): """ 初始化流量模拟器 Args: base_flow: 基础流量 seasonal_amp: 季节性波动幅度 daily_amp: 日变化波动幅度 random_factor: 随机波动幅度 """ self.base_flow = base_flow self.seasonal_amp = seasonal_amp self.daily_amp = daily_amp self.random_factor = random_factor def generate_daily_runoff(self, river_id, day_offset=0): """ 生成一天24小时的流量数据 Args: river_id: 河流ID day_offset: 日期偏移,用于生成不同的随机种子 Returns: list: 24小时的流量数据 """ hourly_data = [] # 为每条河流设置一个固定的随机种子 random.seed(hash(f"{river_id}_{day_offset}")) for hour in range(24): # 计算日变化影响(使用正弦函数模拟) daily_variation = self.daily_amp * math.sin(2 * math.pi * (hour - 6) / 24) # 添加随机波动 random_variation = random.uniform(-self.random_factor, self.random_factor) # 计算总流量 runoff = max(0, self.base_flow + daily_variation + random_variation) hourly_data.append({ 'index': river_id, 'hour': hour, 'runoff': round(runoff, 2) }) return hourly_data class RiverCzmlGenerator: def __init__(self, start_time="2024-06-01T00:00:00Z"): self.start_time = start_time self._init_base_document() def _init_base_document(self): """初始化CZML文档的基本结构""" self.czml_doc = [{ "id": "document", "name": "River Network Visualization", "version": "1.0", "clock": { "interval": f"{self.start_time}/{self._get_end_time()}", "currentTime": self.start_time, "multiplier": 3600, "range": "LOOP_STOP", "step": "SYSTEM_CLOCK_MULTIPLIER" } }] def _get_end_time(self): start = datetime.strptime(self.start_time, "%Y-%m-%dT%H:%M:%SZ") end = start + timedelta(days=1) return end.strftime("%Y-%m-%dT%H:%M:%SZ") def _calculate_color(self, runoff, min_runoff, max_runoff): """计算流量对应的颜色""" ratio = (runoff - min_runoff) / (max_runoff - min_runoff) if max_runoff > min_runoff else 0 # 使用更丰富的颜色渐变:从浅蓝到深蓝 r = int(135 - (75 * ratio)) # 135 -> 60 g = int(206 - (156 * ratio)) # 206 -> 50 b = int(235 - (35 * ratio)) # 235 -> 200 a = 255 return [r, g, b, a] def _flatten_coordinates(self, coordinates): """展平MultiLineString坐标""" flattened = [] for line_string in coordinates: for coord in line_string: flattened.extend([coord[0], coord[1], 0]) return flattened def generate_czml(self, geojson_data, runoff_data): """生成CZML文档""" # 构建river_id到runoff数据的映射 runoff_map = {} all_runoffs = [] for data in runoff_data: river_id = data['river_id'] if river_id not in runoff_map: runoff_map[river_id] = [] runoff_map[river_id].append(data) all_runoffs.append(data['runoff']) min_runoff = min(all_runoffs) max_runoff = max(all_runoffs) for feature in geojson_data['features']: river_id = feature['properties']['Index'] if river_id not in runoff_map: continue river_runoffs = sorted(runoff_map[river_id], key=lambda x: x['hour']) color_property = [] for runoff_data in river_runoffs: time_offset = runoff_data['hour'] * 3600 color_property.append(time_offset) color = self._calculate_color(runoff_data['runoff'], min_runoff, max_runoff) color_property.extend(color) river_entity = { "id": f"river-{river_id}", "name": f"River {river_id}", "polyline": { "positions": { "cartographicDegrees": self._flatten_coordinates( feature['geometry']['coordinates'] ) }, "material": { "polylineOutline": { "color": { "epoch": self.start_time, "rgba": color_property } } }, "width": 3, "clampToGround": True } } self.czml_doc.append(river_entity) return self.czml_doc def generate_sample_geojson(num_rivers=5): """ 生成示例河网GeoJSON数据 Args: num_rivers: 河流数量 Returns: dict: GeoJSON数据 """ features = [] # 设置基础区域范围 base_lon = 120.0 base_lat = 30.0 for i in range(num_rivers): # 为每条河流生成随机的弯曲线段 num_points = random.randint(5, 10) line_coordinates = [] # 生成主河道 current_lon = base_lon + random.uniform(-0.5, 0.5) current_lat = base_lat + random.uniform(-0.5, 0.5) points = [] for j in range(num_points): points.append([current_lon, current_lat]) # 添加随机偏移 current_lon += random.uniform(0.02, 0.05) current_lat += random.uniform(-0.02, 0.02) line_coordinates.append(points) # 可能添加支流 if random.random() < 0.5: branch_points = [] branch_start = random.randint(1, len(points)-2) current_lon = points[branch_start][0] current_lat = points[branch_start][1] for j in range(random.randint(3, 6)): branch_points.append([current_lon, current_lat]) current_lon += random.uniform(-0.03, 0.03) current_lat += random.uniform(0.02, 0.04) line_coordinates.append(branch_points) feature = { "type": "Feature", "properties": { "river_id": str(i+1) }, "geometry": { "type": "MultiLineString", "coordinates": line_coordinates } } features.append(feature) return { "type": "FeatureCollection", "features": features } def main(): # 生成示例GeoJSON数据 num_rivers = 5 geojson_data = generate_sample_geojson(num_rivers) # 生成流量数据 simulator = RunoffSimulator() runoff_data = [] for i in range(num_rivers): river_id = str(i+1) runoff_data.extend(simulator.generate_daily_runoff(river_id)) # 生成CZML generator = RiverCzmlGenerator() czml_data = generator.generate_czml(geojson_data, runoff_data) # 保存数据 with open('out/river_network.geojson', 'w', encoding='utf-8') as f: json.dump(geojson_data, f, indent=2) with open('out/river_network.czml', 'w', encoding='utf-8') as f: json.dump(czml_data, f, indent=2) print("生成的文件:") print("1. river_network.geojson - 河网数据") print("2. river_network.czml - CZML动画数据") # 输出一些统计信息 total_points = sum(len(feature['geometry']['coordinates'][0]) for feature in geojson_data['features']) print(f"\n统计信息:") print(f"河流数量: {num_rivers}") print(f"总点数: {total_points}") print(f"流量数据点数: {len(runoff_data)}") if __name__ == '__main__': main()