Skip to content

Commit 7ed1950

Browse files
sebhoerltkchouakitkchouaki
authored
feat: sparse horizon handler for vdf (#276)
* feat: sparse horizon handler for vdf * small fix * feat: vdf sparse horizon handler - file writing and reading (#278) * feat: Using VDFSparseHorizonHandler by default * fix: file writing and reading * feat: towards testing VDFSparseHorizonHandler VS VDFHorizonHandler * chore: cleanup * feat: comparing output events of VDFHorizonHandler and VDFSparseHorizonHandler --------- Co-authored-by: tkchouaki <tarek.chouaki@irt-systemx.fr> * chore: comparing plans instead of events --------- Co-authored-by: Tarek Chouaki <tkchouaki@gmail.com> Co-authored-by: tkchouaki <tarek.chouaki@irt-systemx.fr>
1 parent 678fbb0 commit 7ed1950

5 files changed

Lines changed: 323 additions & 6 deletions

File tree

‎core/src/main/java/org/eqasim/core/simulation/vdf/VDFConfigGroup.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@ public class VDFConfigGroup extends ReflectiveConfigGroup {
4949
private int writeFlowInterval = 0;
5050

5151
public enum HandlerType {
52-
Horizon, Interpolation
52+
Horizon, Interpolation, SparseHorizon
5353
}
5454

5555
private HandlerType handler = HandlerType.Horizon;

‎core/src/main/java/org/eqasim/core/simulation/vdf/VDFModule.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
import org.eqasim.core.simulation.mode_choice.AbstractEqasimExtension;
1212
import org.eqasim.core.simulation.vdf.handlers.VDFHorizonHandler;
1313
import org.eqasim.core.simulation.vdf.handlers.VDFInterpolationHandler;
14+
import org.eqasim.core.simulation.vdf.handlers.VDFSparseHorizonHandler;
1415
import org.eqasim.core.simulation.vdf.handlers.VDFTrafficHandler;
1516
import org.eqasim.core.simulation.vdf.travel_time.VDFTravelTime;
1617
import org.eqasim.core.simulation.vdf.travel_time.function.BPRFunction;
@@ -41,6 +42,10 @@ protected void installEqasimExtension() {
4142
bind(VDFTrafficHandler.class).to(VDFHorizonHandler.class);
4243
addEventHandlerBinding().to(VDFHorizonHandler.class);
4344
break;
45+
case SparseHorizon:
46+
bind(VDFTrafficHandler.class).to(VDFSparseHorizonHandler.class);
47+
addEventHandlerBinding().to(VDFSparseHorizonHandler.class);
48+
break;
4449
case Interpolation:
4550
bind(VDFTrafficHandler.class).to(VDFInterpolationHandler.class);
4651
addEventHandlerBinding().to(VDFInterpolationHandler.class);
@@ -85,6 +90,12 @@ public VDFHorizonHandler provideVDFHorizonHandler(VDFConfigGroup config, Network
8590
return new VDFHorizonHandler(network, scope, config.getHorizon(), getConfig().global().getNumberOfThreads());
8691
}
8792

93+
@Provides
94+
@Singleton
95+
public VDFSparseHorizonHandler provideVDFSparseHorizonHandler(VDFConfigGroup config, Network network, VDFScope scope) {
96+
return new VDFSparseHorizonHandler(network, scope, config.getHorizon(), getConfig().global().getNumberOfThreads());
97+
}
98+
8899
@Provides
89100
@Singleton
90101
public VDFInterpolationHandler provideVDFInterpolationHandler(VDFConfigGroup config, Network network,

‎core/src/main/java/org/eqasim/core/simulation/vdf/handlers/VDFSparseHorizonHandler.java‎

Lines changed: 292 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,292 @@
1+
package org.eqasim.core.simulation.vdf.handlers;
2+
3+
import java.io.DataInputStream;
4+
import java.io.DataOutputStream;
5+
import java.io.File;
6+
import java.io.FileOutputStream;
7+
import java.io.IOException;
8+
import java.net.URL;
9+
import java.util.ArrayList;
10+
import java.util.Collections;
11+
import java.util.Iterator;
12+
import java.util.LinkedList;
13+
import java.util.List;
14+
import java.util.Map;
15+
16+
import org.apache.logging.log4j.LogManager;
17+
import org.apache.logging.log4j.Logger;
18+
import org.eqasim.core.simulation.vdf.VDFScope;
19+
import org.eqasim.core.simulation.vdf.io.VDFReaderInterface;
20+
import org.eqasim.core.simulation.vdf.io.VDFWriterInterface;
21+
import org.matsim.api.core.v01.Id;
22+
import org.matsim.api.core.v01.IdMap;
23+
import org.matsim.api.core.v01.events.LinkEnterEvent;
24+
import org.matsim.api.core.v01.events.handler.LinkEnterEventHandler;
25+
import org.matsim.api.core.v01.network.Link;
26+
import org.matsim.api.core.v01.network.Network;
27+
import org.matsim.core.utils.io.IOUtils;
28+
29+
import com.google.common.base.Verify;
30+
31+
public class VDFSparseHorizonHandler implements VDFTrafficHandler, LinkEnterEventHandler {
32+
private final VDFScope scope;
33+
34+
private final Network network;
35+
private final int horizon;
36+
private final int numberOfThreads;
37+
38+
private final IdMap<Link, List<Double>> counts = new IdMap<>(Link.class);
39+
40+
private final static Logger logger = LogManager.getLogger(VDFSparseHorizonHandler.class);
41+
42+
private record LinkState(List<Integer> time, List<Double> count) {
43+
}
44+
45+
private List<IdMap<Link, LinkState>> state = new LinkedList<>();
46+
47+
public VDFSparseHorizonHandler(Network network, VDFScope scope, int horizon, int numberOfThreads) {
48+
this.scope = scope;
49+
this.network = network;
50+
this.horizon = horizon;
51+
this.numberOfThreads = numberOfThreads;
52+
53+
for (Id<Link> linkId : network.getLinks().keySet()) {
54+
counts.put(linkId, new ArrayList<>(Collections.nCopies(scope.getIntervals(), 0.0)));
55+
}
56+
}
57+
58+
@Override
59+
public synchronized void handleEvent(LinkEnterEvent event) {
60+
processEnterLink(event.getTime(), event.getLinkId());
61+
}
62+
63+
public void processEnterLink(double time, Id<Link> linkId) {
64+
int i = scope.getIntervalIndex(time);
65+
double currentValue = counts.get(linkId).get(i);
66+
counts.get(linkId).set(i, currentValue + 1);
67+
}
68+
69+
@Override
70+
public IdMap<Link, List<Double>> aggregate(boolean ignoreIteration) {
71+
while (state.size() > horizon) {
72+
state.remove(0);
73+
}
74+
75+
logger.info(String.format("Starting aggregation of %d slices", state.size()));
76+
77+
// Transform counts into state object
78+
if (!ignoreIteration) {
79+
IdMap<Link, LinkState> newState = new IdMap<>(Link.class);
80+
state.add(newState);
81+
82+
for (Map.Entry<Id<Link>, List<Double>> entry : counts.entrySet()) {
83+
double total = 0.0;
84+
85+
for (double value : entry.getValue()) {
86+
total += value;
87+
}
88+
89+
if (total > 0.0) {
90+
LinkState linkState = new LinkState(new ArrayList<>(), new ArrayList<>());
91+
newState.put(entry.getKey(), linkState);
92+
93+
int timeIndex = 0;
94+
for (double count : entry.getValue()) {
95+
if (count > 0.0) {
96+
linkState.time.add(timeIndex);
97+
linkState.count.add(count);
98+
}
99+
100+
timeIndex++;
101+
}
102+
}
103+
}
104+
}
105+
106+
IdMap<Link, List<Double>> aggregated = new IdMap<>(Link.class);
107+
108+
for (Id<Link> linkId : network.getLinks().keySet()) {
109+
// Reset current counts
110+
counts.put(linkId, new ArrayList<>(Collections.nCopies(scope.getIntervals(), 0.0)));
111+
112+
// Initialize aggregated counts
113+
aggregated.put(linkId, new ArrayList<>(Collections.nCopies(scope.getIntervals(), 0.0)));
114+
}
115+
116+
// Aggregate
117+
Iterator<Id<Link>> linkIterator = network.getLinks().keySet().iterator();
118+
119+
Runnable worker = () -> {
120+
Id<Link> currentLinkId = null;
121+
122+
while (true) {
123+
// Fetch new link in queue
124+
synchronized (linkIterator) {
125+
if (linkIterator.hasNext()) {
126+
currentLinkId = linkIterator.next();
127+
} else {
128+
break; // Done
129+
}
130+
}
131+
132+
// Go through history for this link and aggregate by time slot
133+
for (int k = 0; k < state.size(); k++) {
134+
LinkState historyItem = state.get(k).get(currentLinkId);
135+
List<Double> linkAggregator = aggregated.get(currentLinkId);
136+
137+
if (historyItem != null) {
138+
for (int i = 0; i < historyItem.count.size(); i++) {
139+
int timeIndex = historyItem.time.get(i);
140+
linkAggregator.set(timeIndex,
141+
linkAggregator.get(timeIndex) + historyItem.count.get(i) / (double) state.size());
142+
}
143+
}
144+
}
145+
}
146+
};
147+
148+
if (numberOfThreads < 2) {
149+
worker.run();
150+
} else {
151+
List<Thread> threads = new ArrayList<>(numberOfThreads);
152+
153+
for (int k = 0; k < numberOfThreads; k++) {
154+
threads.add(new Thread(worker));
155+
}
156+
157+
for (int k = 0; k < numberOfThreads; k++) {
158+
threads.get(k).start();
159+
}
160+
161+
try {
162+
for (int k = 0; k < numberOfThreads; k++) {
163+
threads.get(k).join();
164+
}
165+
} catch (InterruptedException e) {
166+
throw new RuntimeException(e);
167+
}
168+
}
169+
170+
logger.info(String.format(" Finished aggregation"));
171+
172+
return aggregated;
173+
}
174+
175+
@Override
176+
public VDFReaderInterface getReader() {
177+
return new Reader();
178+
}
179+
180+
@Override
181+
public VDFWriterInterface getWriter() {
182+
return new Writer();
183+
}
184+
185+
public class Reader implements VDFReaderInterface {
186+
@Override
187+
public void readFile(URL inputFile) {
188+
state.clear();
189+
190+
try {
191+
DataInputStream inputStream = new DataInputStream(IOUtils.getInputStream(inputFile));
192+
193+
Verify.verify(inputStream.readDouble() == scope.getStartTime());
194+
Verify.verify(inputStream.readDouble() == scope.getEndTime());
195+
Verify.verify(inputStream.readDouble() == scope.getIntervalTime());
196+
Verify.verify(inputStream.readInt() == scope.getIntervals());
197+
Verify.verify(inputStream.readInt() == horizon);
198+
199+
int slices = (int) inputStream.readInt();
200+
int links = (int) inputStream.readInt();
201+
202+
List<Id<Link>> linkIds = new ArrayList<>(links);
203+
for (int linkIndex = 0; linkIndex < links; linkIndex++) {
204+
linkIds.add(Id.createLinkId(inputStream.readUTF()));
205+
}
206+
207+
logger.info(String.format("Loading %d slices with %d links", slices, links));
208+
209+
for (int sliceIndex = 0; sliceIndex < slices; sliceIndex++) {
210+
IdMap<Link, LinkState> slice = new IdMap<>(Link.class);
211+
state.add(slice);
212+
213+
int sliceLinkCount = inputStream.readInt();
214+
215+
logger.info(String.format("Slice %d/%d, Reading %d link states", sliceIndex+1, slices, sliceLinkCount));
216+
217+
for (int sliceLinkIndex = 0; sliceLinkIndex < sliceLinkCount; sliceLinkIndex++) {
218+
int linkIndex = inputStream.readInt();
219+
int linkStateSize = inputStream.readInt();
220+
221+
LinkState linkState = new LinkState(new ArrayList<>(linkStateSize),
222+
new ArrayList<>(linkStateSize));
223+
slice.put(linkIds.get(linkIndex), linkState);
224+
225+
for (int i = 0; i < linkStateSize; i++) {
226+
linkState.time.add(inputStream.readInt());
227+
linkState.count.add(inputStream.readDouble());
228+
}
229+
}
230+
231+
logger.info(String.format(" Slice %d: %d obs", sliceIndex,
232+
sliceLinkCount));
233+
}
234+
235+
Verify.verify(inputStream.available() == 0);
236+
inputStream.close();
237+
} catch (IOException e) {
238+
throw new RuntimeException(e);
239+
}
240+
}
241+
}
242+
243+
public class Writer implements VDFWriterInterface {
244+
@Override
245+
public void writeFile(File outputFile) {
246+
try {
247+
DataOutputStream outputStream = new DataOutputStream(new FileOutputStream(outputFile.toString()));
248+
249+
outputStream.writeDouble(scope.getStartTime());
250+
outputStream.writeDouble(scope.getEndTime());
251+
outputStream.writeDouble(scope.getIntervalTime());
252+
outputStream.writeInt(scope.getIntervals());
253+
outputStream.writeInt(horizon);
254+
outputStream.writeInt(state.size());
255+
outputStream.writeInt(counts.size());
256+
257+
List<Id<Link>> linkIds = new ArrayList<>(counts.keySet());
258+
for (int linkIndex = 0; linkIndex < linkIds.size(); linkIndex++) {
259+
outputStream.writeUTF(linkIds.get(linkIndex).toString());
260+
}
261+
262+
logger.info(String.format("About to write %d slices", state.size()));
263+
264+
for (int sliceIndex = 0; sliceIndex < state.size(); sliceIndex++) {
265+
IdMap<Link, LinkState> slice = state.get(sliceIndex);
266+
outputStream.writeInt(slice.size());
267+
268+
int sliceLinkIndex = 0;
269+
for (Id<Link> linkId : linkIds) {
270+
LinkState linkState = slice.get(linkId);
271+
if(linkState == null) {
272+
continue;
273+
}
274+
outputStream.writeInt(linkIds.indexOf(linkId));
275+
outputStream.writeInt(linkState.count.size());
276+
277+
for (int i = 0; i < linkState.count.size(); i++) {
278+
outputStream.writeInt(linkState.time.get(i));
279+
outputStream.writeDouble(linkState.count.get(i));
280+
}
281+
sliceLinkIndex += 1;
282+
}
283+
assert sliceLinkIndex == slice.size();
284+
}
285+
286+
outputStream.close();
287+
} catch (IOException e) {
288+
throw new RuntimeException(e);
289+
}
290+
}
291+
}
292+
}

‎core/src/main/java/org/eqasim/core/simulation/vdf/utils/AdaptConfigForVDF.java‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
package org.eqasim.core.simulation.vdf.utils;
22

3-
import org.eqasim.core.components.config.EqasimConfigGroup;
43
import org.eqasim.core.simulation.EqasimConfigurator;
54
import org.eqasim.core.simulation.vdf.VDFConfigGroup;
65
import org.eqasim.core.simulation.vdf.engine.VDFEngineConfigGroup;
@@ -25,9 +24,8 @@ public static void adaptConfigForVDF(Config config, boolean engine) {
2524
VDFConfigGroup.getOrCreate(config).setWriteInterval(1);
2625
VDFConfigGroup.getOrCreate(config).setWriteFlowInterval(1);
2726

28-
// VDF: Set capacity factor instead (We retrieve it form the Eqasim config group)
29-
EqasimConfigGroup eqasimConfigGroup = (EqasimConfigGroup) config.getModules().get(EqasimConfigGroup.GROUP_NAME);
30-
VDFConfigGroup.getOrCreate(config).setCapacityFactor(eqasimConfigGroup.getSampleSize());
27+
VDFConfigGroup vdfConfigGroup = VDFConfigGroup.getOrCreate(config);
28+
vdfConfigGroup.setHandler(VDFConfigGroup.HandlerType.SparseHorizon);
3129

3230
if(engine) {
3331
// VDF Engine: Add config group

‎core/src/test/java/org/eqasim/TestSimulationPipeline.java‎

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,8 @@
5555

5656
import com.google.inject.Inject;
5757
import com.google.inject.Provider;
58+
import org.matsim.utils.eventsfilecomparison.ComparisonResult;
59+
import org.matsim.utils.eventsfilecomparison.EventsFileComparator;
5860

5961
public class TestSimulationPipeline {
6062

@@ -377,6 +379,7 @@ public void testTransitWithAbstractAccess() throws CommandLine.ConfigurationExce
377379
}
378380

379381
public void runVdf() throws CommandLine.ConfigurationException, IOException, InterruptedException {
382+
// This one will use the SparseHorizon handler
380383
AdaptConfigForVDF.main(new String[] {
381384
"--input-config-path", "melun_test/input/config.xml",
382385
"--output-config-path", "melun_test/input/config_vdf.xml",
@@ -387,6 +390,20 @@ public void runVdf() throws CommandLine.ConfigurationException, IOException, Int
387390

388391
runMelunSimulation("melun_test/input/config_vdf.xml", "melun_test/output_vdf");
389392

393+
394+
// We force this one to use the legacy horizon handler
395+
AdaptConfigForVDF.main(new String[] {
396+
"--input-config-path", "melun_test/input/config.xml",
397+
"--output-config-path", "melun_test/input/config_vdf_horizon.xml",
398+
"--engine", "true",
399+
"--config:eqasim:vdf_engine.generateNetworkEvents", "true",
400+
"--config:eqasim:vdf.handler", "Horizon"
401+
});
402+
403+
runMelunSimulation("melun_test/input/config_vdf_horizon.xml", "melun_test/output_vdf_horizon");
404+
405+
assert CRCChecksum.getCRCFromFile("melun_test/output_vdf_horizon/output_plans.xml.gz") == CRCChecksum.getCRCFromFile("melun_test/output_vdf/output_plans.xml.gz");
406+
390407
RunStandaloneModeChoice.main(new String[]{
391408
"--config-path", "melun_test/input/config_vdf.xml",
392409
"--config:standaloneModeChoice.outputDirectory", "melun_test/output_mode_choice_vdf",
@@ -425,7 +442,6 @@ public void testPipeline() throws Exception {
425442

426443
@Test
427444
public void testBaseDeterminism() throws Exception {
428-
Logger logger = LogManager.getLogger(TestSimulationPipeline.class);
429445
Config config = ConfigUtils.loadConfig("melun_test/input/config.xml");
430446
runMelunSimulation(config, "melun_test/output_determinism_1", null, 2);
431447

0 commit comments

Comments
 (0)
⚡