Update map for alternatives incrementally

To avoid having to build the whole HDF5 map from scratch every time.

ruth/data/map.py

@@ -16,7 +16,7 @@
 from osmnx import load_graphml, save_graphml, graph_from_bbox
 from networkx.exception import NetworkXNoPath
 
-from .hdf5_writer import save_graph_to_hdf5
+from .hdf5_writer import get_osmid_from_data, save_graph_to_hdf5
 from ..log import console_logger as cl
 from ..metaclasses import Singleton
 from ..data.segment import Route, Segment, SegmentId, SpeedKph
@@ -330,6 +330,10 @@ def update_temporary_max_speeds(self, timestamp: datetime):
         nx.set_edge_attributes(self.current_network, values=new_current_speeds, name='current_speed')
         nx.set_edge_attributes(self.current_network, values=new_travel_times, name='current_travel_time')
 
+    def get_hdf5_edge_id(self, segment_id: SegmentId) -> int:
+        (node_from, node_to) = segment_id
+        return get_osmid_from_data(self.current_network[node_from][node_to])
+
 
 def admin_level_to_road_filter(admin_level):  # TODO: where to put it?
     """Create a road filter based on administrative level."""

ruth/simulator/kernels.py

@@ -1,10 +1,10 @@
 import logging
 import random
-from typing import List, Optional, Tuple
+from typing import Dict, List, Optional, Tuple
 
 from .ptdr import PTDRInfo
 from ..data.map import Map, osm_routes_to_segment_ids
-from ..data.segment import Route
+from ..data.segment import Route, SegmentId, SpeedMps
 from ..utils import is_root_debug_logging
 from ..vehicle import Vehicle
 from ..zeromq.src.client import Message
@@ -24,6 +24,12 @@ def load_map(self, routing_map: Map):
         """
         self.routing_map = routing_map
 
+    def update_map(self, map: Map, segments: Dict[SegmentId, Optional[SpeedMps]]):
+        """
+        Update speeds of the passed segments in a previously passed map.
+        """
+        pass
+
     def compute_alternatives(self, vehicles: List[Vehicle], k: int) -> List[
         Optional[AlternativeRoutes]]:
         """
@@ -60,6 +66,16 @@ def load_map(self, routing_map: Map):
         self.client.broadcast(Message(kind="load-map", data=map_path))
         self.routing_map = routing_map
 
+    def update_map(self, map: Map, segments: Dict[SegmentId, Optional[SpeedMps]]):
+        """
+        Update speeds of the passed segments in a previously passed map.
+        """
+        self.client.broadcast(Message(kind="update-map", data=[{
+            "edge_id": map.get_hdf5_edge_id(segment_id),
+            "speed": speed if speed is not None else map.get_current_max_speed(segment_id[0],
+                                                                               segment_id[1])
+        } for (segment_id, speed) in segments.items()]))
+
     def compute_alternatives(self, vehicles: List[Vehicle], k: int) -> List[
         Optional[AlternativeRoutes]]:
         messages = [Message(kind="alternatives", data={
@@ -77,7 +93,8 @@ def compute_alternatives(self, vehicles: List[Vehicle], k: int) -> List[
         if is_root_debug_logging():
             logging.debug(f"Response from worker: {results}")
         remapped_routes = [
-            [[self.routing_map.hdf5_to_osm_id(node_id) for node_id in route] for route in result["routes"]]
+            [[self.routing_map.hdf5_to_osm_id(node_id) for node_id in route] for route in
+             result["routes"]]
             for result in results
         ]
         return remapped_routes

ruth/simulator/singlenode.py

@@ -65,7 +65,7 @@ def simulate(
                 if self.current_offset - last_map_update >= self.sim.setting.map_update_freq_s:
                     updated_speeds = self.sim.global_view.take_segment_speeds()
                     self.sim.routing_map.update_current_speeds(updated_speeds)
-                    alternatives_provider.load_map(self.sim.routing_map)
+                    alternatives_provider.update_map(self.sim.routing_map, updated_speeds)
 
                     last_map_update = self.current_offset