postprocessing

results

exception EndOfStream

Bases: ValueError

class ResultDataset(init_data: dict, updates: Iterable[Dict], timeline_info: TimelineInfo | None = None, schema: AttributeSchema | None = None)

Bases: object

static get_slicing_strategy(**kwargs)

slice(entity_group, timestamp: int | str | datetime | None = None, attribute: str | None = None, entity_selector=None, key='id')

class ReversibleUpdate(timestamp: 'int', iteration: 'int', dataset: 'str', entity_group: 'str', indices: 'np.ndarray', update: 'EntityData', reverse_update: 't.Optional[EntityData]' = None, next: 'ReversibleUpdate' = None, prev: 'ReversibleUpdate' = None)

Bases: object

apply(state: TrackedState)

calculate_reverse_update(state: TrackedState)

dataset: str

entity_group: str

indices: ndarray

iteration: int

next: ReversibleUpdate = None

prev: ReversibleUpdate = None

reverse_update: Dict[str, UniformAttributeData | CSRAttributeData] | None = None

revert(state: TrackedState)

timestamp: int

update: Dict[str, UniformAttributeData | CSRAttributeData]

class SimulationResults(init_data_dir: Path, updates_dir: Path, update_pattern='t(?P<timestamp>\\d+)_(?P<iteration>\\d+)_(?P<dataset>\\w+)\\.json', attributes: AttributeSchema | Sequence[AttributeSpec] = (), timeline_info: TimelineInfo = None)

Bases: object

datasets: Dict[str, Path]

get_dataset(name)

updates: Dict[str, List[UpdateFile]]

use(plugin)

class SingleAttributeSlicingStrategy(state: TimeProgressingState, dataset: str, entity_group: str, timeline_info: TimelineInfo)

Bases: SlicingStrategy

slice(attribute: str | None = None, **_)

class SingleEntitySlicingStrategy(state: TimeProgressingState, dataset: str, entity_group: str, timeline_info: TimelineInfo)

Bases: SlicingStrategy

slice(entity_selector=None, key='id', **_)

class SingleTimestampSlicingStrategy(state: TimeProgressingState, dataset: str, entity_group: str, timeline_info: TimelineInfo)

Bases: SlicingStrategy

slice(timestamp: int | str | datetime | None = None, **_)

class SlicingStrategy(state: TimeProgressingState, dataset: str, entity_group: str, timeline_info: TimelineInfo)

Bases: object

abstractmethod slice(timestamp: int | str | datetime | None = None, attribute: str | None = None, entity_selector=None, key='id', **_)

class TimeProgressingState(schema: AttributeSchema | None = None, logger=None)

Bases: TrackedState

add_init_data(init_data: Dict)

add_updates_to_timeline(updates: Iterable[Dict])

get_timestamps(dataset, entity_group=None) → List[int]

move_to(timestamp)

streams: Dict[tuple[str, str], UpdateStream]

class UpdateFile(dataset: 'str', timestamp: 'int', iteration: 'int', path: 'Path')

Bases: object

dataset: str

iteration: int

path: Path

timestamp: int

class UpdateStream(updates: Sequence[ReversibleUpdate] | None = None)

Bases: object

current: ReversibleUpdate | None

insert_after(update: ReversibleUpdate)

next() → ReversibleUpdate

prev() → ReversibleUpdate

merge_updates(*updates: dict)

sqlite_results

SQLite-based simulation results reader.

Provides a SimulationResults-compatible interface for reading simulation results from SQLite databases created by the SQLite storage backend.

class SQLiteSimulationResults(database_path: Path, init_data_dir: Path | None = None, attributes: AttributeSchema | Sequence[AttributeSpec] = (), timeline_info: TimelineInfo = None)

Bases: object

Read simulation results from SQLite database.

Provides the same interface as SimulationResults but reads from a SQLite database instead of individual JSON files. Compatible with movici-viewer.

Example:

>>> # With init_data_dir
>>> results = SQLiteSimulationResults(
...     database_path=Path("simulation.db"),
...     init_data_dir=Path("init_data"),
... )
>>> dataset = results.get_dataset("transport_network")
>>>
>>> # With initial datasets stored in database
>>> results = SQLiteSimulationResults(database_path=Path("simulation.db"))
>>> dataset = results.get_dataset("transport_network")

close()

Close database connection.

get_dataset(name: str) → ResultDataset

Get a dataset with its initial state and all updates.

Parameters:

name – Dataset name

Returns:

ResultDataset with initial data and updates

Raises:

ValueError – If dataset not found

get_datasets() → List[str]

Get list of all available datasets.

Returns:

List of dataset names

get_timestamps(dataset_name: str) → List[int]

Get all timestamps for a dataset.

Parameters:

dataset_name – Name of the dataset

Returns:

List of timestamps in ascending order

use(plugin)

Register a plugin with the schema.

Parameters:

plugin – Plugin to register

detect_results_format(updates_path: Path) → Literal['sqlite', 'json']

Detect whether results are stored in SQLite or JSON format.

Parameters:

updates_path – Path to updates directory or database file

Returns:

“sqlite” if SQLite database found, “json” otherwise

get_simulation_results(init_data_dir: Path | None = None, updates_path: Path | None = None, attributes: AttributeSchema | Sequence[AttributeSpec] = (), timeline_info: TimelineInfo = None, update_pattern: str = 't(?P<timestamp>\\d+)_(?P<iteration>\\d+)_(?P<dataset>\\w+)\\.json')

Factory function to get appropriate SimulationResults instance.

Automatically detects whether results are in SQLite or JSON format and returns the appropriate reader.

Parameters:

• init_data_dir – Directory containing initial dataset JSON files (optional for SQLite if database contains initial datasets)

• updates_path – Path to updates directory or SQLite database

• attributes – Schema for attributes (optional)

• timeline_info – Timeline information (optional)

• update_pattern – Regex pattern for JSON files (only used if JSON format detected)

Returns:

SQLiteSimulationResults or SimulationResults depending on detected format