testing#

dummy#

class DummyModel(model_config: dict)#

Bases: TrackedModel

close = <Mock id='139783110527824'>#

initialize = <Mock id='139783110540400'>#

install = <Mock id='139783110528448'>#

classmethod reset_mocks()#

setup = <Mock id='139783110531472'>#

shutdown = <Mock id='139783110529168'>#

update = <Mock id='139783110535648'>#

helpers#

assert_dataset_dicts_equal(a, b, rtol=1e-05, atol=1e-08)#

Deep compares two nested structures (such as dict) and asserts that they are equivalent. lists and numpy.ndarray``s are compared using ``numpy.isequal or numpy.isclose with equal_nan=True

Parameters:

a – the left dictionary object
b – the right dictionary object
rtol – relative tolerance used as in numpy.isclose
atol – absolute tolerance used as in numpy.isclose

assert_equivalent_data_mask(a, b)#

compare_dataset_dicts(a, b, rtol=1e-05, atol=1e-08)#

create_entity_group_with_data(entity_type: Union[T, Type[T]], data: dict) → T#

data_mask_compare(data_mask)#

dataset_data_to_numpy(data: Union[dict, ndarray, list])#

dataset_dicts_equal(a, b, rtol=1e-05, atol=1e-08)#

get_attribute(name='attr', **kwargs)#

list_dir(path: Path)#

model_schema#

model_config_validator(model_schema: dict)#

model_tester#

class ModelTester(model, settings: Optional[Settings] = None, init_data_handler=None, tmp_dir=None, schema: Optional[Union[AttributeSchema, Sequence[AttributeSpec], Plugin]] = None, raise_on_premature_shutdown=False)#

Bases: object

add_init_data(name: str, data: Union[dict, str, Path])#

close()#

initialize()#

new_time(timestamp: int)#

classmethod run_scenario(model: Type[Model], model_name: str, scenario: dict, rtol=1e-05, atol=1e-08, use_new_time=True, global_schema: Optional[Any] = None)#

update(timestamp: int, data: Optional[dict], **msg_kwargs)#

update_series(timestamp: int, data_series: Sequence[Optional[dict]], **msg_kwargs)#

class NumpyPreProcessor(model: TrackedModel, settings: Settings, schema=None)#

Bases: PreProcessor

process_input(input_data: Optional[dict]) → Optional[bytes]#

process_result(result: Tuple[Optional[bytes], Optional[int]]) → Tuple[Optional[dict], Optional[int]]#

class Plugin(*args, **kwargs)#

Bases: Protocol

install(obj: Extensible)#

class PreProcessor(model: Model, settings: Settings, schema=None)#

Bases: object

close(message: QuitMessage)#

initialize(data_handler: InitDataHandler) → DataMask#

new_time(message: NewTimeMessage)#

process_input(input_data: Optional[dict]) → Optional[bytes]#

process_result(result: Tuple[Optional[bytes], Optional[int]]) → Tuple[Optional[dict], Optional[int]]#

update(msg: UpdateMessage, data) → Tuple[Optional[dict], Optional[int]]#

update_series(msg: UpdateSeriesMessage, data_series) → Tuple[Optional[dict], Optional[int]]#

compare_results(expected: Sequence[Tuple[int, Optional[dict], Optional[int]]], results: Sequence[Tuple[int, Optional[dict], Optional[int]]], rtol=1e-05, atol=1e-08) → List[Tuple[int, Dict[str, str]]]#

format_errors(errors: List[Tuple[int, Dict[str, str]]])#

read_schema(schema: Optional[Union[AttributeSchema, Sequence[AttributeSpec], Plugin]]) → AttributeSchema#

road_network#

class Links(id: 't.Sequence[int]', from_idx: 't.Sequence[int]', to_idx: 't.Sequence[int]')#

Bases: object

classmethod create(links: List[Tuple[int, int]], id_offset=0, node_idx_offset=0)#

from_idx: Sequence[int]#

id: Sequence[int]#

to_idx: Sequence[int]#

class Nodes(id: 't.Sequence[int]', x: 't.Sequence[float]', y: 't.Sequence[float]')#

Bases: object

classmethod create(nodes: List[Tuple[float, float]], id_offset=0)#

duplicate(id_offset)#

id: Sequence[int]#

x: Sequence[float]#

y: Sequence[float]#

class RoadNetworkGenerator(nodes: List[Tuple[float, float]], links: List[Tuple[int, int]], geom_offset=(155000, 463000), max_speed=1, lanes=1, capacity=10)#

Bases: object

generate()#

static generate_link_entities(links: Links, nodes: Nodes, ref_prefix='') → dict#

static generate_node_entities(nodes: Nodes, ref_prefix='')#

generate_road_segments(links: Links, transport_nodes: Nodes)#

generate_transport_nodes(nodes: Nodes)#

generate_virtual_links(links: Links, virtual_nodes: Nodes, transport_nodes: Nodes)#

generate_virtual_nodes(nodes: Nodes)#

generate_road_network(nodes, links, geom_offset=(155000, 463000), max_speed=1, lanes=1, capacity=10)#

Module contents#

class DummyModel(model_config: dict)#

Bases: TrackedModel

close = <Mock id='139783110527824'>#

initialize = <Mock id='139783110540400'>#

install = <Mock id='139783110528448'>#

classmethod reset_mocks()#

setup = <Mock id='139783110531472'>#

shutdown = <Mock id='139783110529168'>#

update = <Mock id='139783110535648'>#

Bases: object

add_init_data(name: str, data: Union[dict, str, Path])#

close()#

initialize()#

new_time(timestamp: int)#

classmethod run_scenario(model: Type[Model], model_name: str, scenario: dict, rtol=1e-05, atol=1e-08, use_new_time=True, global_schema: Optional[Any] = None)#

update(timestamp: int, data: Optional[dict], **msg_kwargs)#

update_series(timestamp: int, data_series: Sequence[Optional[dict]], **msg_kwargs)#

class RoadNetworkGenerator(nodes: List[Tuple[float, float]], links: List[Tuple[int, int]], geom_offset=(155000, 463000), max_speed=1, lanes=1, capacity=10)#

Bases: object

generate()#

static generate_link_entities(links: Links, nodes: Nodes, ref_prefix='') → dict#

static generate_node_entities(nodes: Nodes, ref_prefix='')#

generate_road_segments(links: Links, transport_nodes: Nodes)#

generate_transport_nodes(nodes: Nodes)#

generate_virtual_links(links: Links, virtual_nodes: Nodes, transport_nodes: Nodes)#

generate_virtual_nodes(nodes: Nodes)#

assert_dataset_dicts_equal(a, b, rtol=1e-05, atol=1e-08)#

Deep compares two nested structures (such as dict) and asserts that they are equivalent. lists and numpy.ndarray``s are compared using ``numpy.isequal or numpy.isclose with equal_nan=True

Parameters:

a – the left dictionary object
b – the right dictionary object
rtol – relative tolerance used as in numpy.isclose
atol – absolute tolerance used as in numpy.isclose

assert_equivalent_data_mask(a, b)#

compare_dataset_dicts(a, b, rtol=1e-05, atol=1e-08)#

create_entity_group_with_data(entity_type: Union[T, Type[T]], data: dict) → T#

data_mask_compare(data_mask)#

dataset_data_to_numpy(data: Union[dict, ndarray, list])#

dataset_dicts_equal(a, b, rtol=1e-05, atol=1e-08)#

generate_road_network(nodes, links, geom_offset=(155000, 463000), max_speed=1, lanes=1, capacity=10)#

get_attribute(name='attr', **kwargs)#

list_dir(path: Path)#

model_config_validator(model_schema: dict)#