import pyarrow as pa
import pyarrow.parquet as pq
from quadtree import QuadTree
from libcuckoo import cuckoo_hashclass StreamingGeoIndex:def __init__(self, shard_size=1e6):"""初始化流式空间索引"""self.shard_size = shard_sizeself.current_shard = 0self.mem_table = cuckoo_hash()  # 内存索引self.local_index = QuadTree()   # 本地磁盘索引self.dist_index = {}            # 分布式索引def ingest(self, geodata: pa.Table):"""实时数据摄入"""# 内存表更新for i in range(len(geodata)):wkt = geodata['geometry'][i].as_py()self.mem_table.insert(wkt, geodata['id'][i].as_py())# 定期持久化if len(self.mem_table) > self.shard_size * 0.8:self._flush_to_disk()def _flush_to_disk(self):"""数据刷写到磁盘"""# 构建本地QuadTree索引for item in self.mem_table.items():self.local_index.insert(item.geometry, item.id)# 生成GeoHash分片键geohash = self._compute_geohash(self.mem_table.bbox())self.dist_index[geohash] = f'shard_{self.current_shard}.parquet'# 写入Parquet文件pq.write_table(self.mem_table.to_arrow(), f'data/shard_{self.current_shard}.parquet')self.current_shard += 1self.mem_table.clear()def query(self, geometry, level='memory'):"""多级空间查询"""results = set()# 内存级查询if level in ['memory', 'all']:results.update(self.mem_table.query(geometry))# 本地级查询if level in ['local', 'all']:results.update(self.local_index.query(geometry))# 分布式查询if level == 'all':geohash = self._compute_geohash(geometry.bounds)for shard in self._get_related_shards(geohash):table = pq.read_table(f'data/{shard}')results.update(self._query_parquet(table, geometry))return list(results)def _query_parquet(self, table, geometry):"""查询Parquet文件"""# 使用PyArrow计算引擎加速import pyarrow.compute as pcmask = pc.geo.within(table['geometry'], geometry)return table.filter(mask)['id'].to_pylist()# 使用示例
if __name__ == "__main__":# 模拟流式数据data = pa.table({'id': range(100000),'geometry': [f"POINT({x} {y})" for x,y in zip(np.random.uniform(0,100,100000),np.random.uniform(0,100,100000))]})index = StreamingGeoIndex()for batch in data.to_batches(1000):index.ingest(pa.Table.from_batches([batch]))# 执行查询from shapely import boxresults = index.query(box(30,30,70,70))print(f"查询结果数量: {len(results)}")

import pandas as pd
from pyspark.sql import functions as F
from pyspark.sql.types import *class TrajectoryProcessor:"""分布式轨迹处理引擎"""def __init__(self, spark):self.spark = sparkdef compress(self, df, tolerance=100):"""分布式轨迹压缩"""# 定义UDF@F.pandas_udf(ArrayType(StructType([StructField("ts", TimestampType()),StructField("lon", DoubleType()),StructField("lat", DoubleType())])))def douglas_peucker(points):from trajectorytools import compressreturn compress(points, tolerance)return df.groupBy("traj_id").agg(douglas_peucker(F.collect_list(F.struct("timestamp", "longitude", "latitude"))).alias("compressed_traj"))def extract_features(self, df):"""轨迹特征提取"""# 停留点检测stay_points = self._detect_stay_points(df)# 移动特征计算movement_features = df.groupBy("traj_id").agg(F.countDistinct("timestamp").alias("point_count"),F.max("speed").alias("max_speed"),F.avg("speed").alias("avg_speed"),self._haversine_udf(F.first("longitude"), F.first("latitude"),F.last("longitude"), F.last("latitude")).alias("distance"))return stay_points.join(movement_features, "traj_id")def _detect_stay_points(self, df):"""停留点检测"""# 使用窗口函数检测停留window = Window.partitionBy("traj_id").orderBy("timestamp")return df.withColumn("time_diff", F.unix_timestamp("timestamp") - F.unix_timestamp(F.lag("timestamp").over(window))).withColumn("dist_diff",self._haversine_udf(F.lag("longitude").over(window),F.lag("latitude").over(window),F.col("longitude"),F.col("latitude"))).groupBy("traj_id", "stay_cluster").agg(F.avg("longitude").alias("center_lon"),F.avg("latitude").alias("center_lat"),F.min("timestamp").alias("start_time"),F.max("timestamp").alias("end_time")).filter("end_time - start_time >= 300")  # 至少5分钟@staticmethoddef _haversine_udf(lon1, lat1, lon2, lat2):"""Haversine距离UDF"""return F.udf(lambda a,b,c,d: haversine((a,b),(c,d)),DoubleType())(lon1, lat1, lon2, lat2)# Spark使用示例
if __name__ == "__main__":from pyspark.sql import SparkSessionspark = SparkSession.builder \.appName("TrajectoryProcessing") \.getOrCreate()# 加载轨迹数据df = spark.read.parquet("hdfs://trajectories/*.parquet")processor = TrajectoryProcessor(spark)# 轨迹压缩compressed = processor.compress(df, tolerance=50)compressed.write.parquet("hdfs://compressed_trajs/")# 特征提取features = processor.extract_features(df)features.write.parquet("hdfs://traj_features/")

import networkx as nx
from ortools.constraint_solver import routing_enums_pb2
from ortools.constraint_solver import pywrapcpclass MultimodalRouter:"""多模态路径规划引擎"""def __init__(self, road_graph, transit_graph):self.road_graph = road_graphself.transit_graph = transit_graphself.transfer_nodes = self._find_transfer_nodes()def _find_transfer_nodes(self):"""识别换乘节点"""road_nodes = set(self.road_graph.nodes())transit_nodes = set(self.transit_graph.nodes())return list(road_nodes & transit_nodes)def plan_journey(self, origin, destination, modes, time_window):"""多模态行程规划"""# 构建统一图模型combined_graph = nx.compose_all([self.road_graph,self.transit_graph,self._build_transfer_edges()])# 设置OR-Tools求解器manager = pywrapcp.RoutingIndexManager(len(combined_graph.nodes()),1,  # 车辆数list(combined_graph.nodes()).index(origin),list(combined_graph.nodes()).index(destination))routing = pywrapcp.RoutingModel(manager)# 定义代价函数def time_callback(from_index, to_index):from_node = manager.IndexToNode(from_index)to_node = manager.IndexToNode(to_index)return combined_graph[from_node][to_node]['time']transit_callback_index = routing.RegisterTransitCallback(time_callback)routing.SetArcCostEvaluatorOfAllVehicles(transit_callback_index)# 添加模式约束self._add_mode_constraints(routing, manager, modes)# 设置时间窗口self._add_time_windows(routing, manager, time_window)# 求解search_parameters = pywrapcp.DefaultRoutingSearchParameters()search_parameters.first_solution_strategy = (routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC)solution = routing.SolveWithParameters(search_parameters)if solution:return self._get_solution_path(routing, manager, solution)return Nonedef _add_mode_constraints(self, routing, manager, allowed_modes):"""添加交通方式约束"""pass  # 实现略def _add_time_windows(self, routing, manager, time_window):"""添加时间窗约束"""pass  # 实现略def _get_solution_path(self, routing, manager, solution):"""解析解决方案"""pass  # 实现略# 使用示例
if __name__ == "__main__":# 加载路网数据road_net = nx.read_gpickle("data/road_network.pkl")transit_net = nx.read_gpickle("data/transit_network.pkl")router = MultimodalRouter(road_net, transit_net)# 定义行程参数journey = router.plan_journey(origin="node_1234",destination="node_5678",modes=["walk", "bus", "subway"],time_window=(36000, 39600)  # 10:00-11:00)print("最优行程方案:")for step in journey:print(f"{step['time']}: {step['mode']} from {step['from']} to {step['to']}")

import cupy as cp
from shapely import STRtree
from rtree import indexclass HybridTopologyEngine:"""混合索引拓扑引擎"""def __init__(self, geometries):self.geometries = geometriesself.strtree = STRtree(geometries)self.rtree = index.Index()# 构建GPU加速结构self._build_gpu_index()def _build_gpu_index(self):"""构建GPU加速索引"""# 转换几何为GPU数组self.gpu_boxes = cp.array([geom.bounds for geom in self.geometries])# 预计算几何属性self.gpu_props = cp.array([(geom.area, geom.length) for geom in self.geometries])def query(self, geometry, predicate='intersects'):"""混合索引查询"""# CPU预筛选candidate_ids = list(self.rtree.intersection(geometry.bounds))# GPU精确计算results = self._gpu_compute(geometry.bounds,[self.geometries[i] for i in candidate_ids],predicate)return [self.geometries[i] for i in results]def _gpu_compute(self, bounds, candidates, predicate):"""GPU加速计算"""# 转换查询几何边界query_box = cp.array(bounds)# 计算空间关系overlaps = cp.zeros(len(candidates), dtype=bool)# 空间关系矩阵计算for i in range(len(candidates)):geom_box = cp.array(candidates[i].bounds)# 边界框相交测试if not (query_box[2] < geom_box[0] or query_box[0] > geom_box[2] orquery_box[3] < geom_box[1] orquery_box[1] > geom_box[3]):overlaps[i] = True# 精确计算候选集exact_matches = []for i in cp.where(overlaps)[0].get():if getattr(candidates[i], predicate)(geometry):exact_matches.append(i)return exact_matches# 使用示例
if __name__ == "__main__":from shapely.geometry import *# 创建测试数据polygons = [Polygon([(i,j),(i+1,j),(i+1,j+1),(i,j+1)])for i in range(10) for j in range(10)]engine = HybridTopologyEngine(polygons)# 执行查询query_geom = Point(5.5, 5.5).buffer(0.8)results = engine.query(query_geom, 'intersects')print(f"查询结果数量: {len(results)}")

import torch
import torch.nn as nn
from torch_geometric.nn import TemporalConv
from torch_geometric_temporal import recurrentclass SpatioTemporalPredictor(nn.Module):"""时空联合预测模型"""def __init__(self, input_dim, hidden_dim, output_dim, num_nodes):super().__init__()# 空间特征提取self.spatial_conv = nn.Sequential(nn.Conv2d(input_dim, hidden_dim, kernel_size=3, padding=1),nn.ReLU(),nn.MaxPool2d(2))# 时序特征提取self.temporal_conv = TemporalConv(hidden_dim, hidden_dim, kernel_size=3)# 图注意力机制self.attention = nn.MultiheadAttention(hidden_dim, num_heads=4)# 预测头self.predictor = nn.Sequential(nn.Linear(hidden_dim, hidden_dim//2),nn.ReLU(),nn.Linear(hidden_dim//2, output_dim))def forward(self, x, edge_index):"""前向传播"""batch_size, timesteps = x.shape[:2]# 空间特征提取spatial_features = []for t in range(timesteps):feat = self.spatial_conv(x[:, t])spatial_features.append(feat.flatten(start_dim=1))spatial_features = torch.stack(spatial_features, dim=1)# 时序特征提取temporal_features = self.temporal_conv(spatial_features)# 图注意力node_features = temporal_features.view(batch_size, timesteps, -1)attn_output, _ = self.attention(node_features, node_features, node_features)# 预测return self.predictor(attn_output[:, -1])# 模型服务封装
class PredictionService:"""预测模型服务"""def __init__(self, model_path):self.model = torch.jit.load(model_path)self.preprocessor = FeaturePreprocessor()async def predict(self, request):"""处理预测请求"""# 特征工程features = self.preprocessor.transform(request)# 执行预测with torch.no_grad():prediction = self.model(features)return {"prediction": prediction.numpy().tolist(),"timestamp": datetime.now().isoformat()}async def update(self, new_data):"""在线更新模型"""# 增量训练逻辑pass# FastAPI集成示例
if __name__ == "__main__":from fastapi import FastAPIimport uvicornapp = FastAPI()service = PredictionService("model.pt")@app.post("/predict")async def predict_endpoint(request: dict):return await service.predict(request)@app.post("/update")async def update_endpoint(data: dict):return await service.update(data)uvicorn.run(app, host="0.0.0.0", port=8000)