aequilibrae

class aequilibrae.AequilibraeMatrix

Matrix class

static random_name() → Path

Returns a random name for a matrix with root in the temp directory of the user

>>> from aequilibrae.matrix import AequilibraeMatrix

>>> mat = AequilibraeMatrix()
>>> str(mat.random_name())
'/tmp/aequilibrae/Aequilibrae_matrix_...'

close()

Removes matrix from memory and flushes all data to disk, or closes the OMX file if that is the case

columns() → ndarray

Returns column vector for the matrix in the computational view

Computational view needs to be set to a single matrix core

Returns:

object (np.ndarray): the column totals for the matrix currently on the computational view

>>> from aequilibrae.matrix import AequilibraeMatrix

>>> project = create_example(project_path)

>>> mat = AequilibraeMatrix()
>>> mat.load(Path(project_path) / 'matrices/skims.omx')
>>> mat.computational_view(["distance_blended"])
>>> mat.columns()
array([357.54256811, 357.45109051, 310.88655449, 276.6783439 ,
       266.70388637, 270.62976319, 266.32888632, 279.6897402 ,
       285.89821842, 242.79743295, 252.34085912, 301.78116548,
       302.97058146, 270.61855294, 264.59944248, 257.83842251,
       276.63310578, 257.74513863, 281.15724257, 271.63886077,
       264.62215032, 252.79791125, 273.18139747, 282.7636574 ])

>>> project.close()

computational_view(core_list: List[str] = None)

Creates a memory view for a list of matrices that is compatible with Cython memory buffers

It allows for AequilibraE matrices to be used in all parallelized algorithms within AequilibraE

In case of OMX matrices, the computational view is held only in memory

Arguments:

core_list (list): List with the names of all matrices that need to be in the buffer

>>> from aequilibrae.matrix import AequilibraeMatrix

>>> zones_in_the_model = 3317
>>> names_list = ['Car trips', 'pt trips', 'DRT trips', 'bike trips', 'walk trips']

>>> mat = AequilibraeMatrix()
>>> mat.create_empty(zones=zones_in_the_model,
...                  matrix_names=names_list)
>>> mat.computational_view(['bike trips', 'walk trips'])
>>> mat.view_names
['bike trips', 'walk trips']

copy(output_name: str = None, cores: List[str] = None, names: List[str] = None, compress: bool = None, memory_only: bool = True)

Copies a list of cores (or all cores) from one matrix file to another one

Arguments:

output_name (str): Name of the new matrix file. If none is provided, returns a copy in memory only

cores (list): List of the matrix cores to be copied

names (list, Optional): List with the new names for the cores. Defaults to current names

compress (bool, Optional): Whether you want to compress the matrix or not. Defaults to False. Not yet implemented

memory_only (bool, Optional): Whether you want to keep the matrix copy in memory only. Defaults to True

>>> from aequilibrae.matrix import AequilibraeMatrix

>>> zones_in_the_model = 3317
>>> names_list = ['Car trips', 'pt trips', 'DRT trips', 'bike trips', 'walk trips']

>>> mat = AequilibraeMatrix()
>>> mat.create_empty(zones=zones_in_the_model,
...                  matrix_names=names_list)

>>> mat.copy(Path(my_folder_path) / 'copy_of_my_matrix.aem',
...          cores=['bike trips', 'walk trips'],
...          names=['bicycle', 'walking'],
...          memory_only=False)
<aequilibrae.matrix.aequilibrae_matrix.AequilibraeMatrix object at 0x...>

>>> mat2 = AequilibraeMatrix()
>>> mat2.load(Path(my_folder_path) / 'copy_of_my_matrix.aem')
>>> mat2.cores
2

create_empty(file_name: Path = None, zones: int = None, matrix_names: List[str] = None, data_type: dtype = <class 'numpy.float64'>, index_names: List[str] = None, compressed: bool = False, memory_only: bool = True)

Creates an empty matrix in the AequilibraE format

Arguments:

file_name (Path): Local path to the matrix file

zones (int): Number of zones in the model (Integer). Maximum number of zones in a matrix is 4,294,967,296

matrix_names (list): A regular Python list of names of the matrix. Limit is 50 characters each. Maximum number of cores per matrix is 256

data_type (np.dtype, Optional): Data type of the matrix as NUMPY data types (np.int32, np.int64, np.float32, np.float64). Defaults to np.float64

index_names (list, Optional): A regular Python list of names for indices. Limit is 20 characters each. Maximum number of indices per matrix is 256

compressed (bool, Optional): Whether it is a flat matrix or a compressed one (Boolean - Not yet implemented)

memory_only (bool, Optional): Whether you want to keep the matrix copy in memory only. Defaults to True

>>> from aequilibrae.matrix import AequilibraeMatrix

>>> zones_in_the_model = 3317
>>> names_list = ['Car trips', 'pt trips', 'DRT trips', 'bike trips', 'walk trips']

>>> mat = AequilibraeMatrix()
>>> mat.create_empty(zones=zones_in_the_model,
...                  matrix_names=names_list)
>>> mat.num_indices
1
>>> mat.zones
3317

create_from_omx(omx_path: str, file_path: str = None, cores: List[str] = None, mappings: List[str] = None, robust: bool = True, compressed: bool = False, memory_only: bool = True) → None

Creates an AequilibraeMatrix from an original OpenMatrix

Arguments:

omx_path (Path): Path to the OMX file one wants to import

file_path (Path, Optional): Path for the output AequilibraeMatrix

cores (list, Optional): List of matrix cores to be imported

mappings (list, Optional): List of the matrix mappings (i.e. indices, centroid numbers) to be imported

robust (bool, Optional): Boolean for whether AequilibraE should try to adjust the names for cores and indices in case they are too long. Defaults to True

compressed (bool, Optional): Boolean for whether we should compress the output matrix. Not yet implemented

memory_only (bool, Optional): Whether you want to keep the matrix copy in

memory only. Defaults to True

create_from_trip_list(path_to_file: str, from_column: str, to_column: str, list_cores: List[str]) → str

Creates an AequilibraeMatrix from a trip list csv file The output is saved in the same folder as the trip list file

Arguments:

path_to_file (str): Path for the trip list csv file

from_column (str): trip list file column containing the origin zones numbers

to_column (str): trip list file column containing the destination zones numbers

list_cores (list): list of core columns in the trip list file

export(output_name: Path, cores: List[str] = None)

Exports the matrix to other formats, rather than AEM. Formats currently supported: CSV, OMX

When exporting to AEM or OMX, the user can chose to export only a set of cores, but all indices are exported

When exporting to CSV, the active index will be used, and all cores will be exported as separate columns in the output file

Arguments:

output_name (Path): Path to the output file

cores (list): Names of the cores to be exported.

>>> from aequilibrae.matrix import AequilibraeMatrix

>>> zones_in_the_model = 3317
>>> names_list = ['Car trips', 'pt trips', 'DRT trips', 'bike trips', 'walk trips']

>>> mat = AequilibraeMatrix()
>>> mat.create_empty(zones=zones_in_the_model,
...                  matrix_names=names_list)

>>> mat.export(Path(my_folder_path) / 'my_new_path.omx', ['Car trips', 'bike trips'])

get_matrix(core: str, copy=False) → ndarray

Returns the data for a matrix core

Arguments:

core (str): name of the matrix core to be returned

copy (bool, Optional): return a copy of the data. Defaults to False

Returns:

object (np.ndarray): NumPy array

is_omx()

Returns True if matrix data source is OMX, False otherwise

load(file_path: Path)

Loads matrix from disk. All cores and indices are load. First index is default.

Arguments:

file_path (str): Path to AEM or OMX file on disk

>>> from aequilibrae.matrix import AequilibraeMatrix

>>> project = create_example(project_path)

>>> mat = AequilibraeMatrix()
>>> mat.load(Path(project_path) / 'matrices/skims.omx')
>>> mat.computational_view()
>>> mat.names
['distance_blended', 'time_final']

>>> project.close()

nan_to_num()

Converts all NaN values in all cores in the computational view to zeros

>>> from aequilibrae.matrix import AequilibraeMatrix

>>> nan_matrix = np.empty((3,3))
>>> nan_matrix[:] = np.nan

>>> index = np.arange(1, 4, dtype=np.int32)

>>> mat = AequilibraeMatrix()
>>> mat.create_empty(file_name=Path(my_folder_path) / "matrices/nan_matrix.aem",
...                  zones=3,
...                  matrix_names=["only_nan"])
>>> mat.index[:] = index[:]
>>> mat.matrix["only_nan"][:, :] = nan_matrix[:, :]
>>> mat.computational_view()
>>> mat.nan_to_num()
>>> mat.get_matrix("only_nan")
array([[0., 0., 0.],
       [0., 0., 0.],
       [0., 0., 0.]])

rows() → ndarray

Returns row vector for the matrix in the computational view

Computational view needs to be set to a single matrix core

Returns:

object (np.ndarray): the row totals for the matrix currently on the computational view

>>> from aequilibrae.matrix import AequilibraeMatrix

>>> project = create_example(project_path)

>>> mat = AequilibraeMatrix()
>>> mat.load(Path(project_path) / 'matrices/skims.omx')
>>> mat.computational_view(["distance_blended"])
>>> mat.rows()
array([357.68202084, 358.68778868, 310.68285491, 275.87964738,
       265.91709918, 268.60184371, 267.32264726, 281.3793747 ,
       286.15085073, 242.60308705, 252.1776242 , 305.56774194,
       303.58100777, 270.48841269, 263.20417379, 253.92665702,
       277.1655432 , 258.84368258, 280.65697316, 272.7651157 ,
       264.06806038, 252.87533845, 273.45639965, 281.61102767])

>>> project.close()

save(names=(), file_name=None) → None

Saves matrix data back to file.

If working with AEM file, it flushes data to disk. If working with OMX, requires new names.

Arguments:

names (tuple(str), Optional): New names for the matrices. Required if working with OMX files

file_name (str, Optional): Local path to the matrix file

setDescription(matrix_description: str)

Sets description for the matrix

Arguments:

matrix_description (str): Text with matrix description. Maximum length is 144 characters

>>> from aequilibrae.matrix import AequilibraeMatrix

>>> zones_in_the_model = 3317

>>> mat = AequilibraeMatrix()
>>> mat.create_empty(file_name=Path(my_folder_path) / 'my_matrix.aem',
...                  zones=zones_in_the_model,
...                  memory_only=False)
>>> mat.setDescription('This is a text')
>>> mat.save()
>>> mat.close()

>>> mat = AequilibraeMatrix()
>>> mat.load(Path(my_folder_path) / 'my_matrix.aem')
>>> mat.description.decode('utf-8')
'This is a text'

setName(matrix_name: str)

Sets the name for the matrix itself. Only works for matrices in disk.

Arguments:

matrix_name (str): matrix name. Maximum length is 50 characters

>>> from aequilibrae.matrix import AequilibraeMatrix

>>> zones_in_the_model = 3317

>>> mat = AequilibraeMatrix()
>>> mat.create_empty(file_name=Path(my_folder_path) / 'my_matrix.omx',
...                  zones=zones_in_the_model,
...                  memory_only=False)
>>> mat.setName('This is my example')
>>> mat.save()
>>> mat.close()

set_index(index_to_set: str) → None

Sets the standard index to be the one the user wants to have be the one being used in all operations during run time. The first index is ALWAYS the default one every time the matrix is instantiated

Arguments:

index_to_set (str): Name of the index to be used. The default index name is ‘main_index’

>>> from aequilibrae.matrix import AequilibraeMatrix

>>> zones_in_the_model = 3317
>>> names_list = ['Car trips', 'pt trips', 'DRT trips', 'bike trips', 'walk trips']
>>> index_list = ['tazs', 'census']

>>> mat = AequilibraeMatrix()
>>> mat.create_empty(zones=zones_in_the_model,
...                  matrix_names=names_list,
...                  index_names=index_list )
>>> mat.num_indices
2
>>> mat.current_index
'tazs'
>>> mat.set_index('census')
>>> mat.current_index
'census'

class aequilibrae.AssignmentPaths(table_name: str, project=None)

Class for accessing path files optionally generated during assignment.

static get_path_for_destination_from_files(path_o: DataFrame, path_o_index: DataFrame, destination: int)

for a given path file and path index file, and a given destination, return the path links in o-d order

get_path_for_destination(origin: int, destination: int, iteration: int, traffic_class_id: str)

Return all link ids, i.e. the full path, for a given destination

read_path_file(origin: int, iteration: int, traffic_class_id: str) -> (<class 'pandas.DataFrame'>, <class 'pandas.DataFrame'>)

class aequilibrae.AssignmentResults

Assignment result holder for a single TrafficClass with multiple user classes

get_graph_to_network_mapping()

get_load_results() → DataFrame

Translates the assignment results from the graph format into the network format

Returns:

dataset (pd.DataFrame): Pandas DataFrame data with the traffic class assignment results

get_sl_results() → DataFrame

prepare(graph: Graph, matrix: AequilibraeMatrix) → None

Prepares the object with dimensions corresponding to the assignment matrix and graph objects

Arguments:

graph (Graph): Needs to have been set with number of centroids and list of skims (if any)

matrix (AequilibraeMatrix): Matrix properly set for computation with matrix.computational_view(:obj:`list`)

reset() → None

Resets object to prepared and pre-computation state

set_cores(cores: int) → None

Sets number of cores (threads) to be used in computation

Value of zero sets number of threads to all available in the system, while negative values indicate the number of threads to be left out of the computational effort.

Resulting number of cores will be adjusted to a minimum of zero or the maximum available in the system if the inputs result in values outside those limits

Arguments:

cores (int): Number of cores to be used in computation

total_flows() → None

Totals all link flows for this class into a single link load

Results are placed into total_link_loads class member

class aequilibrae.Graph(*args, **kwargs)

available_skims() → List[str]

Returns graph fields that are available to be set as skims.

Returns:

list (str): Skimmeable field names

compute_path(origin: int, destination: int, early_exit: bool = False, a_star: bool = False, heuristic: str | None = None)

Returns the results from path computation result holder.

Arguments:

origin (int): origin for the path

destination (int): destination for the path

early_exit (bool): stop constructing the shortest path tree once the destination is found. Doing so may cause subsequent calls to ‘update_trace’ to recompute the tree. Default is False.

a_star (bool): whether or not to use A* over Dijkstra’s algorithm. When True, ‘early_exit’ is always True. Default is False.

heuristic (str): heuristic to use if a_star is enabled. Default is None.

compute_skims(cores: int | None = None)

Returns the results from network skimming result holder.

Arguments:

cores (Union[int, None]): number of cores (threads) to be used in computation

create_compressed_link_network_mapping()

Create three arrays providing a mapping of compressed ID to link ID.

Uses sparse compression. Index ‘idx’ by the by compressed ID and compressed ID + 1, the network IDs are then in the range idx[id]:idx[id + 1].

Links not in the compressed graph are not contained within the ‘data’ array.

‘node_mapping’ provides an easy way to check if a node index is present within the compressed graph. If the value is -1 then the node has been removed, either by compression of dead end link removal. If the value is greater than or equal to 0, then that value is the compressed node index.

>>> project = create_example(project_path)

>>> project.network.build_graphs()

>>> graph = project.network.graphs['c']
>>> graph.prepare_graph(np.arange(1,25))

>>> idx, data, node_mapping = graph.create_compressed_link_network_mapping()

>>> project.close()

Returns:

idx (np.array): index array for data

data (np.array): array of link ids

node_mapping: (np.array): array of node_mapping ids

default_types(tp: str)

Returns the default integer and float types used for computation

Arguments:

tp (str): data type. ‘int’ or ‘float’

exclude_links(links: list) → None

Excludes a list of links from a graph by setting their B node equal to their A node

Arguments:

links (list): List of link IDs to be excluded from the graph

load_from_disk(filename: str) → None

Loads graph from disk

Arguments:

filename (str): Path to file

prepare_graph(centroids: ndarray | None = None, remove_dead_ends: bool = True) → None

Prepares the graph for a computation for a certain set of centroids.

Under the hood, if sets all centroids to have IDs from 1 through n, which should correspond to the index of the matrix being assigned.

This is what enables having any node IDs as centroids, and it relies on the inference that all links connected to these nodes are centroid connectors.

Arguments:

centroids (np.ndarray or None, optional): Array with centroid IDs. Mandatory type Int64, unique and positive.

remove_dead_ends (bool, optional): Whether or not to remove dead ends from the graph. Defaults to True.

reverse()

save_compressed_correspondence(path, mode_name, mode_id)

Save graph and nodes_to_indices to disk

save_to_disk(filename: str) → None

Saves graph to disk

Arguments:

filename (str): Path to file. Usual file extension is aeg.

set_blocked_centroid_flows(block_centroid_flows) → None

Chooses whether we want to block paths to go through centroids or not. Default value is True.

Arguments:

block_centroid_flows (bool): Blocking or not paths to go through centroids.

set_graph(cost_field) → None

Sets the field to be used for path computation

Arguments:

cost_field (str): Field name. Must be numeric

set_skimming(skim_fields: list) → None

Sets the list of skims to be computed

Skimming with A* may produce results that differ from traditional Dijkstra’s due to its use a heuristic.

Arguments:

skim_fields (list): Fields must be numeric

class aequilibrae.GravityApplication(project=None, **kwargs)

Applies a synthetic gravity model.

Model is an instance of SyntheticGravityModel class.

Impedance is an instance of AequilibraEMatrix.

Vectors are a pandas DataFrame.

>>> import pandas as pd
>>> from aequilibrae.distribution import SyntheticGravityModel, GravityApplication

>>> project = create_example(project_path)

# We define the model we will use
>>> model = SyntheticGravityModel()

# Before adding a parameter to the model, you need to define the model functional form
# You can select one of GAMMA, EXPO or POWER.
>>> model.function = "GAMMA"

# Only the parameter(s) applicable to the chosen functional form will have any effect
>>> model.alpha = 0.1
>>> model.beta = 0.0001

# We load the impedance matrix
>>> matrix = project.matrices.get_matrix("skims")
>>> matrix.computational_view(["distance_blended"])

# We create the vectors we will use
>>> query = "SELECT zone_id, population, employment FROM zones;"
>>> with project.db_connection as conn:
...     df = pd.read_sql(query, conn)
>>> df.sort_values(by="zone_id", inplace=True)
>>> df.set_index("zone_id", inplace=True)

# You create the vectors you would have
>>> df = df.assign(productions=df.population * 3.0)
>>> df = df.assign(attractions=df.employment * 4.0)
>>> vectors = df[["productions", "attractions"]]

# Balance the vectors
>>> vectors.loc[:, "attractions"] *= vectors["productions"].sum() / vectors["attractions"].sum()

# Create the problem object
>>> args = {"impedance": matrix,
...         "vectors": vectors,
...         "row_field": "productions",
...         "model": model,
...         "column_field": "attractions",
...         "nan_as_zero":True
...         }
>>> gravity = GravityApplication(**args)

# Solve and save the outputs
>>> gravity.apply()
>>> gravity.output.export(project_path / 'matrices' / 'gravity_omx.omx')

>>> project.close()

apply()

Runs the Gravity Application instance as instantiated

Resulting matrix is the output class member

save_to_project(name: str, file_name: str, project=None) → None

Saves the matrix output to the project file

Arguments:

name (str): Name of the desired matrix record

file_name (str): Name for the matrix file name. AEM and OMX supported

project (Project, Optional): Project we want to save the results to. Defaults to the active project

class aequilibrae.GravityCalibration(project=None, **kwargs)

Calibrate a traditional gravity model

Available deterrence function forms are: ‘EXPO’, ‘POWER’ or ‘GAMMA’.

>>> from aequilibrae.distribution import GravityCalibration

>>> project = create_example(project_path)

# We load the demand matrix
>>> demand = project.matrices.get_matrix("demand_omx")
>>> demand.computational_view()

# We load the skim matrix
>>> skim = project.matrices.get_matrix("skims")
>>> skim.computational_view(["time_final"])

>>> args = {"matrix": demand,
...         "impedance": skim,
...         "row_field": "productions",
...         "function": 'expo',
...         "nan_as_zero": True}
>>> gravity = GravityCalibration(**args)

# Solve and save outputs
>>> gravity.calibrate()
>>> gravity.model.save(project_path / 'dist_expo_model.mod')

>>> project.close()

calibrate()

Calibrate the model

Resulting model is in output class member

class aequilibrae.Ipf(project=None, **kwargs)

Iterative proportional fitting procedure

>>> from aequilibrae.distribution import Ipf

>>> project = create_example(project_path)

>>> matrix = project.matrices.get_matrix("demand_omx")
>>> matrix.computational_view()

>>> vectors = pd.DataFrame(
...     {"productions": np.zeros(matrix.zones), "attractions": np.zeros(matrix.zones)},
...     index=matrix.index,
... )

>>> vectors["productions"] = matrix.rows()
>>> vectors["attractions"] = matrix.columns()

>>> ipf_args = {"matrix": matrix,
...             "vectors": vectors,
...             "row_field": "productions",
...             "column_field": "attractions",
...             "nan_as_zero": False}

>>> fratar = Ipf(**ipf_args)
>>> fratar.fit()

# We can get back to our OMX matrix in the end
>>> fratar.output.export(Path(my_folder_path) / "to_omx_output.omx")

>>> project.close()

fit()

Runs the IPF instance problem to adjust the matrix

Resulting matrix is the output class member

save_to_project(name: str, file_name: str, project=None) → MatrixRecord

Saves the matrix output to the project file

Arguments:

name (str): Name of the desired matrix record

file_name (str): Name for the matrix file name. AEM and OMX supported

project (Project, Optional): Project we want to save the results to. Defaults to the active project

class aequilibrae.Log(project_base_path: Path)

API entry point to the log file contents

>>> project = Project()
>>> project.new(project_path)

>>> log = project.log()

# We get all entries for the log file
>>> entries = log.contents()

# Or clear everything (NO UN-DOs)
>>> log.clear()

>>> project.close()

clear()

Clears the log file. Use it wisely

contents() → list

Returns contents of log file

Returns:

log_contents (list): List with all entries in the log file

class aequilibrae.Matrices(project)

Gateway into the matrices available/recorded in the model

check_exists(name: str) → bool

Checks whether a matrix with a given name exists

Returns:

exists (bool): Does the matrix exist?

clear_database() → None

Removes records from the matrices database that do not exist in disk

delete_record(matrix_name: str) → None

Deletes a Matrix Record from the model and attempts to remove from disk

get_matrix(matrix_name: str) → AequilibraeMatrix

Returns an AequilibraE matrix available in the project

Raises an error if matrix does not exist

Arguments:

matrix_name (str): Name of the matrix to be loaded

Returns:

matrix (AequilibraeMatrix): Matrix object

get_record(matrix_name: str) → MatrixRecord

Returns a model Matrix Record for manipulation in memory

list() → DataFrame

List of all matrices available

Returns:

df (pd.DataFrame): Pandas DataFrame listing all matrices available in the model

new_record(name: str, file_name: str, matrix=None) → MatrixRecord

Creates a new record for a matrix in disk, but does not save it

If the matrix file is not already on disk, it will fail

Arguments:

name (str): Name of the matrix

file_name (str): Name of the file on disk

Returns:

matrix_record (MatrixRecord): A matrix record that can be manipulated in memory before saving

reload()

Discards all memory matrices in memory and loads recreate them

update_database() → None

Adds records to the matrices database for matrix files found on disk

class aequilibrae.NetworkSkimming(graph, origins=None, project=None)

>>> from aequilibrae.paths.network_skimming import NetworkSkimming

>>> project = create_example(project_path)
>>> project.network.build_graphs(modes=["c"])

>>> graph = project.network.graphs['c']
>>> graph.set_graph("distance")
>>> graph.set_skimming("distance")

>>> skm = NetworkSkimming(graph)
>>> skm.execute()

# The skim report (if any error generated) is available here
>>> skm.report
[]

# To access the skim matrix directly from its temporary file
>>> matrix = skm.results.skims

# Or you can save the results to disk
>>> skm.save_to_project('skimming_result_omx', 'omx')

>>> project.close()

doWork()

execute()

Runs the skimming process as specified in the graph

save_to_project(name: str, format='omx', project=None) → None

Saves skim results to the project folder and creates record in the database

Arguments:

name (str): Name of the matrix. Same value for matrix record name and file (plus extension)

format (str, Optional): File format (‘aem’ or ‘omx’). Default is ‘omx’

project (Project, Optional): Project we want to save the results to. Defaults to the active project

set_cores(cores: int) → None

Sets number of cores (threads) to be used in computation

Value of zero sets number of threads to all available in the system, while negative values indicate the number of threads to be left out of the computational effort.

Resulting number of cores will be adjusted to a minimum of zero or the maximum available in the system if the inputs result in values outside those limits

Arguments:

cores (int): Number of cores to be used in computation

signal = <aequilibrae.utils.python_signal.PythonSignal object>

class aequilibrae.Parameters(path: Path | None = None)

Global parameters module.

Parameters are used in many procedures, and are often defined in the parameters.yml file ONLY.

Parameters are organized in the following groups:

• assignment

• distribution

• network * links * modes * nodes * osm * gmns

• osm

• system

Please observe that OSM information handled on network is not the same on the OSM group.

>>> from aequilibrae.parameters import Parameters

>>> project = Project()
>>> project.new(project_path)

>>> p = Parameters()

>>> p.parameters['system']['logging_directory'] =  "/path_to/other_logging_directory"
>>> p.parameters['osm']['overpass_endpoint'] = "http://192.168.0.110:32780/api"
>>> p.parameters['osm']['max_query_area_size'] = 10000000000
>>> p.parameters['osm']['sleeptime'] = 0
>>> p.write_back()

>>> # You can also restore the software default values
>>> p.restore_default()

>>> project.close()

classmethod load_default()

restore_default()

Restores parameters to generic default

write_back()

Writes the parameters back to file

file_default: Path = PosixPath('/home/runner/work/aequilibrae/aequilibrae/aequilibrae/parameters.yml')

class aequilibrae.PathResults

Path computation result holder

>>> from aequilibrae.paths.results import PathResults

>>> project = create_example(project_path)
>>> project.network.build_graphs()

# Mode c is car in this project
>>> car_graph = project.network.graphs['c']

# minimize distance
>>> car_graph.set_graph('distance')

# If you want to compute skims
# It does increase path computation time substantially
>>> car_graph.set_skimming(['distance', 'free_flow_time'])

>>> res = PathResults()
>>> res.prepare(car_graph)
>>> res.compute_path(1, 17)

# Update all the outputs mentioned above for destination 9. Same origin: 1
>>> res.update_trace(9)

# clears all computation results
>>> res.reset()

>>> project.close()

compute_path(origin: int, destination: int, early_exit: bool = False, a_star: bool = False, heuristic: str | None = None) → None

Computes the path between two nodes in the network.

A* heuristics are currently only valid distance cost fields.

Arguments:

origin (int): Origin for the path

destination (int): Destination for the path

early_exit (bool): Stop constructing the shortest path tree once the destination is found. Doing so may cause subsequent calls to update_trace to recompute the tree. Default is False.

a_star (bool): Whether or not to use A* over Dijkstra’s algorithm. When True, early_exit is always True. Default is False.

heuristic (str): Heuristic to use if a_star is enabled. Default is None.

get_heuristics() → List[str]

Return the available heuristics.

prepare(graph: Graph) → None

Prepares the object with dimensions corresponding to the graph object

Arguments:

graph (Graph): Needs to have been set with number of centroids and list of skims (if any)

reset() → None

Resets object to prepared and pre-computation state

set_heuristic(heuristic: str) → None

Set the heuristics to be used in A*. Must be one of get_heuristics().

Arguments:

heuristic (str): Heuristic to use in A*.

update_trace(destination: int) → None

Updates the path’s nodes, links, skims and mileposts

If the previously computed path had early_exit enabled, update_trace will check if the destination has already been found, if not the shortest path tree will be recomputed with the early_exit argument passed on.

If the previously computed path had a_star enabled, update_trace always recompute the path.

Arguments:

destination (int): ID of the node we are computing the path too

class aequilibrae.Project

AequilibraE project class

Create Project

>>> new_project = Project()
>>> new_project.new(project_path)

# Safely closes the project
>>> new_project.close()

Open Project

>>> existing_project = Project()
>>> existing_project.open(project_path)

>>> existing_project.close()

classmethod from_path(project_folder)

activate()

check_file_indices() → None

Makes results_database.sqlite and the matrices folder compatible with project database

clone_scenario(scenario_name: str, description: str = '')

Clones the active scenario.

Arguments:

scenario_name (str): scenario name

description (str): useful scenario description

close() → None

Safely closes the project

create_empty_scenario(scenario_name: str, description: str = '')

Creates an empty scenario, without any links, nodes, and zones.

Arguments:

scenario_name (str): scenario name

description (str): useful scenario description

deactivate()

list_scenarios()

Lists the existing scenarios.

Returns:

scenarios (pd.DataFrame): Pandas DataFrame with existing scenarios

log() → Log

Returns a log object

allows the user to read the log or clear it

new(project_path: str) → None

Creates a new project

Arguments:

project_path (str): Full path to the project data folder. If folder exists, it will fail

open(project_path: str) → None

Loads project from disk

Arguments:

project_path (str): Full path to the project data folder. If the project inside does not exist, it will fail.

upgrade(ignore_project: bool = False, ignore_transit: bool = False, ignore_results: bool = False)

Find and apply all applicable migrations.

All database upgrades are applied within a single transaction.

Optionally ignore specific databases. This is useful when a database is known to be incompatible with some migrations but you’d still like to upgrade the others. Take care when ignoring a database. For a particular version of aequilibrae, it is assumed that all migrations have been applied successfully or the project was created with the latest schema, skipping/ignoring migrations will likely lead to issues/broken assumptions.

If skipping a specific migration is required, use the aequilibrae.project.tools.MigrationManager object directly. Consult it’s documentation page for details. Take care when skipping migrations.

Arguments:

ignore_project (bool, optional): Ignore the project database. No direct migrations will be

applied. Defaults to False.

ignore_transit (bool, optional): Ignore the transit database. No direct migrations will be

applied. Defaults to False.

ignore_results (bool, optional): Ignore the results database. No direct migrations will be

applied. Defaults to False.

use_scenario(scenario_name: str)

Switch the active scenario.

Arguments:

scenario_name (str): name of the scenario to be activated

property about: About

property db_connection

property db_connection_spatial

property logger: Logger

property matrices: Matrices

property network: Network

property parameters: dict

property path_to_file

property project_base_path

property project_parameters: Parameters

property results: Results

property results_connection

property run

Load and return the AequilibraE run module with the default arguments from parameters.yml partially applied.

Refer to run/__init__.py file within the project folder for documentation.

property transit: Transit

property transit_connection

property zoning

class aequilibrae.SkimResults

Network skimming result holder.

>>> from aequilibrae.paths.results import SkimResults

>>> project = create_example(project_path)
>>> project.network.build_graphs()

# Mode c is car in this project
>>> car_graph = project.network.graphs['c']

# minimize travel time
>>> car_graph.set_graph('free_flow_time')

# Skims travel time and distance
>>> car_graph.set_skimming(['free_flow_time', 'distance'])

>>> res = SkimResults()
>>> res.prepare(car_graph)

>>> res.skims.export(project_path / "skim_matrices.omx")

>>> project.close()

prepare(graph: Graph)

Prepares the object with dimensions corresponding to the graph objects

Arguments:

graph (Graph): Needs to have been set with number of centroids and list of skims (if any)

class aequilibrae.SyntheticGravityModel

Simple class object to represent synthetic gravity models

load(file_name)

Loads model from disk. Extension is *.mod

save(file_name)

Saves model to disk in yaml format. Extension is *.mod

class aequilibrae.TrafficAssignment(project=None)

Traffic assignment class.

For a comprehensive example on use, see the Use examples page.

>>> from aequilibrae.paths import TrafficAssignment, TrafficClass

>>> project = create_example(project_path)
>>> project.network.build_graphs()

>>> graph = project.network.graphs['c'] # we grab the graph for cars
>>> graph.set_graph('free_flow_time') # let's say we want to minimize time
>>> graph.set_skimming(['free_flow_time', 'distance']) # And will skim time and distance
>>> graph.set_blocked_centroid_flows(True)

>>> proj_matrices = project.matrices

>>> demand = proj_matrices.get_matrix("demand_omx")

# We will only assign one user class stored as 'matrix' inside the OMX file
>>> demand.computational_view(['matrix'])

# Creates the assignment class
>>> assigclass = TrafficClass("car", graph, demand)

>>> assig = TrafficAssignment()

# The first thing to do is to add at list of traffic classes to be assigned
>>> assig.set_classes([assigclass])

# Then we set the volume delay function
>>> assig.set_vdf("BPR")  # This is not case-sensitive

# And its parameters
>>> assig.set_vdf_parameters({"alpha": "b", "beta": "power"})

# The capacity and free flow travel times as they exist in the graph
>>> assig.set_capacity_field("capacity")
>>> assig.set_time_field("free_flow_time")

# And the algorithm we want to use to assign
>>> assig.set_algorithm('bfw')

>>> assig.max_iter = 10
>>> assig.rgap_target = 0.00001

>>> assig.execute() # we then execute the assignment

# If you want, it is possible to access the convergence report
>>> convergence_report = pd.DataFrame(assig.assignment.convergence_report)

# Assignment results can be viewed as a Pandas DataFrame
>>> results_df = assig.results()

# Information on the assignment setup can be recovered with
>>> info = assig.info()

# Or save it directly to the results database
>>> results = assig.save_results(table_name='base_year_assignment')

# skims are here
>>> avg_skims = assigclass.results.skims # blended ones
>>> last_skims = assigclass._aon_results.skims # those for the last iteration

>>> project.close()

add_class(traffic_class: TrafficClass) → None

Adds a traffic class to the assignment

Arguments:

traffic_class (TrafficClass): Traffic class

add_preload(preload: DataFrame, name: str = None) → None

Given a dataframe of ‘link_id’, ‘direction’ and ‘preload’, merge into current preloads dataframe.

Arguments:

preload (pd.DataFrame): dataframe mapping ‘link_id’ & ‘direction’ to ‘preload’ name (str): Name for particular preload (optional - default name will be chosen if not specified)

algorithms_available() → list

Returns all algorithms available for use

Returns:

list: List of string values to be used with set_algorithm

execute(log_specification=True) → None

Processes assignment

get_skim_results() → list

Prepares the assignment skim results for all classes

Returns:

skim list (list): Lists of all skims with the results for each class

info() → dict

Returns information for the traffic assignment procedure

Dictionary contains keys ‘Algorithm’, ‘Classes’, ‘Computer name’, ‘Procedure ID’, ‘Maximum iterations’ and ‘Target RGap’.

The classes key is also a dictionary with all the user classes per traffic class and their respective matrix totals

Returns:

info (dict): Dictionary with summary information

log_specification()

report() → DataFrame

Returns the assignment convergence report

Returns:

DataFrame (pd.DataFrame): Convergence report

results() → DataFrame

Prepares the assignment results as a Pandas DataFrame

Returns:

DataFrame (pd.DataFrame): Pandas DataFrame with all the assignment results indexed on link_id

save_results(table_name: str, keep_zero_flows=True, project=None) → None

Saves the assignment results to results_database.sqlite

Method fails if table exists

Arguments:

table_name (str): Name of the table to hold this assignment result

keep_zero_flows (bool): Whether we should keep records for zero flows. Defaults to True

project (Project, Optional): Project we want to save the results to. Defaults to the active project

save_select_link_flows(table_name: str, project=None) → None

Saves the select link link flows for all classes into the results database.

Arguments:

table_name (str): Name of the table being inserted to. Note the traffic class

project (Project, Optional): Project we want to save the results to. Defaults to the active project

save_select_link_matrices(matrix_name: str, project=None) → None

Saves the Select Link matrices for each TrafficClass in the current TrafficAssignment class into OMX format.

Arguments:

name (str): name of the matrices

project (Project, Optional): Project we want to save the results to. Defaults to the active project

save_select_link_results(name: str) → None

Saves both the Select Link matrices and flow results at the same time, using the same name.

Arguments:

name (str): name of the matrices

save_skims(matrix_name: str, which_ones='final', format='omx', project=None) → None

Saves the skims (if any) to the skim folder and registers in the matrix list

Arguments:

name (str): Name of the matrix record to hold this matrix (same name used for file name)

which_ones (str, Optional): ‘final’: Results of the final iteration,

‘blended’: Averaged results for all iterations, ‘all’: Saves skims for both the final iteration and the blended ones. Default is ‘final’

format (str, Optional): File format (‘aem’ or ‘omx’). Default is ‘omx’

project (Project, Optional): Project we want to save the results to.

Defaults to the active project

select_link_flows() → Dict[str, DataFrame]

Returns a dataframe of the select link flows for each class

set_algorithm(algorithm: str)

Chooses the assignment algorithm. e.g. ‘frank-wolfe’, ‘bfw’, ‘msa’

‘fw’ is also accepted as an alternative to ‘frank-wolfe’

Arguments:

algorithm (str): Algorithm to be used

set_capacity_field(capacity_field: str) → None

Sets the graph field that contains link capacity for the assignment period -> e.g. ‘capacity1h’

Arguments:

capacity_field (str): Field name

set_classes(classes: List[TrafficClass]) → None

Sets Traffic classes to be assigned

Arguments:

classes (List[TrafficClass]): List of Traffic classes for assignment

set_cores(cores: int) → None

Allows one to set the number of cores to be used AFTER traffic classes have been added

Inherited from AssignmentResultsBase

Arguments:

cores (int): Number of CPU cores to use

set_path_file_format(file_format: str) → None

Specify path saving format. Either parquet or feather.

Arguments:

file_format (str): Name of file format to use for path files

set_save_path_files(save_it: bool) → None

Turn path saving on or off.

Arguments:

save_it (bool): Boolean to indicate whether paths should be saved

set_time_field(time_field: str) → None

Sets the graph field that contains free flow travel time -> e.g. ‘fftime’

Arguments:

time_field (str): Field name

set_vdf(vdf_function: str) → None

Sets the Volume-delay function to be used

Arguments:

vdf_function (str): Name of the VDF to be used

set_vdf_parameters(par: dict) → None

Sets the parameters for the Volume-delay function.

Parameter values can be scalars (same values for the entire network) or network field names (link-specific values) - Examples: {‘alpha’: 0.15, ‘beta’: 4.0} or {‘alpha’: ‘alpha’, ‘beta’: ‘beta’}

The Akcelik VDF parameter ‘tau’ value has typical 8 factor absorbed into it. Users should supply 8 * tau to match other common usages. Additionally the standard 0.25 factor can be overridden by supplying the ‘alpha’ parameter.

Arguments:

par (dict): Dictionary with all parameters for the chosen VDF

skim_congested(skim_fields=None, return_matrices=False) → dict | None

Skims the congested network. The user can add a list of skims to be computed, which will be added to the congested time and the assignment cost from the last iteration of the assignment.

The matrices are always stored internally in the AequilibraE objects to be saved to the project if needed. If return_matrices is set to True, the matrices are also returned.

Arguments:

skim_fields (Union[None, str]): Name of the skims to use. If None, uses default only return_matrices (Bool): Returns a dictionary with skims. Defaults to False.

all_algorithms = ['all-or-nothing', 'msa', 'frank-wolfe', 'fw', 'cfw', 'bfw']

bpr_parameters = ['alpha', 'beta']

class aequilibrae.TrafficClass(name: str, graph: Graph, matrix: AequilibraeMatrix)

Traffic class for equilibrium traffic assignment

>>> from aequilibrae.paths import TrafficClass

>>> project = create_example(project_path)
>>> project.network.build_graphs()

>>> graph = project.network.graphs['c'] # we grab the graph for cars
>>> graph.set_graph('free_flow_time') # let's say we want to minimize time
>>> graph.set_skimming(['free_flow_time', 'distance']) # And will skim time and distance
>>> graph.set_blocked_centroid_flows(False)

>>> proj_matrices = project.matrices

>>> demand = proj_matrices.get_matrix("demand_omx")
>>> demand.computational_view()

>>> tc = TrafficClass("car", graph, demand)
>>> tc.set_pce(1.3)

>>> project.close()

set_fixed_cost(field_name: str, multiplier=1)

Sets value of time

Arguments:

field_name (str): Name of the graph field with fixed costs for this class

multiplier (Union[float, int]): Multiplier for the fixed cost. Defaults to 1 if not set

set_pce(pce: float | int) → None

Sets Passenger Car equivalent

Arguments:

pce (Union[float, int]): PCE. Defaults to 1 if not set

set_select_links(links: Dict[str, List[Tuple[int, int]]])

Set the selected links. Checks if the links and directions are valid. Translates link_id and direction into unique link id used in compact graph. Supply links=None to disable select link analysis.

Arguments:

links (Union[None, Dict[str, List[Tuple[int, int]]]]): name of link set and Link IDs and directions to be used in select link analysis

set_vot(value_of_time: float) → None

Sets value of time

Arguments:

value_of_time (Union[float, int]): Value of time. Defaults to 1 if not set

skim_congested(skim_fields=None)

Skims the congested network. The user can add a list of skims to be computed, which will be added to the congested time and the assignment cost from the last iteration of the assignment.

Arguments:

skim_fields (Union[None, str]): Name of the skims to use. If None, uses default only

property info: dict

class aequilibrae.VDF

Volume-Delay function

>>> from aequilibrae.paths import VDF

>>> vdf = VDF()
>>> vdf.functions_available()
['bpr', 'bpr2', 'conical', 'inrets', 'akcelik']

functions_available() → list

returns a list of all functions available

class aequilibrae.allOrNothing(class_name: str, matrix: AequilibraeMatrix, graph: Graph, results: AssignmentResults)

doWork()

execute()

func_assig_thread(origin, all_threads)

signal = <aequilibrae.utils.python_signal.PythonSignal object>

Modules

context
distribution
log
matrix
parameters
paths
project
reference_files
transit
utils