data_collector

concurrent

class LimitedThreadPoolExecutor(max_workers=None, thread_name_prefix='', initializer=None, initargs=())

Bases: ThreadPoolExecutor

Similar to ThreadPoolExecutor, but blocks on submit when all workers are busy

submit(function, *args, **kwargs)
exception MultipleException(exceptions: Sequence[Exception])

Bases: Exception

static format_exception(e: Exception)
class MultipleFutures(iterable: Iterable[Future] = ())

Bases: object

Keep track of multiple concurrent.Futures

add(fut: Future)
done()
exception()
wait()

data_collector

class DataCollector(model_config: dict)

Bases: SimpleModel

classmethod add_storage_strategy(name, strategy: Type[StorageStrategy])
aggregate: bool = False
close(**_)
flush(moment: Moment, origin: str | None)
get_storage_strategy(settings: Settings, logger: Logger)
initialize(settings: Settings, logger: Logger, **_) → DataMask
maybe_flush(moment: Moment, origin, trigger)
new_time(new_time: Moment, **_)
state: t.Optional[TrackedState] = None
strategies: t.Dict[str, t.Type[StorageStrategy]] = {'file': <class 'movici_simulation_core.models.data_collector.data_collector.FileStorageStrategy'>, 'sqlite': <class 'movici_simulation_core.models.data_collector.sqlite_strategy.SQLiteStorageStrategy'>}
strategy: StorageStrategy
submit(fn, *args, **kwargs)
update(moment: Moment, data: dict | None, message: UpdateMessage) → Tuple[dict | None, Moment | None]
class FileStorageStrategy(directory: Path, filename_template='t{timestamp}_{iteration}_{name}')

Bases: StorageStrategy

classmethod choose(model_config: dict, settings: Settings, **_) → StorageStrategy
initialize()
reset_iterations(model: DataCollector)
store(info: UpdateInfo)
class UpdateInfo(name: 'str', timestamp: 'int', iteration: 'int', data: 'dict', origin: 't.Optional[str]' = None)

Bases: object

data: dict
full_data()
iteration: int
name: str
origin: str | None = None
timestamp: int

sqlite_strategy

SQLite storage strategy for DataCollector model.

This module provides SQLiteStorageStrategy which stores simulation updates in a SQLite database instead of individual JSON files.

class SQLiteStorageStrategy(database_path: Path, settings: Settings)

Bases: StorageStrategy

Storage strategy that persists simulation updates to an SQLite database.

Features:

  • Thread-safe concurrent writes (uses internal locking)

  • Efficient binary storage of numpy arrays

  • Support for CSR sparse arrays

  • Single database file instead of thousands of JSON files

  • Fast indexed queries by timestamp, iteration, dataset

classmethod choose(model_config: dict, settings: Settings, logger: Logger) → SQLiteStorageStrategy

Factory method to create SQLiteStorageStrategy from configuration.

Parameters:

  • model_config – DataCollector model configuration

  • settings – Global simulation settings

  • logger – Logger instance

Returns:

SQLiteStorageStrategy instance

Raises:

ValueError – If neither database_path nor storage_dir is configured

close()

Clean up database connections.

Called when simulation ends. Ensures all connections are properly closed.

initialize()

Initialize the database.

Also stores initial datasets from Settings.data_dir for self-contained archives.

reset_iterations(model)

Reset the iteration counter.

Called when a new timestamp starts to reset iteration numbering.

Parameters:

model – DataCollector instance

store(info)

Store a simulation update in the database.

This method is called from worker threads by the DataCollector’s thread pool. The underlying SimulationDatabase uses locking to ensure thread-safe writes.

Parameters:

info –

UpdateInfo instance containing:

  • name: Dataset name

  • timestamp: Simulation timestamp

  • iteration: Iteration number

  • data: Update data dictionary

  • origin: Optional model identifier

strategy

class StorageStrategy

Bases: object

classmethod choose(model_config: dict, settings: Settings, logger: Logger) → StorageStrategy
close()
initialize()
reset_iterations(model: DataCollector)
store(info: UpdateInfo)