traffic_demand_calculation#

common#

class ArgDefaultDict#: Bases: defaultdict

class Contributor#: Bases: object

class DemandEstimation(global_params: Sequence[GlobalContributor] = (), local_params: Sequence[LocalContributor] = ())#

Bases: object

close()#

initialize(mapper: LocalMapper)#

setup(*, state: TrackedState, settings: Settings, schema: AttributeSchema, logger: Logger)#

update(matrix: ndarray, force_update: bool = False, *, moment: Moment) → ndarray#

class GlobalContributor(parameter: str, csv_tape: CsvTape)#

Bases: Contributor

get_value()#

has_changes()#

update_factor(factor: float, *, moment: Moment) → float#

class LocalContributor#

Bases: Contributor

close()#

has_changes()#

initialize(mapper: LocalMapper)#

setup(*, state: TrackedState, settings: Settings, schema: AttributeSchema, logger: Logger)#

update_demand(matrix: ndarray, force_update: bool = False, *, moment: Moment)#

class LocalMapper(demand_geometry: Geometry)#

Bases: object

get_nearest(target_entities) → ndarray#: The resulting array length matches the demand entities, the values match indices in the target entity group

global_contributors#

class GlobalElasticityParameter(parameter: str, csv_tape: CsvTape, elasticity: float)#

Bases: GlobalContributor

Formulation: F_ij = (GP_n / GP_(n-1))_i ** (eta) * (GP_n / GP_(n-1))_j ** (eta): F_ij: Multiplication factor for demand from node i to node j GP: Global Parameter n: iteration number (eg: year) eta: elasticity

get_factor(value)#

reset_value()#

update_factor(factor: float, **_) → float#

class ScalarParameter(parameter: str, csv_tape: CsvTape)#

Bases: GlobalContributor

update_factor(factor: float, **_) → float#

local_contributors#

class Investment(seconds, entity_id, multiplier)#

Bases: NamedTuple

entity_id: int#: Alias for field number 1

multiplier: float#: Alias for field number 2

seconds: int#: Alias for field number 0

class InvestmentContributor(investments: Sequence[Investment], demand_node_index: Index)#

Bases: LocalContributor

update_demand(matrix: ndarray, force_update: bool = False, *, moment: Moment)#

class LocalEffectsContributor(info: LocalParameterInfo)#

Bases: LocalContributor

calculate_contribution(new_values, old_values)#: Calculate the contribution (P_ij/P_ij_old)**elasticity to the demand change factor

calculate_values()#: Calculate parameter values P[] so that P_ij can be reconstructed, this can be a 1d-array which together with self._indices can reconstruct P_ij (See eg. NearestValue) or a 2d-array containing every P_ij (See RouteCostFactor). P_ij will then be used to calculate the contribution (P_ij/P_ij_old)**elasticity to the demand change factor

has_changes() → bool#

old_value: ndarray | None = None#

update_demand(matrix: ndarray, force_update: bool = False, **_) → ndarray#

class LocalParameterInfo(target_dataset: str, target_entity_group: str, target_geometry: str, target_attribute: movici_simulation_core.core.attribute.UniformAttribute | movici_simulation_core.core.attribute.CSRAttribute, elasticity: float)#

Bases: object

elasticity: float#

target_attribute: UniformAttribute | CSRAttribute#

target_dataset: str#

target_entity_group: str#

target_geometry: str#

class NearestValue(info: LocalParameterInfo)#

Bases: LocalEffectsContributor

calculate_contribution(new_values, old_values)#: Calculate the contribution (P_ij/P_ij_old)**elasticity to the demand change factor

calculate_values()#: Calculate parameter values P[] so that P_ij can be reconstructed, this can be a 1d-array which together with self._indices can reconstruct P_ij (See eg. NearestValue) or a 2d-array containing every P_ij (See RouteCostFactor). P_ij will then be used to calculate the contribution (P_ij/P_ij_old)**elasticity to the demand change factor

initialize(mapper: LocalMapper)#

setup(state: TrackedState, **_)#

class RouteCostFactor(info: LocalParameterInfo)#

Bases: LocalEffectsContributor, ShortestPathMixin

This effect calculator computes the paths between pairs of demand nodes connected by a route. It calculates the sum of the given _attribute on this route and returns that.

calculate_values()#: Calculate parameter values P[] so that P_ij can be reconstructed, this can be a 1d-array which together with self._indices can reconstruct P_ij (See eg. NearestValue) or a 2d-array containing every P_ij (See RouteCostFactor). P_ij will then be used to calculate the contribution (P_ij/P_ij_old)**elasticity to the demand change factor

initialize(mapper: LocalMapper)#

setup(*, state: TrackedState, logger: Logger, **_)#

class ShortestPathMixin#

Bases: object

static deduplicate_nodes(meth)#

When mapping a demand OD matrix onto a target network, such as the effect of travel time on roads (target network) on the waterway demand (demand OD matrix). There may be duplicate mapped road virtual nodes, since this is an N -> M mapping, where M <= N. As an optimization, these duplicates can be detected by this decorator, and only the unique indices are passed to the decorated method as the first positional argument after self. The decorated method is expected to return a MxM matrix and this decorator expands it back to NxN

This only works for methods in classes that inherit from LocalEffectsContributor

initialize_network()#

setup_state(state: TrackedState, info: LocalParameterInfo)#

calculate_localized_contribution_1d(values, old_values, indices, elasticity)#

get_ratio_for_node(node_i, values, old_values, indices)#

model#

class TrafficDemandCalculation(model_config: dict)#

Bases: TrackedModel

Implementation of the demand estimation model. Reads a csv with scenario parameters. Calculates changes in the od matrices based on change of scenario parameters in time.

Asgarpour, S., Konstantinos, K., Hartmann, A., and Neef, R. (2021). Modeling interdependent infrastructures under future scenarios. Work in Progress.

auto_reset = 8#

configure_demand_nodes(state: TrackedState, schema: AttributeSchema)#

demand_estimation: DemandEstimation#

get_global_elasticity_contributors(data_handler: InitDataHandler) → List[GlobalContributor]#

static get_global_parameters_tape(data_handler: InitDataHandler, name: str) → CsvTape#

get_investment_contributors()#

get_local_contributors(state: TrackedState, schema: AttributeSchema) → List[LocalContributor]#

get_scenario_multipliers()#

initialize(state: TrackedState)#

The initialize method is called when all of the state’s INIT attribute arrays are filled with data. This may be during the model engines initialization phase or during t=0. Data that is required for the model to initialize attribute may be published in another model’s t0-update, and the TrackedModelAdapter can wait for this to happen before calling initialize. When the simulation progresses to t>0 before the model’s INIT attributes have been filled, an Exception is raised, indicating that the model was not ready yet.

Model.initialize may raise NotReady to indicate that it does not have its required input data yet. This is for example useful if a model has a number OPT`ional required attributes of which at least one must be set. The model would check whether this is the case, and raise `NotReady if it is not. Once a model has succesfully run its initialize method, this method will not be called again for the duration of the simulation.

Parameters:: state – The model’s TrackedState object, managed by the TrackedModelAdapter

new_time(state: TrackedState, moment: Moment)#: Called for every change of timestamp during a simulation run. This method is called before checking whether the state is ready for INIT or PUB and may be called before the initialize and update methods have been called the first time.

proceed_tape(moment: Moment)#

setup(state: TrackedState, settings: Settings, schema: AttributeSchema, init_data_handler: InitDataHandler, logger: Logger, **_)#

In setup, a model receives a state object, it’s config and other parameters. The goal of setup is to prepare the state by giving it information of the attributes it needs to track (by subscribing (INIT/SUB/OPT) or publishing (PUB) attributes) from which datasets. These attributes may be grouped together in EntityGroup classes or created directly. The main entry points for registering are:

state.add_dataset() for registering a bunch of EntityGroup classes for a certain dataset name at once

state.add_entity_group() for registering a single EntityGroup class (or instance) for a dataset name

state.register_attribute() for registering a single attribute in a dataset/entity_group combination

During setup there is no data available in the state. These will be downloaded automatically by the TrackedModelAdapter. However, additional datasets may be requested directly through the init_data_handler parameter.

Parameters:

state – The model’s TrackedState object, managed by the TrackedModelAdapter
settings – global settings
schema – The AttributeSchema with all registered attributes
init_data_handler – an InitDataHandler that may be used to retrieve additional datasets
logger – a logging.Logger instance

shutdown(state: TrackedState) → None#: Called when a simulation ends (either due to it being finished or one of the models raises an exception). The model may implement this method to clean up local resources. This method may be called before the initialize and update methods have been called the first time

update(state: TrackedState, moment: Moment) → Moment | None#

The update method is called for every update coming from the model engine. However it is only called the first time once all PUB attributes have their arrays filled with data. When the simulation progresses to t>0 before the model’s SUB attributes have been filled, an Exception is raised, indicating that the model was not ready yet.

Parameters:

state – The model’s TrackedState object, managed by the TrackedModelAdapter
moment – The current simulation Moment

Returns:

an optional Moment indicating the next time a model want to be woken up, as per the model engine’s protocol

convert_v1_v2(config)#

traffic_demand_calculation#

common#

global_contributors#

local_contributors#

model#

Module contents#