Visualizer for exiting matsim solutions. Improvement to matsim engine

README.md

@@ -18,3 +18,5 @@ python main.py path/to/config.xml
 ```
 
 Two scenarios are provided in this repository: equil, sioux. The output for that given simulation will be in the output folder.
+
+https://www.matsim.org/gallery/munich

hub_function_to_matsim_activity.py

@@ -2,7 +2,7 @@ import hub.helpers.constants as cte
 
 class HubFunctionToMatsimActivity:
   """
-  Hub function to nrcan construction function class
+  Hub function to matsim activity construction function class
   """
   def __init__(self):
     self._dictionary = {
@@ -28,43 +28,43 @@ class HubFunctionToMatsimActivity:
       cte.SECONDARY_SCHOOL: 'edu',
       cte.UNIVERSITY: 'edu',
       cte.LABORATORY_AND_RESEARCH_CENTER: 'edu',
-      cte.STAND_ALONE_RETAIL: 'secondary',
+      cte.STAND_ALONE_RETAIL: 'work,secondary',
       cte.HOSPITAL: 'work',
       cte.OUT_PATIENT_HEALTH_CARE: 'work',
       cte.HEALTH_CARE: 'work',
-      cte.RETIREMENT_HOME_OR_ORPHANAGE: 'home',
-      cte.COMMERCIAL: 'secondary',
-      cte.STRIP_MALL: 'secondary',
-      cte.SUPERMARKET: 'secondary',
-      cte.RETAIL_SHOP_WITHOUT_REFRIGERATED_FOOD: 'secondary',
-      cte.RETAIL_SHOP_WITH_REFRIGERATED_FOOD: 'secondary',
-      cte.RESTAURANT: 'secondary',
-      cte.QUICK_SERVICE_RESTAURANT: 'secondary',
-      cte.FULL_SERVICE_RESTAURANT: 'secondary',
-      cte.HOTEL: 'work',
-      cte.HOTEL_MEDIUM_CLASS: 'work',
-      cte.SMALL_HOTEL: 'work',
-      cte.LARGE_HOTEL: 'work',
+      cte.RETIREMENT_HOME_OR_ORPHANAGE: 'home,secondary',
+      cte.COMMERCIAL: 'work,secondary',
+      cte.STRIP_MALL: 'work,secondary',
+      cte.SUPERMARKET: 'work,secondary',
+      cte.RETAIL_SHOP_WITHOUT_REFRIGERATED_FOOD: 'work,secondary',
+      cte.RETAIL_SHOP_WITH_REFRIGERATED_FOOD: 'work,secondary',
+      cte.RESTAURANT: 'work,secondary',
+      cte.QUICK_SERVICE_RESTAURANT: 'work,secondary',
+      cte.FULL_SERVICE_RESTAURANT: 'work,secondary',
+      cte.HOTEL: 'work,secondary',
+      cte.HOTEL_MEDIUM_CLASS: 'work,secondary',
+      cte.SMALL_HOTEL: 'work,secondary',
+      cte.LARGE_HOTEL: 'work,secondary',
       cte.DORMITORY: 'work',
-      cte.EVENT_LOCATION: 'secondary',
-      cte.CONVENTION_CENTER: 'secondary',
-      cte.HALL: 'secondary',
-      cte.GREEN_HOUSE: 'secondary',
+      cte.EVENT_LOCATION: 'work,secondary',
+      cte.CONVENTION_CENTER: 'work,secondary',
+      cte.HALL: 'work',
+      cte.GREEN_HOUSE: 'work',
       cte.INDUSTRY: 'work',
       cte.WORKSHOP: 'work',
       cte.WAREHOUSE: 'work',
       cte.WAREHOUSE_REFRIGERATED: 'work',
-      cte.SPORTS_LOCATION: 'secondary',
-      cte.SPORTS_ARENA: 'secondary',
-      cte.GYMNASIUM: 'secondary',
-      cte.MOTION_PICTURE_THEATRE: 'secondary',
-      cte.MUSEUM: 'secondary',
-      cte.PERFORMING_ARTS_THEATRE: 'secondary',
-      cte.TRANSPORTATION: 'secondary',
+      cte.SPORTS_LOCATION: 'work,secondary',
+      cte.SPORTS_ARENA: 'work,secondary',
+      cte.GYMNASIUM: 'work,secondary',
+      cte.MOTION_PICTURE_THEATRE: 'work,secondary',
+      cte.MUSEUM: 'work,secondary',
+      cte.PERFORMING_ARTS_THEATRE: 'work,secondary',
+      cte.TRANSPORTATION: 'work',
       cte.AUTOMOTIVE_FACILITY: 'work',
       cte.PARKING_GARAGE: 'work',
-      cte.RELIGIOUS: 'secondary',
-      cte.NON_HEATED: 'secondary',
+      cte.RELIGIOUS: 'work,secondary',
+      cte.NON_HEATED: 'work',
       cte.DATACENTER: 'work',
       cte.FARM: 'work'
     }

main.py

@@ -6,6 +6,7 @@ from hub.imports.usage_factory import UsageFactory
 from hub.helpers.dictionaries import Dictionaries
 
 from matsim_engine import MatSimEngine
+from matsim_visualizer import MatSimVisualizer
 
 try:
   file_path = (Path(__file__).parent / 'input_files' / 'summerschool_all_buildings.geojson')
@@ -32,6 +33,9 @@ try:
 
   MatSimEngine(city, 'output_files')
 
+  # visualizer = MatSimVisualizer('output/output_network.xml.gz', 'output/output_events.xml.gz')
+  # visualizer.visualize()
+
 except Exception as ex:
   print('error: ', ex)
   print('[simulation abort]')

matsim_engine.py

@@ -30,11 +30,13 @@ class MatSimEngine:
     # 1- Facilities
     # TODO: Add the ability for a building to have multiple activities
     for building in city.buildings:
-      activity = hub_function_to_matsim.dictionary[building.function]
-      schedule = matsim_schedule.dictionary[activity]
+      activity = hub_function_to_matsim.dictionary[building.function].split(',')
+      schedule = matsim_schedule.dictionary[activity[0]]
       start_time, end_time = schedule.split('-')
 
       new_id = 0
+
+      # TODO: building.grounds (TBD)
       for surface in building.surfaces:
         for coord in surface.solid_polygon.coordinates:
           new_id += 1
@@ -45,8 +47,11 @@ class MatSimEngine:
         '@id': building.name,
         '@x': str(building.centroid[0]),
         '@y': str(building.centroid[1]),
-        'activity': {
-          '@type': activity,
+        'activity': []
+      }
+      for new_activity in activity:
+        facility['activity'].append({
+          '@type': new_activity,
           'capacity': {
             '@value': '4.0'  # TODO: Replace with a proper value taken from function
           },
@@ -55,8 +60,8 @@ class MatSimEngine:
             '@start_time': start_time,
             '@end_time': end_time
           }
-        }
-      }
+        })
+
       facilities_dict['facilities']['facility'].append(facility)
 
     gdf = gpd.GeoDataFrame(
@@ -81,6 +86,9 @@ class MatSimEngine:
     subprocess.run(command)
 
     # 3- Population
+
+    # 3.1 - Public Transport
+
     # 4- Config Generation
 
   def run(self):

matsim_visualizer.py

@@ -0,0 +1,113 @@
+import gzip
+
+import xmltodict
+import networkx as nx
+import matplotlib.pyplot as plt
+from matplotlib.animation import FuncAnimation
+from collections import defaultdict
+import matplotlib.cm as cm
+import matplotlib.colors as colors
+
+class MatSimVisualizer():
+  def __init__(self, network_file_path, events_file_path):
+    self.network_file_path = network_file_path
+    self.events_file_path = events_file_path
+    self.G = nx.Graph()
+    self.pos = None
+    self.traffic_per_tick = defaultdict(lambda: defaultdict(int))
+    self.cumulative_traffic = defaultdict(lambda: defaultdict(int))
+    self.cmap = cm.viridis
+    self.norm = None
+
+  def load_data(self):
+    # Load network data
+    with gzip.open(self.network_file_path, 'rb') as file:
+      network_doc = xmltodict.parse(file.read().decode('utf-8'))
+
+    # Parse nodes
+    self.nodes = {node['@id']: (float(node['@x']), float(node['@y'])) for node in
+                  network_doc['network']['nodes']['node']}
+
+    # Parse links
+    self.links = [{
+      'id': link['@id'],
+      'from': link['@from'],
+      'to': link['@to']
+    } for link in network_doc['network']['links']['link']]
+
+    link_state = defaultdict(list)
+
+    # Load and parse the events file
+    with gzip.open(self.events_file_path, 'rb') as file:
+      events_doc = xmltodict.parse(file.read().decode('utf-8'))
+
+    for event in events_doc['events']['event']:
+      link_id = event.get('@link')
+      event_type = event.get('@type')
+      tick = float(event.get('@time'))
+      vehicle_id = event.get('@vehicle')
+
+      if link_id is not None and event_type is not None and tick is not None:
+        if event_type == 'entered link' or event_type == 'vehicle enters traffic':
+          self.traffic_per_tick[tick][link_id] += 1
+          link_state[link_id].append(vehicle_id)
+        elif event_type == 'left link' or event_type == 'vehicle leaves traffic':
+          self.traffic_per_tick[tick][link_id] -= 1
+          link_state[link_id].remove(vehicle_id)
+
+    for link in self.links:
+      self.cumulative_traffic[0][link['id']] = 0
+
+    # Accumulate the counts to get the total number of vehicles on each link up to each tick
+    actual_tick = 0
+    sorted_ticks = sorted(self.traffic_per_tick.keys())
+    for tick in sorted_ticks:
+      if actual_tick not in self.cumulative_traffic:
+        # Start with the vehicle counts of the previous tick
+        self.cumulative_traffic[actual_tick] = defaultdict(int, self.cumulative_traffic.get(actual_tick - 1, {}))
+
+      # Apply the changes recorded for the current tick
+      for link_id, change in self.traffic_per_tick[tick].items():
+        self.cumulative_traffic[actual_tick][link_id] += change
+
+      actual_tick += 1  # Move to the next tick
+
+  def create_graph(self):
+    for node_id, coords in self.nodes.items():
+      self.G.add_node(node_id, pos=coords)
+    for link in self.links:
+      self.G.add_edge(link['from'], link['to'])
+    self.pos = nx.get_node_attributes(self.G, 'pos')
+
+  def setup_color_mapping(self):
+    # Find max traffic to setup the normalization instance
+    max_traffic = max(max(self.cumulative_traffic[tick].values()) for tick in self.cumulative_traffic)
+    self.norm = colors.Normalize(vmin=0, vmax=max_traffic)
+
+  def update(self, frame_number):
+    tick = sorted(self.cumulative_traffic.keys())[frame_number]
+    traffic_data = self.cumulative_traffic[tick]
+
+    edge_colors = [self.cmap(self.norm(traffic_data.get(link['id'], 0))) for link in self.links]
+    edge_widths = [2 + self.norm(traffic_data.get(link['id'], 0)) * 3 for link in self.links]
+
+    plt.cla()
+    nx.draw(self.G, self.pos, node_size=10, node_color='blue', width=edge_widths, edge_color=edge_colors, with_labels=False,
+            edge_cmap=self.cmap)
+
+    plt.title(f"Time: {tick}")
+
+  def visualize(self):
+    self.load_data()
+    self.create_graph()
+    self.setup_color_mapping()
+
+    fig, ax = plt.subplots()
+
+    sm = plt.cm.ScalarMappable(cmap=self.cmap, norm=self.norm)
+    sm.set_array([])
+    plt.colorbar(sm, ax=ax, label='Traffic Density')
+
+    ani = FuncAnimation(fig, self.update, frames=len(self.cumulative_traffic), repeat=False)
+    ani.save('traffic_animation.gif', writer='ffmpeg', fps=5)
+    plt.show()