traffic_assignment_calculation

dataset

class CargoTrackSegmentEntity(name: str = None)

Bases: TrackSegmentEntity

attributes: Dict[str, AttributeField] = {'capacity': <movici_simulation_core.core.attribute.AttributeField object>, 'cargo_allowed': <movici_simulation_core.core.attribute.AttributeField object>, 'cargo_average_time': <movici_simulation_core.core.attribute.AttributeField object>, 'cargo_max_speed': <movici_simulation_core.core.attribute.AttributeField object>}
capacity
cargo_allowed
cargo_average_time
cargo_max_speed
property max_speed
class DemandNodeEntity(name: str = None)

Bases: PointEntity

attributes: Dict[str, AttributeField] = {'cargo_demand': <movici_simulation_core.core.attribute.AttributeField object>, 'passenger_demand': <movici_simulation_core.core.attribute.AttributeField object>}
cargo_demand
passenger_demand
class PassengerTrackSegmentEntity(name: str = None)

Bases: TrackSegmentEntity

attributes: Dict[str, AttributeField] = {'capacity': <movici_simulation_core.core.attribute.AttributeField object>, 'passenger_average_time': <movici_simulation_core.core.attribute.AttributeField object>, 'passenger_max_speed': <movici_simulation_core.core.attribute.AttributeField object>}
capacity
property max_speed
passenger_average_time
passenger_max_speed
class TrackSegmentEntity(name: str = None)

Bases: TrafficTransportSegmentEntity

attributes: Dict[str, AttributeField] = {'_max_speed': <movici_simulation_core.core.attribute.AttributeField object>, 'average_time': <movici_simulation_core.core.attribute.AttributeField object>, 'cargo_flow': <movici_simulation_core.core.attribute.AttributeField object>, 'passenger_flow': <movici_simulation_core.core.attribute.AttributeField object>}
average_time
cargo_flow
passenger_flow
class TrafficTransportSegmentEntity(name: str = None)

Bases: TransportSegmentEntity

additional_time
attributes: Dict[str, AttributeField] = {'additional_time': <movici_simulation_core.core.attribute.AttributeField object>, 'average_time': <movici_simulation_core.core.attribute.AttributeField object>, 'cargo_flow': <movici_simulation_core.core.attribute.AttributeField object>, 'delay_factor': <movici_simulation_core.core.attribute.AttributeField object>, 'passenger_car_unit': <movici_simulation_core.core.attribute.AttributeField object>, 'passenger_flow': <movici_simulation_core.core.attribute.AttributeField object>, 'volume_to_capacity': <movici_simulation_core.core.attribute.AttributeField object>}
average_time
cargo_flow
delay_factor
passenger_car_unit
passenger_flow
volume_to_capacity

model

class CargoTrackModality

Bases: TrackModality

cargo_allowed(model: Model) → UniformAttribute | None
get_capacities(model: Model)
get_demands(model: Model) → Tuple[ndarray, ndarray]
initialize_parameters(model: Model)
process_capacity(model: Model, init=False) → bool
publish_attributes: Sequence[PublishAttribute] = (PublishAttribute(name='cargo_flow', target=None, correction_value=0), PublishAttribute(name='congested_time', target='cargo_average_time', correction_value=1000000000.0))
transport_segment_entity

alias of CargoTrackSegmentEntity

transport_type: str = 'cargo_tracks'
class ModalityStrategy

Bases: object

free_flow_time(model: Model) → ndarray
get_capacities(model: Model)
get_demands(model: Model) → Tuple[ndarray, ndarray]
initialize_parameters(model: Model)
process_capacity(model: Model, init=False) → bool
process_free_flow_time(model: Model, init=False) → bool
process_max_speed(model: Model, init=False) → bool
publish_attributes: Sequence[PublishAttribute] = (PublishAttribute(name='passenger_flow', target=None, correction_value=0), PublishAttribute(name='cargo_flow', target=None, correction_value=0), PublishAttribute(name='passenger_car_unit', target=None, correction_value=0), PublishAttribute(name='volume_to_capacity', target=None, correction_value=0), PublishAttribute(name='delay_factor', target=None, correction_value=1), PublishAttribute(name='congested_time', target='average_time', correction_value=1000000000.0))
publish_results(model: Model, results: AssignmentResultCollection)
setup_state(model: Model, state: TrackedState, dataset_name: str)
transport_segment_entity

alias of TrafficTransportSegmentEntity

transport_type: str
class Model(model_config)

Bases: TrackedModel

Calculates traffic attributes on roads

cargo_pcu: float
demand_nodes: DemandNodeEntity | None = None
classmethod get_schema_attributes() → Iterable[AttributeSpec]
initialize(**_)

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

modality: ModalityStrategy
project: ProjectWrapper | None = None
setup(state: TrackedState, settings: Settings, **_)

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(**_) → 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

transport_nodes: PointEntity | None = None
transport_segments: TrafficTransportSegmentEntity | None = None
update(**_) → 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

vdf_alpha: float | str
vdf_beta: float
class PassengerTrackModality

Bases: TrackModality

get_demands(model: Model) → Tuple[ndarray, ndarray]
publish_attributes: Sequence[PublishAttribute] = (PublishAttribute(name='passenger_flow', target=None, correction_value=0), PublishAttribute(name='congested_time', target='passenger_average_time', correction_value=1000000000.0))
transport_segment_entity

alias of PassengerTrackSegmentEntity

transport_type: str = 'passenger_tracks'
class PublishAttribute(name: 'str', target: 't.Optional[str]' = None, correction_value: 'float' = 0)

Bases: object

correction_value: float = 0
name: str
target: str | None = None
class RoadModality

Bases: ModalityStrategy

transport_type: str = 'roads'
class TrackModality

Bases: ModalityStrategy

initialize_parameters(model: Model)
publish_attributes: Sequence[PublishAttribute] = (PublishAttribute(name='passenger_flow', target=None, correction_value=0), PublishAttribute(name='cargo_flow', target=None, correction_value=0), PublishAttribute(name='congested_time', target='average_time', correction_value=1000000000.0))
transport_segment_entity

alias of TrackSegmentEntity

transport_type: str = 'tracks'
class WaterwayModality

Bases: ModalityStrategy

alpha(model: Model) → ndarray

We want waterway segments with locks to have an additional waiting time t_wait We set the free_flow_times of these segments as

t_ff’ = t_ff + t_wait

To convert it to vdf terms:

t_ff’ = t_ff + s’ a = r / t_ff’ b = 4.9

With these aequilibrae can compute: vdf = t_ff’ * (1 + a * volume_over_capacity**b)

free_flow_time(model: Model) → ndarray
initialize_parameters(model: Model)
transport_type: str = 'waterways'
convert_v1_v2(config)
get_matrix(csr_array: TrackedCSRArray)

Module contents