pytsc.backends.cityflow package

Submodules

pytsc.backends.cityflow module

pytsc.backends.cityflow module

pytsc.backends.cityflow module

pytsc.backends.cityflow module

pytsc.backends.cityflow module

pytsc.backends.cityflow module

pytsc.backends.cityflow module

pytsc.backends.cityflow module

pytsc.backends.cityflow module

pytsc.backends.cityflow module

pytsc.backends.cityflow module

pytsc.backends.cityflow.config module

class pytsc.backends.cityflow.config.Config(scenario, **kwargs)[source]

Bases: pytsc.common.config.BaseConfig

Configuration class for the CityFlow simulator.

Parameters

  • scenario (str) – The name of the scenario.

  • **kwargs – Additional keyword arguments.

create_and_save_cityflow_cfg()[source]
class pytsc.backends.cityflow.config.DisruptedConfig(scenario, mode='train', **kwargs)[source]

Bases: pytsc.backends.cityflow.config.Config

Configuration class for the CityFlow simulator with disruptions.

Parameters

  • scenario (str) – The name of the scenario.

  • mode (str) – The mode of the simulator (train, test, etc.).

  • **kwargs – Additional keyword arguments.

set_domain_class(domain_class)[source]

pytsc.backends.cityflow.grid_disruptor module

class pytsc.backends.cityflow.grid_disruptor.CityFlowGridDisruptor(scenario, disruption_ratio, speed_reduction_factor, replicate_no)[source]

Bases: object

Disruptor for CityFlow grid networks. This class modifies the speed of lanes in a grid network based on a specified disruption ratio and speed reduction factor.

Parameters

  • scenario (str) – The name of the scenario.

  • disruption_ratio (float) – The ratio of traffic signals to disrupt.

  • speed_reduction_factor (float) – The factor by which to reduce the speed of disrupted lanes.

  • replicate_no (int) – The replicate number for the disrupted network.

generate_disrupted_network()[source]

Generate the disrupted network by selecting edges to disrupt, lowering their speed, and saving the disrupted network to a file.

pytsc.backends.cityflow.grid_generator module

class pytsc.backends.cityflow.grid_generator.CityFlowGridNetworkGenerator(cityflow_dir, nrows, ncols, mean_flow_rates=[600])[source]

Bases: object

Generates a grid network for CityFlow simulation.

Parameters

  • cityflow_dir (str) – Path to the CityFlow installation directory.

  • nrows (int) – Number of rows in the grid.

  • ncols (int) – Number of columns in the grid.

  • mean_flow_rates (list) – List of mean flow rates for the arterial and side streets.

col_distance = 200
generate_roadnet_files()[source]
intersection_width = 20
lane_max_speed = 11.11
n_left_lanes = 0
n_right_lanes = 0
n_straight_lanes = 1
row_distance = 200
veh_headway_time = 2
veh_max_speed = 11.11
veh_min_gap = 2.5
class pytsc.backends.cityflow.grid_generator.CityFlowOneWayGridNetworkGenerator(cityflow_dir, nrows, ncols, mean_flow_rates=[600])[source]

Bases: pytsc.backends.cityflow.grid_generator.CityFlowGridNetworkGenerator

Generates a one-way grid network for CityFlow simulation.

Parameters

  • cityflow_dir (str) – Path to the CityFlow installation directory.

  • nrows (int) – Number of rows in the grid.

  • ncols (int) – Number of columns in the grid.

  • mean_flow_rates (list) – List of mean flow rates for the arterial and side streets.

col_distance = 200
intersection_width = 20
lane_max_speed = 11.11
n_left_lanes = 0
n_right_lanes = 0
n_straight_lanes = 2
row_distance = 200
veh_headway_time = 2
veh_max_speed = 11.11
veh_min_gap = 2.5
pytsc.backends.cityflow.grid_generator.filter_roads(roads)[source]

pytsc.backends.cityflow.metrics module

class pytsc.backends.cityflow.metrics.MetricsParser(parsed_network, simulator, traffic_signals)[source]

Bases: pytsc.common.metrics.BaseMetricsParser

Network wide metrics

Parameters

  • parsed_network (ParsedNetwork) – Parsed network object.

  • simulator (CityFlowSimulator) – CityFlow simulator object.

  • traffic_signals (dict) – Dictionary of traffic signals.

property average_travel_time

The average travel time of vehicles in the network.

Returns

The average travel time of vehicles.

Return type

float

property density

The density of vehicles in the network.

Returns

The density of vehicles.

Return type

float

property density_map

The density map is a matrix that represents the density of vehicles between traffic signals in the network. It is calculated based on the occupancy of lanes connecting the traffic signals.

Returns

The density map of the network.

Return type

np.ndarray

property flickering_signal

The flickering signal is the average of the phase changed signals of all traffic signals in the network. It indicates how often the traffic signals are changing phases.

Returns

The flickering signal.

Return type

float

get_step_stats()[source]

Get the step statistics for the simulation.

Returns

A dictionary containing various statistics for the current step.

Return type

dict

property mean_delay

The mean delay of vehicles in the network.

Returns

The mean delay of vehicles.

Return type

float

property mean_speed

The mean speed of vehicles in the network.

Returns

The mean speed of vehicles.

Return type

float

property mst

The maximum spanning tree (MST) of the density map.

Returns

The maximum spanning tree of the density map.

Return type

np.ndarray

property n_queued

The total number of queued vehicles in the network.

Returns

The total number of queued vehicles.

Return type

int

property n_queued_norm

The normalized number of queued vehicles in the network.

Returns

The normalized number of queued vehicles.

Return type

float

property network_flow

The network flow is the product of the density and the normalized mean speed.

Returns

The network flow.

Return type

float

property norm_mean_speed

The normalized mean speed of vehicles in the network.

Returns

The normalized mean speed of vehicles.

Return type

float

property pressure

The pressure of the network is the sum of the pressure of all traffic signals in the network. It indicates the congestion level of the network.

Returns

The pressure of the network.

Return type

float

property pressures

The pressure of each traffic signal in the network.

Returns

A list of pressure values for each traffic signal.

Return type

list

property time_step

The time step of the simulation.

Returns

The time step of the simulation.

Return type

float

pytsc.backends.cityflow.network_parser module

class pytsc.backends.cityflow.network_parser.NetworkParser(config)[source]

Bases: pytsc.common.network_parser.BaseNetworkParser

Parses the network file and extracts essential information and stores it in the config.

Parameters

config (Config) – Configuration object containing the network file path.

property adjacency_matrix

Computes the adjacency matrix for the traffic signal network.

Returns

A 2D array representing the adjacency matrix of the network.

Return type

np.ndarray

property directional_lanes

Extracts a dictionary of directional lanes for each traffic signal.

Returns

A dictionary where keys are traffic signal IDs and values are dictionaries

mapping end intersection IDs to lists of lane IDs.

Return type

dict

property distance_matrix

Computes the distance matrix for the traffic signal network.

Returns

A 2D array representing the distance between traffic signals.

Return type

np.ndarray

property edge_features

Extracts edge features for the traffic signal network.

Returns

A 3D array of edge features, where each element represents

the number of lanes and average lane length between traffic signals.

Return type

np.ndarray

property in_degrees

Computes the in-degrees for each traffic signal in the network.

Returns

A 1D array representing the in-degrees of traffic signals.

Return type

np.ndarray

property intersections

Extracts the intersections from the network.

Returns

A sorted list of intersection objects.

Return type

list

property k_hop_neighbors

Computes the k-hop neighbors for each traffic signal in the network.

Returns

A dictionary where keys are traffic signal IDs and values are dictionaries

Return type

dict

property lane_angles

Extracts angles for each lane in the network.

Returns

A dictionary where keys are lane IDs and values are angles in degrees.

Return type

dict

property lane_indices

Extracts lane indices for each lane in the network.

Returns

A dictionary where keys are lane IDs and values are lane indices.

Return type

dict

property lane_lengths

Extracts lengths for each lane in the network.

Returns

A dictionary where keys are lane IDs and values are lengths.

Return type

dict

property lane_max_speeds

Extracts maximum speeds for each lane in the network.

Returns

A dictionary where keys are lane IDs and values are maximum speeds.

Return type

dict

property lanes

Extracts the IDs of all lanes in the network.

Returns

A sorted list of lane IDs.

Return type

list

property neighbors_lanes

Extracts a dictionary of neighbors and their connecting lanes for each traffic signal.

Returns

A dictionary where keys are traffic signal IDs and values are dictionaries

mapping neighbor traffic signal IDs to lists of connecting lane IDs.

Return type

dict

property neighbors_offsets

Computes the offsets for each traffic signal based on the travel time to its neighbors.

Returns

A dictionary where keys are traffic signal IDs and values are dictionaries

mapping neighbor traffic signal IDs to their respective offsets.

Return type

dict

property network_boundary

Returns the network boundary.

Returns

A tuple containing the minimum and maximum coordinates of the network.

Return type

tuple

property norm_network_boundary

Returns the normalized network boundary.

Returns

A list containing the width and height of the network.

Return type

list

property out_degrees

Computes the out-degrees for each traffic signal in the network.

Returns

A 1D array representing the out-degrees of traffic signals.

Return type

np.ndarray

plot_network(figsize=(12, 12))[source]

Plots the network using NetworkX and Matplotlib.

Parameters

figsize (tuple) – Size of the figure to plot.

property roads

Extracts the roads from the network.

Returns

A sorted list of road objects.

Return type

list

property traffic_signal_ids

Extracts the IDs of all traffic signals in the network.

Returns

A sorted list of traffic signal IDs.

Return type

list

property ts_coordinates

Extracts coordinates for each traffic signal.

Returns

A dictionary where keys are traffic signal IDs and values are lists of coordinates.

Return type

dict

property ts_norm_coordinates

Extracts normalized coordinates for each traffic signal.

Returns

A dictionary where keys are traffic signal IDs and values are lists of normalized coordinates.

Return type

dict

property ts_phase_to_inc_out_lanes

Extracts the mapping of incoming lanes to outgoing lanes for each traffic signal phase.

Returns

A dictionary where keys are traffic signal IDs and values are dictionaries

mapping phase indices to dictionaries of incoming lanes and their corresponding outgoing lanes.

Return type

dict

pytsc.backends.cityflow.randomize_flows module

pytsc.backends.cityflow.retriever module

class pytsc.backends.cityflow.retriever.Retriever(simulator)[source]

Bases: pytsc.common.retriever.BaseRetriever

Data retrieval class for the CityFlow simulator.

Parameters

simulator (CityFlowSimulator) – CityFlow simulator object.

retrieve_lane_measurements()[source]

Compute lane measurements based on the results from the simulator.

Returns

Dictionary containing lane measurements.

Return type

dict

retrieve_sim_measurements()[source]

Compute simulation measurements based on the results from the simulator.

Returns

Dictionary containing simulation measurements.

Return type

dict

retrieve_ts_measurements()[source]

Retrieve traffic signal measurements from the simulator.

Returns

Dictionary containing traffic signal measurements.

Return type

dict

pytsc.backends.cityflow.simulator module

class pytsc.backends.cityflow.simulator.Simulator(parsed_network)[source]

Bases: pytsc.common.simulator.BaseSimulator

CityFlow simulator class.

Parameters

parsed_network (ParsedNetwork) – Parsed network object containing network information.

close_simulator()[source]

Close the CityFlow simulator.

property is_terminated

Check if the simulation is terminated.

Returns

True if the simulation is terminated, False otherwise.

Return type

bool

retrieve_step_measurements()[source]

Retrieve step measurements from the simulator.

Returns

Dictionary containing step measurements.

Return type

dict

property sim_step

Get the current simulation step.

Returns

Current simulation step.

Return type

int

property sim_time

Get the current simulation time.

Returns

Current simulation time.

Return type

float

simulator_step(n_steps)[source]

Perform a simulation step.

start_simulator()[source]

Start the CityFlow simulator.

pytsc.backends.cityflow.traffic_signal module

class pytsc.backends.cityflow.traffic_signal.TSController(id, config, simulator)[source]

Bases: pytsc.common.traffic_signal.BaseTSController

Traffic signal controller for CityFlow simulator.

Parameters

  • id (str) – Traffic signal ID.

  • config (dict) – Configuration dictionary containing traffic signal parameters.

  • simulator (Simulator) – Simulator object containing simulation parameters and network information.

switch_phase(phase_index)[source]

Switch the traffic light phase to the specified index.

Parameters

phase_index (int) – Index of the phase to switch to.

class pytsc.backends.cityflow.traffic_signal.TSProgram(id, config, simulator)[source]

Bases: pytsc.common.traffic_signal.BaseTSProgram

Traffic signal program for CityFlow simulator.

Parameters

  • id (str) – Traffic signal ID.

  • config (dict) – Configuration dictionary containing traffic signal parameters.

  • simulator (Simulator) – Simulator object containing simulation parameters and network information.

start_phase_index = 0
class pytsc.backends.cityflow.traffic_signal.TrafficSignal(id, config, simulator)[source]

Bases: pytsc.common.traffic_signal.BaseTrafficSignal

Traffic signal class for CityFlow simulator.

Parameters

  • id (str) – Traffic signal ID.

  • config (dict) – Configuration dictionary containing traffic signal parameters.

  • simulator (Simulator) – Simulator object containing simulation parameters and network information.

action_to_phase(phase_index)[source]

Convert the action to the corresponding phase index.

Parameters

phase_index (int) – Index of the phase to convert to.

debug = False
get_controller_action(controller)[source]

Get the action for the specified controller.

Parameters

controller (str) – The type of controller to get actions for.

Returns

Dictionary containing the action for the specified controller.

Return type

dict

update_stats(sub_results)[source]

Update the traffic signal statistics based on the sub-results from the simulator.

Parameters

sub_results (dict) – Dictionary containing sub-results from the simulator.

pytsc.backends.cityflow.trip_generator module

class pytsc.backends.cityflow.trip_generator.CityFlowOneWayTripGenerator(scenario, start_time, end_time, inter_mu_ns, inter_sigma_ns, inter_mu_ew, inter_sigma_ew, disrupted=False, edge_weights=None, **kwargs)[source]

Bases: pytsc.backends.cityflow.trip_generator.CityFlowTripGenerator

Generates vehicle flows for a CityFlow simulation scenario with one-way streets.

Parameters

  • scenario (str) – The name of the scenario.

  • start_time (int) – The start time for flow generation.

  • end_time (int) – The end time for flow generation.

  • inter_mu_ns (float) – The mean of the Gaussian distribution for inter-arrival times on north-south roads.

  • inter_sigma_ns (float) – The standard deviation of the Gaussian distribution for inter-arrival times on north-south roads.

  • inter_mu_ew (float) – The mean of the Gaussian distribution for inter-arrival times on east-west roads.

  • inter_sigma_ew (float) – The standard deviation of the Gaussian distribution for inter-arrival times on east-west roads.

  • disrupted (bool) – Whether the network is disrupted or not.

  • edge_weights (dict) – Edge weights for route calculation. If None, calculated based on lane max speeds.

  • **kwargs – Additional configuration parameters.

generate_flows(filepath, replicate_no=None)[source]

Generate vehicle flows for the CityFlow simulation scenario with one-way streets.

Parameters

  • filepath (str) – The path to save the generated flows.

  • replicate_no (int) – The replicate number for the simulation.

class pytsc.backends.cityflow.trip_generator.CityFlowRandomizedTripGenerator(scenario, start_time, end_time, **kwargs)[source]

Bases: pytsc.backends.cityflow.trip_generator.CityFlowTripGenerator

Generates vehicle flows for a CityFlow simulation scenario with randomized routes.

Parameters

  • scenario (str) – The name of the scenario.

  • start_time (int) – The start time for flow generation.

  • end_time (int) – The end time for flow generation.

  • inter_mu (float) – The mean of the Gaussian distribution for inter-arrival times.

  • inter_sigma (float) – The standard deviation of the Gaussian distribution for inter-arrival times.

  • seed (int) – Random seed for reproducibility.

  • edge_weights (dict) – Edge weights for route calculation. If None, calculated based on lane max speeds.

  • turn_probs (list) – List of probabilities for turning left, right, or going straight.

  • **kwargs – Additional configuration parameters.

generate_flows(filepath, flow_type='low', replicate_no=None)[source]

Generate vehicle flows for the CityFlow simulation scenario with randomized routes.

Parameters

  • filepath (str) – The path to save the generated flows.

  • flow_type (str) – The type of flow to generate (“low”, “medium”, “high”).

  • replicate_no (int) – The replicate number for the simulation.

get_flow_rates()[source]

Get flow rates and route proportions for each road in the network.

turns = ['turn_left', 'turn_right', 'go_straight']
vehicle_data = {'headwayTime': 1.5, 'length': 5.0, 'maxNegAcc': 4.5, 'maxPosAcc': 2.0, 'maxSpeed': 11.11, 'minGap': 2.5, 'usualNegAcc': 4.5, 'usualPosAcc': 2.0, 'width': 2.0}

Generates vehicle flows for a CityFlow simulation scenario. NOTE: Traffic signal network is assumed to be a grid network.

Parameters

  • scenario (str) – The name of the scenario.

  • start_time (int) – The start time for flow generation.

  • end_time (int) – The end time for flow generation.

  • inter_mu (float) – The mean of the Gaussian distribution for inter-arrival times.

  • inter_sigma (float) – The standard deviation of the Gaussian distribution for inter-arrival times.

  • seed (int) – Random seed for reproducibility.

  • edge_weights (dict) – Edge weights for route calculation. If None, calculated based on lane max speeds.

  • turn_probs (list) – List of probabilities for turning left, right, or going straight.

  • **kwargs – Additional configuration parameters.

class pytsc.backends.cityflow.trip_generator.CityFlowTripGenerator(scenario, start_time, end_time, inter_mu, inter_sigma, seed=0, edge_weights=None, turn_probs=[0.1, 0.3, 0.6], **kwargs)[source]

Bases: pytsc.common.trip_generator.TripGenerator

generate_flows(filepath, replicate_no=None)[source]
turns = ['turn_left', 'turn_right', 'go_straight']
vehicle_data = {'headwayTime': 1.5, 'length': 5.0, 'maxNegAcc': 4.5, 'maxPosAcc': 2.0, 'maxSpeed': 11.11, 'minGap': 2.5, 'usualNegAcc': 4.5, 'usualPosAcc': 2.0, 'width': 2.0}

Generates vehicle flows for a CityFlow simulation scenario. NOTE: Traffic signal network is assumed to be a grid network.

Parameters

  • scenario (str) – The name of the scenario.

  • start_time (int) – The start time for flow generation.

  • end_time (int) – The end time for flow generation.

  • inter_mu (float) – The mean of the Gaussian distribution for inter-arrival times.

  • inter_sigma (float) – The standard deviation of the Gaussian distribution for inter-arrival times.

  • seed (int) – Random seed for reproducibility.

  • edge_weights (dict) – Edge weights for route calculation. If None, calculated based on lane max speeds.

  • turn_probs (list) – List of probabilities for turning left, right, or going straight.

  • **kwargs – Additional configuration parameters.

class pytsc.backends.cityflow.trip_generator.FlowDisruptedCityFlowTripGenerator(scenario, start_time, end_time, inter_mu, inter_sigma, disruption_ratio=0.1, **kwargs)[source]

Bases: pytsc.backends.cityflow.trip_generator.CityFlowTripGenerator

generate_flows(filepath, replicate_no=None)[source]
class pytsc.backends.cityflow.trip_generator.IntervalCityFlowTripGenerator(scenario, start_time, end_time, inter_mu, inter_sigma, seed=0, edge_weights=None, turn_probs=[0.1, 0.3, 0.6], **kwargs)[source]

Bases: pytsc.backends.cityflow.trip_generator.CityFlowTripGenerator

generate_flows(filepath, replicate_no, interval_duration=360, shape=1.5, scale=300)[source]

Generate flows with the mean flow rate specified for every interval_duration seconds.

class pytsc.backends.cityflow.trip_generator.LinkDisruptedCityFlowTripGenerator(scenario, start_time, end_time, inter_mu, inter_sigma, disruption_ratio=0.1, **kwargs)[source]

Bases: pytsc.backends.cityflow.trip_generator.CityFlowTripGenerator

generate_flows(filepath, replicate_no=None)[source]
class pytsc.backends.cityflow.trip_generator.VariableDemandTripGenerator(scenario, start_time, end_time, inter_mus, inter_sigmas, edge_weights, disrupted=False, turn_probs=[0.3333333333333333, 0.3333333333333333, 0.3333333333333333], **kwargs)[source]

Bases: pytsc.backends.cityflow.trip_generator.CityFlowTripGenerator

Generates vehicle flows for a CityFlow simulation scenario with variable demand.

Parameters

  • scenario (str) – The name of the scenario.

  • start_time (int) – The start time for flow generation.

  • end_time (int) – The end time for flow generation.

  • inter_mus (dict) – Dictionary of mean inter-arrival times for each edge.

  • inter_sigmas (dict) – Dictionary of standard deviation of inter-arrival times for each edge.

  • disrupted (bool) – Whether the network is disrupted or not.

  • turn_probs (list) – List of probabilities for turning left, right, or going straight.

  • **kwargs – Additional configuration parameters.

generate_flows(filepath, replicate_no=None)[source]

Generate vehicle flows for the CityFlow simulation scenario.

Parameters

  • filepath (str) – The path to save the generated flows.

  • replicate_no (int) – The replicate number for the simulation.

pytsc.backends.cityflow.trip_generator.detect_turn_direction(prev_road, curr_road)[source]

Detects the turn direction based on the previous and current road IDs.

Parameters

  • prev_road (str) – ID of the previous road.

  • curr_road (str) – ID of the current road.

Returns

Turn direction (“turn_left”, “turn_right”, “go_straight”).

Return type

str

Module contents