postprocessing#
results#
- exception EndOfStream#
Bases:
ValueError
- class ResultDataset(init_data: dict, updates: Iterable[Dict], timeline_info: TimelineInfo | 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 = None)#
Bases:
object- get_dataset(name)#
- 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- abstract slice(timestamp: int | str | datetime | None = None, attribute: str | None = 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: Sequence[ReversibleUpdate] | None = 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: TimelineInfo | 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 = None)#
Bases:
object- get_dataset(name)#
- use(plugin)#
- 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 UpdateStream(updates: Sequence[ReversibleUpdate] | None = None)#
Bases:
object- insert_after(update: ReversibleUpdate)#
- next() ReversibleUpdate#
- prev() ReversibleUpdate#
- merge_updates(*updates: dict)#