aequilibrae.distribution package¶
Submodules¶
aequilibrae.distribution.gravity_application module¶
- class aequilibrae.distribution.gravity_application.GravityApplication(**kwargs)¶
Bases:
object
Applies a synthetic gravity model
Model is an instance of SyntheticGravityModel class Impedance is an instance of AequilibraEMatrix Row and Column vectors are instances of AequilibraeData
import pandas as pd import sqlite3 from aequilibrae.matrix import AequilibraeMatrix from aequilibrae.matrix import AequilibraeData from aequilibrae.distribution import SyntheticGravityModel from aequilibrae.distribution import GravityApplication # We define the model we will use model = SyntheticGravityModel() # Before adding a parameter to the model, you need to define the model functional form model.function = "GAMMA" # "EXPO" or "POWER" # Only the parameter(s) applicable to the chosen functional form will have any effect model.alpha = 0.1 model.beta = 0.0001 # Or you can load the model from a file model.load('path/to/model/file') # We load the impedance matrix matrix = AequilibraeMatrix() matrix.load('path/to/impedance_matrix.aem') matrix.computational_view(['distance']) # We create the vectors we will use conn = sqlite3.connect('path/to/demographics/database') query = "SELECT zone_id, population, employment FROM demographics;" df = pd.read_sql_query(query,conn) index = df.zone_id.values[:] zones = index.shape[0] # You create the vectors you would have df = df.assign(production=df.population * 3.0) df = df.assign(attraction=df.employment * 4.0) # We create the vector database args = {"entries": zones, "field_names": ["productions", "attractions"], "data_types": [np.float64, np.float64], "memory_mode": True} vectors = AequilibraeData() vectors.create_empty(**args) # Assign the data to the vector object vectors.productions[:] = df.production.values[:] vectors.attractions[:] = df.attraction.values[:] vectors.index[:] = zones[:] # Balance the vectors vectors.attractions[:] *= vectors.productions.sum() / vectors.attractions.sum() # Create the problem object args = {"impedance": matrix, "rows": vectors, "row_field": "productions", "model": model, "columns": vectors, "column_field": "attractions", "output": 'path/to/output/matrix.aem', "nan_as_zero":True } gravity = GravityApplication(**args) # Solve and save the outputs gravity.apply() gravity.output.export('path/to/omx_file.omx') with open('path.to/report.txt', 'w') as f: for line in gravity.report: f.write(f'{line}
‘)
- __init__(**kwargs)¶
Instantiates the Ipf problem
- Args:
model (
SyntheticGravityModel
): Synthetic gravity model to applyimpedance (
AequilibraeMatrix
): Impedance matrix to be usedrows (
AequilibraeData
): Vector object with data for row totalsrow_field (
str
): Field name that contains the data for the row totalscolumns (
AequilibraeData
): Vector object with data for column totalscolumn_field (
str
): Field name that contains the data for the column totalscore_name (
str
, optional): Name for the output matrix core. Defaults to “gravity”parameters (
str
, optional): Convergence parameters. Defaults to those in the parameter filenan_as_zero (
bool
, optional): If Nan values should be treated as zero. Defaults to True- Results:
output (
AequilibraeMatrix
): Result Matrixreport (
list
): Iteration and convergence reporterror (
str
): Error description
- apply()¶
Runs the Gravity Application instance as instantiated
Resulting matrix is the output class member
- save_to_project(name: str, file_name: str) None ¶
Saves the matrix output to the project file
- Args:
name (
str
): Name of the desired matrix record file_name (str
): Name for the matrix file name. AEM and OMX supported
aequilibrae.distribution.gravity_calibration module¶
Algorithms to calibrate synthetic gravity models with power and exponential functions
The procedures implemented in this code are some of those suggested in Modelling Transport, 4th Edition, Ortuzar and Willumsen, Wiley 2011
- class aequilibrae.distribution.gravity_calibration.GravityCalibration(**kwargs)¶
Bases:
object
Calibrate a traditional gravity model
Available deterrence function forms are: ‘EXPO’ or ‘POWER’. ‘GAMMA’
from aequilibrae.matrix import AequilibraeMatrix from aequilibrae.distribution import GravityCalibration # We load the impedance matrix matrix = AequilibraeMatrix() matrix.load('path/to/trip_matrix.aem') matrix.computational_view(['total_trips']) # We load the impedance matrix impedmatrix = AequilibraeMatrix() impedmatrix.load('path/to/impedance_matrix.aem') impedmatrix.computational_view(['traveltime']) # Creates the problem args = {"matrix": matrix, "impedance": impedmatrix, "row_field": "productions", "function": 'expo', "nan_as_zero":True } gravity = GravityCalibration(**args) # Solve and save outputs gravity.calibrate() gravity.model.save('path/to/dist_expo_model.mod') with open('path.to/report.txt', 'w') as f: for line in gravity.report: f.write(f'{line}\n')
- __init__(**kwargs)¶
Instantiates the Gravity calibration problem
- Args:
matrix (
AequilibraeMatrix
): Seed/base trip matriximpedance (
AequilibraeMatrix
): Impedance matrix to be usedfunction (
str
): Function name to be calibrated. “EXPO” or “POWER”parameters (
str
, optional): Convergence parameters. Defaults to those in the parameter filenan_as_zero (
bool
, optional): If Nan values should be treated as zero. Defaults to True- Results:
model (
SyntheticGravityModel
): Calibrated modelreport (
list
): Iteration and convergence reporterror (
str
): Error description
- calibrate()¶
Calibrate the model
Resulting model is in output class member
aequilibrae.distribution.ipf module¶
- class aequilibrae.distribution.ipf.Ipf(**kwargs)¶
Bases:
object
Iterative proportional fitting procedure
import pandas as pd from aequilibrae.distribution import Ipf from aequilibrae.matrix import AequilibraeMatrix from aequilibrae.matrix import AequilibraeData matrix = AequilibraeMatrix() # Here we can create from OMX or load from an AequilibraE matrix. matrix.create_from_omx(path/to/aequilibrae_matrix, path/to/omxfile) # The matrix will be operated one (see the note on overwriting), so it does # not make sense load an OMX matrix source_vectors = pd.read_csv(path/to/CSVs) zones = source_vectors.zone.shape[0] args = {"entries": zones, "field_names": ["productions", "attractions"], "data_types": [np.float64, np.float64], "memory_mode": True} vectors = AequilibraEData() vectors.create_empty(**args) vectors.productions[:] = source_vectors.productions[:] vectors.attractions[:] = source_vectors.attractions[:] # We assume that the indices would be sorted and that they would match the matrix indices vectors.index[:] = source_vectors.zones[:] args = { "matrix": matrix, "rows": vectors, "row_field": "productions", "columns": vectors, "column_field": "attractions", "nan_as_zero": False} fratar = Ipf(**args) fratar.fit() # We can get back to our OMX matrix in the end fratar.output.export(path/to_omx/output.omx) fratar.output.export(path/to_aem/output.aem)
- __init__(**kwargs)¶
Instantiates the Ipf problem
- Args:
matrix (
AequilibraeMatrix
): Seed Matrixrows (
AequilibraeData
): Vector object with data for row totalsrow_field (
str
): Field name that contains the data for the row totalscolumns (
AequilibraeData
): Vector object with data for column totalscolumn_field (
str
): Field name that contains the data for the column totalsparameters (
str
, optional): Convergence parameters. Defaults to those in the parameter filenan_as_zero (
bool
, optional): If Nan values should be treated as zero. Defaults to True- Results:
output (
AequilibraeMatrix
): Result Matrixreport (
list
): Iteration and convergence reporterror (
str
): Error description
- 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) aequilibrae.project.data.matrix_record.MatrixRecord ¶
Saves the matrix output to the project file
- Args:
name (
str
): Name of the desired matrix record file_name (str
): Name for the matrix file name. AEM and OMX supported