testing#

dummy#

class DummyModel(model_config: dict)#

Bases: TrackedModel

close = <Mock id='140234443832432'>#
initialize = <Mock id='140234443828976'>#
install = <Mock id='140234443831904'>#
classmethod reset_mocks()#
setup = <Mock id='140234443830320'>#
shutdown = <Mock id='140234443832144'>#
update = <Mock id='140234443830752'>#

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: T | Type[T], data: dict, state: TrackedState | None = None) → T#
data_mask_compare(data_mask)#
dataset_data_to_numpy(data: 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: Settings | None = None, init_data_handler=None, tmp_dir=None, schema: AttributeSchema | Sequence[AttributeSpec] | Plugin | None = None, raise_on_premature_shutdown=False)#

Bases: object

add_init_data(name: str, data: 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: Any | None = None)#
update(timestamp: int, data: dict | None, **msg_kwargs)#
update_series(timestamp: int, data_series: Sequence[dict | None], **msg_kwargs)#
class NumpyPreProcessor(model: TrackedModel, settings: Settings, schema=None)#

Bases: PreProcessor

process_input(input_data: dict | None) → bytes | None#
process_result(result: Tuple[bytes | None, int | None]) → Tuple[dict | None, int | None]#
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: dict | None) → bytes | None#
process_result(result: Tuple[bytes | None, int | None]) → Tuple[dict | None, int | None]#
update(msg: UpdateMessage, data) → Tuple[dict | None, int | None]#
update_series(msg: UpdateSeriesMessage, data_series) → Tuple[dict | None, int | None]#
compare_results(expected: Sequence[Tuple[int, dict | None, int | None]], results: Sequence[Tuple[int, dict | None, int | None]], rtol=1e-05, atol=1e-08) → List[Tuple[int, Dict[str, str]]]#
format_errors(errors: List[Tuple[int, Dict[str, str]]])#
read_schema(schema: AttributeSchema | Sequence[AttributeSpec] | Plugin | None) → AttributeSchema#

road_network#

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_node_entities(nodes: Nodes, ref_prefix='')#
generate_road_segments(links: Links, transport_nodes: Nodes)#
generate_transport_nodes(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='140234443832432'>#
initialize = <Mock id='140234443828976'>#
install = <Mock id='140234443831904'>#
classmethod reset_mocks()#
setup = <Mock id='140234443830320'>#
shutdown = <Mock id='140234443832144'>#
update = <Mock id='140234443830752'>#
class ModelTester(model, settings: Settings | None = None, init_data_handler=None, tmp_dir=None, schema: AttributeSchema | Sequence[AttributeSpec] | Plugin | None = None, raise_on_premature_shutdown=False)#

Bases: object

add_init_data(name: str, data: 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: Any | None = None)#
update(timestamp: int, data: dict | None, **msg_kwargs)#
update_series(timestamp: int, data_series: Sequence[dict | None], **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_node_entities(nodes: Nodes, ref_prefix='')#
generate_road_segments(links: Links, transport_nodes: Nodes)#
generate_transport_nodes(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: T | Type[T], data: dict, state: TrackedState | None = None) → T#
data_mask_compare(data_mask)#
dataset_data_to_numpy(data: 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)#