postprocessing#

results#

exception EndOfStream#: Bases: ValueError

class ResultDataset(init_data: dict, updates: Iterable[Dict], timeline_info: Optional[TimelineInfo] = None)#

Bases: object

static get_slicing_strategy(**kwargs)#

slice(entity_group, timestamp: Optional[Union[int, str, datetime]] = None, attribute: Optional[str] = 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: Optional[Dict[str, Union[UniformAttributeData, CSRAttributeData]]] = None#

revert(state: TrackedState)#

timestamp: int#

update: Dict[str, Union[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: Union[AttributeSchema, Sequence[AttributeSpec]] = (), timeline_info: Optional[TimelineInfo] = None)#

Bases: object

get_dataset(name)#

use(plugin)#

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

Bases: SlicingStrategy

slice(attribute: Optional[str] = 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: Optional[Union[int, str, datetime]] = None, **_)#

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

Bases: object

abstract slice(timestamp: Optional[Union[int, str, datetime]] = None, attribute: Optional[str] = None, entity_selector=None, key='id', **_)#

class TimeProgressingState(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)#

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

Bases: object

dataset: str#

iteration: int#

path: Path#

timestamp: int#

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

Bases: object

insert_after(update: ReversibleUpdate)#

next() → ReversibleUpdate#

prev() → ReversibleUpdate#

merge_updates(*updates: dict)#

Module contents#

class ResultDataset(init_data: dict, updates: Iterable[Dict], timeline_info: Optional[TimelineInfo] = None)#

Bases: object

static get_slicing_strategy(**kwargs)#

slice(entity_group, timestamp: Optional[Union[int, str, datetime]] = None, attribute: Optional[str] = None, entity_selector=None, key='id')#