pytsc.controllers package

Submodules

pytsc.controllers module

pytsc.controllers module

pytsc.controllers module

pytsc.controllers module

pytsc.controllers.controllers module

class pytsc.controllers.controllers.BaseController(traffic_signal, round_robin=True, **kwargs)

Bases: abc.ABC

Base class for traffic signal controllers. This class defines the interface for different traffic signal controllers and provides common functionality for controller management.

Parameters

• traffic_signal (TrafficSignal) – Traffic signal object containing simulation parameters and network information.

• round_robin (bool) – Flag to indicate if round-robin scheduling is used.

• **kwargs – Additional keyword arguments for controller configuration.

class pytsc.controllers.controllers.FixedTimeController(traffic_signal, green_time=25)

Bases: pytsc.controllers.controllers.BaseController

Fixed Time Controller.

Parameters

• traffic_signal (TrafficSignal) – Traffic signal object containing simulation parameters and network information.

• green_time (int) – Duration of the green phase in seconds.

get_action(inp)

Select the next action based on fixed time intervals.

Parameters

inp (dict) – Input data containing network and traffic signal information.

Returns

Index of the selected action.

Return type

int

class pytsc.controllers.controllers.GreedyController(traffic_signal)

Bases: pytsc.controllers.controllers.BaseController

Greedy Controller.

Parameters

traffic_signal (TrafficSignal) – Traffic signal object containing simulation parameters and network information.

get_action(inp)

Select the next action based on the greedy algorithm.

Parameters

inp (dict) – Input data containing network and traffic signal information.

Returns

Index of the selected action.

Return type

int

class pytsc.controllers.controllers.MaxPressureController(traffic_signal)

Bases: pytsc.controllers.controllers.BaseController

Max Pressure Controller.

Parameters

traffic_signal (TrafficSignal) – Traffic signal object containing simulation parameters and network information.

get_action(inp)

Select the next action based on the max pressure algorithm.

Parameters

inp (dict) – Input data containing network and traffic signal information.

Returns

Index of the selected action.

Return type

int

class pytsc.controllers.controllers.RandomController(traffic_signal)

Bases: pytsc.controllers.controllers.BaseController

Randomly selects an action from the available actions.

Parameters

traffic_signal (TrafficSignal) – Traffic signal object containing simulation parameters and network information.

get_action(inp)

Select a random action from the available actions.

Parameters

inp (dict) – Input data containing network and traffic signal information.

Returns

Index of the selected action.

Return type

int

class pytsc.controllers.controllers.SOTLController(traffic_signal, theta=3, mu=4, phi_min=5)

Bases: pytsc.controllers.controllers.BaseController

SOTL (Saturation Over Time Limit) Controller.

Args: traffic_signal (TrafficSignal): Traffic signal object containing simulation parameters and network information. theta (int): threshold for vehicles on red phase lanes mu (int): threshold for vehicles on green phase lanes phi_min (int): minimum green time

get_action(inp)

Select the next action based on the SOTL algorithm.

Parameters

inp (dict) – Input data containing network and traffic signal information.

pytsc.controllers.evaluate module

class pytsc.controllers.evaluate.Evaluate(scenario, simulator_backend, controller_name, add_env_args, add_controller_args, **kwargs)

Bases: object

Base class for evaluating traffic signal controllers. :param scenario: Name of the scenario to evaluate. :type scenario: str :param simulator_backend: Simulator backend to use (e.g., “cityflow”, “sumo”). :type simulator_backend: str :param controller_name: Name of the controller to evaluate. :type controller_name: str :param add_env_args: Additional arguments for the environment. :type add_env_args: dict :param add_controller_args: Additional arguments for the controller. :type add_controller_args: dict :param **kwargs: Additional keyword arguments.

action_space = 'phase_selection'

run(hours, save_stats=False, plot_stats=False, output_folder=None)

Run the evaluation for a specified number of hours. :param hours: Number of hours to evaluate. :type hours: int :param save_stats: Flag to save statistics. :type save_stats: bool :param plot_stats: Flag to plot statistics. :type plot_stats: bool :param output_folder: Folder to save output files. :type output_folder: str

class pytsc.controllers.evaluate.RLEvaluate(scenario, simulator_backend, controller_name, add_env_args, add_controller_args, **kwargs)

Bases: pytsc.controllers.evaluate.Evaluate

Class for evaluating traffic signal controllers using reinforcement learning. :param scenario: Name of the scenario to evaluate. :type scenario: str :param simulator_backend: Simulator backend to use (e.g., “cityflow”, “sumo”). :type simulator_backend: str :param controller_name: Name of the controller to evaluate. :type controller_name: str :param add_env_args: Additional arguments for the environment. :type add_env_args: dict :param add_controller_args: Additional arguments for the controller. :type add_controller_args: dict :param **kwargs: Additional keyword arguments.

action_space = 'phase_switch'

run(hours, save_stats=False, plot_stats=False, output_folder=None)

Run the evaluation for a specified number of hours. :param hours: Number of hours to evaluate. :type hours: int :param save_stats: Flag to save statistics. :type save_stats: bool :param plot_stats: Flag to plot statistics. :type plot_stats: bool :param output_folder: Folder to save output files. :type output_folder: str

pytsc.controllers.rl_controller module

class pytsc.controllers.rl_controller.GraphAttentionComm(adjacency_matrix, hidden_dim=64, n_heads=4)

Bases: torch.nn.modules.module.Module

forward(x, dropout=0.0)

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

training: bool

class pytsc.controllers.rl_controller.LaneAggregator(static_feat_dim=9, dynamic_feat_dim=10, phase_id_dim=20, hidden_dim=64, n_heads=4, device='cpu')

Bases: torch.nn.modules.module.Module

forward(static_feats, dynamic_feats, phase_ids)

static_feats: (bs * n_agents, max_n_lanes, lane_feature_dim) dynamic_feats: (bs * n_agents, max_n_lanes, pos_mat_dim) phase_ids: (bs * n_agents, 1, phase_id_dim) # hidden_state: (bs * n_agents, hidden_dim)

training: bool

class pytsc.controllers.rl_controller.LaneAttentionAggregator(static_feat_dim=9, dynamic_feat_dim=10, phase_id_dim=20, hidden_dim=64, n_heads=4, device='cpu')

Bases: torch.nn.modules.module.Module

forward(static_feats, dynamic_feats, phase_ids)

static_feats: (bs * n_agents, max_n_lanes, lane_feature_dim) dynamic_feats: (bs * n_agents, max_n_lanes, pos_mat_dim) phase_ids: (bs * n_agents, 1, phase_id_dim) # hidden_state: (bs * n_agents, hidden_dim)

training: bool

class pytsc.controllers.rl_controller.MixedRLController(tsc_env, **kwargs)

Bases: pytsc.controllers.rl_controller.RLController

model_paths = ['pytsc/controllers/single_generalized_agent_1.thpytsc/controllers/multi_generalized_agent_1.th']

class pytsc.controllers.rl_controller.MultiGeneralizedAgentRLController(tsc_env, **kwargs)

Bases: pytsc.controllers.rl_controller.RLController

model_paths = ['pytsc/controllers/multi_generalized_agent_1.th']

class pytsc.controllers.rl_controller.MultiGeneralizedGraphAgentRLController(tsc_env, **kwargs)

Bases: pytsc.controllers.rl_controller.RLController

graph = True

model_paths = ['pytsc/controllers/multi_generalized_graph_agent_1.th']

class pytsc.controllers.rl_controller.MultiGeneralizedVarRobustAgentRLController(tsc_env, **kwargs)

Bases: pytsc.controllers.rl_controller.RLController

model_paths = ['pytsc/controllers/multi_generalized_var_robust_agent_1.th']

class pytsc.controllers.rl_controller.MultiGeneralizedVarRobustGraphAgentRLController(tsc_env, **kwargs)

Bases: pytsc.controllers.rl_controller.RLController

graph = True

model_paths = ['pytsc/controllers/multi_generalized_var_robust_graph_agent_1.th']

class pytsc.controllers.rl_controller.RLController(tsc_env, **kwargs)

Bases: object

get_action(hidden_states)

graph = False

hidden_dim = 64

init_hidden()

model_paths = ['pytsc/controllers/agent.th']

n_actions = 2

class pytsc.controllers.rl_controller.SingleGeneralizedAgentRLController(tsc_env, **kwargs)

Bases: pytsc.controllers.rl_controller.RLController

model_paths = ['pytsc/controllers/single_generalized_agent_1.th']

class pytsc.controllers.rl_controller.SpecializedMARLController(tsc_env, **kwargs)

Bases: pytsc.controllers.rl_controller.RLController

graph = True

model_paths = ['pytsc/controllers/specialized_marl_agent_1.th']

class pytsc.controllers.rl_controller.TSCAgent(n_agents, n_actions, hidden_dim, max_n_lanes)

Bases: torch.nn.modules.module.Module

forward(static_lane_feats, pos_mats, phase_ids, hidden_state, **kwargs)

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

init_hidden()

training: bool

class pytsc.controllers.rl_controller.TSCAgentEnsemble(n_agents, n_actions, hidden_dim, max_n_lanes, n_models, adjacency_matrix, static_lane_feats, graph, action_selection_strategy='mean')

Bases: torch.nn.modules.module.Module

init_hidden()

max_n_controlled_lanes = 16

max_phases = 20

select_action(obs, hidden_states, avail_actions)

training: bool

class pytsc.controllers.rl_controller.TSCGraphAgent(n_agents, n_actions, hidden_dim, max_n_lanes, adjacency_matrix, graph, n_heads=4)

Bases: torch.nn.modules.module.Module

forward(static_lane_feats, pos_mats, phase_ids, hidden_state, dropout=0.0)

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

init_hidden()

training: bool

Module contents