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.
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
A world configuration is a container for the world configuration, including: - world_id: Optional[UUID] - The unique identifier for the world. If not provided, a new one will be generated. - name: Optional[str] - A human-readable alias for the world.
world_id = Field(default_factory=(uuid.uuid7), description='The unique identifier for the world. If not provided, a new one will be generated.')
class-attribute
instance-attribute
¶
name = Field(default=None, description='A human-readable alias for the world')
class-attribute
instance-attribute
¶
model_config = dict(arbitrary_types_allowed=True)
class-attribute
instance-attribute
¶
StorageConfig
¶
Bases: BaseModel
Storage Backend Context for configuring local or cloud storage access
Includes
- uri: str - The URI location for the storage backend
- namespace: str - The desired namespace for the catalog
- io_config: IOConfig - The access credentials or the daft session/catalog
uri = Field(default='./archetype_data', description='The URI location for the storage backend (str or Path)')
class-attribute
instance-attribute
¶
namespace = Field(default='archetypes')
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='Configuration for the native I/O layer, e.g. credentials for accessing cloud storage systems.')
class-attribute
instance-attribute
¶
model_config = dict(arbitrary_types_allowed=True)
class-attribute
instance-attribute
¶
is_async
property
¶
use_lancedb
property
¶
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 all services together with correct dependency ordering.
Usage
container = ServiceContainer() world = await container.world_service.create_world(config, storage_config) await container.simulation_service.step(world.world_id, run_config)
Pass registry_path to enable persistent world discovery across
server restarts. The long-running archetype serve process uses
the registry to rehydrate previously created worlds on startup.
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)