Python API
Auto-generated from source docstrings via mkdocstrings.
Core¶
Component
¶
Base class for all archetype components, extending LanceModel with additional utilities.
get_type_by_name(name)
classmethod
¶
Find a Component subclass (at any depth) by its class name.
Walks the full subclass tree — __subclasses__() only returns
direct subclasses, so a naive lookup misses components defined as
subclasses of intermediate base classes.
Raises on ambiguity: two Component classes sharing a name would make name-addressed payloads (CLI/API spawn commands) land rows in whichever archetype import order happened to favor — a silent wrong-table write. Ambiguity must be loud.
from_dict(data)
classmethod
¶
Create a component instance from a dictionary.
The dict must either include a "type" key naming a concrete
Component subclass, or this method must be called on a concrete
subclass directly. Calling Component.from_dict({...}) without
a "type" key raises ValueError rather than silently
constructing a base Component with lost type information.
get_prefix()
classmethod
¶
Generate a standardized prefix for this component type's fields.
to_pyarrow_schema()
classmethod
¶
Convert this component type to a PyArrow schema. Uses LanceModel's built-in to_arrow_schema method.
get_prefixed_schema()
classmethod
¶
Get this component's PyArrow schema with prefixed field names. This is used when combining multiple components into an archetype schema.
to_row_dict()
¶
AsyncProcessor
¶
Bases: iAsyncProcessor
process(df, **input_kwargs)
async
¶
Async version of process method. Override this in subclasses.
AsyncWorld
¶
Bases: iAsyncWorld
add_hook(event_type, fn, *, mode='blocking')
¶
Register a handler for lifecycle events.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
event_type
|
type[_HookEventT]
|
Event dataclass to listen for: PreTick, PostTick, OnSpawn, OnDespawn, OnComponentAdded, or OnComponentRemoved. |
required |
fn
|
AsyncHookHandler[_HookEventT]
|
Async callable that accepts the matching event instance. |
required |
mode
|
FireMode
|
"blocking" awaits the handler inline. "spawn" schedules it with asyncio.create_task so the tick does not wait. |
'blocking'
|
Returns:
| Type | Description |
|---|---|
HookHandle
|
HookHandle to pass back to remove_hook when the handler should be |
HookHandle
|
unregistered. |
Example
from archetype.core.hooks import PostTick
async def log_tick(event: PostTick) -> None: print(f"tick {event.tick} done, {len(event.results)} archetypes")
handle = world.add_hook(PostTick, log_tick)
... later ...¶
world.remove_hook(handle)
step(run_config, **input_kwargs)
async
¶
Executes one full, parallel simulation tick.
Tick Lifecycle
- pre_tick hook fires
- For each archetype (parallel): a. Query previous state (tick N-1) b. Materialize mutations (spawn/despawn caches) c. Execute processors (priority order) d. Persist to store
- Increment tick
- post_tick hook fires (tick is now N+1)
Note: Messages enqueued in tick N are dequeued in tick N+1.
Resources
¶
Type-safe container for world-level shared state.
Usage
world.resources.insert(SimConfig(gravity=9.8)) world.resources.insert(broker)
In processor¶
config = resources.require(SimConfig) broker = resources.get(CommandBroker) # Returns None if not present
Configuration¶
WorldConfig
¶
Bases: BaseModel
World identity and initial state.
Carries the world's identity (id, name) and its mutable simulation state (tick counter, entity registry, mutation caches). Passed to the world constructor so all state is explicit and serializable.
world_id = Field(default_factory=(uuid.uuid7), description='Unique identifier for the world. Auto-generated if omitted.')
class-attribute
instance-attribute
¶
name = Field(default=None, description='Human-readable alias for the world')
class-attribute
instance-attribute
¶
run_id = Field(default=None, description='Active run identifier')
class-attribute
instance-attribute
¶
tick = Field(default=0, description='Current simulation tick')
class-attribute
instance-attribute
¶
next_entity_id = Field(default=1, description='Next entity ID to assign')
class-attribute
instance-attribute
¶
entity2sig = Field(default_factory=dict, description='Entity ID → archetype signature mapping')
class-attribute
instance-attribute
¶
spawn_cache = Field(default_factory=dict, description='Pending spawn rows keyed by archetype signature')
class-attribute
instance-attribute
¶
despawn_cache = Field(default_factory=dict, description='Pending despawn entity IDs keyed by archetype signature')
class-attribute
instance-attribute
¶
lineage = Field(default_factory=list, description="Ancestor read segments as (world_id, run_id, up_to_tick), ascending. Reads for ticks <= up_to_tick resolve to that ancestor's rows; later ticks belong to this world's own run. Populated by fork.")
class-attribute
instance-attribute
¶
model_config = dict(arbitrary_types_allowed=True)
class-attribute
instance-attribute
¶
StorageConfig
¶
Bases: BaseModel
Storage backend configuration for app/runtime storage resolution.
Includes
- uri: str - The URI location for the storage backend
- namespace: str - The desired namespace for the catalog
- io_config: IOConfig - Native Daft I/O configuration for storage reads/writes
uri = Field(default='./archetype_db', description='The URI location for the storage backend (str or Path)')
class-attribute
instance-attribute
¶
namespace = Field(default='ecs')
class-attribute
instance-attribute
¶
backend = Field(default=(StorageBackend.LANCEDB), description="Storage backend engine: 'iceberg' or 'lancedb (default)'")
class-attribute
instance-attribute
¶
io_config = Field(default=None, description='Native Daft I/O configuration passed to Iceberg storage read/write operations.')
class-attribute
instance-attribute
¶
model_config = dict(arbitrary_types_allowed=True)
class-attribute
instance-attribute
¶
RunConfig
¶
Bases: BaseModel
A run represents the configuration of a sequence of world.steps, and configures the runtime options for the world.
Carries configuration for the run, including: - run_id: UUID - The unique identifier for the run sequence, a uuid7 - num_steps: int - The number of steps to execute in the run sequence - debug: bool - Whether or not to enable debug mode - validate: bool - Whether or not to enable validation mode
Add ergonomic named constructors, e.g. RunConfig.dev(steps=1, debug=True)
and RunConfig.benchmark(steps, explain=False) to reduce call-site verbosity.
App¶
ServiceContainer
¶
Wires services together with correct dependency ordering.
Each service owns its internal composition. The container only handles service-to-service wiring.
Usage
container = ServiceContainer() world = await container.world_service.create_world(config, storage_config) await container.simulation_service.step(world.world_id, run_config)
shutdown()
async
¶
Gracefully shut down all services.
Command
¶
Bases: BaseModel
id = Field(default_factory=(uuid.uuid7))
class-attribute
instance-attribute
¶
tick = 0
class-attribute
instance-attribute
¶
actor_id = None
class-attribute
instance-attribute
¶
type = CommandType.CUSTOM
class-attribute
instance-attribute
¶
payload = Field(default_factory=dict)
class-attribute
instance-attribute
¶
priority = 0
class-attribute
instance-attribute
¶
version = 1
class-attribute
instance-attribute
¶
seq = Field(default_factory=(lambda: next(_SEQ)))
class-attribute
instance-attribute
¶
model_config = dict(frozen=True, arbitrary_types_allowed=True)
class-attribute
instance-attribute
¶
serialize_uuids(v, info)
¶
arrow_schema()
classmethod
¶
Return a canonical Arrow schema suitable for Parquet.
to_arrow()
¶
__lt__(other)
¶
CommandType
¶
Bases: StrEnum
Command types for the command broker.
The broker is the universal simulation interface supporting: - Entity-level mutations (spawn, despawn, components) - Processor mutations (hot-swap behavior) - Simulation-level operations (recursive simulation, rollouts)