Spaces:
Running
Running
File size: 4,543 Bytes
a58f1b7 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 |
# modules/ai_link_flow_emulator.py
# TripAI – AI Emulator for Link Flows
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Dict, Tuple
import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestRegressor
from .route_assignment import aon_assignment, Network
@dataclass
class LinkFlowEmulator:
"""
AI emulator for link flows under scaled OD demand.
Attributes
----------
model : RandomForestRegressor
Multi-output regressor mapping demand scale -> link flows.
link_ids : np.ndarray
IDs of links in the same order as training outputs.
base_total_demand : float
Total baseline car OD (for reference).
"""
model: RandomForestRegressor
link_ids: np.ndarray
base_total_demand: float
def _generate_training_scenarios(
base_od: np.ndarray,
network: Network,
n_scenarios: int = 20,
low_scale: float = 0.7,
high_scale: float = 1.3,
) -> Tuple[np.ndarray, np.ndarray]:
"""
Generate training scenarios by scaling baseline OD and performing
AON assignment to obtain link flows.
Parameters
----------
base_od : np.ndarray
Baseline OD matrix (veh/h).
network : Network
n_scenarios : int
Number of random scaling scenarios.
low_scale : float
Minimum demand scale factor.
high_scale : float
Maximum demand scale factor.
Returns
-------
X : np.ndarray
Feature matrix of shape (n_scenarios, 1) – the demand scale.
Y : np.ndarray
Target matrix of shape (n_scenarios, n_links) – link flows.
"""
n_zones = base_od.shape[0]
n_links = len(network.links)
scales = np.random.uniform(low_scale, high_scale, size=n_scenarios)
X = scales.reshape(-1, 1)
Y = np.zeros((n_scenarios, n_links), dtype=float)
# Build index -> (from_zone, to_zone) map to reuse AON logic
# We will call the existing aon_assignment with scaled OD each time.
# Convert base OD to DataFrame with synthetic zone index 0..n-1.
zones = np.arange(n_zones)
base_od_df = pd.DataFrame(base_od, index=zones, columns=zones)
for idx, s in enumerate(scales):
od_scaled = base_od_df * s
flows_df = aon_assignment(od_scaled, network)
Y[idx, :] = flows_df["flow_vehph"].to_numpy()
return X, Y
def train_link_flow_emulator(
base_car_od: np.ndarray,
network: Network,
n_scenarios: int = 20,
) -> tuple[LinkFlowEmulator, pd.DataFrame]:
"""
Train a simple RandomForest-based emulator that maps a single
scalar 'demand scale' to resulting link flows.
Parameters
----------
base_car_od : np.ndarray
Baseline car OD matrix (veh/h).
network : Network
n_scenarios : int
Number of training scenarios.
Returns
-------
emulator : LinkFlowEmulator
training_history : pd.DataFrame
Scenario scales and corresponding total flows, for diagnostics.
"""
X, Y = _generate_training_scenarios(base_car_od, network, n_scenarios=n_scenarios)
model = RandomForestRegressor(
n_estimators=200,
max_depth=12,
random_state=42,
)
model.fit(X, Y)
link_ids = network.links.index.to_numpy()
base_total = float(base_car_od.sum())
# Build training history for inspection
total_flows = Y.sum(axis=1)
training_history = pd.DataFrame(
{
"scale": X.flatten(),
"total_link_flow": total_flows,
}
)
emulator = LinkFlowEmulator(
model=model,
link_ids=link_ids,
base_total_demand=base_total,
)
return emulator, training_history
def predict_link_flows(
emulator: LinkFlowEmulator,
scale: float,
network: Network,
) -> pd.DataFrame:
"""
Predict link flows for a new demand scale using the trained emulator.
Parameters
----------
emulator : LinkFlowEmulator
scale : float
Multiplicative scaling factor relative to baseline OD.
network : Network
Returns
-------
pd.DataFrame
Link table with predicted flows in column 'flow_vehph_emulated'.
"""
X_new = np.array([[scale]], dtype=float)
y_pred = emulator.model.predict(X_new).flatten()
links = network.links.copy()
links["flow_vehph_emulated"] = y_pred
return links
|