add Route object based route specification

Author: toruseo

tests/test_other_functions.py

@@ -1114,7 +1114,7 @@ def test_route_vehicle_methods():
     assert equal_tolerance(tt_from_vehicle_route, tt_from_route_by_departure_time)
 
 
-def test_route_enforce_route():
+def test_route_enforce_route_old():
     W = World(
         name="",
         deltan=5,
@@ -1154,6 +1154,78 @@ def test_route_enforce_route():
     for l in r2:
         assert df[df["link"]==l.name]["traffic_volume"].values[0] == 2130
 
+def test_route_enforce_route_by_route_object():
+
+    W = World(
+        name="looproute",    # Scenario name
+        deltan=5,   # Simulation aggregation unit delta n
+        tmax=2400,  # Total simulation time (s)
+        print_mode=1, save_mode=0, show_mode=1,    # Various options
+        random_seed=0    # Set the random seed
+    )
+
+    # Define the scenario
+    ## Create nodes
+    W.addNode(name="O", x=0, y=0)
+    W.addNode("D", 2, 0)
+    W.addNode("A", 1, 0)
+    W.addNode("B", 1.5, 1)
+    W.addNode("C", 0.5, 1)
+    W.addNode("E", 1, -1)
+    ## Create links between nodes
+    W.addLink(None, "O", "A", length=500, free_flow_speed=20, number_of_lanes=1)
+    W.addLink(None, "A", "D", length=500, free_flow_speed=20, number_of_lanes=1)
+    W.addLink(None, "A", "B", length=500, free_flow_speed=20, number_of_lanes=1)
+    W.addLink(None, "B", "C", length=500, free_flow_speed=20, number_of_lanes=1)
+    W.addLink(None, "C", "A", length=500, free_flow_speed=20, number_of_lanes=1)
+    W.addLink(None, "O", "E", length=1000, free_flow_speed=20, number_of_lanes=1)
+    W.addLink(None, "E", "D", length=1000, free_flow_speed=20, number_of_lanes=1)
+    W.addLink(None, "O", "D", length=500, free_flow_speed=40, number_of_lanes=1)
+
+    ## Create OD traffic demand between nodes
+    W.adddemand("O","D", t_start=0, t_end=1000, volume=400)
+
+    #W.show_network()
+
+    route_straight = W.defRoute(["O-A", "A-D"])
+    route_detour_loop = W.defRoute(["O-A", "A-B", "B-C", "C-A", "A-D"])
+    route_detour_loop_triple = W.defRoute(["O-A", "A-B", "B-C", "C-A", "A-B", "B-C", "C-A", "A-B", "B-C", "C-A", "A-D"])
+    route_detour = W.defRoute(["O-E", "E-D"])
+
+    for i,veh in enumerate(W.VEHICLES.values()):
+        if i%4 == 0:
+            veh.enforce_route(route_straight)
+        elif i%4 == 1:
+            veh.enforce_route(route_detour)
+        elif i%4 == 2:
+            veh.enforce_route(route_detour_loop)
+        else:
+            veh.enforce_route(route_detour_loop_triple)
+
+    # Run the simulation to the end
+    W.exec_simulation()
+
+    # Print summary of simulation result
+    W.analyzer.print_simple_stats()
+
+    W.analyzer.network_average()
+    df = W.analyzer.link_to_pandas()
+    print(df)
+    
+    assert df.query("link == 'O-A'")["traffic_volume"].item() == 300
+    assert df.query("link == 'A-D'")["traffic_volume"].item() == 300
+    assert df.query("link == 'A-B'")["traffic_volume"].item() == 400
+    assert df.query("link == 'B-C'")["traffic_volume"].item() == 400
+    assert df.query("link == 'C-A'")["traffic_volume"].item() == 400
+    assert df.query("link == 'O-E'")["traffic_volume"].item() == 100
+    assert df.query("link == 'E-D'")["traffic_volume"].item() == 100
+    assert df.query("link == 'O-D'")["traffic_volume"].item() == 0
+
+    assert W.VEHICLES["0"].traveled_route()[0] == route_straight
+    assert W.VEHICLES["1"].traveled_route()[0] == route_detour
+    assert W.VEHICLES["2"].traveled_route()[0] == route_detour_loop
+    assert W.VEHICLES["3"].traveled_route()[0] == route_detour_loop_triple
+
 def test_construct_time_space_network():
     W = World(
         name="",

uxsim/uxsim.py

@@ -370,8 +370,8 @@ def __init__(s, W, name, start_node, end_node, length, free_flow_speed=20, jam_d
         ----------
         W : object
             The world to which the link belongs.
-        name : str
-            The name of the link.
+        name : str | None
+            The name of the link. If None, the name is automatically generated as "{start_node.name}-{end_node.name}".
         start_node : str | Node
             The name of the start node of the link.
         end_node : str | Node
@@ -572,6 +572,8 @@ def __init__(s, W, name, start_node, end_node, length, free_flow_speed=20, jam_d
 
         s.id = len(s.W.LINKS)
         s.name = name
+        if s.name == None:
+            s.name = s.start_node.name+"-"+s.end_node.name
         s.auto_rename = auto_rename
         if s.name in s.W.LINKS_NAME_DICT.keys():
             if auto_rename:
@@ -959,10 +961,12 @@ def __init__(s, W, orig, dest, departure_time, name=None, route_pref=None, route
         else:
             s.route_pref = {l.id:route_pref[l] for l in route_pref.keys()}
 
-        #these links will be always chosen or not chosen when choosing next link at each node
+        #these links will be always chosen or not chosen when choosing next link at each node. This is not used anymore. keeping for backward compatibility
         s.links_prefer = [s.W.get_link(l) for l in links_prefer]
         s.links_avoid = [s.W.get_link(l) for l in links_avoid]
 
+        s.specified_route = None
+
         #行き止まりに行ってしまったときにトリップを止める
         s.trip_abort = trip_abort
         s.flag_trip_aborted = 0
@@ -1135,6 +1139,11 @@ def enforce_route(s, route, set_avoid=False):
         set_avoid : bool
             If True, the vehicle only travel the links in `route` argument. This is very strict. If the route is not consistent (e.g., the route does not completely connect between the origin and the destination), it will raise an exception.
         """
+        route = [s.W.get_link(l) for l in route]
+
+        s.specified_route = route
+
+        #for backward compatibility
         s.links_prefer = [s.W.get_link(l) for l in route]
         if set_avoid:
             s.links_avoid = [s.W.get_link(l) for l in s.W.LINKS if l not in s.links_prefer]
@@ -1182,29 +1191,42 @@ def route_next_link_choice(s):
         """
         Select a next link from the current link.
         """
-        if s.dest != s.link.end_node:
-            outlinks = list(s.link.end_node.outlinks.values())
+        #print("aaa", end="")
+        if s.specified_route == None:
+            if s.dest != s.link.end_node:
+                outlinks = list(s.link.end_node.outlinks.values())
 
-            if len(outlinks):
+                if len(outlinks):
 
-                #if links_prefer is given and available at the node, select only from the links in the list. if links_avoid is given, select links not in the list.
-                if set(outlinks) & set(s.links_prefer):
-                    outlinks = sorted(set(outlinks) & set(s.links_prefer), key=lambda l:l.name)
-                if set(outlinks) & set(s.links_avoid):
-                    outlinks = sorted(set(outlinks) - set(s.links_avoid), key=lambda l:l.name)
+                    #if links_prefer is given and available at the node, select only from the links in the list. if links_avoid is given, select links not in the list.
+                    if set(outlinks) & set(s.links_prefer):
+                        outlinks = sorted(set(outlinks) & set(s.links_prefer), key=lambda l:l.name)
+                    if set(outlinks) & set(s.links_avoid):
+                        outlinks = sorted(set(outlinks) - set(s.links_avoid), key=lambda l:l.name)
 
-                preference = np.array([s.route_pref[l.id] for l in outlinks], dtype=float)
-                if s.W.hard_deterministic_mode == False:
-                    if sum(preference) > 0:
-                        s.route_next_link = s.W.rng.choice(outlinks, p=preference/sum(preference))
+                    preference = np.array([s.route_pref[l.id] for l in outlinks], dtype=float)
+                    if s.W.hard_deterministic_mode == False:
+                        if sum(preference) > 0:
+                            s.route_next_link = s.W.rng.choice(outlinks, p=preference/sum(preference))
+                        else:
+                            s.route_next_link = s.W.rng.choice(outlinks)
                     else:
-                        s.route_next_link = s.W.rng.choice(outlinks)
-                else:
-                    s.route_next_link = max(zip(preference, outlinks), key=lambda x:x[0])[1]
+                        s.route_next_link = max(zip(preference, outlinks), key=lambda x:x[0])[1]
 
+                else:
+                    s.route_next_link = None
+        else:
+            outlinks = list(s.link.end_node.outlinks.values())
+            if len(outlinks):
+                number_of_traveled_links = len(s.log_t_link)-1
+                if s.specified_route[number_of_traveled_links] in outlinks:
+                    s.route_next_link = s.specified_route[number_of_traveled_links]
+                else:
+                    raise ValueError(f"Vehicle {s.name}: specified route {s.specified_route} is inconsistent at the current link {s.link}. Debug info: {number_of_traveled_links=}, {outlinks=}")
             else:
                 s.route_next_link = None
 
+
     def add_dest(s, dest, order=-1):
         """
         Add a destination to the vehicle's destination list.
@@ -2700,6 +2722,9 @@ def __iter__(s):
     def __len__(s):
         return len(s.links)
 
+    def __getitem__(s, index):
+        return s.links[index]
+    
     def __eq__(self, other):
         """
         Override `==` operator. If the links of two route are the same, then the routes are the same.