smarts.core.utils.sumo_utils module

Importing this module “redirects” the import to the “real” sumolib. This is available for convenience and to reduce code duplication as sumolib lives under SUMO_HOME.

class smarts.core.utils.sumo_utils.DomainWrapper(traci_conn, domain: Domain, attribute_name)[source]: Wraps traci.Domain type for the TraciConn utility

class smarts.core.utils.sumo_utils.LocalSumoProcess(sumo_port)[source]

Connects to a local sumo process.

generate(base_params: List[str], sumo_binary: Literal['sumo', 'sumo-gui'] = 'sumo')[source]: Generate the process.

property host: str: The port this process is associated with.

poll() → int | None[source]: Poll the underlying process.

property port: int: The port this process is associated with.

terminate(kill)[source]: Terminate this process.

wait(timeout=None)[source]: Wait on the underlying process.

class smarts.core.utils.sumo_utils.RemoteSumoProcess(remote_host, remote_port)[source]

Connects to a sumo server.

generate(base_params: List[str], sumo_binary: Literal['sumo', 'sumo-gui'] = 'sumo')[source]: Generate the process.

property host: str: The port this process is associated with.

poll() → int | None[source]: Poll the underlying process.

property port: int: The port this process is associated with.

terminate(kill: bool)[source]: Terminate this process.

wait(timeout: float | None = None) → int[source]: Wait on the underlying process.

class smarts.core.utils.sumo_utils.SumoProcess[source]

A simplified utility representing a SUMO process.

abstract generate(base_params: List[str], sumo_binary: Literal['sumo', 'sumo-gui'] = 'sumo')[source]: Generate the process.

abstract property host: str: The port this process is associated with.

abstract poll() → int | None[source]: Poll the underlying process.

abstract property port: int: The port this process is associated with.

abstract terminate(kill: bool)[source]: Terminate this process.

abstract wait(timeout: float | None = None) → int[source]: Wait on the underlying process.

class smarts.core.utils.sumo_utils.TraciConn(sumo_process: SumoProcess, host: str = 'localhost', name: str = '')[source]

A simplified utility for connecting to a SUMO process.

close_traci_and_pipes(wait: bool = True, kill: bool = True)[source]: Safely closes all connections. We should expect this method to always work without throwing

connect(timeout: float, minimum_traci_version: int, minimum_sumo_version: Tuple[int, ...], debug: bool = False)[source]: Attempt a connection with the SUMO process.

property connected: bool: Check if the connection is still valid.

property hostname: str: Get the used TraCI port.

must_reset()[source]: If the version of sumo will have errors if just reloading such that it must be reset.

property port: int | None: Get the used TraCI port.

property sumo_version: Tuple[int, ...]: Get the current SUMO version as a tuple.

teardown()[source]: Clean up all resources.

property viable: bool: If making a connection to the sumo process is still viable.