Select link correctness fix (#447)

* Revert "Disable select link (#443)" This reverts commit 0cd3d48. * Add Kai Tang's test and data * Potential select link fix * Test formatting * Fix tests imports * Add select link test This test asserts that the results of the select link on the links 7, and 13 are the same as the results of the assignment. These links were chosen for this particular network to cover all paths used. * Prevent data races in select link results Memory for the multi-threaded runs are now allocated in MuliThreadedAoN along side the rest of the multi-threaded memory. * Installs package to run documentation pipeline * installing * Install pandoc --------- Co-authored-by: Pedro Camargo <[email protected]>
AequilibraE · Sep 29, 2023 · 477f397 · 477f397
1 parent 039fee7
commit 477f397
Show file tree

Hide file tree

Showing 10 changed files with 129 additions and 74 deletions.
diff --git a/.github/workflows/documentation.yml b/.github/workflows/documentation.yml
@@ -29,12 +29,13 @@ jobs:
         pip install -r docs/requirements-docs.txt
         python -m pip install sphinx-gallery --user
         sudo apt update
-        sudo apt install -y --fix-missing libsqlite3-mod-spatialite libspatialite-dev
+        sudo apt install -y --fix-missing libsqlite3-mod-spatialite libspatialite-dev pandoc
         sudo ln -s /usr/lib/x86_64-linux-gnu/mod_spatialite.so /usr/lib/x86_64-linux-gnu/mod_spatialite
 
     - name: Compile library
       run: |
         python setup.py build_ext --inplace
+        pip install .
 
     - name: Check history of versions
       run: |

diff --git a/aequilibrae/paths/AoN.pyx b/aequilibrae/paths/AoN.pyx
@@ -97,11 +97,10 @@ def one_to_all(origin, matrix, graph, result, aux_result, curr_thread):
         bint select_link = False
 
     if result._selected_links:
-
         has_flow_mask = aux_result.has_flow_mask[curr_thread, :]
-        sl_od_matrix_view = aux_result.temp_sl_od_matrix[:, origin_index, :, :]
-        sl_link_loading_view = aux_result.temp_sl_link_loading[:, :, :]
-        link_list = aux_result.select_links[:, :]
+        sl_od_matrix_view = aux_result.temp_sl_od_matrix[curr_thread, :, origin_index, :, :]
+        sl_link_loading_view = aux_result.temp_sl_link_loading[curr_thread, :, :, :]
+        link_list = aux_result.select_links[:, :]  # Read only, don't need to slice on curr_thread
         select_link = True
     #Now we do all procedures with NO GIL
     with nogil:

diff --git a/aequilibrae/paths/linear_approximation.py b/aequilibrae/paths/linear_approximation.py
@@ -412,12 +412,14 @@ def execute(self):
                             # The temp has an index associated with the link_set name
                             copy_three_dimensions(
                                 c.results.select_link_od.matrix[name],  # matrix being written into
-                                c._aon_results.temp_sl_od_matrix[idx, :, :, :],  # results after the iteration
+                                np.sum(aon.aux_res.temp_sl_od_matrix, axis=0)[
+                                    idx, :, :, :
+                                ],  # results after the iteration
                                 self.cores,  # core count
                             )
                             copy_two_dimensions(
                                 c.results.select_link_loading[name],  # ouput matrix
-                                c._aon_results.temp_sl_link_loading[idx, :, :],  # matrix 1
+                                np.sum(aon.aux_res.temp_sl_link_loading, axis=0)[idx, :, :],  # matrix 1
                                 self.cores,  # core count
                             )
                     flows.append(c.results.total_link_loads)
@@ -447,15 +449,15 @@ def execute(self):
                             # The temp flows have an index associated with the link_set name
                             linear_combination_skims(
                                 c.results.select_link_od.matrix[name],  # output matrix
-                                c._aon_results.temp_sl_od_matrix[idx, :, :, :],  # matrix 1
+                                np.sum(aon.aux_res.temp_sl_od_matrix, axis=0)[idx, :, :, :],  # matrix 1
                                 c.results.select_link_od.matrix[name],  # matrix 2 (previous iteration)
                                 self.stepsize,  # stepsize
                                 self.cores,  # core count
                             )
 
                             linear_combination(
                                 c.results.select_link_loading[name],  # output matrix
-                                c._aon_results.temp_sl_link_loading[idx, :, :],  # matrix 1
+                                np.sum(aon.aux_res.temp_sl_link_loading, axis=0)[idx, :, :],  # matrix 1
                                 c.results.select_link_loading[name],  # matrix 2 (previous iteration)
                                 self.stepsize,  # stepsize
                                 self.cores,  # core count

diff --git a/aequilibrae/paths/multi_threaded_aon.py b/aequilibrae/paths/multi_threaded_aon.py
@@ -38,8 +38,20 @@ def prepare(self, graph, results):
             self.has_flow_mask = np.zeros((results.cores, graph.compact_num_links), dtype=bool)
             # Copying the select link matrices from results
             self.select_links = results.select_links
-            self.temp_sl_od_matrix = results.temp_sl_od_matrix
-            self.temp_sl_link_loading = results.temp_sl_link_loading
+            self.temp_sl_od_matrix = np.zeros(
+                (
+                    results.cores,
+                    len(results._selected_links),
+                    graph.num_zones,
+                    graph.num_zones,
+                    results.classes["number"],
+                ),
+                dtype=graph.default_types("float"),
+            )
+            self.temp_sl_link_loading = np.zeros(
+                (results.cores, len(results._selected_links), graph.compact_num_links, results.classes["number"]),
+                dtype=graph.default_types("float"),
+            )
 
         if results.num_skims > 0:
             self.temporary_skims = np.zeros((results.cores, results.compact_nodes, results.num_skims), dtype=ftype)

diff --git a/aequilibrae/paths/results/assignment_results.py b/aequilibrae/paths/results/assignment_results.py
@@ -109,6 +109,7 @@ def prepare(self, graph: Graph, matrix: AequilibraeMatrix) -> None:
         self.crosswalk = np.zeros(graph.graph.shape[0], self.__integer_type)
         self.crosswalk[graph.graph.__supernet_id__.values] = graph.graph.__compressed_id__.values
         self.__graph_ids = graph.graph.__supernet_id__.values
+        self.__graph_compressed_ids = graph.graph.__compressed_id__.values
         self.__redim()
         self.__graph_id__ = graph.__id__
 
@@ -130,16 +131,6 @@ def prepare(self, graph: Graph, matrix: AequilibraeMatrix) -> None:
                 -1,
                 dtype=graph.default_types("int"),
             )
-            # 4d dimensions: link_set, origins, destinations, subclass
-            self.temp_sl_od_matrix = np.zeros(
-                (len(self._selected_links), graph.num_zones, graph.num_zones, self.classes["number"]),
-                dtype=graph.default_types("float"),
-            )
-            # 3d dimensions: link_set, link_id, subclass
-            self.temp_sl_link_loading = np.zeros(
-                (len(self._selected_links), graph.compact_num_links, self.classes["number"]),
-                dtype=graph.default_types("float"),
-            )
 
             sl_idx = {}
             for i, val in enumerate(self._selected_links.items()):
@@ -150,8 +141,14 @@ def prepare(self, graph: Graph, matrix: AequilibraeMatrix) -> None:
                 # Multidimensional arrays where each row has different lengths
                 self.select_links[i][: len(arr)] = arr
                 # Correctly sets the dimensions for the final output matrices
-                self.select_link_od.matrix[name] = self.temp_sl_od_matrix[i]
-                self.select_link_loading[name] = self.temp_sl_link_loading[i]
+                self.select_link_od.matrix[name] = np.zeros(
+                    (graph.num_zones, graph.num_zones, self.classes["number"]),
+                    dtype=graph.default_types("float"),
+                )
+                self.select_link_loading[name] = np.zeros(
+                    (graph.compact_num_links, self.classes["number"]),
+                    dtype=graph.default_types("float"),
+                )
 
             # Overwrites previous arrays on assignment results level with the index to access that array in Cython
             self._selected_links = sl_idx
@@ -277,7 +274,6 @@ def get_load_results(self) -> AequilibraeData:
     def get_sl_results(self) -> AequilibraeData:
         # Set up the name for each column. Each set of select links has a column for ab, ba, total flows
         # for each subclass contained in the TrafficClass
-        raise NotImplementedError("Select link is currently disabled. See issue #442")
         fields = [
             e
             for name in self._selected_links.keys()
@@ -299,7 +295,7 @@ def get_sl_results(self) -> AequilibraeData:
             # Link flows initialised
             link_flows = np.full((self.links, self.classes["number"]), np.nan)
             # maps link flows from the compressed graph to the uncompressed graph
-            assign_link_loads(link_flows, self.select_link_loading[name], self.crosswalk, self.cores)
+            assign_link_loads(link_flows, self.select_link_loading[name], self.__graph_compressed_ids, self.cores)
             for i, n in enumerate(self.classes["names"]):
                 # Directional Flows
                 res.data[name + "_" + n + "_ab"][m.network_ab_idx] = np.nan_to_num(link_flows[m.graph_ab_idx, i])
@@ -309,6 +305,7 @@ def get_sl_results(self) -> AequilibraeData:
                 res.data[name + "_" + n + "_tot"] = np.nan_to_num(res.data[name + "_" + n + "_ab"]) + np.nan_to_num(
                     res.data[name + "_" + n + "_ba"]
                 )
+
         return res
 
     def save_to_disk(self, file_name=None, output="loads") -> None:

diff --git a/aequilibrae/paths/traffic_assignment.py b/aequilibrae/paths/traffic_assignment.py
@@ -687,8 +687,6 @@ def select_link_flows(self) -> Dict[str, pd.DataFrame]:
         """
         Returns a dataframe of the select link flows for each class
         """
-
-        raise NotImplementedError("Select link is currently disabled. See issue #442")
         class_flows = []  # stores the df for each class
         for cls in self.classes:
             # Save OD_matrices
@@ -715,7 +713,6 @@ def save_select_link_flows(self, table_name: str, project=None) -> None:
             Defaults to the active project
         """
 
-        raise NotImplementedError("Select link is currently disabled. See issue #442")
         if not project:
             project = self.project or get_active_project()
         df = self.select_link_flows()
@@ -748,7 +745,6 @@ def save_select_link_matrices(self, file_name: str) -> None:
         Saves the Select Link matrices for each TrafficClass in the current TrafficAssignment class
         """
 
-        raise NotImplementedError("Select link is currently disabled. See issue #442")
         for cls in self.classes:
             # Save OD_matrices
             if cls._selected_links is None:
@@ -770,7 +766,5 @@ def save_select_link_results(self, name: str) -> None:
         :Arguments:
             **name** (:obj:`str`): name of the matrices
         """
-
-        raise NotImplementedError("Select link is currently disabled. See issue #442")
         self.save_select_link_flows(name)
         self.save_select_link_matrices(name)
diff --git a/aequilibrae/paths/traffic_class.py b/aequilibrae/paths/traffic_class.py
@@ -139,7 +139,6 @@ def set_select_links(self, links: Dict[str, List[Tuple[int, int]]]):
         :Arguments:
             **links** (:obj:`Union[None, Dict[str, List[Tuple[int, int]]]]`): name of link set and
              Link IDs and directions to be used in select link analysis"""
-        raise NotImplementedError("Select link is currently disabled. See issue #442")
         self._selected_links = {}
         for name, link_set in links.items():
             if len(name.split(" ")) != 1:

diff --git a/docs/source/examples/full_workflows/plot_forecasting.py b/docs/source/examples/full_workflows/plot_forecasting.py
@@ -324,44 +324,44 @@
 assig.max_iter = 500
 assig.rgap_target = 0.00001
 
-# #%%
-# # **OPTIONAL**
-
-# # If we want to execute select link analysis on a particular TrafficClass, we set the links we are analyzing.
-# # The format of the input select links is a dictionary (str: list[tuple]).
-# # Each entry represents a separate set of selected links to compute. The str name will name the set of links.
-# # The list[tuple] is the list of links being selected, of the form (link_id, direction), as it occurs in the Graph.
-# # Direction can be 0, 1, -1. 0 denotes bi-directionality
-# # For example, let's use Select Link on two sets of links:
-
-# # %%
-# select_links = {
-#     "Leaving node 1": [(1, 1), (2, 1)],
-#     "Random nodes": [(3, 1), (5, 1)],
-# }
-# # We call this command on the class we are analyzing with our dictionary of values
-# assigclass.set_select_links(select_links)
-
-# assig.execute()  # we then execute the assignment
-
-# # %%
-# # Now let us save our select link results, all we need to do is provide it with a name
-# # In addition to exporting the select link flows, it also exports the Select Link matrices in OMX format.
-# assig.save_select_link_results("select_link_analysis")
-
-# # %%
-# # Say we just want to save our select link flows, we can call:
-# assig.save_select_link_flows("just_flows")
-
-# # Or if we just want the SL matrices:
-# assig.save_select_link_matrices("just_matrices")
-# # Internally, the save_select_link_results calls both of these methods at once.
-
-# # We could export it to CSV or AequilibraE data, but let's put it directly into the results database
-# assig.save_results("future_year_assignment")
-
-# # And save the skims
-# assig.save_skims("future_year_assignment_skims", which_ones="all", format="omx")
+#%%
+# **OPTIONAL**
+
+# If we want to execute select link analysis on a particular TrafficClass, we set the links we are analyzing.
+# The format of the input select links is a dictionary (str: list[tuple]).
+# Each entry represents a separate set of selected links to compute. The str name will name the set of links.
+# The list[tuple] is the list of links being selected, of the form (link_id, direction), as it occurs in the Graph.
+# Direction can be 0, 1, -1. 0 denotes bi-directionality
+# For example, let's use Select Link on two sets of links:
+
+# %% 
+select_links = {
+    "Leaving node 1": [(1, 1), (2, 1)],
+    "Random nodes": [(3, 1), (5, 1)],
+}
+# We call this command on the class we are analyzing with our dictionary of values
+assigclass.set_select_links(select_links)
+
+assig.execute()  # we then execute the assignment
+
+# %%
+# Now let us save our select link results, all we need to do is provide it with a name
+# In addition to exporting the select link flows, it also exports the Select Link matrices in OMX format.
+assig.save_select_link_results("select_link_analysis")
+
+# %%
+# Say we just want to save our select link flows, we can call:
+assig.save_select_link_flows("just_flows")
+
+# Or if we just want the SL matrices:
+assig.save_select_link_matrices("just_matrices")
+# Internally, the save_select_link_results calls both of these methods at once.
+
+# We could export it to CSV or AequilibraE data, but let's put it directly into the results database
+assig.save_results("future_year_assignment")
+
+# And save the skims
+assig.save_skims("future_year_assignment_skims", which_ones="all", format="omx")
 
 #%%
 # We can also plot convergence

diff --git a/tests/aequilibrae/paths/test_select_link.py b/tests/aequilibrae/paths/test_select_link.py
@@ -3,15 +3,15 @@
 import zipfile
 from os.path import join, dirname
 from tempfile import gettempdir
-from unittest import TestCase, skip
-
+from unittest import TestCase
+import pandas as pd
 import numpy as np
 
 from aequilibrae import TrafficAssignment, TrafficClass, Graph, Project, PathResults
+from aequilibrae.matrix import AequilibraeMatrix
 from ...data import siouxfalls_project
 
 
-@skip("Select link is currently disabled. See issue #442")
 class TestSelectLink(TestCase):
     def setUp(self) -> None:
         os.environ["PATH"] = os.path.join(gettempdir(), "temp_data") + ";" + os.environ["PATH"]
@@ -38,7 +38,6 @@ def setUp(self) -> None:
         self.assignment.set_time_field("free_flow_time")
         self.assignment.max_iter = 1
         self.assignment.set_algorithm("msa")
-        self.assignment.set_cores(1)
 
     def tearDown(self) -> None:
         self.matrix.close()
@@ -149,6 +148,58 @@ def test_link_out_of_bounds(self):
         self.assignclass = TrafficClass("car", self.car_graph, self.matrix)
         self.assertRaises(ValueError, self.assignclass.set_select_links, {"test": [(78, 1), (1, 1)]})
 
+    def test_kaitang(self):
+        proj_path = os.path.join(gettempdir(), "test_traffic_assignment_path_files" + uuid.uuid4().hex)
+        os.mkdir(proj_path)
+        zipfile.ZipFile(join(dirname(siouxfalls_project), "KaiTang.zip")).extractall(proj_path)
+
+        link_df = pd.read_csv(os.path.join(proj_path, "link.csv"))
+        centroids_array = np.array([7, 8, 11])
+
+        net = link_df.copy()
+
+        g = Graph()
+        g.network = net
+        g.network_ok = True
+        g.status = "OK"
+        g.mode = "a"
+        g.prepare_graph(centroids_array)
+        g.set_blocked_centroid_flows(False)
+        g.set_graph("fft")
+
+        aem_mat = AequilibraeMatrix()
+        aem_mat.load(os.path.join(proj_path, "demand_a.aem"))
+        aem_mat.computational_view(["a"])
+
+        assign_class = TrafficClass("class_a", g, aem_mat)
+        assign_class.set_fixed_cost("a_toll")
+        assign_class.set_vot(1.1)
+        assign_class.set_select_links(links={"trace": [(7, 0), (13, 0)]})
+
+        assign = TrafficAssignment()
+        assign.set_classes([assign_class])
+        assign.set_vdf("BPR")
+        assign.set_vdf_parameters({"alpha": "alpha", "beta": "beta"})
+        assign.set_capacity_field("capacity")
+        assign.set_time_field("fft")
+        assign.set_algorithm("bfw")
+        assign.max_iter = 100
+        assign.rgap_target = 0.0001
+
+        # 4.execute
+        assign.execute()
+
+        # 5.receive results
+        assign_flow_res_df = assign.results().reset_index(drop=False).fillna(0)
+        select_link_flow_df = assign.select_link_flows().reset_index(drop=False).fillna(0)
+
+        pd.testing.assert_frame_equal(
+            assign_flow_res_df[["link_id", "a_ab", "a_ba", "a_tot"]],
+            select_link_flow_df.rename(
+                columns={"class_a_trace_a_ab": "a_ab", "class_a_trace_a_ba": "a_ba", "class_a_trace_a_tot": "a_tot"}
+            ),
+        )
+
 
 def create_od_mask(demand: np.array, graph: Graph, sl):
     res = PathResults()

diff --git a/tests/data/KaiTang.zip b/tests/data/KaiTang.zip