pytsc.common package

Submodules

pytsc.common module

pytsc.common.actions module

class pytsc.common.actions.BaseActionSpace(config, traffic_signals)[source]

Bases: abc.ABC

Base class for action spaces in traffic signal control. This class defines the interface for different action spaces and provides common functionality for action space management.

Parameters

config (Config) – Configuration object containing simulation parameters.
traffic_signals (dict) – Dictionary of traffic signals in the network.

abstract apply(actions)[source]

Apply the given actions to the traffic signals.

Parameters: actions (list) – List of actions to be applied to the traffic signals.

abstract get_mask()[source]

Get the action mask for each traffic signal.

Returns: List of action masks for each traffic signal.
Return type: list

abstract get_size()[source]

Get the size of the action space.

Returns: Size of the action space.
Return type: int

get_trad_controller_actions(controller)[source]

Get the actions for traditional controllers.

Parameters: controller (str) – The type of controller to get actions for.
Returns: List of actions for each traffic signal.
Return type: list

class pytsc.common.actions.CentralizedActionSpace(individual_action_space)[source]

Bases: pytsc.common.actions.BaseActionSpace

In this child class, actions are the phase index to switch to for each traffic signal.

Parameters: individual_action_space (BaseActionSpace) – The action space for each traffic signal.

apply(action)[source]

Apply the given action to the traffic signals.

Parameters: action (int) – The action to be applied to the traffic signals.

decode_action(action)[source]

Decode the action into individual actions for each traffic signal.

Parameters: action (int) – The action to be decoded.
Returns: List of individual actions for each traffic signal.
Return type: list

get_mask()[source]

Get the action mask for each traffic signal.

Returns: List of action masks for each traffic signal.
Return type: list

get_size()[source]

Get the size of the action space.

Returns: Size of the action space.
Return type: int

class pytsc.common.actions.PhaseSelectionActionSpace(config, traffic_signals)[source]

Bases: pytsc.common.actions.BaseActionSpace

In this child class, actions are the phase index to switch to for each traffic signal.

Parameters

config (Config) – Configuration object containing simulation parameters.
traffic_signals (dict) – Dictionary of traffic signals in the network.

apply(actions)[source]

Apply the given actions to the traffic signals.

Parameters: actions (list) – List of actions to be applied to the traffic signals.

get_mask()[source]

Get the action mask for each traffic signal.

Returns: List of action masks for each traffic signal.
Return type: list

get_size()[source]

Get the size of the action space.

Returns: Size of the action space.
Return type: int

class pytsc.common.actions.PhaseSwitchActionSpace(config, traffic_signals)[source]

Bases: pytsc.common.actions.BaseActionSpace

In this child class, actions are binary: 0 for no switch (remain in the current phase) 1 for switch (move to the next phase in a round-robin manner).

apply(actions)[source]

Apply the given actions to the traffic signals.

Parameters: actions (list) – List of actions to be applied to the traffic signals.

get_mask()[source]

Get the action mask for each traffic signal.

Returns: List of action masks for each traffic signal.
Return type: list

get_size()[source]

Get the size of the action space.

Returns: Size of the action space.
Return type: int

get_trad_controller_actions(controller)[source]

Get the actions for traditional controllers.

Parameters: controller (str) – The type of controller to get actions for.
Returns: List of actions for each traffic signal.
Return type: list

pytsc.common.config module

class pytsc.common.config.BaseConfig(scenario, debug=False, **kwargs)[source]

Bases: abc.ABC

Base class for configuration management in traffic signal control. This class handles loading and merging configuration files for different scenarios and simulator backends.

Parameters

scenario (str) – Name of the scenario for which the configuration is being loaded.
debug (bool) – Flag to enable debug mode. If True, detailed logs will be printed.
**kwargs – Additional configuration parameters to override default values.

pytsc.common.metrics module

class pytsc.common.metrics.BaseMetricsParser[source]

Bases: abc.ABC

Base class for metrics parsers in traffic signal control. This class defines the interface for different metrics parsers and provides common functionality for metrics management.

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

abstract get_step_stats()[source]

Get the statistics for the current simulation step.

Returns: Dictionary containing the statistics for the current step.
Return type: dict

pytsc.common.network_parser module

class pytsc.common.network_parser.BaseNetworkParser(config)[source]

Bases: abc.ABC

Base class for network parsers in traffic signal control. This class defines the interface for different network parsers and provides common functionality for network management.

Parameters

config (Config) – Configuration object containing simulation parameters.
traffic_signal_ids (list) – List of traffic signal IDs in the network.

pytsc.common.observations module

class pytsc.common.observations.BaseObservationSpace(config, parsed_network, traffic_signals, simulator_backend)[source]

Bases: abc.ABC

Base class for observation spaces in traffic signal control. This class defines the interface for different observation spaces and provides common functionality for observation space management.

Parameters

config (Config) – Configuration object containing simulation parameters.
parsed_network (ParsedNetwork) – Parsed network object containing network information.
traffic_signals (dict) – Dictionary of traffic signals in the network.
simulator_backend (str) – The simulator backend to be used (e.g., “cityflow”, “sumo”).

static high()[source]

static low()[source]

class pytsc.common.observations.LaneFeatures(config, parsed_network, traffic_signals, simulator_backend)[source]

Bases: pytsc.common.observations.BaseObservationSpace

Each incoming lane is divided into bins of v_size + min_gap (7.5) meters. Visibility of each traffic signal is in terms of # of bins, not meters. -1: no vehicle in bin 0: vehicle in bin, not moving >0: vehicle in bin, moving Observation:

position matrix of each incoming lane

one hot encoding of current phase

get_observation_info()[source]: Get information about the observation space. :returns: Dictionary containing information about the observation space. :rtype: dict

get_observations()[source]: Each observation includes the lane features of each incoming lane and the one-hot encoding of the current phase. :returns: List of observations for each traffic signal. :rtype: list

get_per_agent_lane_features()[source]: Get lane features for each agent (traffic signal). Each agent’s lane features are padded to the maximum number of controlled lanes. :returns: List of lane features for each agent. :rtype: list

get_size()[source]: Get the size of the observation space. :returns: Size of the observation space. :rtype: int

get_state()[source]: Get the state for each traffic signal. Each state includes the lane features of each incoming lane and the one-hot encoding of the current phase. :returns: List of states for each traffic signal. :rtype: list

get_state_size()[source]: Get the size of the state space. :returns: Size of the state space. :rtype: int

max_lane_length = 500

max_lane_speed = 15.0

max_lanes_per_direction = 6

max_n_controlled_lanes = 16

max_phases = 20

reset_dropped_lanes()[source]: Randomly drop lanes based on the dropout probability. This method is called at the beginning of each episode.

class pytsc.common.observations.PositionMatrix(config, parsed_network, traffic_signals, simulator_backend)[source]

Bases: pytsc.common.observations.BaseObservationSpace

Each incoming lane is divided into bins of v_size + min_gap (7.5) meters. Visibility of each traffic signal is in terms of # of bins, not meters. -1: no vehicle in bin 0: vehicle in bin, not moving >0: vehicle in bin, moving Observation:

position matrix of each incoming lane

one hot encoding of current phase

get_observation_info()[source]: Get information about the observation space. :returns: Dictionary containing information about the observation space. :rtype: dict

get_observations()[source]: Get observations for each traffic signal. Each observation includes the position matrix of each incoming lane and the one-hot encoding of the current phase. :returns: List of observations for each traffic signal. :rtype: list

get_per_agent_lane_features()[source]: Get lane features for each agent (traffic signal). Each agent’s lane features are padded to the maximum number of controlled lanes. :returns: List of lane features for each agent. :rtype: list

get_size()[source]: Get the size of the observation space. :returns: Size of the observation space. :rtype: int

get_state()[source]: Get the state for each traffic signal. Each state includes the position matrix of each incoming lane and the one-hot encoding of the current phase. :returns: List of states for each traffic signal. :rtype: list

get_state_size()[source]: Get the size of the state space. :returns: Size of the state space. :rtype: int

max_lane_length = 500

max_lane_speed = 15.0

max_lanes_per_direction = 6

max_n_controlled_lanes = 16

max_phases = 20

pytsc.common.retriever module

class pytsc.common.retriever.BaseRetriever(simulator)[source]

Bases: abc.ABC

Base class for data retrieval in traffic signal control. This class defines the interface for different data retrieval methods and provides common functionality for data management.

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

abstract retrieve_lane_measurements()[source]

Retrieve lane measurements from the simulator.

Returns: Dictionary containing lane measurements.
Return type: dict

abstract retrieve_sim_measurements()[source]

Retrieve simulation measurements from the simulator.

Returns: Dictionary containing simulation measurements.
Return type: dict

abstract retrieve_ts_measurements()[source]

Retrieve traffic signal measurements from the simulator.

Returns: Dictionary containing traffic signal measurements.
Return type: dict

pytsc.common.reward module

class pytsc.common.reward.BaseRewardFunction(metrics, traffic_signals)[source]

Bases: abc.ABC

Base class for reward functions in traffic signal control. This class defines the interface for different reward functions and provides common functionality for reward management.

Parameters

metrics (Metrics) – Metrics object containing simulation parameters.
traffic_signals (dict) – Dictionary of traffic signals in the network.

abstract get_global_reward()[source]

Calculate the global reward based on pressure and flickering.

Returns: Global reward for the network.
Return type: float

abstract get_local_reward()[source]

Get the local reward for each traffic signal.

Returns: List of local rewards for each traffic signal.
Return type: list

class pytsc.common.reward.MaxPressure(metrics, traffic_signals)[source]

Bases: pytsc.common.reward.BaseRewardFunction

Max Pressure reward function.

Parameters

metrics (Metrics) – Metrics object containing simulation parameters.
traffic_signals (dict) – Dictionary of traffic signals in the network.

get_global_reward()[source]

Calculate the global reward based on pressure and flickering.

Returns: Global reward for the network.
Return type: float

get_local_reward()[source]

Calculate the local reward for each traffic signal based on pressure and flickering.

Returns: List of local rewards for each traffic signal.
Return type: list

class pytsc.common.reward.QueueLength(metrics, traffic_signals)[source]

Bases: pytsc.common.reward.BaseRewardFunction

Queue Length reward function.

Parameters

metrics (Metrics) – Metrics object containing simulation parameters.
traffic_signals (dict) – Dictionary of traffic signals in the network.

get_global_reward()[source]

Calculate the global reward based on pressure and flickering.

Returns: Global reward for the network.
Return type: float

get_local_reward()[source]

Calculate the local reward for each traffic signal based on queue length and flickering.

Returns: List of local rewards for each traffic signal.
Return type: list

pytsc.common.simulator module

class pytsc.common.simulator.BaseSimulator(parsed_network)[source]

Bases: abc.ABC

Base class for traffic signal control simulators. This class defines the interface for different simulators and provides common functionality for simulation management.

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

abstract close_simulator()[source]

Close the simulator.

Returns: True if the simulator closed successfully, False otherwise.
Return type: bool

abstract is_terminated()[source]

Check if the simulation has terminated.

Returns: True if the simulation has terminated, False otherwise.
Return type: bool

abstract retrieve_step_measurements()[source]

Retrieve measurements for the current simulation step.

Returns: Dictionary containing measurements for the current step.
Return type: dict

abstract simulator_step()[source]

Perform a simulation step.

Returns: True if the simulation step was successful, False otherwise.
Return type: bool

abstract start_simulator()[source]

Start the simulator.

Returns: True if the simulator started successfully, False otherwise.
Return type: bool

pytsc.common.traffic_signal module

class pytsc.common.traffic_signal.BaseTSController(id, config, simulator)[source]

Bases: abc.ABC

Base class for traffic signal controllers. :param id: Unique identifier for the traffic signal controller. :type id: str :param config: Configuration dictionary containing traffic signal parameters. :type config: dict :param simulator: Simulator object containing simulation parameters and network information. :type simulator: Simulator

property current_phase

Get the current phase of the traffic signal controller.

Returns: Current phase of the traffic signal controller.
Return type: str

property current_phase_index

Get the index of the current phase in the traffic signal controller.

Returns: Index of the current phase in the traffic signal controller.
Return type: int

get_allowable_phase_switches()[source]

Get the allowable phase switches based on the current phase and time spent on it.

Returns: List of allowable phase switches for the traffic signal controller.
Return type: list

property green_phase_indices

Get the green phase indices of the traffic signal controller.

Returns: List of green phase indices in the traffic signal controller.
Return type: list

property n_phases

Get the number of phases in the traffic signal controller.

Returns: Number of phases in the traffic signal controller.
Return type: int

property next_green_phase_index

Get the index of the next green phase in the traffic signal controller.

Returns: Index of the next green phase in the traffic signal controller.
Return type: int

property next_phase_index

Get the index of the next phase in the traffic signal controller.

Returns: Index of the next phase in the traffic signal controller.
Return type: int

property norm_time_on_phase

Get the normalized time spent on the current phase of the traffic signal controller.

Returns: Normalized time spent on the current phase of the traffic signal controller.
Return type: float

property phase_indices

Get the phase indices of the traffic signal controller.

Returns: List of phase indices in the traffic signal controller.
Return type: list

property phase_one_hot

Get the one-hot encoding of the current phase of the traffic signal controller.

Returns: One-hot encoding of the current phase of the traffic signal controller.
Return type: list

property phases_min_max_times

Get the minimum and maximum times for each phase in the traffic signal controller.

Returns: Dictionary containing minimum and maximum times for each phase.
Return type: dict

abstract switch_phase()[source]: Switch the phase of the traffic signal controller.

property time_on_phase

Get the time spent on the current phase of the traffic signal controller.

Returns: Time spent on the current phase of the traffic signal controller.
Return type: int

property yellow_phase_indices

Get the yellow phase indices of the traffic signal controller.

Returns: List of yellow phase indices in the traffic signal controller.
Return type: list

property yellow_time

Get the yellow time of the traffic signal controller.

Returns: Yellow time of the traffic signal controller.
Return type: int

class pytsc.common.traffic_signal.BaseTSProgram(id, config, simulator)[source]

Bases: abc.ABC

Base class for traffic signal programs. :param id: Unique identifier for the traffic signal program. :type id: str :param config: Configuration dictionary containing traffic signal parameters. :type config: dict :param simulator: Simulator object containing simulation parameters and network information. :type simulator: Simulator

set_initial_phase(phase_index)[source]: Set the initial phase of the traffic signal program. :param phase_index: Index of the initial phase. :type phase_index: int

update_current_phase(phase_index)[source]: Update the current phase of the traffic signal program. :param phase_index: Index of the new phase. :type phase_index: int

class pytsc.common.traffic_signal.BaseTrafficSignal(id, config, simulator)[source]

Bases: abc.ABC

Base class for traffic signals in the network. :param id: Unique identifier for the traffic signal. :type id: str :param config: Configuration dictionary containing traffic signal parameters. :type config: dict :param simulator: Simulator object containing simulation parameters and network information. :type simulator: Simulator

abstract action_to_phase(action)[source]: Convert action to phase index. :param action: Action index. :type action: int

init_rule_based_controllers()[source]: Initialize rule-based controllers for the traffic signal.

abstract update_stats()[source]: Update the statistics of the traffic signal.

class pytsc.common.traffic_signal.TLSFreePhaseSelectLogic(controller)[source]

Bases: object

phase_dict: {phase_idx: {“min_time”: int, “max_time”: int}}} NOTES: - The phase indices must be ordered such that the phase list starts with a green phase and every green phase must be followed by a yellow phase. - Step size is fixed to be 5 seconds.

get_allowable_phase_switches(time_on_phase)[source]

Get the allowable phase switches based on the current phase and time spent on it.

Parameters: time_on_phase (int) – Time spent on the current phase.
Returns: List of allowable phase switches for the traffic signal controller.
Return type: list

class pytsc.common.traffic_signal.TLSRoundRobinPhaseSelectLogic(controller)[source]

Bases: pytsc.common.traffic_signal.TLSFreePhaseSelectLogic

Round robin phase selection logic for traffic signal controllers.

Parameters: controller (BaseTSController) – Traffic signal controller object.

get_allowable_phase_switches(time_on_phase)[source]

Get the allowable phase switches based on the current phase and time spent on it.

Parameters: time_on_phase (int) – Time spent on the current phase.
Returns: List of allowable phase switches for the traffic signal controller.
Return type: list

pytsc.common.trip_generator module

class pytsc.common.trip_generator.TripGenerator[source]

Bases: abc.ABC

Base class for trip generators in traffic signal control. This class defines the interface for different trip generators and provides common functionality for trip generation.

generate_flows()[source]: Generate flows for the trip generator.

pytsc.common.utils module

class pytsc.common.utils.EnvLogger[source]

Bases: object

A simple logger class to log messages to the console. This class ensures that the logger is initialized only once and provides methods to log messages at different levels (info, warning, etc.).

static get_logger(name='pytsc.common.utils', level=20)[source]

static log(level, msg)[source]

static log_info(msg)[source]

static log_warning(msg)[source]

logger = None

static set_log_level(level)[source]

pytsc.common.utils.calculate_vehicle_bin_index(n_bins, lane_length, vehicle_position)[source]

Calculate the bin index for a vehicle based on its position in the lane. :param n_bins: Number of bins. :type n_bins: int :param lane_length: Length of the lane. :type lane_length: float :param vehicle_position: Position of the vehicle in the lane. :type vehicle_position: float

Returns: Bin index for the vehicle.
Return type: int

pytsc.common.utils.compute_max_spanning_tree(density_map)[source]

pytsc.common.utils.flatten_list(list_of_lists)[source]: Flattens a list of lists into a single list.

pytsc.common.utils.generate_weibull_flow_rates(shape, scale, max_rate, num_segments)[source]

Generate flow rates based on a Weibull distribution. :param shape: Shape parameter for the Weibull distribution. :type shape: float :param scale: Scale parameter for the Weibull distribution. :type scale: float :param max_rate: Maximum flow rate. :type max_rate: float :param num_segments: Number of segments to divide the flow rates into. :type num_segments: int

Returns: Flow rates for each segment.
Return type: np.ndarray

pytsc.common.utils.pad_array(array, size, pad_value=0)[source]: Pads the given array with the specified padding value so that it has the specified size. If the array is already larger than the specified size, it is returned unmodified.

pytsc.common.utils.pad_list(inp_list, size, pad_value=0)[source]: Pads the given list with the specified padding value so that it has the specified size. If the list is already larger than the specified size, it is returned unmodified.

pytsc.common.utils.recursively_update_dict(d, u)[source]: Recursively update a dictionary with another dictionary.

pytsc.common.utils.sort_alphanumeric_ids(ids)[source]

pytsc.common.utils.validate_input_against_allowed(inp_cfg_str, allowed_cfg_strs)[source]

pytsc.common package

Submodules

pytsc.common module

pytsc.common module

pytsc.common module

pytsc.common module

pytsc.common module

pytsc.common module

pytsc.common module

pytsc.common module

pytsc.common module

pytsc.common module

pytsc.common module

pytsc.common module

pytsc.common.actions module

pytsc.common.config module

pytsc.common.metrics module

pytsc.common.network_parser module

pytsc.common.observations module

pytsc.common.retriever module

pytsc.common.reward module

pytsc.common.simulator module

pytsc.common.traffic_signal module

pytsc.common.trip_generator module

pytsc.common.utils module

Module contents