Data#

modelly.data holds two ways to collect data from a run: the DataCollector (model, agent, agent-type and table reporters, called once per step) and the dataset recorders of modelly.data.recorders (named datasets recorded on a schedule into memory, JSON, Parquet or SQL).

Module	Contents
`modelly.data.collector`	`DataCollector`
`modelly.data.reporters`	reporter normalisation and the row-collection paths
`modelly.data.io`	Arrow and Parquet export, DuckDB queries
`modelly.data.recorders`	`DataRegistry`, the datasets and the recorders

DataCollector#

DataCollector takes add_table_row from modelly.data.tables and the get_*_dataframe methods from modelly.data.frames; they are listed under DataCollector.

Modelly Data Collection Module.

DataCollector is meant to provide a simple, standard way to collect data generated by a Modelly model. It collects four types of data: model-level data, agent-level data, agent-type-level data, and tables.

A DataCollector is instantiated with three dictionaries of reporter names and associated variable names or functions for each, one for model-level data, one for agent-level data, and one for agent-type-level data; a fourth dictionary provides table names and columns. Variable names are converted into functions which retrieve attributes of that name.

When the collect() method is called, each model-level function is called, with the model as the argument, and the results associated with the relevant variable. Then the agent-level functions are called on each agent, and the agent-type-level functions are called on each agent of the specified type.

Additionally, other objects can write directly to tables by passing in an appropriate dictionary object for a table row.

The DataCollector then stores the data it collects in dictionaries:

model_vars maps each reporter to a list of its values
tables maps each table to a dictionary, with each column as a key with a list as its value.
_agent_records maps each model step to a list of each agent’s id and its values.
_agenttype_records maps each model step to a dictionary of agent types, each containing a list of each agent’s id and its values.

Finally, DataCollector can create a pandas DataFrame from each collection, and a pyarrow Table or a Parquet file (to_arrow, to_parquet).

class DataCollector(model_reporters=None, agent_reporters=None, agenttype_reporters=None, tables=None)[source]#

Class for collecting data generated by a Modelly model.

A DataCollector is instantiated with dictionaries of names of model-, agent-, and agent-type-level variables to collect, associated with attribute names or functions which actually collect them. When the collect(…) method is called, it collects these attributes and executes these functions one by one and stores the results.

Instantiate a DataCollector with lists of model, agent, and agent-type reporters.

Both model_reporters, agent_reporters, and agenttype_reporters accept a dictionary mapping a variable name to either an attribute name, a function, a method of a class/instance, a partial function, or a function with parameters placed in a list.

Model reporters can take five types of arguments:

{"agent_count": lambda m: len(m.agents)}                  # lambda
{"agent_count": self.get_agent_count}                     # bound method
{"agent_count": Model.get_agent_count}                    # method of a class
{"model_attribute": "model_attribute"}                    # attribute name
{"agent_count": functools.partial(count_agents, multiplier=2)}
{"Model_Function": [function, [param_1, param_2]]}        # function + params

Agent reporters can similarly take:

{"energy": "energy"}                                      # attribute name
{"energy": lambda a: a.energy}                            # lambda
{"agent_action": self.do_action}                          # bound method
{"agent_action": Agent.do_action}                         # method of a class
{"energy": functools.partial(get_energy, scale=2)}
{"Agent_Function": [function, [param_1, param_2]]}        # function + params

Agenttype reporters take a dictionary mapping agent types to dictionaries of reporter names and attributes/funcs/methods, similar to agent_reporters:

{Wolf: {"energy": lambda a: a.energy}}

The tables arg accepts a dictionary mapping names of tables to lists of columns. For example, if we want to allow agents to write their age when they are destroyed (to keep track of lifespans), it might look like:

{"Lifespan": ["unique_id", "age"]}

Parameters:

model_reporters – Dictionary of reporter names and attributes/funcs/methods.
agent_reporters – Dictionary of reporter names and attributes/funcs/methods.
agenttype_reporters – Dictionary of agent types to dictionaries of reporter names and attributes/funcs/methods.
tables – Dictionary of table names to lists of column names.

Notes

If you want to pickle your model you must not use lambda functions.
If your model includes a large number of agents, it is recommended to use attribute names for the agent reporter, as it will be faster.
agenttype_reporters collect only agents of exactly the given type. Subclasses are not included, so a reporter keyed on a superclass does not pick up its subclasses. Add a separate reporter per concrete type if you need subclass data.

collect(model)[source]#: Collect all the data for the given model object.

add_table_row(table_name, row, ignore_missing=False)#

Add a row dictionary to a specific table.

A row with missing required columns leaves the table unchanged.

Parameters:

table_name – Name of the table to append a row to.
row – A dictionary of the form {column_name: value…}
ignore_missing – If True, fill any missing columns with Nones; if False, throw an error if any columns are missing

get_agent_vars_dataframe()#

Create a pandas DataFrame from the agent variables.

The DataFrame has one column for each variable, with two additional columns for tick and agent_id.

get_agenttype_vars_dataframe(agent_type)#

Create a pandas DataFrame from the agent-type variables for a specific agent type.

The DataFrame has one column for each variable, with two additional columns for tick and agent_id.

Parameters:: agent_type – The type of agent to get the data for.

get_model_vars_dataframe()#

Create a pandas DataFrame from the model variables.

The DataFrame has one column for each model variable, and the index is (implicitly) the model tick.

get_table_dataframe(table_name)#

Create a pandas DataFrame from a particular table.

Parameters:: table_name – The name of the table to convert.

to_arrow(kind: str = 'model', name: Any = None)#: This collector’s kind collection as a pyarrow.Table (see to_arrow()).

to_parquet(path: str | Path, kind: str = 'model', name: Any = None) → Path#: Write this collector’s kind collection to Parquet (see to_parquet()).

Arrow, Parquet and DuckDB#

DataCollector.to_arrow and DataCollector.to_parquet call these functions; query runs SQL over Arrow tables, DataFrames and Parquet files. They need pip install "modelly-abm[experiments]".

Arrow and Parquet export of a DataCollector, and SQL over the results with DuckDB.

pyarrow and duckdb are optional: pip install "modelly-abm[experiments]". Parquet files are written atomically (a temporary file in the same folder, then os.replace), so a reader never sees half a file, on any platform.

Example

>>> collector.to_parquet("runs/model.parquet")
>>> query("SELECT avg(Gini) FROM model", model=collector.to_arrow())

to_arrow(collector: Any, kind: str = 'model', name: Any = None)[source]#

One collection of collector as a pyarrow.Table.

Parameters:

collector – a DataCollector.
kind – "model", "agent" (with Step and AgentID columns), "agenttype" or "table".
name – the agent class for "agenttype", the table name for "table".

Example

>>> table = to_arrow(model.datacollector, "agent")

write_parquet_atomic(table: Any, path: str | Path) → Path[source]#

Write a pyarrow.Table to path atomically; return the path.

The table goes to a temporary file next to path first, then os.replace moves it into place, so a crash leaves no partial file.

to_parquet(collector: Any, path: str | Path, kind: str = 'model', name: Any = None) → Path[source]#

Write one collection of collector to a Parquet file, atomically.

Arguments are those of to_arrow(); returns the path written.

Example

>>> to_parquet(model.datacollector, "out/agents.parquet", "agent")

query(sql: str, connection: Any = None, **tables: Any)[source]#

Run sql with DuckDB over named tables; return a DuckDB relation.

Each keyword names a table in the query: a pyarrow.Table, a pandas DataFrame, or a path (or glob) of Parquet files.

Example

>>> query("SELECT count(*) FROM runs", runs="out/*.parquet").fetchone()

tables_of(collector: Any) → Mapping[str, Any][source]#: Every non-empty collection of collector by name: model, agent and its tables.

Reporters#

The functions behind the collector: reporter normalisation and the fast row-collection paths.

Reporter plumbing for modelly.data.collector.DataCollector.

A reporter is what the user hands the collector: an attribute name, a callable, a bound method, a functools.partial or a [function, [params]] list. The functions here normalise them once at registration and then gather values as cheaply as possible on every collect:

attribute reporters are gathered with a single operator.attrgetter per agent, so the per-agent work happens in C;
a value is copied only when it is mutable, so the common scalar case costs a set lookup instead of a deepcopy call.

snapshot(value: Any) → Any[source]#

Return value itself when immutable, otherwise a deep copy.

Stored values must not change when the model mutates them on a later step.

check_list_reporter(name: str, reporter: list) → None[source]#: Validate that a list-style reporter has the form [callable, [params]].

validate_model_reporter(name: str, reporter: Any, model: Any) → None[source]#

Run a model reporter once so a broken definition fails loudly on first collect.

Raises:

ValueError – If reporter has an invalid list format
AttributeError – If model attribute doesn’t exist
RuntimeError – If reporter execution fails

compile_model_reporter(name: str, reporter: Any) → Callable[[Any], Any][source]#

Turn a model reporter definition into a plain callable(model).

The type dispatch happens here, once, instead of on every collect. List reporters are re-checked on every call: the list is user-owned and may have been edited after validation.

getattr_or_none(model: Any, name: str) → Any[source]#: getattr with a None default, as model attribute reporters have always behaved.

collect_model_value(name: str, reporter: Any, model: Any) → Any[source]#

Evaluate one model reporter against model and return a stored-safe value.

List reporters are re-checked on every call: the list is user-owned and may have been edited after validation.

make_agent_reporter(name: str, reporter: Any) → Callable[[Any], Any][source]#

Normalise an agent reporter definition into a one-argument callable.

Attribute names become a getter that raises a descriptive AttributeError and carries the attribute in _modelly_attribute so collection can use the C fast path. [function, [params]] lists are bound. Anything else is already a suitable callable and is returned as is.

collect_agent_rows(agents: Iterable[Any], reporters: dict[str, Callable], time: float) → list[tuple][source]#

Return one (time, unique_id, *values) row per agent.

Values are gathered one reporter at a time over all agents (a C-level map per column) and the columns are zipped into rows, so the per-agent Python work is close to zero. A column is copied only if it holds a mutable value. Attribute reporters use one attrgetter; a missing attribute re-runs the reporter’s own getter to produce the descriptive error.

When every reporter is a Column of the agents’ single class and agents is that class’s whole population, the rows are sliced straight out of the column table in table order, with no per-agent work at all.

Datasets#

DataRegistry: the named datasets a model exposes to its recorders.

class DataRegistry[source]#

A registry for data sets.

Initialize the registry.

add_dataset(dataset: DataSet)[source]#

Add a dataset to the registry.

Parameters:: dataset – the dataset to register
Raises:: RuntimeError – if a dataset with the same name is already registered

create_dataset(dataset_type, name, *args, fields: str | list[str] | None = None, **kwargs) → DataSet[source]#: Create a dataset of the specified type and add it to the registry.

track_agents(agents: AbstractAgentSet, name: str, fields: str | list[str] | None = None, *, use_dirty_flag: bool = False) → AgentDataSet[source]#: Track the specified fields for the agents in the AgentSet.

track_model(model: Model, name: str, fields: str | list[str] | None = None) → ModelDataSet[source]#: Track the specified fields in the model.

track_agents_numpy(agent_type: type[~modelly.core.agent.Agent], name: str, fields: str | list[str] | None = None, n: int = 100, dtype: ~numpy.dtype | type = <class 'numpy.float64'>) → NumpyAgentDataSet[source]#

Track agent fields using NumPy storage.

Parameters:

agent_type – The agent class to install properties on
name – Name of the dataset
fields – Attribute names to track
n – Initial capacity
dtype – NumPy dtype for the array

Returns:

The created NumpyAgentDataSet

close()[source]#: Close all datasets.

get(name: str, default=None) → DataSet | None[source]#: Get a dataset by name.

The DataSet protocol and the attrgetter-based BaseDataSet.

class DataSet(*args, **kwargs)[source]#

Protocol for all data collection classes.

All datasets raise RuntimeError if accessed after being closed.

property data: Any#: Return collected data.

close()[source]#: Close the dataset.

record(recorder: BaseDataRecorder, configuration: DatasetConfig | None = None) → DataSet[source]#: Record the collected data.

class BaseDataSet(name, *, fields: str | list[str] | None = None)[source]#

Abstract base class for data sets.

Parameters:

name – the name of the data set
fields – the fields to collect

Initialize a base data set.

abstract property data#: Return the data of the dataset.

close()[source]#: Cleanup and close the data set.

record(recorder: BaseDataRecorder, configuration: DatasetConfig | None = None) → DataSet[source]#: Record the collected data.

Datasets read from Python objects: agents, the model, and hand-written table rows.

class AgentDataSet(name: str, agents: AbstractAgentSet[A], *, use_dirty_flag: bool = False, fields: str | list[str] | None = None)[source]#

Bases: BaseDataSet, Generic

Data set for agents data.

Parameters:

name – the name of the data set
agents – the agents to collect data from
fields – fields to collect
use_dirty_flag – if True, enables manual dirty-flag caching optimization. Default is False (always recompute snapshot).

Init. of AgentDataSet.

property data: list[dict[str, Any]]#: Return the data of the dataset.

set_dirty_flag() → None[source]#

Mark the dataset as dirty.

Only meaningful if use_dirty_flag=True. The next access to .data will recompute the snapshot.

close()[source]#: Close the data set.

class ModelDataSet(name, model: M, fields: str | list[str] | None = None)[source]#

Bases: BaseDataSet, Generic

Data set for model data.

Parameters:

name – the name of the data set
model – the model to collect data from
fields – the fields to collect.

Init of ModelDataSet.

property data: dict[str, Any]#: Return the data of the dataset.

close()[source]#: Close the data set.

class TableDataSet(name, fields: str | list[str] | None = None)[source]#

Bases: object

A Table DataSet.

Parameters:

name – the name of the data set
fields – string or list of strings specifying the columns

fixme: this needs a closer look: it now follows the datacollector, so you just add a row.

Init.

add_row(row: dict[str, Any])[source]#

Add a row to the table.

Parameters:

row – the row to add

Raises:

RuntimeError – if the dataset has been closed
ValueError – if the row is empty
ValueError – if the row is missing required fields
ValueError – if the row contains unexpected fields

property data: list[dict[str, Any]]#: Return the data of the dataset.

close()[source]#: Close the data set.

record(recorder: BaseDataRecorder, configuration: DatasetConfig | None = None) → DataSet[source]#: Record the collected data.

NumpyAgentDataSet: agent attributes stored as columns of a contiguous NumPy array.

class NumpyAgentDataSet(name: str, agent_type: type[A], fields: str | list[str] | None = None, n: int = 100, dtype: np.dtype | type = <class 'numpy.float64'>)[source]#

A NumPy array data set for storing agent data.

Uses swap-with-last removal to keep data contiguous, allowing views.

Note

This class does not inherit from BaseDataSet because it doesn’t use the attrgetter-based collection mechanism.

Initialize the dataset.

Parameters:

name – Name of the dataset
agent_type – The agent class to install properties on
fields – attribute names to track
n – Initial capacity
dtype – NumPy dtype for the array

Raises:

ValueError – if no attributes are specified

add_agent(agent: A) → int[source]#

Add an agent to the dataset.

Parameters:: agent – The agent to add
Returns:: The index assigned to the agent
Raises:: RuntimeError if the dataset has been closed. –

remove_agent(agent: A) → None[source]#

Remove an agent from the dataset using swap-with-last.

Parameters:: agent – The agent to remove
Raises:: ValueError if the agent is not in the dataset –

property data: ndarray#: Return active agent data as a VIEW (no copy).

Warning

Modifying the returned array modifies the underlying data.

property data_copy: ndarray#: Return a copy of active agent data.

property agent_ids: ndarray#: Return the agent ids as a view (no copy).

property active_agents: list[A]#: Return list of all active agents (order matches data rows).

close() → None[source]#: Close the dataset and remove properties from agent class.

record(recorder: BaseDataRecorder, configuration: DatasetConfig | None = None) → DataSet[source]#: Record the collected data.

Recorders#

Collection schedule for one dataset: interval, time window and sliding window.

Configuration for dataset collection behavior.

Variables:

interval (int | float) – Collection frequency in time units
start_time (int | float) – First time to begin collection
end_time (int | float | None) – Last time to collect (None = collect forever)
window_size (int | None) – Max snapshots to keep (None = keep all, N = sliding window)
enabled (bool) – Whether collection is active
_next_collection (int | float) – Next scheduled collection time (internal)

Examples

# Collect every step from 0 to infinity DatasetConfig(interval=1, start_time=0)

# Collect every 5 steps from 100 to 500 DatasetConfig(interval=5, start_time=100, end_time=500)

# Collect every step, keep only last 1000 snapshots DatasetConfig(interval=1, start_time=0, window_size=1000)

# Warmup phase: skip first 100 steps, collect 100-200 DatasetConfig(interval=1, start_time=100, end_time=200)

should_collect(current_time: float) → bool[source]#

Check if we should collect at this time.

Parameters:: current_time – The current simulation time
Returns:: True if collection should happen, False otherwise

update_next_collection(current_time: float) → None[source]#

Update the next collection time.

Parameters:: current_time – The current simulation time

BaseDataRecorder for the Custom Data Recorders.

Subclasses must implement:

_store_dataset_snapshot: Store a single dataset snapshot
_get_dataset_dataframe: Retrieve stored data as DataFrame
clear: Clear stored data
summary: Return collection status summary

The base class handles:

Observable subscription and signal handling
Collection interval management
Dataset configuration
Enable/disable logic

Key Design Principles:

ABC-based design for custom storage backends
Separation of concerns: BaseDataRecorder(owns config) vs CustomRecorder(owns storage)
Observable-based collection

class BaseDataRecorder(model: Model, config: dict[str, DatasetConfig | dict[str, Any]] | None = None)[source]#

Base class for data recorders.

Initialize the recorder.

Parameters:

model – The model to observe.
config – Config mapping dataset names to configuration. Values can be dicts or DatasetConfig objects.

collect() → None[source]#

Manually trigger collection (for manual mode or testing).

Normally you don’t need to call this - it’s handled via observables. But you can call it manually if needed.

abstractmethod clear(dataset_name: str | None = None) → None[source]#: Clear stored data.

enable_dataset(dataset_name: str) → None[source]#: Enable collection for a dataset.

disable_dataset(dataset_name: str) → None[source]#: Disable collection for a dataset.

add_dataset(dataset: DataSet, configuration: DatasetConfig | None = None) → None[source]#

Add a dataset to the collection.

Parameters:

dataset (DataSet) – Dataset to record
configuration (DatasetConfig|None) – Configuration for recording the dataset

abstractmethod get_table_dataframe(name: str) → DataFrame[source]#: Convert stored data to pandas DataFrame.

get_all_dataframes() → dict[str, DataFrame][source]#: Get DataFrames for all datasets.

abstractmethod summary() → dict[str, Any][source]#: Get collection status summary.

DataRecorder: the default recorder, keeping snapshots in memory.

class DatasetStorage(blocks: deque = <factory>, metadata: dict[str, ~typing.Any]=<factory>, total_rows: int = 0, estimated_size_bytes: int = 0)[source]#

Bases: object

Storage container with sliding window support.

class DataRecorder(model: Model, config: dict[str, DatasetConfig | dict[str, Any]] | None = None)[source]#

Bases: BaseDataRecorder

In-memory data recorder (default implementation).

Initialize the recorder and subscribe to model observables.

clear(dataset_name: str | None = None) → None[source]#: Clear stored data.

get_table_dataframe(name: str) → DataFrame[source]#: Convert stored data to pandas DataFrame.

estimate_memory_usage() → float[source]#: Estimate current memory usage in MB.

summary() → dict[str, Any][source]#: Get collection status summary.

JSONDataRecorder: snapshots kept as JSON-ready records and saved to .json files.

class NumpyJSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)[source]#

Bases: JSONEncoder

JSON Encoder that handles Numpy types.

Constructor for JSONEncoder, with sensible defaults.

If skipkeys is false, then it is a TypeError to attempt encoding of keys that are not str, int, float, bool or None. If skipkeys is True, such items are simply skipped.

If ensure_ascii is true, the output is guaranteed to be str objects with all incoming non-ASCII characters escaped. If ensure_ascii is false, the output can contain non-ASCII characters.

If check_circular is true, then lists, dicts, and custom encoded objects will be checked for circular references during encoding to prevent an infinite recursion (which would cause an RecursionError). Otherwise, no such check takes place.

If allow_nan is true, then NaN, Infinity, and -Infinity will be encoded as such. This behavior is not JSON specification compliant, but is consistent with most JavaScript based encoders and decoders. Otherwise, it will be a ValueError to encode such floats.

If sort_keys is true, then the output of dictionaries will be sorted by key; this is useful for regression tests to ensure that JSON serializations can be compared on a day-to-day basis.

If indent is a non-negative integer, then JSON array elements and object members will be pretty-printed with that indent level. An indent level of 0 will only insert newlines. None is the most compact representation.

If specified, separators should be an (item_separator, key_separator) tuple. The default is (’, ‘, ‘: ‘) if indent is None and (‘,’, ‘: ‘) otherwise. To get the most compact JSON representation, you should specify (‘,’, ‘:’) to eliminate whitespace.

If specified, default is a function that gets called for objects that can’t otherwise be serialized. It should return a JSON encodable version of the object or raise a TypeError.

default(obj)[source]#: Convert Numpy types to native Python types.

class JSONDataRecorder(model, config: dict[str, DatasetConfig | dict[str, Any]] | None = None, output_dir='.')[source]#

Bases: BaseDataRecorder

Store data as JSON files.

Initialize JSON Recorder.

get_table_dataframe(name: str) → DataFrame[source]#: Convert stored JSON-like data to DataFrame.

clear(dataset_name: str | None = None) → None[source]#: Clear data.

summary() → dict[str, Any][source]#: Get summary.

save_to_json()[source]#: Save all data to JSON files.

ParquetDataRecorder: snapshots buffered and written to .parquet files.

class ParquetDataRecorder(model, config: dict[str, DatasetConfig | dict[str, Any]] | None = None, output_dir: str = '.')[source]#

Bases: BaseDataRecorder

Store collected data in Parquet files.

Initialize Parquet storage backend.

get_table_dataframe(name: str) → DataFrame[source]#: Read data from Parquet file.

clear(dataset_name: str | None = None) → None[source]#: Delete Parquet files.

summary() → dict[str, Any][source]#: Get collection status summary.

SQLDataRecorder: snapshots stored as rows of a SQLite database.

class SQLDataRecorder(model, config: dict[str, DatasetConfig | dict[str, Any]] | None = None, db_path: str = ':memory:')[source]#

Bases: BaseDataRecorder

Store collected data in SQLite database.

Initialize SQL storage backend.

get_table_dataframe(name: str) → DataFrame[source]#: Convert stored data to pandas DataFrame.

query(sql: str) → DataFrame[source]#: Execute custom SQL query.

clear(dataset_name: str | None = None) → None[source]#: Clear stored data by dropping tables.

summary() → dict[str, Any][source]#: Get collection status summary.