Merge remote-tracking branch 'origin/helmet5-experimental' into feat/park-and-ride-choice-model

Author: hsl-petrhaj

Scripts/assignment/emme_bindings/mock_project.py

@@ -712,6 +712,9 @@ def id(self):
     def outgoing_segments(self, include_hidden=False):
         return (s for s in self.network.transit_segments(include_hidden)
             if s.i_node is self)
+    
+    def incoming_links(self):
+        return (l for l in self.network.links() if l.j_node is self)
 
 class Link(NetworkObject):
     def __init__(self, 
@@ -825,6 +828,7 @@ def __init__(self, network: Network, line: TransitLine, link: Link):
             self, network, network._extra_attr["TRANSIT_SEGMENT"])
         self.line = line
         self.link = link
+        self.loop_index = 1
         self.allow_alightings = False
         self.allow_boardings = False
         self.transit_time_func = 0

Scripts/datahandling/zonedata.py

@@ -83,6 +83,10 @@ def __init__(self, data_dir: str, zone_numbers: numpy.array):
         self.nr_zones = len(self.zone_numbers)
         self.nr_zones_hs15 = len(self.zone_numbers_hs15)
         self["population_density"] = pop / landdata["builtar"]
+        # Filter indexes where population_density > 50000
+        high_population_density = self["population_density"].index[self["population_density"] > 50000]
+        if len(high_population_density) > 0:
+            log.warn(f"Zone(s) {list(high_population_density)} have an abnormally high population density. Make sure that the builtar values in the .lnd file are calculated correctly.")
         wp = workdata.pop("total")
         self["workplaces"] = wp
         ShareChecker({})["Workplace shares"] = workdata.sum(axis="columns")

Scripts/utils/modify_network.py

@@ -1,34 +1,94 @@
+import pandas as pd
 import parameters.assignment as param
 
+# Fallback logger for use within EMME
+try:
+    import utils.log as log
+except ImportError:
+    class FallbackLogger:
+        def info(self, message):
+            print(f"[INFO] {message}")
+
+        def debug(self, message):
+            print(f"[DEBUG] {message}")
+
+        def warn(self, message):
+            print(f"[WARN] {message}")
+
+    log = FallbackLogger()
+
 # Utility functions that modify a network. Functions take a network as input and return the modified network
 
 def add_bus_stops(network):
+    # Initialize an empty dictionary to store line IDs and maximum stop distances
+    data = {"line_id": [], "maximum_stop_distance": [], "is_motorway": [], "loops": []}
+    high_distance_lines = []
+    looped_lines = []
+
     for line in network.transit_lines():
-            if line.mode.id in param.stop_codes:
-                stop_codes = param.stop_codes[line.mode.id]
-                for segment in line.segments():
-                    is_stop = segment.i_node.data2 in stop_codes
-                    if line.mode.id in "de":
-                        # Non-HSL bus lines
-                        not_hsl = segment.i_node.label not in param.hsl_area
-                        if line.id[-1] == '1':
-                            # Line starts in HSL area
-                            segment.allow_alightings = not_hsl and is_stop
-                            segment.allow_boardings = is_stop
-                        elif line.id[-1] == '2':
-                            # Line ends in HSL area
-                            segment.allow_alightings = is_stop
-                            segment.allow_boardings = not_hsl and is_stop
-                        elif line.id[-1] == '3':
-                            #Line is circular and possibly goes through HSL area
-                            #not perfect solution, out-in-in-out allows for in-in routes as well
-                            segment.allow_alightings = is_stop
-                            segment.allow_boardings = is_stop
-                        else:
-                            raise ValueError(
-                                "Unknown direction code for line " + line.id)
-                    else:
+        if line.mode.id in param.stop_codes:
+            stop_codes = param.stop_codes[line.mode.id]
+            stop_distance = 0
+            max_stop_distance = 0
+            is_motorway = 0
+            loop = 0
+
+            for segment in line.segments():
+                if segment.loop_index > 1 and loop == 0:
+                    loop = 1
+                    looped_lines.append(line.id)
+                segment_length = segment.link.length
+                linktype = segment.link.type % 100
+                if linktype in param.roadclasses and is_motorway == 0:
+                    # Car link with standard attributes
+                    roadclass = param.roadclasses[linktype]
+                    if roadclass.type == "motorway":
+                        is_motorway = 1
+
+                stop_distance += segment_length
+                is_stop = segment.i_node.data2 in stop_codes
+
+                if is_stop:
+                    if stop_distance > max_stop_distance:
+                        max_stop_distance = stop_distance
+                    stop_distance = 0
+
+                if line.mode.id in "de":
+
+                    not_hsl = segment.i_node.label not in param.hsl_area
+                    if line.id[-1] == '1':
+                        segment.allow_alightings = not_hsl and is_stop
+                        segment.allow_boardings = is_stop
+                    elif line.id[-1] == '2':
+                        segment.allow_alightings = is_stop
+                        segment.allow_boardings = not_hsl and is_stop
+                    elif line.id[-1] == '3':
                         segment.allow_alightings = is_stop
                         segment.allow_boardings = is_stop
+                    else:
+                        raise ValueError(
+                            "Unknown direction code for line " + line.id)
+                else:
+                    segment.allow_alightings = is_stop
+                    segment.allow_boardings = is_stop
+
+            # Append data for the current line
+            data["line_id"].append(line.id)
+            data["maximum_stop_distance"].append(max_stop_distance)
+            data["is_motorway"].append(is_motorway)
+            data["loops"].append(loop)
+
+            if line.mode.id in "bg" and max_stop_distance > 3 and not is_motorway: # and int(line.id[0]) < 6
+                log.debug(f"Line: {line.id},\t Maximum distance between consecutive stops: {max_stop_distance:.2f}")
+                high_distance_lines.append(line.id)
+
+    # TODO: Print to results folder
+    max_stop_distances = pd.DataFrame(data)
+
+    if high_distance_lines:
+        log.info(f"{len(high_distance_lines)} HSL line(s) have a maximum stop distance greater than 3 km and no motorway sections.")
 
-    return network
+    if looped_lines:
+        log.warn(f"Line(s) {looped_lines} traverse over the same links multiple times.")
+
+    return network

Scripts/utils/validate_network.py

@@ -62,6 +62,23 @@ def validate(network, fares=None):
     unofficial_nodes = set()
     nr_links = 0
     nr_zero_gradients = 0
+    
+    # Make sure that cars, pedestrians cyclists can access all centroids
+    unaccessible_centroids = []
+
+    required_modes = {'a', 'f', 'c'}  # Modes for pedestrians, cyclists, and cars
+
+    for centroid in network.centroids():
+        accessible_modes = {mode.id for link in centroid.incoming_links() for mode in link.modes}
+        if not required_modes.issubset(accessible_modes):
+            unaccessible_centroids.append(centroid.id)
+    
+    if unaccessible_centroids:
+        msg = f"Centroids {unaccessible_centroids} are not accessible by pedestrians, cyclists and/or passenger cars."
+        log.error(msg)
+        raise ValueError(msg)
+
+    
     for link in network.links():
         nr_links += 1
         if not link.modes:
@@ -132,13 +149,13 @@ def validate(network, fares=None):
             raise ValueError(msg)
 
         try:
-            if link['@kaltevuus'] == 0:
+            if link['@kaltevuus'] == 0 and not link.i_node.is_centroid and not link.j_node.is_centroid:
                 nr_zero_gradients += 1
         except KeyError:
-            raise ValueError("Gradients not defined. Use an extra_links file with @kaltevuus,\
- or create extra attribute @kaltevuus with zeros.\
- @kaltevuus is used to model the effect of hills on bicycle route choice,\
- setting @kaltevuus to 0 on links will ignore hills.")
+            raise ValueError("Gradients not defined. Use an extra_links file with @kaltevuus, " +
+                             "or create extra attribute @kaltevuus with zeros. " +
+                             "@kaltevuus is used to model the effect of hills on bicycle route choice, " +
+                             "setting @kaltevuus to 0 on links will ignore hills.")
     zero_gradient_ratio = (nr_zero_gradients/nr_links)*100
     if zero_gradient_ratio > 20:
         msg = "{} % of links have a gradient of 0.".format(zero_gradient_ratio)