Creates tests for distribution procedures

qaequilibrae/modules/distribution_procedures/apply_gravity_procedure.py

@@ -12,10 +12,7 @@ def __init__(self, parentThread, **kwargs):
         self.report = []
 
     def doWork(self):
-        try:
-            self.gravity.apply()
-            self.output = self.gravity.output
-            self.report = self.gravity.report
-        except ValueError as e:
-            self.error = e
+        self.gravity.apply()
+        self.output = self.gravity.output
+        self.report = self.gravity.report
         self.jobFinished.emit("apply_gravity")

qaequilibrae/modules/distribution_procedures/distribution_models_dialog.py

@@ -35,7 +35,7 @@
 
 
 class DistributionModelsDialog(QtWidgets.QDialog, FORM_CLASS):
-    def __init__(self, qgs_proj, mode=None):
+    def __init__(self, qgs_proj, mode=None, testing=False):
         QtWidgets.QDialog.__init__(self)
         self.iface = qgs_proj.iface
         self.setupUi(self)
@@ -104,6 +104,8 @@ def __init__(self, qgs_proj, mode=None):
         self.user_chosen_model = None
         self.update_model_parameters()
 
+        self.testing = testing
+
     def load_matrices(self):
         self.matrices = list_matrices(self.project.matrices.fldr)
 
@@ -270,7 +272,6 @@ def add_to_table(self, dictio, table):
                 table.setItem(i, 1, QTableWidgetItem(str(dictio[data_name].cores)))
 
     def browse_outfile(self, file_type):
-
         file_types = {
             "aed": ["AequilibraE dataset", ["Aequilibrae dataset(*.aed)"], ".aed"],
             "mod": ["Model file", ["Model file(*.mod)"], ".mod"],
@@ -306,8 +307,9 @@ def add_job_to_queue(self):
                 atra_field = self.cob_atra_field.currentText()
 
             if self.job == "ipf":
-                out_name = self.browse_outfile("aem")
-                if out_name is not None:
+                if not self.testing:
+                    self.out_name = self.browse_outfile("aem")
+                if self.out_name is not None:
                     args = {
                         "matrix": seed_matrix,
                         "rows": prod_vec,
@@ -319,8 +321,9 @@ def add_job_to_queue(self):
                     worker_thread = IpfProcedure(qgis.utils.iface.mainWindow(), **args)
 
             if self.job == "apply":
-                out_name = self.browse_outfile("aem")
-                if out_name is not None:
+                if not self.testing:
+                    self.out_name = self.browse_outfile("aem")
+                if self.out_name is not None:
                     for i in range(1, self.table_model.rowCount()):
                         if str(self.table_model.item(i, 0).text()) == "Alpha":
                             self.model.alpha = float(self.table_model.cellWidget(i, 1).value())
@@ -334,14 +337,15 @@ def add_job_to_queue(self):
                         "model": self.model,
                         "columns": atra_vec,
                         "column_field": atra_field,
-                        "output": out_name,
+                        "output": self.out_name,
                         "nan_as_zero": self.chb_empty_as_zero.isChecked(),
                     }
                     worker_thread = ApplyGravityProcedure(qgis.utils.iface.mainWindow(), **args)
 
             if self.job == "calibrate":
-                out_name = self.browse_outfile("mod")
-                if out_name is not None:
+                if not self.testing:
+                    self.out_name = self.browse_outfile("mod")
+                if self.out_name is not None:
                     if self.rdo_expo.isChecked():
                         func_name = "EXPO"
                     if self.rdo_power.isChecked():
@@ -362,7 +366,7 @@ def add_job_to_queue(self):
             self.chb_empty_as_zero.setEnabled(False)
             if worker_thread is None:
                 return
-            self.add_job_to_list(worker_thread, out_name)
+            self.add_job_to_list(worker_thread, self.out_name)
         else:
             qgis.utils.iface.messageBar().pushMessage(self.tr("Procedure error: "), self.error, level=3)
 
@@ -407,6 +411,7 @@ def check_data(self):
             if self.cob_imped_field.currentIndex() < 0:
                 self.error = self.tr("Impedance matrix is missing")
 
+        print(self.error)
         if self.error is not None:
             return False
         else:
@@ -431,7 +436,7 @@ def job_finished_from_thread(self, success):
         self.exit_procedure()
 
     def exit_procedure(self):
-        if self.report is not None:
+        if self.report is not None and not self.testing:
             dlg2 = ReportDialog(self.iface, self.report)
             dlg2.show()
             dlg2.exec_()

qaequilibrae/modules/distribution_procedures/ipf_procedure.py

@@ -14,10 +14,7 @@ def __init__(self, parentThread, **kwargs):
         self.report = []
 
     def doWork(self):
-        try:
-            self.ipf.fit()
-            self.report = self.ipf.report
-            self.output = self.ipf.output
-        except ValueError as e:
-            self.error = e
+        self.ipf.fit()
+        self.report = self.ipf.report
+        self.output = self.ipf.output
         self.jobFinished.emit("finishedIPF")

test/data/SiouxFalls_project/mod_negative_exponential_X.mod

@@ -0,0 +1,4 @@
+SyntheticGravityModel:
+  alpha: null
+  beta: 0.020709580776383137
+  function: EXPO

test/test_distribution_procedures.py

@@ -0,0 +1,204 @@
+from os.path import isfile, splitext, basename
+
+import numpy as np
+import openmatrix as omx
+import pytest
+from PyQt5.QtCore import Qt
+from aequilibrae.matrix import AequilibraeData, AequilibraeMatrix
+
+from qaequilibrae.modules.distribution_procedures.distribution_models_dialog import DistributionModelsDialog
+from qaequilibrae.modules.paths_procedures.traffic_assignment_dialog import TrafficAssignmentDialog
+
+
+def run_traffic_assignment(ae_with_project, qtbot, ext):
+    dialog = TrafficAssignmentDialog(ae_with_project)
+
+    assignment_result = f"TrafficAssignment_DP_{ext}"
+    dialog.output_scenario_name.setText(assignment_result)
+    dialog.cob_matrices.setCurrentText("demand.aem")
+
+    dialog.tbl_core_list.selectRow(0)
+    dialog.cob_mode_for_class.setCurrentIndex(0)
+    dialog.ln_class_name.setText("car")
+    dialog.pce_setter.setValue(1.0)
+    dialog.chb_check_centroids.setChecked(False)
+    qtbot.mouseClick(dialog.but_add_class, Qt.LeftButton)
+
+    dialog.cob_skims_available.setCurrentText("free_flow_time")
+    qtbot.mouseClick(dialog.but_add_skim, Qt.LeftButton)
+    dialog.cob_skims_available.setCurrentText("distance")
+    qtbot.mouseClick(dialog.but_add_skim, Qt.LeftButton)
+
+    dialog.cob_vdf.setCurrentText("BPR")
+    dialog.cob_capacity.setCurrentText("capacity")
+    dialog.cob_ffttime.setCurrentText("free_flow_time")
+    dialog.cb_choose_algorithm.setCurrentText("bfw")
+    dialog.max_iter.setText("500")
+    dialog.rel_gap.setText("0.001")
+    dialog.tbl_vdf_parameters.cellWidget(0, 1).setText("0.15")
+    dialog.tbl_vdf_parameters.cellWidget(1, 1).setText("4.0")
+
+    dialog.run()
+
+
+def test_ipf(ae_with_project, qtbot):
+    dataset_name = "test/data/SiouxFalls_project/synthetic_future_vector.aed"
+    dataset = AequilibraeData()
+    dataset.load(dataset_name)
+
+    data_name = splitext(basename(dataset_name))[0]
+
+    dialog = DistributionModelsDialog(ae_with_project, mode="ipf", testing=True)
+
+    file_path = "test/data/SiouxFalls_project/demand_ipf.aem"
+    dialog.out_name = file_path
+    dialog.outfile = file_path
+    dialog.datasets[data_name] = dataset
+
+    dialog.load_comboboxes(dialog.datasets.keys(), dialog.cob_prod_data)
+    dialog.load_comboboxes(dialog.datasets.keys(), dialog.cob_atra_data)
+
+    temp = list(dialog.matrices["name"])
+    demand_idx = temp.index("demand.aem")
+    dialog.cob_seed_mat.setCurrentIndex(demand_idx)
+    dialog.cob_seed_field.setCurrentText("matrix")
+
+    dialog.cob_prod_data.setCurrentText("synthetic_future_vector")
+    dialog.cob_prod_field.setCurrentText("origins")
+    dialog.cob_atra_data.setCurrentText("synthetic_future_vector")
+    dialog.cob_atra_field.setCurrentText("destinations")
+
+    qtbot.mouseClick(dialog.but_queue, Qt.LeftButton)
+    qtbot.mouseClick(dialog.but_run, Qt.LeftButton)
+
+    dialog.close()
+
+    assert isfile(file_path)
+
+    mat = AequilibraeMatrix()
+    mat.load(file_path)
+
+    assert mat.matrix["matrix"].shape == (24, 24)
+    assert np.sum(np.nan_to_num(mat.matrix["matrix"])[:, :]) > 360600
+
+
+@pytest.mark.parametrize(
+    ("is_negative", "is_power", "file1", "file2", "ext"),
+    [
+        (True, False, "mod_negative_exponential", "", "A"),
+        (False, True, "", "mod_inverse_power", "B"),
+        (True, True, "mod_negative_exponential", "mod_inverse_power", "C"),
+    ],
+)
+def test_calibrate_gravity(ae_with_project, qtbot, is_negative, is_power, file1, file2, ext):
+    run_traffic_assignment(ae_with_project, qtbot, ext)
+
+    dialog = DistributionModelsDialog(ae_with_project, mode="calibrate", testing=True)
+
+    dialog.path = "test/data/SiouxFalls_project/"
+
+    temp = list(dialog.matrices["name"])
+    imped_idx = temp.index(f"TrafficAssignment_DP_{ext}_car")
+    demand_idx = temp.index("omx")
+    dialog.cob_imped_mat.setCurrentIndex(imped_idx)
+    dialog.cob_imped_field.setCurrentText("free_flow_time_final")
+    dialog.cob_seed_mat.setCurrentIndex(demand_idx)
+    dialog.cob_seed_field.setCurrentText("matrix")
+
+    if is_negative:
+        f1 = f"test/data/SiouxFalls_project/{file1}_{ext}.mod"
+
+        dialog.out_name = f1
+
+        dialog.rdo_expo.setChecked(True)
+
+        qtbot.mouseClick(dialog.but_queue, Qt.LeftButton)
+
+    if is_power:
+        f2 = f"test/data/SiouxFalls_project/{file2}_{ext}.mod"
+
+        dialog.out_name = f2
+
+        dialog.rdo_power.setChecked(True)
+
+        qtbot.mouseClick(dialog.but_queue, Qt.LeftButton)
+
+    qtbot.mouseClick(dialog.but_run, Qt.LeftButton)
+
+    dialog.close()
+
+    if is_negative:
+        assert isfile(f1)
+
+        file_text = ""
+        with open(f1, "r", encoding="utf-8") as file:
+            for line in file.readlines():
+                file_text += line
+
+        assert "alpha: null" in file_text
+        assert "function: EXPO" in file_text
+
+    if is_power:
+        assert isfile(f2)
+
+        file_text = ""
+        with open(f2, "r", encoding="utf-8") as file:
+            for line in file.readlines():
+                file_text += line
+
+        assert "beta: null" in file_text
+        assert "function: POWER" in file_text
+
+
+@pytest.mark.parametrize(
+    ("is_negative", "is_power", "is_gamma", "ext"),
+    [(True, False, False, "X"), (False, True, False, "Y"), (False, False, True, "Z")],
+)
+def test_apply_gravity(ae_with_project, qtbot, is_negative, is_power, is_gamma, ext):
+    dataset_name = "test/data/SiouxFalls_project/synthetic_future_vector.aed"
+    dataset = AequilibraeData()
+    dataset.load(dataset_name)
+
+    data_name = splitext(basename(dataset_name))[0]
+
+    dialog = DistributionModelsDialog(ae_with_project, mode="apply", testing=True)
+
+    dialog.datasets[data_name] = dataset
+    dialog.load_comboboxes(dialog.datasets.keys(), dialog.cob_prod_data)
+    dialog.load_comboboxes(dialog.datasets.keys(), dialog.cob_atra_data)
+
+    temp = list(dialog.matrices["name"])
+    imped_idx = temp.index(f"trafficassignment_dp_x_car_omx")
+    dialog.cob_imped_mat.setCurrentIndex(imped_idx)
+    dialog.cob_imped_field.setCurrentText("free_flow_time_final")
+
+    dialog.cob_prod_data.setCurrentText("synthetic_future_vector")
+    dialog.cob_prod_field.setCurrentText("origins")
+    dialog.cob_atra_data.setCurrentText("synthetic_future_vector")
+    dialog.cob_atra_field.setCurrentText("destinations")
+
+    if is_negative:
+        dialog.model.load(f"test/data/SiouxFalls_project/mod_negative_exponential_X.mod")
+        dialog.update_model_parameters()
+    elif is_power:
+        dialog.model.function = "POWER"
+        dialog.model.alpha = 0.02718039228535631
+        dialog.update_model_parameters()
+    elif is_gamma:
+        dialog.model.alpha = 0.02718039228535631
+        dialog.model.beta = 0.020709580776383137
+
+    file_path = f"test/data/SiouxFalls_project/matrices/ADJ-TrafficAssignment_DP_{ext}.omx"
+    dialog.out_name = file_path
+
+    qtbot.mouseClick(dialog.but_queue, Qt.LeftButton)
+    qtbot.mouseClick(dialog.but_run, Qt.LeftButton)
+
+    dialog.close()
+
+    assert isfile(file_path)
+
+    mtx = omx.open_file(file_path)
+    mtx = mtx["gravity"][:]
+    assert mtx.shape == (24, 24)  # matrix shape
+    assert mtx.sum() > 0  # matrix is not null

test/test_load_matrix.py

@@ -7,23 +7,23 @@
 
 def load_layers():
     import csv
-    path_to_csv = 'test/data/SiouxFalls_project/SiouxFalls_od.csv'
-    datalayer = QgsVectorLayer('None?delimiter=,', "open_layer", 'memory')
+
+    path_to_csv = "test/data/SiouxFalls_project/SiouxFalls_od.csv"
+    datalayer = QgsVectorLayer("None?delimiter=,", "open_layer", "memory")
 
     fields = [
-        QgsField('O', QVariant.Int),
-        QgsField('D', QVariant.Int),
-        QgsField('Ton', QVariant.Double),
+        QgsField("O", QVariant.Int),
+        QgsField("D", QVariant.Int),
+        QgsField("Ton", QVariant.Double),
     ]
     datalayer.dataProvider().addAttributes(fields)
     datalayer.updateFields()
 
-    with open(
-        path_to_csv, 'r') as file:
+    with open(path_to_csv, "r") as file:
         reader = csv.DictReader(file)
         for row in reader:
-            origin = int(row['O'])
-            destination = int(row['D'])
+            origin = int(row["O"])
+            destination = int(row["D"])
             tons = float(row["Ton"])
 
             feature = QgsFeature()
@@ -36,6 +36,7 @@ def load_layers():
     else:
         QgsProject.instance().addMapLayer(datalayer)
 
+
 def test_matrix_menu(ae_with_project, qtbot):
     from qaequilibrae.modules.matrix_procedures.load_matrix_dialog import LoadMatrixDialog
     from test.test_qaequilibrae_menu_with_project import check_if_new_active_window_matches_class
@@ -48,6 +49,7 @@ def handle_trigger():
     QTimer.singleShot(10, handle_trigger)
     action.trigger()
 
+
 def test_save_matrix(ae_with_project, qtbot):
     file_name = "test/data/SiouxFalls_project/test_matrix.aem"
     load_layers()
@@ -62,11 +64,12 @@ def test_save_matrix(ae_with_project, qtbot):
     dialog.prepare_final_matrix()
 
     from aequilibrae.matrix import AequilibraeMatrix
+
     mat = AequilibraeMatrix()
     mat.load(file_name)
-    
-    assert mat.matrix["ton"].shape == (24,24)
-    assert np.sum(np.nan_to_num(mat.matrix["ton"])[:,:]) == 360600
+
+    assert mat.matrix["ton"].shape == (24, 24)
+    assert np.sum(np.nan_to_num(mat.matrix["ton"])[:, :]) == 360600
     assert (mat.index == np.arange(1, 25)).all()
 
     dialog.close()