"""Contains the merge network class.""" from flow.networks.base import Network from flow.core.params import InitialConfig from flow.core.params import TrafficLightParams from numpy import pi, sin, cos INFLOW_EDGE_LEN = 200 # length of the inflow edges (needed for resets) VEHICLE_LENGTH = 5 ADDITIONAL_NET_PARAMS = { # length of the merge edge "merge_length": 75, # length of the highway leading to the merge "pre_merge_length": 200, # merge taper "merge_taper_length": 100, # length of the highway past the merge "post_merge_length": 1000, # length of bottle neck downstreem "bottleneck_length": 500, # ramp storage length: minimum Storage length of at least 450 feet (Use formula or graph for individual ramp calculations.) # 450 feet is for 240 vehicles per hour on a one lane ramp "ramp_storage_length": 200, # number of lanes in the merge "merge_lanes": 2, # number of lanes in the highway "highway_lanes": 3, # taper lane "merge_taper_lanes": 4, # max speed limit of the network "speed_limit": 65, # merge speed limit "ramp_speed_limit": 65, # angle of ramp "ramp_angle" : pi/6 } class MergeMeteringNetwork(Network): """Network class for highways with a single in-merge. This network consists of a single or multi-lane highway network with an on-ramp with a variable number of lanes that can be used to generate periodic perturbation. Requires from net_params: * **merge_length** : length of the merge edge * **pre_merge_length** : length of the highway leading to the merge * **post_merge_length** : length of the highway past the merge * **merge_lanes** : number of lanes in the merge * **highway_lanes** : number of lanes in the highway * **speed_limit** : max speed limit of the network Usage ----- >>> from flow.core.params import NetParams >>> from flow.core.params import VehicleParams >>> from flow.core.params import InitialConfig >>> from flow.networks import MergeNetwork >>> >>> network = MergeNetwork( >>> name='merge', >>> vehicles=VehicleParams(), >>> net_params=NetParams( >>> additional_params={ >>> 'merge_length': 100, >>> 'pre_merge_length': 200, >>> 'post_merge_length': 100, >>> 'merge_lanes': 1, >>> 'highway_lanes': 1, >>> 'speed_limit': 30 >>> }, >>> ) >>> ) """ def __init__(self, name, vehicles, net_params, initial_config=InitialConfig(), traffic_lights=TrafficLightParams()): """Initialize a merge network.""" for p in ADDITIONAL_NET_PARAMS.keys(): if p not in net_params.additional_params: raise KeyError('Network parameter "{}" not supplied'.format(p)) super().__init__(name, vehicles, net_params, initial_config, traffic_lights) def specify_nodes(self, net_params): """See parent class.""" angle = net_params.additional_params["ramp_angle"] merge = net_params.additional_params["merge_length"] merge_taper_length = net_params.additional_params["merge_taper_length"] storage_length = net_params.additional_params["ramp_storage_length"] premerge = net_params.additional_params["pre_merge_length"] postmerge = net_params.additional_params["post_merge_length"] bottleneck_length = net_params.additional_params["bottleneck_length"] nodes = [ { "id": "inflow_highway", "x": -INFLOW_EDGE_LEN, "y": 0 }, { "id": "center", "y": 0, "x": 0 }, { "id": "merge_begin", "y": 0, "x": premerge, "radius": 10 }, { "id": "merge_end", "y": 0, "x": premerge + merge_taper_length, "radius": 10 }, { "id": "bottleneck", "y": 0, "x": premerge + merge_taper_length + postmerge }, { "id": "outflow_highway", "y": 0, "x": premerge + merge_taper_length + postmerge + bottleneck_length }, { "id": "stopbar_ramp", "x": premerge - merge * cos(angle), "y": -merge * sin(angle) }, { "id": "inflow_ramp", "x": premerge - (merge + storage_length) * cos(angle), "y": -(merge + storage_length) * sin(angle) }, ] return nodes def specify_edges(self, net_params): """See parent class.""" merge = net_params.additional_params["merge_length"] merge_taper_length = net_params.additional_params["merge_taper_length"] storage_length = net_params.additional_params["ramp_storage_length"] premerge = net_params.additional_params["pre_merge_length"] postmerge = net_params.additional_params["post_merge_length"] bottleneck_length = net_params.additional_params["bottleneck_length"] edges = [{ "id": "inflow_highway", "type": "highwayType", "from": "inflow_highway", "to": "center", "length": INFLOW_EDGE_LEN }, { "id": "center", "type": "highwayType", "from": "center", "to": "merge_begin", "length": premerge }, { "id": "merge_begin", "type": "merge_taper", "from": "merge_begin", "to": "merge_end", "length": merge_taper_length }, { "id": "merge_end", "type": "highwayType", "from": "merge_end", "to": "bottleneck", "length": postmerge }, { "id": "bottleneck", "type": "capacityDrop", "from": "bottleneck", "to": "outflow_highway", "length": bottleneck_length }, { "id": "inflow_ramp", "type": "mergeType", "from": "inflow_ramp", "to": "stopbar_ramp", "length": storage_length }, { "id": "stopbar_ramp", "type": "mergeType", "from": "stopbar_ramp", "to": "merge_begin", "length": merge }] return edges def specify_types(self, net_params): """See parent class.""" h_lanes = net_params.additional_params["highway_lanes"] m_lanes = net_params.additional_params["merge_lanes"] speed = net_params.additional_params["speed_limit"] types = [{ "id": "highwayType", "numLanes": h_lanes, "speed": speed }, { "id": "capacityDrop", "numLanes": h_lanes - 1, "speed": speed }, { "id": "merge_taper", "numLanes": h_lanes + 1, "speed": speed }, { "id": "mergeType", "numLanes": m_lanes, "speed": speed }] return types def specify_routes(self, net_params): """See parent class.""" rts = { "inflow_highway": ["inflow_highway", "center", "merge_begin", "merge_end", "bottleneck"], "center": ["center", "merge_begin", "merge_end", "bottleneck"], "merge_begin": ["merge_begin", "merge_end", "bottleneck"], "merge_end": ["merge_end", "bottleneck"], "bottleneck": ["bottleneck"], "inflow_ramp": ["inflow_ramp", "stopbar_ramp", "merge_begin", "merge_end", "bottleneck"], "stopbar_ramp": ["stopbar_ramp", "merge_begin", "merge_end", "bottleneck"] } return rts # def specify_edge_starts(self): # """See parent class.""" # premerge = self.net_params.additional_params["pre_merge_length"] # postmerge = self.net_params.additional_params["post_merge_length"] # storage_length = self.net_params.additional_params["ramp_storage_length"] # bottleneck_length = self.net_params.additional_params["bottleneck_length"] # edgestarts = [("inflow_highway", 0), ("left", INFLOW_EDGE_LEN + 0.1), # ("center", INFLOW_EDGE_LEN + premerge + 22.6), # ("bottleneck", # INFLOW_EDGE_LEN + premerge + postmerge + 22.6), # ("inflow_ramp", # storage_length + premerge + postmerge + 22.6), # ("stopbar_ramp", # 2 * storage_length + premerge + postmerge + 22.7)] # return edgestarts # def specify_internal_edge_starts(self): # """See parent class.""" # premerge = self.net_params.additional_params["pre_merge_length"] # postmerge = self.net_params.additional_params["post_merge_length"] # storage_length = self.net_params.additional_params["ramp_storage_length"] # bottleneck_length = self.net_params.additional_params["bottleneck_length"] # internal_edgestarts = [ # (":left", INFLOW_EDGE_LEN), (":center", # INFLOW_EDGE_LEN + premerge + 0.1), # (":stopbar_ramp", 2 * storage_length + premerge + postmerge + 22.6) # ] # return internal_edgestarts