API Reference

This page groups and expands the public API by area. Each section includes a brief description of the modules followed by auto-generated API docs (mkdocstrings). Use the grouped structure to locate functionality quickly, then dive into the detailed signatures and docstrings.

Contents

• How to read this reference
• Core
• Components
• Systems
• Utilities
• Levels (Authoring)
• Rendering
• Examples
• Gym Environment
• Registries and enums

How to read this reference

• Modules are grouped by domain. Each entry links to the module’s classes and functions.

• Headings provide a short summary of what the module does before the auto-doc.

• Code examples for common tasks live in the Functional API and Guides pages; this reference focuses on signatures and behavior.

• Tip: If module headings are too large in the Table of Contents, set heading_level: 3 in mkdocs.yml under mkdocstrings options so modules render as H3 under these H2 sections.

Core

State and stepping

• grid_universe.state: Immutable State dataclass holding component stores, meta, and RNG seed.

• grid_universe.step: The main reducer that applies one action and orchestrates all systems in the correct order.

grid_universe.state

Core immutable ECS State dataclass.

This module defines the frozen :class:State object that represents the entire game / simulation snapshot at a single turn. All systems are pure functions that take a previous State plus inputs (e.g. an Action) and return a new State; no mutation happens in-place. This makes the engine deterministic, easy to test, and friendly to functional style reducers.

Design notes:

• Component stores are persistent maps (pyrsistent.PMap) keyed by EntityID. Absence of a key means the entity does not currently possess that component.
• Effect components (Immunity, Phasing, Speed, TimeLimit, UsageLimit) are referenced by :class:grid_universe.components.properties.Status which holds ordered effect_ids. Several systems (status tick, GC) walk those references.
• The prev_position and trail auxiliary stores are populated by dedicated systems to enable path‑based effects (e.g. trail rendering or damage-on-cross mechanics).
• win / lose flags are mutually exclusive terminal markers. The reducer short‑circuits on terminal states.

Google‑style docstrings throughout the codebase refer back to this structure; see :mod:grid_universe.step for how the reducer orchestrates systems.

State dataclass

Immutable ECS world state.

Instances are value objects; every transition creates a new State. Only include persistent / serializable data here (no open handles or caches). Systems should be pure functions that accept and return State.

Attributes:

Name	Type	Description
width	int	Grid width in tiles.
height	int	Grid height in tiles.
move_fn	MoveFn	Movement candidate function used to resolve move actions.
objective_fn	ObjectiveFn	Predicate evaluated after each step to set win.
immunity	PMap[EntityID, Immunity]	Effect component map.
phasing	PMap[EntityID, Phasing]	Effect component map.
speed	PMap[EntityID, Speed]	Effect component map.
time_limit	PMap[EntityID, TimeLimit]	Effect limiter map (remaining steps).
usage_limit	PMap[EntityID, UsageLimit]	Effect limiter map (remaining uses).
agent	PMap[EntityID, Agent]	Player / AI controllable entity marker components.
appearance	PMap[EntityID, Appearance]	Rendering metadata (glyph, layering, groups).
blocking	PMap[EntityID, Blocking]	Entities that prevent movement into their tile.
collectible	PMap[EntityID, Collectible]	Items that can be picked up.
collidable	PMap[EntityID, Collidable]	Entities that can collide (triggering damage, cost, etc.).
cost	PMap[EntityID, Cost]	Movement or interaction cost applied when entered.
damage	PMap[EntityID, Damage]	Passive damage applied on collision / contact.
dead	PMap[EntityID, Dead]	Marker for logically removed entities (awaiting GC).
exit	PMap[EntityID, Exit]	Tiles that can satisfy the objective when conditions met.
health	PMap[EntityID, Health]	Health pools for damage / lethal checks.
inventory	PMap[EntityID, Inventory]	Item/key collections carried by entities.
key	PMap[EntityID, Key]	Keys that can unlock Locked components.
lethal_damage	PMap[EntityID, LethalDamage]	Immediate kill damage sources (pits, hazards).
locked	PMap[EntityID, Locked]	Lock descriptors requiring matching keys.
moving	PMap[EntityID, Moving]	Entities currently undergoing movement (inter-step state).
pathfinding	PMap[EntityID, Pathfinding]	Agents with pathfinding goals and cached paths.
portal	PMap[EntityID, Portal]	Teleport endpoints / pairs.
position	PMap[EntityID, Position]	Current grid position of entities.
pushable	PMap[EntityID, Pushable]	Entities that can be displaced by push actions.
required	PMap[EntityID, Required]	Items/conditions needed to satisfy Exit / objective.
rewardable	PMap[EntityID, Rewardable]	Components conferring score rewards when collected or triggered.
status	PMap[EntityID, Status]	Ordered list container referencing effect component ids.
prev_position	PMap[EntityID, Position]	Snapshot of positions before movement this step.
trail	PMap[Position, PSet[EntityID]]	Positions traversed this step mapped to entity ids.
turn	int	Turn counter (0-based).
score	int	Accumulated score.
turn_limit	int | None	Optional maximum number of turns allowed. When set, reaching this number triggers a lose state unless already win. None disables the limit.
win	bool	True if objective met.
lose	bool	True if losing condition met.
message	str | None	Optional informational / terminal message.
seed	int | None	Base RNG seed for deterministic rendering or procedural systems.

Source code in grid_universe/state.py

@dataclass(frozen=True)
class State:
    """Immutable ECS world state.

    Instances are *value objects*; every transition creates a new ``State``.
    Only include persistent / serializable data here (no open handles or
    caches). Systems should be pure functions that accept and return ``State``.

    Attributes:
        width (int): Grid width in tiles.
        height (int): Grid height in tiles.
        move_fn (MoveFn): Movement candidate function used to resolve move actions.
        objective_fn (ObjectiveFn): Predicate evaluated after each step to set ``win``.
        immunity (PMap[EntityID, Immunity]): Effect component map.
        phasing (PMap[EntityID, Phasing]): Effect component map.
        speed (PMap[EntityID, Speed]): Effect component map.
        time_limit (PMap[EntityID, TimeLimit]): Effect limiter map (remaining steps).
        usage_limit (PMap[EntityID, UsageLimit]): Effect limiter map (remaining uses).
        agent (PMap[EntityID, Agent]): Player / AI controllable entity marker components.
        appearance (PMap[EntityID, Appearance]): Rendering metadata (glyph, layering, groups).
        blocking (PMap[EntityID, Blocking]): Entities that prevent movement into their tile.
        collectible (PMap[EntityID, Collectible]): Items that can be picked up.
        collidable (PMap[EntityID, Collidable]): Entities that can collide (triggering damage, cost, etc.).
        cost (PMap[EntityID, Cost]): Movement or interaction cost applied when entered.
        damage (PMap[EntityID, Damage]): Passive damage applied on collision / contact.
        dead (PMap[EntityID, Dead]): Marker for logically removed entities (awaiting GC).
        exit (PMap[EntityID, Exit]): Tiles that can satisfy the objective when conditions met.
        health (PMap[EntityID, Health]): Health pools for damage / lethal checks.
        inventory (PMap[EntityID, Inventory]): Item/key collections carried by entities.
        key (PMap[EntityID, Key]): Keys that can unlock ``Locked`` components.
        lethal_damage (PMap[EntityID, LethalDamage]): Immediate kill damage sources (pits, hazards).
        locked (PMap[EntityID, Locked]): Lock descriptors requiring matching keys.
        moving (PMap[EntityID, Moving]): Entities currently undergoing movement (inter-step state).
        pathfinding (PMap[EntityID, Pathfinding]): Agents with pathfinding goals and cached paths.
        portal (PMap[EntityID, Portal]): Teleport endpoints / pairs.
        position (PMap[EntityID, Position]): Current grid position of entities.
        pushable (PMap[EntityID, Pushable]): Entities that can be displaced by push actions.
        required (PMap[EntityID, Required]): Items/conditions needed to satisfy Exit / objective.
        rewardable (PMap[EntityID, Rewardable]): Components conferring score rewards when collected or triggered.
        status (PMap[EntityID, Status]): Ordered list container referencing effect component ids.
        prev_position (PMap[EntityID, Position]): Snapshot of positions before movement this step.
        trail (PMap[Position, PSet[EntityID]]): Positions traversed this step mapped to entity ids.
        turn (int): Turn counter (0-based).
        score (int): Accumulated score.
        turn_limit (int | None): Optional maximum number of turns allowed. When
            set, reaching this number triggers a ``lose`` state unless already
            ``win``. ``None`` disables the limit.
        win (bool): True if objective met.
        lose (bool): True if losing condition met.
        message (str | None): Optional informational / terminal message.
        seed (int | None): Base RNG seed for deterministic rendering or procedural systems.
    """

    # Level
    width: int
    height: int
    move_fn: "MoveFn"
    objective_fn: "ObjectiveFn"

    # Components
    ## Effects
    immunity: PMap[EntityID, Immunity] = pmap()
    phasing: PMap[EntityID, Phasing] = pmap()
    speed: PMap[EntityID, Speed] = pmap()
    time_limit: PMap[EntityID, TimeLimit] = pmap()
    usage_limit: PMap[EntityID, UsageLimit] = pmap()
    ## Properties
    agent: PMap[EntityID, Agent] = pmap()
    appearance: PMap[EntityID, Appearance] = pmap()
    blocking: PMap[EntityID, Blocking] = pmap()
    collectible: PMap[EntityID, Collectible] = pmap()
    collidable: PMap[EntityID, Collidable] = pmap()
    cost: PMap[EntityID, Cost] = pmap()
    damage: PMap[EntityID, Damage] = pmap()
    dead: PMap[EntityID, Dead] = pmap()
    exit: PMap[EntityID, Exit] = pmap()
    health: PMap[EntityID, Health] = pmap()
    inventory: PMap[EntityID, Inventory] = pmap()
    key: PMap[EntityID, Key] = pmap()
    lethal_damage: PMap[EntityID, LethalDamage] = pmap()
    locked: PMap[EntityID, Locked] = pmap()
    moving: PMap[EntityID, Moving] = pmap()
    pathfinding: PMap[EntityID, Pathfinding] = pmap()
    portal: PMap[EntityID, Portal] = pmap()
    position: PMap[EntityID, Position] = pmap()
    pushable: PMap[EntityID, Pushable] = pmap()
    required: PMap[EntityID, Required] = pmap()
    rewardable: PMap[EntityID, Rewardable] = pmap()
    status: PMap[EntityID, Status] = pmap()
    ## Extra
    prev_position: PMap[EntityID, Position] = pmap()
    trail: PMap[Position, PSet[EntityID]] = pmap()
    damage_hits: PSet[tuple[EntityID, EntityID, int]] = pset()

    # Status
    turn: int = 0
    score: int = 0
    win: bool = False
    lose: bool = False
    message: Optional[str] = None
    turn_limit: Optional[int] = None

    # RNG
    seed: Optional[int] = None

    @property
    def description(self) -> PMap[str, Any]:
        """Sparse serialization of non‑empty fields.

        Iterates dataclass fields and returns a persistent map including only
        those that are non‑empty (for component maps) or truthy (for scalars).
        Useful for lightweight diagnostics / debugging without dumping large
        empty maps.

        Returns:
            PMap[str, Any]: Persistent map of field name to value for all
            populated fields.
        """
        description: PMap[str, Any] = pmap()
        for field in self.__dataclass_fields__:
            value = getattr(self, field)
            # Skip empty persistent maps to keep output concise. We use a duck
            # type check because mypy cannot infer concrete key/value types for
            # every store here; failing len() should just include the value.
            if isinstance(value, type(pmap())):
                try:  # pragma: no cover - defensive
                    if len(value) == 0:  # pyright: ignore[reportUnknownArgumentType]
                        continue
                except Exception:
                    pass
            description = description.set(field, value)
        return pmap(description)

description property

description

Sparse serialization of non‑empty fields.

Iterates dataclass fields and returns a persistent map including only those that are non‑empty (for component maps) or truthy (for scalars). Useful for lightweight diagnostics / debugging without dumping large empty maps.

Returns:

Type	Description
PMap[str, Any]	PMap[str, Any]: Persistent map of field name to value for all
PMap[str, Any]	populated fields.

grid_universe.step

State reducer and step orchestration.

This module wires together all systems in the correct order to implement a single turn transition given an Action. The exported :func:step is the only public mutation entry point for gameplay progression and is intentionally pure: it returns a new :class:grid_universe.state.State.

Ordering rationale (high level):

1. position_system snapshots previous positions (enables trail / cross checks).
2. Autonomous movers & pathfinding update entities (moving_system / pathfinding_system).
3. status_tick_system decrements effect limits before applying player action.
4. Player action sub‑steps (movement may produce multiple sub‑moves via speed effects).
5. After each sub‑move we run interaction systems (portal, damage, rewards) to allow chained behaviors (e.g. portal then damage at destination).
6. After the entire action we apply GC, tile costs, terminal checks, and bump turn.

All helper _step_* functions are internal and assume validation of inputs.

step

step(state, action, agent_id=None)

Advance the simulation by one logical action.

Resolves autonomous movement, applies the player's chosen Action (which may translate to multiple movement sub‑steps for speed effects), runs interaction / status systems, and returns a new State.

Parameters:

Name	Type	Description	Default
state	State	Previous immutable world state.	required
action	Action	Player action enum value to apply.	required
agent_id	EntityID | None	Explicit agent entity id. If None the first entity in state.agent is used. Raises if no agent exists.	None

Returns:

Name	Type	Description
State	State	Next state snapshot. If the input state is already terminal (win/lose) or invalid the same object may be returned unchanged.

Raises:

Type	Description
ValueError	If there is no agent or the action is not recognized.

Source code in grid_universe/step.py

def step(state: State, action: Action, agent_id: Optional[EntityID] = None) -> State:
    """Advance the simulation by one logical action.

    Resolves autonomous movement, applies the player's chosen ``Action`` (which
    may translate to multiple movement sub‑steps for speed effects), runs
    interaction / status systems, and returns a new ``State``.

    Args:
        state (State): Previous immutable world state.
        action (Action): Player action enum value to apply.
        agent_id (EntityID | None): Explicit agent entity id. If ``None`` the first
            entity in ``state.agent`` is used. Raises if no agent exists.

    Returns:
        State: Next state snapshot. If the input state is already terminal (win/lose)
            or invalid the same object may be returned unchanged.

    Raises:
        ValueError: If there is no agent or the action is not recognized.
    """
    if agent_id is None and (agent_id := next(iter(state.agent.keys()), None)) is None:
        raise ValueError("State contains no agent")

    if agent_id in state.dead:
        return replace(state, lose=True)

    if not is_valid_state(state, agent_id) or is_terminal_state(state, agent_id):
        return state

    # Reset per-action damage hit tracking and trail at the very start of a new step
    state = replace(state, damage_hits=pset(), trail=pmap())

    state = position_system(state)  # before movements
    state = moving_system(state)
    state = pathfinding_system(state)
    state = status_tick_system(state)

    if action in MOVE_ACTIONS:
        state = _step_move(state, action, agent_id)
    elif action == Action.USE_KEY:
        state = _step_usekey(state, action, agent_id)
    elif action == Action.PICK_UP:
        state = _step_pickup(state, action, agent_id)
    elif action == Action.WAIT:
        state = _step_wait(state, action, agent_id)
    else:
        raise ValueError("Action is not valid")

    if action not in MOVE_ACTIONS:
        state = _after_substep(state, action, agent_id)

    return _after_step(state, agent_id)

Actions and types

• grid_universe.actions: Agent actions, including movement and non-movement (use key, pick up, wait).

• grid_universe.types: Core type aliases and enums used across the codebase (EntityID, MoveFn, ObjectiveFn, EffectType, EffectLimit).

• grid_universe.objectives: Built-in objective functions (default, collect, exit, unlock, push), plus a registry for selection by name.

grid_universe.actions

Action enumerations.

Defines both the human readable :class:Action (string enum) used internally. MOVE_ACTIONS is the canonical ordered list of movement actions; checks like if action in MOVE_ACTIONS are preferred over enum name comparisons.

Action

Bases: StrEnum

String enum of player actions.

Enum Members

UP: Move one tile up. DOWN: Move one tile down. LEFT: Move one tile left. RIGHT: Move one tile right. USE_KEY: Attempt to unlock a co‑located locked entity with a matching key. PICK_UP: Collect items (powerups / coins / cores / keys) at the current tile. WAIT: Advance a turn without moving (effects still tick).

Source code in grid_universe/actions.py

class Action(StrEnum):
    """String enum of player actions.

    Enum Members:
        UP: Move one tile up.
        DOWN: Move one tile down.
        LEFT: Move one tile left.
        RIGHT: Move one tile right.
        USE_KEY: Attempt to unlock a co‑located locked entity with a matching key.
        PICK_UP: Collect items (powerups / coins / cores / keys) at the current tile.
        WAIT: Advance a turn without moving (effects still tick).
    """

    UP = auto()
    DOWN = auto()
    LEFT = auto()
    RIGHT = auto()
    USE_KEY = auto()
    PICK_UP = auto()
    WAIT = auto()

grid_universe.types

Common type aliases and enumerations.

MoveFn and ObjectiveFn are central extension points used in the State to allow pluggable movement / win condition behavior.

EffectType

Bases: StrEnum

Effect component categories (reflected in serialized observations).

Source code in grid_universe/types.py

class EffectType(StrEnum):
    """Effect component categories (reflected in serialized observations)."""

    IMMUNITY = auto()
    PHASING = auto()
    SPEED = auto()

EffectLimit

Bases: StrEnum

Limit semantics for effects (time or usage based).

Source code in grid_universe/types.py

class EffectLimit(StrEnum):
    """Limit semantics for effects (time or usage based)."""

    TIME = auto()
    USAGE = auto()

grid_universe.objectives

Objective predicate functions and registry.

Each objective function answers: "Has the agent satisfied the win condition?" They are pure predicates over a :class:State and an agent_id. The main reducer checks state.objective_fn after each full action step to decide whether to set state.win.

Functions here should be fast (O(number of relevant components)).

OBJECTIVE_FN_REGISTRY module-attribute

OBJECTIVE_FN_REGISTRY = {'default': default_objective_fn, 'exit': exit_objective_fn, 'collect': collect_required_objective_fn, 'unlock': all_unlocked_objective_fn, 'push': all_pushable_at_exit_objective_fn}

Name → objective predicate mapping for level configuration.

default_objective_fn

default_objective_fn(state, agent_id)

Collect all required items and reach an exit tile.

Source code in grid_universe/objectives.py

def default_objective_fn(state: State, agent_id: EntityID) -> bool:
    """Collect all required items and reach an exit tile."""
    return collect_required_objective_fn(state, agent_id) and exit_objective_fn(
        state, agent_id
    )

exit_objective_fn

exit_objective_fn(state, agent_id)

Agent stands on any entity possessing an Exit component.

Source code in grid_universe/objectives.py

def exit_objective_fn(state: State, agent_id: EntityID) -> bool:
    """Agent stands on any entity possessing an ``Exit`` component."""
    if agent_id not in state.position:
        return False
    return (
        len(entities_with_components_at(state, state.position[agent_id], state.exit))
        > 0
    )

collect_required_objective_fn

collect_required_objective_fn(state, agent_id)

All entities marked Required have been collected (no longer collectible).

Source code in grid_universe/objectives.py

def collect_required_objective_fn(state: State, agent_id: EntityID) -> bool:
    """All entities marked ``Required`` have been collected (no longer collectible)."""
    return all((eid not in state.collectible) for eid in state.required)

all_unlocked_objective_fn

all_unlocked_objective_fn(state, agent_id)

No remaining locked entities (doors, etc.).

Source code in grid_universe/objectives.py

def all_unlocked_objective_fn(state: State, agent_id: EntityID) -> bool:
    """No remaining locked entities (doors, etc.)."""
    return len(state.locked) == 0

all_pushable_at_exit_objective_fn

all_pushable_at_exit_objective_fn(state, agent_id)

Every Pushable entity currently occupies an exit tile.

Source code in grid_universe/objectives.py

def all_pushable_at_exit_objective_fn(state: State, agent_id: EntityID) -> bool:
    """Every Pushable entity currently occupies an exit tile."""
    for pushable_id in state.pushable:
        if pushable_id not in state.position:
            return False
        if (
            len(
                entities_with_components_at(
                    state, state.position[pushable_id], state.exit
                )
            )
            == 0
        ):
            return False
    return True

Components

Property components (authoring/runtime)

• Appearance controls rendering and layering; Position locates entities. Other components model gameplay (Blocking, Collectible, etc.).

grid_universe.components.properties.appearance

Rendering appearance component.

Appearance controls layering and icon/background behavior when composing tiles. Priority ordering rules:

• For background tiles (background=True) the highest priority value wins.
• For main foreground selection the lowest priority value wins (allows important items to sit on top even if visually small).

icon=True marks entities that render as small corner overlays (e.g. powerups) in addition to the main occupant.

AppearanceName

Bases: StrEnum

Enumeration of built‑in appearance categories.

Source code in grid_universe/components/properties/appearance.py

class AppearanceName(StrEnum):
    """Enumeration of built‑in appearance categories."""

    NONE = auto()
    BOOTS = auto()
    BOX = auto()
    COIN = auto()
    CORE = auto()
    DOOR = auto()
    EXIT = auto()
    FLOOR = auto()
    GEM = auto()
    GHOST = auto()
    HUMAN = auto()
    KEY = auto()
    LAVA = auto()
    LOCK = auto()
    MONSTER = auto()
    PORTAL = auto()
    SHIELD = auto()
    SPIKE = auto()
    WALL = auto()

Appearance dataclass

Visual rendering metadata.

Attributes:

Name	Type	Description
name	AppearanceName	Symbolic appearance identifier.
priority	int	Integer priority used for layering selection.
icon	bool	If True this entity may render as a small corner icon.
background	bool	If True counts as a background layer candidate.

Source code in grid_universe/components/properties/appearance.py

@dataclass(frozen=True)
class Appearance:
    """Visual rendering metadata.

    Attributes:
        name: Symbolic appearance identifier.
        priority: Integer priority used for layering selection.
        icon: If True this entity may render as a small corner icon.
        background: If True counts as a background layer candidate.
    """

    name: AppearanceName
    priority: int = 0
    icon: bool = False
    background: bool = False

grid_universe.components.properties.position

Position component.

Immutable integer grid coordinates. Stored in State.position keyed by entity id. The prev_position store records the prior turn's component for trail / cross detection.

Position dataclass

Grid coordinate.

Attributes:

Name	Type	Description
x	int	Column index (0 at left).
y	int	Row index (0 at top).

Source code in grid_universe/components/properties/position.py

@dataclass(frozen=True)
class Position:
    """Grid coordinate.

    Attributes:
        x: Column index (0 at left).
        y: Row index (0 at top).
    """

    x: int
    y: int

grid_universe.components.properties.agent

Agent marker component.

Presence of :class:Agent designates the controllable player entity. Only one agent is typically present; the reducer will select the first if multiple exist. This component carries no data but enables queries / system routing.

Agent dataclass

Marker (no fields).

Source code in grid_universe/components/properties/agent.py

@dataclass(frozen=True)
class Agent:
    """Marker (no fields)."""

    pass

grid_universe.components.properties.blocking

Blocking component.

Marks an entity as occupying its tile for purposes of movement collision. Ignored for entities with active Phasing effect.

Blocking dataclass

Marker (no data).

Source code in grid_universe/components/properties/blocking.py

@dataclass(frozen=True)
class Blocking:
    """Marker (no data)."""

    pass

grid_universe.components.properties.collectible

Collectible component.

Marks an entity that can be picked up into an agent's inventory. If combined with :class:Rewardable or :class:Required, logic in collectible / objective systems updates score or win conditions at pickup.

Collectible dataclass

Marker (no data).

Source code in grid_universe/components/properties/collectible.py

@dataclass(frozen=True)
class Collectible:
    """Marker (no data)."""

    pass

grid_universe.components.properties.collidable

Collidable component.

Indicates an entity participates in collision interactions (e.g. portal entry). Distinct from Blocking which prevents movement; collidable objects may coexist with pass-through mechanics.

Collidable dataclass

Marker (no data).

Source code in grid_universe/components/properties/collidable.py

@dataclass(frozen=True)
class Collidable:
    """Marker (no data)."""

    pass

grid_universe.components.properties.cost

Tile movement cost component (per-step penalty).

Cost dataclass

Movement cost applied once per logical action when on this tile.

Source code in grid_universe/components/properties/cost.py

@dataclass(frozen=True)
class Cost:
    """Movement cost applied once per logical action when on this tile."""

    amount: int

grid_universe.components.properties.damage

Damage component (non-lethal).

Damage dataclass

Hit point damage applied on contact / crossing.

Source code in grid_universe/components/properties/damage.py

@dataclass(frozen=True)
class Damage:
    """Hit point damage applied on contact / crossing."""

    amount: int

grid_universe.components.properties.dead

Dead marker component (post-mortem).

Dead dataclass

Marker set by health/damage logic when HP reaches zero or lethal hit.

Source code in grid_universe/components/properties/dead.py

@dataclass(frozen=True)
class Dead:
    """Marker set by health/damage logic when HP reaches zero or lethal hit."""

    pass

grid_universe.components.properties.exit

Exit dataclass

Marks an entity as an exit tile / goal location.

Objective predicates typically search for agents reaching any entity with this component. The component has no fields; presence alone is meaningful.

Source code in grid_universe/components/properties/exit.py

@dataclass(frozen=True)
class Exit:
    """Marks an entity as an exit tile / goal location.

    Objective predicates typically search for agents reaching any entity with
    this component. The component has no fields; presence alone is meaningful.
    """

    pass

grid_universe.components.properties.health

Health dataclass

Tracks current and maximum hit points for damage / healing systems.

Attributes:

Name	Type	Description
health	int	Current hit points. Systems should clamp this to [0, max_health].
max_health	int	Upper bound for health; may be used to normalize UI or compute proportional rewards.

Source code in grid_universe/components/properties/health.py

@dataclass(frozen=True)
class Health:
    """Tracks current and maximum hit points for damage / healing systems.

    Attributes:
        health:
            Current hit points. Systems should clamp this to ``[0, max_health]``.
        max_health:
            Upper bound for ``health``; may be used to normalize UI or compute
            proportional rewards.
    """

    health: int
    max_health: int

grid_universe.components.properties.inventory

Inventory dataclass

Set of owned item entity IDs.

The immutable PSet enables O(1) sharing across state copies; adding or removing an item produces a new component instance. Other systems (e.g. keys, rewards) inspect membership for gating logic.

Attributes:

Name	Type	Description
item_ids	PSet[EntityID]	Persistent set of entity identifiers currently held.

Source code in grid_universe/components/properties/inventory.py

@dataclass(frozen=True)
class Inventory:
    """Set of owned item entity IDs.

    The immutable ``PSet`` enables O(1) sharing across state copies; adding or
    removing an item produces a new component instance. Other systems (e.g.
    keys, rewards) inspect membership for gating logic.

    Attributes:
        item_ids:
            Persistent set of entity identifiers currently held.
    """

    item_ids: PSet[EntityID]

grid_universe.components.properties.key

Key item component (pairs with Locked).

Key dataclass

Key id string used to unlock matching locked entities.

Source code in grid_universe/components/properties/key.py

@dataclass(frozen=True)
class Key:
    """Key id string used to unlock matching locked entities."""

    key_id: str  # 'red', 'blue', etc.

grid_universe.components.properties.lethal_damage

LethalDamage dataclass

Marker: entity inflicts fatal damage on contact / interaction.

Systems detecting collisions or overlaps may directly set a target's Health to zero (or apply sufficient damage) when encountering an entity with this component. Presence alone conveys semantics.

Source code in grid_universe/components/properties/lethal_damage.py

@dataclass(frozen=True)
class LethalDamage:
    """Marker: entity inflicts fatal damage on contact / interaction.

    Systems detecting collisions or overlaps may directly set a target's
    ``Health`` to zero (or apply sufficient damage) when encountering an
    entity with this component. Presence alone conveys semantics.
    """

    pass

grid_universe.components.properties.locked

Locked dataclass

Indicates the entity is locked and may require a key to unlock.

Attributes:

Name	Type	Description
key_id	str	Identifier of the key required. An empty string can represent a generic lock (any key) or a permanently locked state depending on objective / system interpretation.

Source code in grid_universe/components/properties/locked.py

@dataclass(frozen=True)
class Locked:
    """Indicates the entity is locked and may require a key to unlock.

    Attributes:
        key_id:
            Identifier of the key required. An empty string can represent a
            generic lock (any key) or a permanently locked state depending on
            objective / system interpretation.
    """

    key_id: str = ""  # If empty, may mean "locked with no key" or generic lock

grid_universe.components.properties.moving

Autonomous movement component.

Moving entities advance automatically each turn along a specified axis and direction with optional bouncing at boundaries and configurable tile step speed (processed before player action). prev_position stores the last position for cross / trail interactions when the movement system updates it.

MovingAxis

Bases: StrEnum

Axis enumeration for autonomous motion.

Source code in grid_universe/components/properties/moving.py

class MovingAxis(StrEnum):
    """Axis enumeration for autonomous motion."""

    HORIZONTAL = auto()
    VERTICAL = auto()

Moving dataclass

Autonomous mover definition.

Attributes:

Name	Type	Description
axis	MovingAxis	Axis of travel (horizontal or vertical).
direction	int	+1 or -1 indicating step direction along the axis.
bounce	bool	Reverse direction at edge if True; otherwise stop at boundary.
speed	int	Tile steps attempted per turn.
prev_position	Optional[Position]	Internal bookkeeping of last position (set by system).

Source code in grid_universe/components/properties/moving.py

@dataclass(frozen=True)
class Moving:
    """Autonomous mover definition.

    Attributes:
        axis: Axis of travel (horizontal or vertical).
        direction: +1 or -1 indicating step direction along the axis.
        bounce: Reverse direction at edge if True; otherwise stop at boundary.
        speed: Tile steps attempted per turn.
        prev_position: Internal bookkeeping of last position (set by system).
    """

    axis: MovingAxis
    direction: int  # 1 or -1
    bounce: bool = True
    speed: int = 1
    prev_position: Optional[Position] = None

grid_universe.components.properties.pathfinding

Pathfinding dataclass

AI movement directive for automated entities.

Specifies how an entity should compute movement objectives each step.

Attributes:

Name	Type	Description
target	Optional[EntityID]	Optional entity ID to follow/approach. If None and type is PATH the system may skip pathfinding or use a default goal.
type	PathfindingType	Strategy: PATH requests full pathfinding (e.g., A*), whereas STRAIGHT_LINE attempts direct movement along axis-aligned shortest displacement without obstacle search.

Source code in grid_universe/components/properties/pathfinding.py

@dataclass(frozen=True)
class Pathfinding:
    """AI movement directive for automated entities.

    Specifies how an entity should compute movement objectives each step.

    Attributes:
        target:
            Optional entity ID to follow/approach. If ``None`` and ``type`` is
            ``PATH`` the system may skip pathfinding or use a default goal.
        type:
            Strategy: ``PATH`` requests full pathfinding (e.g., A*), whereas
            ``STRAIGHT_LINE`` attempts direct movement along axis-aligned shortest
            displacement without obstacle search.
    """

    target: Optional[EntityID] = None
    type: PathfindingType = PathfindingType.PATH

grid_universe.components.properties.portal

Portal dataclass

Teleportation link between two entities.

Attributes:

Name	Type	Description
pair_entity	int	Entity ID of the destination/linked portal. When an entity moves onto this portal, movement systems may relocate it to the paired portal's position (often preserving direction or applying post-teleport rules).

Source code in grid_universe/components/properties/portal.py

@dataclass(frozen=True)
class Portal:
    """Teleportation link between two entities.

    Attributes:
        pair_entity:
            Entity ID of the destination/linked portal. When an entity moves onto
            this portal, movement systems may relocate it to the paired portal's
            position (often preserving direction or applying post-teleport rules).
    """

    pair_entity: int  # Entity ID of the paired portal

grid_universe.components.properties.pushable

Pushable dataclass

Marker indicating the entity can be displaced by another's movement.

Push mechanics typically trigger when an agent attempts to move into a tile occupied by a pushable entity; the system tries to move the pushable entity in the same direction if the next tile is free.

Source code in grid_universe/components/properties/pushable.py

@dataclass(frozen=True)
class Pushable:
    """Marker indicating the entity can be displaced by another's movement.

    Push mechanics typically trigger when an agent attempts to move into a
    tile occupied by a pushable entity; the system tries to move the pushable
    entity in the same direction if the next tile is free.
    """

    pass

grid_universe.components.properties.required

Required dataclass

Marker signifying an entity must satisfy a condition to progress.

Often attached to goal or exit entities to indicate prerequisites (such as possessing certain items) must be met; interpretation is handled by objective or validation systems.

Source code in grid_universe/components/properties/required.py

@dataclass(frozen=True)
class Required:
    """Marker signifying an entity must satisfy a condition to progress.

    Often attached to goal or exit entities to indicate prerequisites (such as
    possessing certain items) must be met; interpretation is handled by
    objective or validation systems.
    """

    pass

grid_universe.components.properties.rewardable

Rewardable dataclass

Specifies a scalar reward granted upon satisfying a condition.

Systems can award the amount (e.g., reinforcement learning signal) when the entity is collected, reached, or otherwise triggered by an agent.

Attributes:

Name	Type	Description
amount	int	Numeric reward value to emit; magnitude and sign semantics are up to the environment integration.

Source code in grid_universe/components/properties/rewardable.py

@dataclass(frozen=True)
class Rewardable:
    """Specifies a scalar reward granted upon satisfying a condition.

    Systems can award the ``amount`` (e.g., reinforcement learning signal)
    when the entity is collected, reached, or otherwise triggered by an agent.

    Attributes:
        amount:
            Numeric reward value to emit; magnitude and sign semantics are up to
            the environment integration.
    """

    amount: int

grid_universe.components.properties.status

Status component.

Holds a set of effect entity ids the holder currently has active. The ordering is not semantically relevant (set semantics) but systems iterate the PSet in deterministic order for reproducibility. Limits (time/usage) are stored on the effect entities themselves.

Status dataclass

Active effect references.

Attributes:

Name	Type	Description
effect_ids	PSet[EntityID]	Persistent set of effect entity ids.

Source code in grid_universe/components/properties/status.py

@dataclass(frozen=True)
class Status:
    """Active effect references.

    Attributes:
        effect_ids: Persistent set of effect entity ids.
    """

    effect_ids: PSet[EntityID]

Effects and limits

• Effect entities are referenced from Status.effect_ids and may include limits.

grid_universe.components.effects.immunity

Immunity effect component (negates incoming damage instances).

Immunity dataclass

Marker (no data).

Source code in grid_universe/components/effects/immunity.py

@dataclass(frozen=True)
class Immunity:
    """Marker (no data)."""

    pass

grid_universe.components.effects.phasing

Phasing dataclass

Effect component: entity ignores blocking collisions.

When present, movement systems treat the entity as non-blocking for the purpose of traversing tiles that would normally be obstructed (e.g. walls or other blocking entities). Other entities may still collide with this entity unless they also have logic that skips blocked checks.

Typically combined with a :class:TimeLimit or :class:UsageLimit to make the phasing temporary.

Source code in grid_universe/components/effects/phasing.py

@dataclass(frozen=True)
class Phasing:
    """Effect component: entity ignores blocking collisions.

    When present, movement systems treat the entity as *non-blocking* for the
    purpose of traversing tiles that would normally be obstructed (e.g.
    walls or other blocking entities). Other entities may still collide with
    this entity unless they also have logic that skips blocked checks.

    Typically combined with a :class:`TimeLimit` or :class:`UsageLimit` to make
    the phasing temporary.
    """

    pass

grid_universe.components.effects.speed

Speed effect component.

Multiplies the number of movement sub‑steps performed for a movement action. Each sub‑step triggers post‑movement interaction systems, allowing rapid chain effects (e.g. portal + damage) within a single logical action.

Speed dataclass

Movement multiplier.

Attributes:

Name	Type	Description
multiplier	int	Positive integer factor applied to base 1 movement steps.

Source code in grid_universe/components/effects/speed.py

@dataclass(frozen=True)
class Speed:
    """Movement multiplier.

    Attributes:
        multiplier: Positive integer factor applied to base 1 movement steps.
    """

    multiplier: int

grid_universe.components.effects.time_limit

TimeLimit dataclass

Decorator effect specifying a maximum number of remaining steps.

Systems decrement the remaining amount each global step; when it reaches zero the wrapped effect (or status) is removed. This enables temporary power-ups (e.g. phasing for 5 turns).

Attributes:

Name	Type	Description
amount	int	Number of future steps for which the associated effect/status is still active. Implementations should treat amount <= 0 as expired.

Source code in grid_universe/components/effects/time_limit.py

@dataclass(frozen=True)
class TimeLimit:
    """Decorator effect specifying a maximum number of remaining steps.

    Systems decrement the remaining ``amount`` each global step; when it
    reaches zero the wrapped effect (or status) is removed. This enables
    temporary power-ups (e.g. phasing for 5 turns).

    Attributes:
        amount:
            Number of *future* steps for which the associated effect/status is
            still active. Implementations should treat ``amount <= 0`` as expired.
    """

    amount: int

grid_universe.components.effects.usage_limit

UsageLimit dataclass

Decorator effect counting down discrete consumptions.

Rather than expiring with time, a UsageLimit is decremented by a system whenever the wrapped effect is used (domain-specific). Typical use cases include limited charges (e.g. three teleports) or a fixed number of phasing moves.

Attributes:

Name	Type	Description
amount	int	Remaining number of uses. When it reaches zero the effect/status should be removed.

Source code in grid_universe/components/effects/usage_limit.py

@dataclass(frozen=True)
class UsageLimit:
    """Decorator effect counting down discrete consumptions.

    Rather than expiring with time, a ``UsageLimit`` is decremented by a
    system whenever the wrapped effect is *used* (domain-specific). Typical
    use cases include limited charges (e.g. three teleports) or a fixed number
    of phasing moves.

    Attributes:
        amount:
            Remaining number of uses. When it reaches zero the effect/status
            should be removed.
    """

    amount: int

grid_universe.components.effects

Effect component aggregates.

This sub-package defines effect components: temporary or conditional modifiers that decorate entities (e.g. immunity, phasing, speed changes) as well as limiting wrappers (usage / time limits). Effects are modeled as plain data objects which systems interpret each step; they do not mutate themselves.

Effect is provided as a convenience union of the runtime modifying effects (currently :class:Immunity, :class:Phasing, :class:Speed). Limit wrappers (:class:TimeLimit, :class:UsageLimit) are kept separate because they can apply to any effect type and are processed by status / GC systems.

Importing::

from grid_universe.components.effects import Effect, Speed

or via the top-level components package::

from grid_universe.components import Speed

Immunity dataclass

Marker (no data).

Source code in grid_universe/components/effects/immunity.py

@dataclass(frozen=True)
class Immunity:
    """Marker (no data)."""

    pass

Phasing dataclass

Effect component: entity ignores blocking collisions.

When present, movement systems treat the entity as non-blocking for the purpose of traversing tiles that would normally be obstructed (e.g. walls or other blocking entities). Other entities may still collide with this entity unless they also have logic that skips blocked checks.

Typically combined with a :class:TimeLimit or :class:UsageLimit to make the phasing temporary.

Source code in grid_universe/components/effects/phasing.py

@dataclass(frozen=True)
class Phasing:
    """Effect component: entity ignores blocking collisions.

    When present, movement systems treat the entity as *non-blocking* for the
    purpose of traversing tiles that would normally be obstructed (e.g.
    walls or other blocking entities). Other entities may still collide with
    this entity unless they also have logic that skips blocked checks.

    Typically combined with a :class:`TimeLimit` or :class:`UsageLimit` to make
    the phasing temporary.
    """

    pass

Speed dataclass

Movement multiplier.

Attributes:

Name	Type	Description
multiplier	int	Positive integer factor applied to base 1 movement steps.

Source code in grid_universe/components/effects/speed.py

@dataclass(frozen=True)
class Speed:
    """Movement multiplier.

    Attributes:
        multiplier: Positive integer factor applied to base 1 movement steps.
    """

    multiplier: int

TimeLimit dataclass

Decorator effect specifying a maximum number of remaining steps.

Systems decrement the remaining amount each global step; when it reaches zero the wrapped effect (or status) is removed. This enables temporary power-ups (e.g. phasing for 5 turns).

Attributes:

Name	Type	Description
amount	int	Number of future steps for which the associated effect/status is still active. Implementations should treat amount <= 0 as expired.

Source code in grid_universe/components/effects/time_limit.py

@dataclass(frozen=True)
class TimeLimit:
    """Decorator effect specifying a maximum number of remaining steps.

    Systems decrement the remaining ``amount`` each global step; when it
    reaches zero the wrapped effect (or status) is removed. This enables
    temporary power-ups (e.g. phasing for 5 turns).

    Attributes:
        amount:
            Number of *future* steps for which the associated effect/status is
            still active. Implementations should treat ``amount <= 0`` as expired.
    """

    amount: int

UsageLimit dataclass

Decorator effect counting down discrete consumptions.

Rather than expiring with time, a UsageLimit is decremented by a system whenever the wrapped effect is used (domain-specific). Typical use cases include limited charges (e.g. three teleports) or a fixed number of phasing moves.

Attributes:

Name	Type	Description
amount	int	Remaining number of uses. When it reaches zero the effect/status should be removed.

Source code in grid_universe/components/effects/usage_limit.py

@dataclass(frozen=True)
class UsageLimit:
    """Decorator effect counting down discrete consumptions.

    Rather than expiring with time, a ``UsageLimit`` is decremented by a
    system whenever the wrapped effect is *used* (domain-specific). Typical
    use cases include limited charges (e.g. three teleports) or a fixed number
    of phasing moves.

    Attributes:
        amount:
            Remaining number of uses. When it reaches zero the effect/status
            should be removed.
    """

    amount: int

Systems

Movement, pathfinding, position

• movement_system: Agent’s single-step application obeying Blocking unless Phasing is active.

• moving_system: Autonomous movers with axis/direction/speed/bounce.

• pathfinding_system: Greedy or A* pursuit.

• position_system: Snapshots positions into prev_position at turn start.

grid_universe.systems.movement

Player (agent) movement system.

Attempts to move the controlled agent to next_pos applying effect logic:

1. If the agent has an active Phasing effect (consuming a usage/time limit) it ignores blocking components entirely.
2. Otherwise the move is allowed only if destination is in-bounds and not blocked by Blocking/Pushable/Collidable entities (push handling occurs in a separate system before this is called).

Returns the original State if movement is not possible; otherwise a new State with updated position (and possibly decremented usage limits).

movement_system

movement_system(state, entity_id, next_pos)

Move agent one tile if allowed.

Parameters:

Name	Type	Description	Default
state	State	Current state.	required
entity_id	EntityID	Agent entity id (ignored if not an agent).	required
next_pos	Position	Desired destination position.	required

Returns:

Name	Type	Description
State	State	Same state if blocked / invalid or updated with new position.

Source code in grid_universe/systems/movement.py

def movement_system(state: State, entity_id: EntityID, next_pos: Position) -> State:
    """Move agent one tile if allowed.

    Args:
        state (State): Current state.
        entity_id (EntityID): Agent entity id (ignored if not an agent).
        next_pos (Position): Desired destination position.

    Returns:
        State: Same state if blocked / invalid or updated with new position.
    """
    if entity_id not in state.agent:
        return state

    if not is_in_bounds(state, next_pos):
        return state  # Out of bounds: don't move

    # Check for phasing
    if entity_id in state.status:
        usage_limit: PMap[EntityID, UsageLimit] = state.usage_limit
        usage_limit, effect_id = use_status_effect_if_present(
            state.status[entity_id].effect_ids,
            state.phasing,
            state.time_limit,
            usage_limit,
        )
        if effect_id is not None:
            # Ignore all blocking, just move
            return replace(
                state,
                position=state.position.set(entity_id, next_pos),
                usage_limit=usage_limit,
            )

    if is_blocked_at(state, next_pos, check_collidable=False):
        return state

    return replace(state, position=state.position.set(entity_id, next_pos))

grid_universe.systems.moving

Autonomous linear movement system.

Updates entities with a Moving component by translating them along their configured axis and direction up to speed tiles per step, bouncing (i.e. reversing direction) if configured and blocked/out-of-bounds.

move

move(state, entity_id, pos, next_pos, state_moving, state_position)

Attempt a single-tile move for a moving entity.

Returns updated moving/position maps and whether movement was blocked.

Source code in grid_universe/systems/moving.py

def move(
    state: State,
    entity_id: EntityID,
    pos: Position,
    next_pos: Position,
    state_moving: PMap[EntityID, Moving],
    state_position: PMap[EntityID, Position],
) -> Tuple[PMap[EntityID, Moving], PMap[EntityID, Position], bool]:
    """Attempt a single-tile move for a moving entity.

    Returns updated moving/position maps and whether movement was blocked.
    """
    moving = state_moving[entity_id]
    blocked = not is_in_bounds(state, next_pos) or is_blocked_at(
        state, next_pos, check_collidable=entity_id in state.blocking
    )
    if blocked:
        # Reverse direction if bouncing, else leave unchanged
        new_direction = moving.direction * (-1 if moving.bounce else 1)
        state_moving = state_moving.set(
            entity_id,
            replace(moving, direction=new_direction, prev_position=pos),
        )
    else:
        state_position = state_position.set(entity_id, next_pos)
        state_moving = state_moving.set(
            entity_id,
            replace(moving, prev_position=pos),
        )
    return state_moving, state_position, blocked

moving_system

moving_system(state)

Advance all moving entities for the current step.

Source code in grid_universe/systems/moving.py

def moving_system(state: State) -> State:
    """Advance all moving entities for the current step."""
    state_position = state.position
    state_moving = state.moving

    for entity_id, moving in state_moving.items():
        pos = state_position.get(entity_id)
        if pos is None:
            continue
        if moving.direction not in (-1, 1):
            raise ValueError(
                f"Invalid moving direction for {entity_id}: {moving.direction}"
            )
        dx, dy = (
            (moving.direction, 0)
            if moving.axis == MovingAxis.HORIZONTAL
            else (0, moving.direction)
        )
        for _ in range(moving.speed):
            pos = state_position[entity_id]
            next_pos = Position(pos.x + dx, pos.y + dy)
            state_moving, state_position, blocked = move(
                state, entity_id, pos, next_pos, state_moving, state_position
            )
            state = add_trail_position(state, entity_id, state_position[entity_id])
            if blocked:
                break

        state = replace(state, position=state_position, moving=state_moving)

    return state

grid_universe.systems.pathfinding

Pathfinding systems.

Provides straight-line heuristic movement and A* shortest path selection for entities with the Pathfinding component. Supports effect-based blocking via usage-limited phasing/immunity status checks before movement.

get_astar_next_position

get_astar_next_position(state, entity_id, target_id)

Compute next step toward target using A* (Manhattan metric).

Ignores collidable/pushable differences and treats only blocking tiles as obstacles. Returns current position if already at goal or no path.

Source code in grid_universe/systems/pathfinding.py

def get_astar_next_position(
    state: State, entity_id: EntityID, target_id: EntityID
) -> Position:
    """Compute next step toward target using A* (Manhattan metric).

    Ignores collidable/pushable differences and treats only blocking tiles as
    obstacles. Returns current position if already at goal or no path.
    """
    start = state.position[entity_id]
    goal = state.position[target_id]

    if start == goal:
        return start

    def in_bounds(pos: Position) -> bool:
        return is_in_bounds(state, pos)

    def is_blocked(pos: Position) -> bool:
        return is_blocked_at(state, pos, check_collidable=False)

    def heuristic(a: Position, b: Position) -> int:
        return abs(a.x - b.x) + abs(a.y - b.y)

    neighbors = [(0, 1), (0, -1), (1, 0), (-1, 0)]

    def get_valid_next_positions(position: Position) -> List[Position]:
        neighbor_positions = [
            Position(position.x + dx, position.y + dy) for dx, dy in neighbors
        ]
        return [
            pos for pos in neighbor_positions if in_bounds(pos) and not is_blocked(pos)
        ]

    frontier: PriorityQueue[Tuple[int, int, Position]] = PriorityQueue()
    prev_pos: Dict[Position, Position] = {}
    cost_so_far: Dict[Position, int] = {start: 0}

    tiebreaker = count()  # Unique sequence count
    frontier.put((0, next(tiebreaker), start))

    while not frontier.empty():
        _, __, current = frontier.get()
        if current == goal:
            break
        for next_pos in get_valid_next_positions(current):
            new_cost = cost_so_far[current] + 1
            if next_pos not in cost_so_far or new_cost < cost_so_far[next_pos]:
                cost_so_far[next_pos] = new_cost
                priority = new_cost + heuristic(next_pos, goal)
                frontier.put((priority, next(tiebreaker), next_pos))
                prev_pos[next_pos] = current

    # Reconstruct path
    if goal not in prev_pos:
        return start  # No path found

    # Walk backwards to get the path
    path: List[Position] = []
    current = goal
    while current != start:
        path.append(current)
        current = prev_pos[current]
    path.reverse()

    if not path:
        return start
    return path[0]

get_straight_line_next_position

get_straight_line_next_position(state, entity_id, target_id)

Choose axis-aligned step maximizing dot product toward target.

Source code in grid_universe/systems/pathfinding.py

def get_straight_line_next_position(
    state: State, entity_id: EntityID, target_id: EntityID
) -> Position:
    """Choose axis-aligned step maximizing dot product toward target."""
    target_vec = position_to_vector(state.position[target_id])
    entity_vec = position_to_vector(state.position[entity_id])
    dvec = vector_subtract(target_vec, entity_vec)
    actions = [(0, 1), (0, -1), (1, 0), (-1, 0)]
    values = [vector_dot_product(pvector(action), dvec) for action in actions]
    best_action = actions[argmax(values)]
    return Position(
        state.position[entity_id].x + best_action[0],
        state.position[entity_id].y + best_action[1],
    )

entity_pathfinding

entity_pathfinding(state, usage_limit, entity_id)

Apply pathfinding for a single entity (straight-line or A*).

Source code in grid_universe/systems/pathfinding.py

def entity_pathfinding(
    state: State, usage_limit: PMap[EntityID, UsageLimit], entity_id: EntityID
) -> State:
    """Apply pathfinding for a single entity (straight-line or A*)."""
    if entity_id not in state.position or entity_id not in state.pathfinding:
        return state

    pathfinding_type = state.pathfinding[entity_id].type
    pathfinding_target = state.pathfinding[entity_id].target

    if pathfinding_target is None:
        return state

    if pathfinding_target in state.status:
        usage_limit, effect_id = use_status_effect_if_present(
            state.status[pathfinding_target].effect_ids,
            state.phasing,
            state.time_limit,
            usage_limit,
        )
        if effect_id is not None:
            return state

    if pathfinding_type == PathfindingType.STRAIGHT_LINE:
        next_pos = get_straight_line_next_position(state, entity_id, pathfinding_target)
    elif pathfinding_type == PathfindingType.PATH:
        next_pos = get_astar_next_position(state, entity_id, pathfinding_target)
    else:
        raise NotImplementedError

    if is_blocked_at(state, next_pos, check_collidable=False) or not is_in_bounds(
        state, next_pos
    ):
        return state

    return replace(state, position=state.position.set(entity_id, next_pos))

pathfinding_system

pathfinding_system(state)

Advance all pathfinding-enabled entities by one tile if possible.

Source code in grid_universe/systems/pathfinding.py

def pathfinding_system(state: State) -> State:
    """Advance all pathfinding-enabled entities by one tile if possible."""
    usage_limit: PMap[EntityID, UsageLimit] = state.usage_limit
    for entity_id in state.pathfinding:
        state = entity_pathfinding(state, usage_limit, entity_id)
    return state

grid_universe.systems.position

Position snapshot system.

Maintains prev_position as an immutable snapshot of all entity positions at the start (or end) of a step. Other systems (e.g. portal teleportation, trail generation, damage-on-crossing) rely on this historical information to detect transitions or movement paths.

position_system

position_system(state)

Snapshot current entity positions.

Parameters:

Name	Type	Description	Default
state	State	Current immutable simulation state.	required

Returns:

Name	Type	Description
State	State	New state with prev_position replaced by a pmap copy of the current position mapping.

Source code in grid_universe/systems/position.py

def position_system(state: State) -> State:
    """Snapshot current entity positions.

    Args:
        state (State): Current immutable simulation state.

    Returns:
        State: New state with ``prev_position`` replaced by a pmap copy of the
            current ``position`` mapping.
    """
    prev_position: Dict[EntityID, Position] = {}
    for eid, pos in state.position.items():
        prev_position[eid] = pos
    return replace(state, prev_position=pmap(prev_position))

Interactions (portal, damage, push, tile)

• portal_system: Teleports collidable entrants to the paired portal.

• damage_system: Applies Damage/LethalDamage on co-location and cross paths; respects Immunity/Phasing.

• push_system: Pushes Pushable objects if destination is free; moves agent and pushable.

• Tile systems: Reward and Cost handling.

grid_universe.systems.portal

Portal teleportation system.

Moves entering collidable entities from a portal to its paired portal's position. An entity is considered entering if its previous position differs from the current one and it is present in the augmented trail for the portal's tile this step.

portal_system_entity

portal_system_entity(state, augmented_trail, portal_id)

Teleport entities entering the specified portal to its pair.

Source code in grid_universe/systems/portal.py

def portal_system_entity(
    state: State, augmented_trail: PMap[Position, PSet[EntityID]], portal_id: EntityID
) -> State:
    """Teleport entities entering the specified portal to its pair."""
    portal = state.portal.get(portal_id)
    portal_position = state.position.get(portal_id)
    if portal_position is None or portal is None:
        return state

    pair_position = state.position.get(portal.pair_entity)
    if pair_position is None:
        return state

    if is_blocked_at(state, pair_position, check_collidable=True):
        return state  # Teleport not possible

    entity_ids = set(augmented_trail.get(portal_position, pset())) & set(
        state.collidable
    )
    entering_entity_ids = {
        eid
        for eid in entity_ids
        if state.prev_position.get(eid) != state.position.get(eid)
        and state.position.get(eid) == portal_position
    }

    state_position = state.position
    for eid in entering_entity_ids:
        state_position = state_position.set(eid, pair_position)
    return replace(state, position=state_position)

portal_system

portal_system(state)

Apply portal teleportation for all portals in the state.

Source code in grid_universe/systems/portal.py

def portal_system(state: State) -> State:
    """Apply portal teleportation for all portals in the state."""
    augmented_trail: PMap[Position, PSet[EntityID]] = get_augmented_trail(
        state, pset(state.collidable)
    )
    for portal_id in state.portal:
        state = portal_system_entity(state, augmented_trail, portal_id)
    return state

grid_universe.systems.damage

Damage / lethal damage resolution system.

Rules (inclusion predicates): * Overlap: target_curr == damager_curr * Swap: target_prev == damager_curr AND target_curr == damager_prev * Trail intersection: (target_trail & damager_trail) != ∅ * Endpoint cross: target_curr == damager_prev AND (target_prev ∈ damager_trail OR damager_prev ∈ target_trail)

Exclusion (takes precedence): * Pure vacated origin: target steps onto damager_prev without swap and no path intersection (no trail overlap, target_prev not in damager_trail, and damager_prev/current not in target_trail)

Additionally

• A damager may harm a specific target at most once per turn (tracked via damage_hits).
• Self‑damage is ignored.
• Trail lookups are cached.

damage_system

damage_system(state)

Resolve damage / lethal interactions for this turn.

O(H * D + T) where

H = # entities with health D = # entities with damage/lethal components T = total trail entries this action

Source code in grid_universe/systems/damage.py

def damage_system(state: State) -> State:
    """Resolve damage / lethal interactions for this turn.

    Complexity: O(H * D + T) where
        H = # entities with health
        D = # entities with damage/lethal components
        T = total trail entries this action
    """
    health: PMap[EntityID, Health] = state.health
    dead: PMap[EntityID, Dead] = state.dead
    usage_limit: PMap[EntityID, UsageLimit] = state.usage_limit
    damage_hits: PSet[DamageHit] = state.damage_hits

    damager_ids = _candidate_damagers(state)
    trail_cache = _build_trail_cache(state)

    # Iterate over snapshot list to avoid issues if component maps structurally change.
    for target_id in list(state.health.keys()):
        health, dead, usage_limit, damage_hits = _apply_damage_for_target(
            state,
            target_id,
            health,
            dead,
            usage_limit,
            damage_hits,
            damager_ids,
            trail_cache,
        )

    return replace(
        state,
        health=health,
        dead=dead,
        usage_limit=usage_limit,
        damage_hits=damage_hits,
    )

grid_universe.systems.push

Push interaction system.

Enables entities (typically agents) to push adjacent entities marked with the Pushable component into the next cell along the interaction vector, provided the destination cell is free of blocking/collidable constraints. Supports multi-entity stacks at the source tile by moving all pushables.

compute_destination

compute_destination(state, current_pos, next_pos)

Compute push destination given current and occupant next positions.

Returns the square beyond next_pos in the movement direction, applying wrap logic if the state's move function is the wrapping one. None if outside bounds and not wrapping.

Source code in grid_universe/systems/push.py

def compute_destination(
    state: State, current_pos: Position, next_pos: Position
) -> Optional[Position]:
    """Compute push destination given current and occupant next positions.

    Returns the square beyond ``next_pos`` in the movement direction, applying
    wrap logic if the state's move function is the wrapping one. ``None`` if
    outside bounds and not wrapping.
    """
    dx = next_pos.x - current_pos.x
    dy = next_pos.y - current_pos.y
    dest_x = next_pos.x + dx
    dest_y = next_pos.y + dy

    if state.move_fn is wrap_around_move_fn:
        return wrap_position(dest_x, dest_y, state.width, state.height)

    target_position = Position(dest_x, dest_y)
    if not is_in_bounds(state, target_position):
        return None

    return target_position

push_system

push_system(state, eid, next_pos)

Attempt to push any pushable entities at next_pos.

Parameters:

Name	Type	Description	Default
state	State	Current immutable state.	required
eid	EntityID	Entity initiating the push (must have a position).	required
next_pos	Position	Adjacent position the entity is trying to move into.	required

Returns:

Name	Type	Description
State	State	Updated state with moved positions if push succeeds; original state otherwise.

Source code in grid_universe/systems/push.py

def push_system(state: State, eid: EntityID, next_pos: Position) -> State:
    """Attempt to push any pushable entities at ``next_pos``.

    Args:
        state (State): Current immutable state.
        eid (EntityID): Entity initiating the push (must have a position).
        next_pos (Position): Adjacent position the entity is trying to move into.

    Returns:
        State: Updated state with moved positions if push succeeds; original state otherwise.
    """
    current_pos = state.position.get(eid)
    if current_pos is None:
        return state

    # Is there a pushable object at next_pos?
    pushable_ids = entities_with_components_at(state, next_pos, state.pushable)
    if not pushable_ids:
        return state  # Nothing to push

    pushable_id = pushable_ids[0]
    push_to = compute_destination(state, current_pos, next_pos)
    if push_to is None:
        return state

    if is_blocked_at(state, push_to, check_collidable=True):
        return state  # Push not possible

    new_position = state.position.set(eid, next_pos)
    for pushable_id in pushable_ids:
        new_position = new_position.set(pushable_id, push_to)
        add_trail_position(state, pushable_id, push_to)

    return replace(state, position=new_position)

grid_universe.systems.tile

Tile interaction systems.

Applies passive score modifications for standing on tiles that carry Rewardable (positive) or Cost (negative) components which are not collected through the collectible system (i.e. non-pickup surfaces).

get_noncollectible_entities

get_noncollectible_entities(state, pos, component_map)

Return entity IDs at pos with a component but not collectible.

Source code in grid_universe/systems/tile.py

def get_noncollectible_entities(
    state: State,
    pos: Position,
    component_map: Union[PMap[EntityID, Rewardable], PMap[EntityID, Cost]],
) -> Set[EntityID]:
    """Return entity IDs at ``pos`` with a component but not collectible."""
    at_pos = entities_at(state, pos)
    ids = set(component_map.keys())
    collectible_ids = set(state.collectible.keys())
    return (at_pos & ids) - collectible_ids

tile_reward_system

tile_reward_system(state, eid)

Increase score for rewardable non-collectible entities at agent tile.

Source code in grid_universe/systems/tile.py

def tile_reward_system(state: State, eid: EntityID) -> State:
    """Increase score for rewardable non-collectible entities at agent tile."""
    pos = state.position.get(eid)
    if not is_valid_state(state, eid) or is_terminal_state(state, eid) or pos is None:
        return state

    reward_ids = get_noncollectible_entities(state, pos, state.rewardable)
    if not reward_ids:
        return state

    score = state.score + sum(state.rewardable[rid].amount for rid in reward_ids)
    return replace(state, score=score)

tile_cost_system

tile_cost_system(state, eid)

Decrease score for cost-bearing non-collectible entities at agent tile.

Source code in grid_universe/systems/tile.py

def tile_cost_system(state: State, eid: EntityID) -> State:
    """Decrease score for cost-bearing non-collectible entities at agent tile."""
    pos = state.position.get(eid)
    if not is_valid_state(state, eid) or is_terminal_state(state, eid) or pos is None:
        return state

    cost_ids = get_noncollectible_entities(state, pos, state.cost)
    if not cost_ids:
        return state

    score = state.score - sum(state.cost[cid].amount for cid in cost_ids)
    return replace(state, score=score)

Status and terminal

• Status tick/GC: Decrement time limits and garbage-collect expired/orphaned effects.

• Terminal: Win/lose conditions.

grid_universe.systems.status

Status effect lifecycle system.

Coordinates ticking and garbage collection of effect entities referenced by a Status component. Supports two limiter decorators:

• TimeLimit: decremented each step.
• UsageLimit: decremented by specific systems upon use (outside this file).

The system performs two phases: 1. Tick: Decrement all time limits for active effects. 2. GC: Remove orphaned or expired effect IDs, pruning both the owning entity's status set and the global entity map.

tick_time_limit

tick_time_limit(state, status, time_limit)

Decrement per-effect time limits present in status.

Source code in grid_universe/systems/status.py

def tick_time_limit(
    state: State,
    status: Status,
    time_limit: PMap[EntityID, TimeLimit],
) -> PMap[EntityID, TimeLimit]:
    """Decrement per-effect time limits present in ``status``."""
    for effect_id in status.effect_ids:
        if effect_id in time_limit:
            time_limit = time_limit.set(
                effect_id, TimeLimit(amount=time_limit[effect_id].amount - 1)
            )
    return time_limit

cleanup_effect

cleanup_effect(effect_id, effect_ids)

Remove effect_id from status if present.

Source code in grid_universe/systems/status.py

def cleanup_effect(
    effect_id: EntityID,
    effect_ids: PSet[EntityID],
) -> PSet[EntityID]:
    """Remove ``effect_id`` from status if present."""
    effect_ids = effect_ids.remove(effect_id)
    return effect_ids

is_effect_expired

is_effect_expired(effect_id, time_limit, usage_limit)

Return True if effect's time or usage limit has reached zero.

Source code in grid_universe/systems/status.py

def is_effect_expired(
    effect_id: EntityID,
    time_limit: PMap[EntityID, TimeLimit],
    usage_limit: PMap[EntityID, UsageLimit],
) -> bool:
    """Return True if effect's time or usage limit has reached zero."""
    if effect_id in time_limit and time_limit[effect_id].amount <= 0:
        return True
    if effect_id in usage_limit and usage_limit[effect_id].amount <= 0:
        return True
    return False

garbage_collect

garbage_collect(state, time_limit, usage_limit, status)

Remove orphaned or expired effects from status and entity maps.

Source code in grid_universe/systems/status.py

def garbage_collect(
    state: State,
    time_limit: PMap[EntityID, TimeLimit],
    usage_limit: PMap[EntityID, UsageLimit],
    status: Status,
) -> Status:
    """Remove orphaned or expired effects from status and entity maps."""
    effect_ids: PSet[EntityID] = status.effect_ids

    # Remove invalid effect_ids by checking all effect component maps using EffectType
    for effect_id in list(effect_ids):
        if all(
            effect_id not in getattr(state, effect_type.name.lower())
            for effect_type in EffectType
        ):
            effect_ids = cleanup_effect(effect_id, effect_ids)

    # Remove expired effect_ids
    for effect_id in list(effect_ids):
        if is_effect_expired(effect_id, time_limit, usage_limit):
            effect_ids = cleanup_effect(effect_id, effect_ids)

    return replace(status, effect_ids=effect_ids)

status_tick_system

status_tick_system(state)

Phase 1: decrement all active time limits.

Source code in grid_universe/systems/status.py

def status_tick_system(state: State) -> State:
    """Phase 1: decrement all active time limits."""
    state_status = state.status
    state_time_limit = state.time_limit

    for _, entity_status in state_status.items():
        state_time_limit = tick_time_limit(state, entity_status, state_time_limit)

    return replace(
        state,
        status=state_status,
        time_limit=state_time_limit,
    )

status_gc_system

status_gc_system(state)

Phase 2: prune orphaned / expired effects from statuses and entities.

Source code in grid_universe/systems/status.py

def status_gc_system(state: State) -> State:
    """Phase 2: prune orphaned / expired effects from statuses and entities."""
    state_status = state.status
    state_time_limit = state.time_limit
    state_usage_limit = state.usage_limit

    for entity_id, entity_status in state_status.items():
        entity_status = garbage_collect(
            state, state_time_limit, state_usage_limit, entity_status
        )
        state_status = state_status.set(entity_id, entity_status)

    return replace(
        state,
        status=state_status,
        time_limit=state_time_limit,
        usage_limit=state_usage_limit,
    )

status_system

status_system(state)

Run tick + GC phases for all statuses (public entry point).

Source code in grid_universe/systems/status.py

def status_system(state: State) -> State:
    """Run tick + GC phases for all statuses (public entry point)."""
    state = status_tick_system(state)
    state = status_gc_system(state)
    return state

grid_universe.systems.terminal

Terminal condition systems.

Defines win/lose evaluation utilities that set state.win or state.lose flags exactly once when conditions are met (objective success or agent death). These flags are side-channel indicators—other systems may short-circuit when the state is already terminal.

win_system

win_system(state, agent_id)

Set win flag if objective function returns True for agent.

Skips evaluation if state already terminal or agent invalid/dead.

Source code in grid_universe/systems/terminal.py

def win_system(state: State, agent_id: EntityID) -> State:
    """Set ``win`` flag if objective function returns True for agent.

    Skips evaluation if state already terminal or agent invalid/dead.
    """
    if not is_valid_state(state, agent_id) or is_terminal_state(state, agent_id):
        return state

    if state.objective_fn(state, agent_id):
        return replace(state, win=True)
    return state

lose_system

lose_system(state, agent_id)

Set lose flag if agent is dead (idempotent).

Source code in grid_universe/systems/terminal.py

def lose_system(state: State, agent_id: EntityID) -> State:
    """Set ``lose`` flag if agent is dead (idempotent)."""
    if agent_id in state.dead and not state.lose:
        return replace(state, lose=True)
    return state

turn_system

turn_system(state, agent_id)

Set lose flag if turn limit is reached.

Source code in grid_universe/systems/terminal.py

def turn_system(state: State, agent_id: EntityID) -> State:
    """Set ``lose`` flag if turn limit is reached."""
    state = replace(state, turn=state.turn + 1)
    if (
        state.turn_limit is not None
        and state.turn >= state.turn_limit
        and not state.win
    ):
        state = replace(state, lose=True)
    return state

Utilities

Grid and ECS helpers

• Grid helpers: bounds, blocking, wrap, push destination math.

• ECS helpers: query entities at a position and with components.

grid_universe.utils.grid

Grid math / collision helpers.

Utility predicates used by movement & push systems. Functions here are pure and intentionally lightweight to keep inner loops fast.

is_in_bounds

is_in_bounds(state, pos)

Return True if pos lies within the level rectangle.

Source code in grid_universe/utils/grid.py

def is_in_bounds(state: State, pos: Position) -> bool:
    """Return True if ``pos`` lies within the level rectangle."""
    return 0 <= pos.x < state.width and 0 <= pos.y < state.height

wrap_position

wrap_position(x, y, width, height)

Toroidal wrap for coordinates (used by wrap movement).

Source code in grid_universe/utils/grid.py

def wrap_position(x: int, y: int, width: int, height: int) -> Position:
    """Toroidal wrap for coordinates (used by wrap movement)."""
    return Position(x % width, y % height)

is_blocked_at

is_blocked_at(state, pos, check_collidable=True, check_pushable=True)

Return True if any blocking entity occupies pos.

Parameters:

Name	Type	Description	Default
state	State	World state.	required
pos	Position	Candidate destination.	required
check_collidable	bool	If True, treat Collidable as blocking (for agent movement); pushing may disable this to allow pushing into collidable tiles.	True

Source code in grid_universe/utils/grid.py

def is_blocked_at(
    state: State,
    pos: Position,
    check_collidable: bool = True,
    check_pushable: bool = True,
) -> bool:
    """Return True if any blocking entity occupies ``pos``.

    Args:
        state (State): World state.
        pos (Position): Candidate destination.
        check_collidable (bool): If True, treat ``Collidable`` as blocking (for agent movement);
            pushing may disable this to allow pushing into collidable tiles.
    """
    ids_at_pos: Set[EntityID] = entities_at(state, pos)
    for other_id in ids_at_pos:
        if (
            other_id in state.blocking
            or (check_pushable and other_id in state.pushable)
            or (check_collidable and other_id in state.collidable)
        ):
            return True
    return False

grid_universe.utils.ecs

ECS convenience queries.

Helper functions for querying entity/component relationships without introducing iteration logic into systems. All functions are pure and operate on the immutable :class:grid_universe.state.State snapshot.

Performance: entities_at uses a cached reverse index of the immutable State.position PMap to provide O(1) lookups per state snapshot.

entities_at

entities_at(state, pos)

Return entity IDs whose position equals pos.

Source code in grid_universe/utils/ecs.py

def entities_at(state: State, pos: Position) -> Set[EntityID]:
    """Return entity IDs whose position equals ``pos``."""
    idx = _position_index(state.position)
    return set(idx.get(pos, ()))

entities_with_components_at

entities_with_components_at(state, pos, *component_stores)

Return IDs at pos possessing all provided component stores.

Source code in grid_universe/utils/ecs.py

def entities_with_components_at(
    state: State, pos: Position, *component_stores: Mapping[EntityID, object]
) -> List[EntityID]:
    """Return IDs at ``pos`` possessing all provided component stores."""
    ids_at_pos: Set[EntityID] = entities_at(state, pos)
    for store in component_stores:
        ids_at_pos &= set(store.keys())
    return list(ids_at_pos)

Status/inventory and health helpers

• Status helpers: finding/consuming effects with limits.

• Inventory helpers: keys lookup; add/remove items.

• Health helpers: damage application and death check.

grid_universe.utils.status

Status effect utility helpers.

Pure helpers for querying, selecting and consuming effects referenced by a Status component. Separation from the system module keeps logic reusable across movement/pathfinding or interaction systems that need to consult or exhaust limited effects (e.g., usage-limited phasing).

has_effect

has_effect(state, effect_id)

Return True if effect_id exists in any runtime effect store.

Source code in grid_universe/utils/status.py

def has_effect(state: State, effect_id: EntityID) -> bool:
    """Return True if ``effect_id`` exists in any runtime effect store."""
    effect_maps: List[EffectMap] = [state.immunity, state.phasing, state.speed]
    for effect in effect_maps:
        if effect_id in effect:
            return True
    return False

valid_effect

valid_effect(state, effect_id)

Return True if effect has no expired time/usage limit.

Source code in grid_universe/utils/status.py

def valid_effect(state: State, effect_id: EntityID) -> bool:
    """Return True if effect has no expired time/usage limit."""
    # Only add effect if its time or usage limit is positive or unlimited
    if effect_id in state.time_limit and state.time_limit[effect_id].amount <= 0:
        return False
    if effect_id in state.usage_limit and state.usage_limit[effect_id].amount <= 0:
        return False
    return True

add_status

add_status(status, effect_id)

Return new Status with effect ID added.

Source code in grid_universe/utils/status.py

def add_status(status: Status, effect_id: EntityID) -> Status:
    """Return new ``Status`` with effect ID added."""
    return Status(effect_ids=status.effect_ids.add(effect_id))

remove_status

remove_status(status, effect_id)

Return new Status with effect ID removed.

Source code in grid_universe/utils/status.py

def remove_status(status: Status, effect_id: EntityID) -> Status:
    """Return new ``Status`` with effect ID removed."""
    return Status(effect_ids=status.effect_ids.remove(effect_id))

get_status_effect

get_status_effect(effect_ids, effects, time_limit, usage_limit)

Select a valid effect from effect_ids matching any provided store.

Selection rules: 1. Filter to effect IDs present in at least one supplied effect map. 2. Drop expired effects (time or usage limit <= 0). 3. Prefer effects without usage limits; otherwise lowest EID yields tie.

Source code in grid_universe/utils/status.py

def get_status_effect(
    effect_ids: PSet[EntityID],
    effects: Union[EffectMap, Sequence[EffectMap]],
    time_limit: PMap[EntityID, TimeLimit],
    usage_limit: PMap[EntityID, UsageLimit],
) -> Optional[EntityID]:
    """Select a valid effect from ``effect_ids`` matching any provided store.

    Selection rules:
    1. Filter to effect IDs present in at least one supplied effect map.
    2. Drop expired effects (time or usage limit <= 0).
    3. Prefer effects without usage limits; otherwise lowest EID yields tie.
    """
    effect_maps: List[EffectMap] = _normalize_effects(effects)

    # Effects present in any of the requested effect stores
    relevant = [
        eid for eid in effect_ids if any(eid in eff_map for eff_map in effect_maps)
    ]
    if not relevant:
        return None

    # Filter out expired effects
    valid: list[EntityID] = []
    for eid in relevant:
        # Expired by time
        if eid in time_limit and time_limit[eid].amount <= 0:
            continue
        # Expired by usage
        if eid in usage_limit and usage_limit[eid].amount <= 0:
            continue
        valid.append(eid)

    if not valid:
        return None

    # Deterministic order
    valid.sort()

    # Prefer effects without usage limits (infinite or time-limited)
    for eid in valid:
        if eid not in usage_limit:
            return eid

    # Otherwise, return the first remaining usage-limited effect
    return valid[0]

use_status_effect

use_status_effect(effect_id, usage_limit)

Consume one use from a usage-limited effect if present.

Source code in grid_universe/utils/status.py

def use_status_effect(
    effect_id: EntityID, usage_limit: PMap[EntityID, UsageLimit]
) -> PMap[EntityID, UsageLimit]:
    """Consume one use from a usage-limited effect if present."""
    if effect_id not in usage_limit:
        return usage_limit
    usage_limit = usage_limit.set(
        effect_id,
        replace(usage_limit[effect_id], amount=usage_limit[effect_id].amount - 1),
    )
    return usage_limit

use_status_effect_if_present

use_status_effect_if_present(effect_ids, effects, time_limit, usage_limit)

Select and consume an effect (if any) returning updated usage map.

Source code in grid_universe/utils/status.py

def use_status_effect_if_present(
    effect_ids: PSet[EntityID],
    effects: Union[EffectMap, Sequence[EffectMap]],
    time_limit: PMap[EntityID, TimeLimit],
    usage_limit: PMap[EntityID, UsageLimit],
) -> Tuple[PMap[EntityID, UsageLimit], Optional[EntityID]]:
    """Select and consume an effect (if any) returning updated usage map."""
    effect_maps: List[EffectMap] = _normalize_effects(effects)
    effect_id = get_status_effect(effect_ids, effect_maps, time_limit, usage_limit)
    if effect_id is not None:
        usage_limit = use_status_effect(effect_id, usage_limit)
    return usage_limit, effect_id

grid_universe.utils.inventory

Inventory manipulation helpers.

add_item

add_item(inventory, item_id)

Return a new inventory with item_id added.

Source code in grid_universe/utils/inventory.py

def add_item(inventory: Inventory, item_id: EntityID) -> Inventory:
    """Return a new inventory with ``item_id`` added."""
    return Inventory(item_ids=inventory.item_ids.add(item_id))

remove_item

remove_item(inventory, item_id)

Return a new inventory with item_id removed.

Source code in grid_universe/utils/inventory.py

def remove_item(inventory: Inventory, item_id: EntityID) -> Inventory:
    """Return a new inventory with ``item_id`` removed."""
    return Inventory(item_ids=inventory.item_ids.remove(item_id))

has_key_with_id

has_key_with_id(inventory, key_store, key_id)

Return ID of a key with key_id if present in inventory else None.

Source code in grid_universe/utils/inventory.py

def has_key_with_id(
    inventory: Inventory, key_store: Mapping[EntityID, Key], key_id: str
) -> EntityID | None:
    """Return ID of a key with ``key_id`` if present in inventory else None."""
    for item_id in inventory.item_ids:
        key = key_store.get(item_id)
        if key and key.key_id == key_id:
            return item_id
    return None

all_keys_with_id

all_keys_with_id(inventory, key_store, key_id)

Return persistent set of all key IDs matching key_id.

Source code in grid_universe/utils/inventory.py

def all_keys_with_id(
    inventory: Inventory, key_store: Mapping[EntityID, Key], key_id: str
) -> PSet[EntityID]:
    """Return persistent set of all key IDs matching ``key_id``."""
    return pset(
        item_id
        for item_id in inventory.item_ids
        if (k := key_store.get(item_id)) and k.key_id == key_id
    )

grid_universe.utils.health

Health and damage helpers.

apply_damage_and_check_death

apply_damage_and_check_death(health_dict, dead_dict, eid, damage, lethal)

Apply damage to entity and mark dead if lethal or HP reaches zero.

Source code in grid_universe/utils/health.py

def apply_damage_and_check_death(
    health_dict: PMap[EntityID, Health],
    dead_dict: PMap[EntityID, Dead],
    eid: EntityID,
    damage: int,
    lethal: bool,
) -> Tuple[PMap[EntityID, Health], PMap[EntityID, Dead]]:
    """Apply damage to entity and mark dead if lethal or HP reaches zero."""
    if eid in health_dict:
        hp = health_dict[eid]
        new_hp = max(0, hp.health - damage)
        health_dict = health_dict.set(
            eid, Health(health=new_hp, max_health=hp.max_health)
        )
        if new_hp == 0 or lethal:
            dead_dict = dead_dict.set(eid, Dead())
            health_dict = health_dict.set(
                eid, Health(health=0, max_health=hp.max_health)
            )
    else:
        if lethal:
            dead_dict = dead_dict.set(eid, Dead())
    return health_dict, dead_dict

Rendering helpers and image ops

• Numpy-based HSV recoloring preserving tone.

• Direction triangle overlays.

grid_universe.utils.image

Image utilities.

Vectorized helpers for lightweight image post-processing used by the renderer.

The primary goal is fast, deterministic recoloring of small RGBA sprite textures (e.g. 16x16 or 32x32) without introducing heavyweight dependencies or per-pixel Python loops. Operations are implemented with NumPy and operate on float32 buffers to balance precision and performance.

Key Functions

recolor_image_keep_tone: Re-hues an image to a target color while preserving original per-pixel luminance (value channel) and optionally original saturation. This enables palette swapping / team coloring while retaining shading.

draw_direction_triangles_on_image

Overlays directional arrow/triangle glyphs onto a texture to visualize movement intent or facing direction, used for debugging pathfinding or animating multi-step movements.

Implementation Notes

The HSV <-> RGB conversions are fully vectorized and attempt to minimize branching; alpha is preserved exactly unless explicitly recolored.

recolor_image_keep_tone

recolor_image_keep_tone(base, target_rgb, keep_saturation=True, saturation_mix=0.0, min_saturation=0.0)

Recolor non-transparent pixels by replacing Hue with target color's Hue, preserving per-pixel Value (brightness/tone). Saturation is preserved by default.

Source code in grid_universe/utils/image.py

def recolor_image_keep_tone(
    base: Image.Image,
    target_rgb: Tuple[int, int, int],
    keep_saturation: bool = True,
    saturation_mix: float = 0.0,
    min_saturation: float = 0.0,
) -> Image.Image:
    """
    Recolor non-transparent pixels by replacing Hue with target color's Hue,
    preserving per-pixel Value (brightness/tone). Saturation is preserved by default.
    """
    if base.mode != "RGBA":
        base = base.convert("RGBA")

    arr: UInt8Array = np.array(base, dtype=np.uint8)
    r8: UInt8Array = arr[..., 0]
    g8: UInt8Array = arr[..., 1]
    b8: UInt8Array = arr[..., 2]
    a8: UInt8Array = arr[..., 3]

    # Normalize to [0,1] float32
    r: FloatArray = r8.astype(np.float32) / 255.0
    g: FloatArray = g8.astype(np.float32) / 255.0
    b: FloatArray = b8.astype(np.float32) / 255.0

    visible: BoolArray = a8 > 0

    # Convert texture to HSV
    _, s, v = _rgb_to_hsv_np(r, g, b)

    # Target hue/saturation
    tr, tg, tb = (
        np.float32(target_rgb[0] / 255.0),
        np.float32(target_rgb[1] / 255.0),
        np.float32(target_rgb[2] / 255.0),
    )
    # Build constant arrays (same shape) with explicit dtype
    r_const: FloatArray = np.full_like(r, tr, dtype=np.float32)
    g_const: FloatArray = np.full_like(g, tg, dtype=np.float32)
    b_const: FloatArray = np.full_like(b, tb, dtype=np.float32)

    th, ts, _tv_unused = _rgb_to_hsv_np(r_const, g_const, b_const)

    # Replace hue with target hue
    h_new: FloatArray = th

    # Saturation strategy
    if keep_saturation and saturation_mix == 0.0:
        s_new: FloatArray = s
    else:
        mix = np.float32(np.clip(saturation_mix, 0.0, 1.0))
        s_new = ((1.0 - mix) * s + mix * ts).astype(np.float32)

    if min_saturation > 0.0:
        s_new = np.maximum(s_new, np.float32(min_saturation)).astype(np.float32)

    # Value stays the same
    v_new: FloatArray = v

    rr, gg, bb = _hsv_to_rgb_np(h_new, s_new, v_new)

    # Write back only for visible pixels
    out: UInt8Array = arr.copy()
    out_r: UInt8Array = (rr * 255.0).astype(np.uint8)
    out_g: UInt8Array = (gg * 255.0).astype(np.uint8)
    out_b: UInt8Array = (bb * 255.0).astype(np.uint8)

    out[..., 0][visible] = out_r[visible]
    out[..., 1][visible] = out_g[visible]
    out[..., 2][visible] = out_b[visible]
    # alpha unchanged
    return Image.fromarray(out, mode="RGBA")

draw_direction_triangles_on_image

draw_direction_triangles_on_image(image, size, dx, dy, count)

Draw 'count' filled triangles pointing (dx, dy) on the given RGBA image. Triangles are centered: the centroid of each triangle is symmetrically arranged around the image center. Spacing is between triangle centroids.

Source code in grid_universe/utils/image.py

def draw_direction_triangles_on_image(
    image: Image.Image, size: int, dx: int, dy: int, count: int
) -> Image.Image:
    """
    Draw 'count' filled triangles pointing (dx, dy) on the given RGBA image.
    Triangles are centered: the centroid of each triangle is symmetrically arranged
    around the image center. Spacing is between triangle centroids.
    """
    if count <= 0 or (dx, dy) == (0, 0):
        return image

    draw = ImageDraw.Draw(image)
    cx, cy = size // 2, size // 2

    # Triangle geometry (relative to size)
    tri_height = max(4, int(size * 0.16))
    tri_half_base = max(3, int(size * 0.10))
    spacing = max(2, int(size * 0.12))  # distance between triangle centroids

    # Axis-aligned direction and perpendicular
    ux, uy = dx, dy  # points toward the triangle tip
    px, py = -uy, ux  # perpendicular (for base width)

    # Offsets for centroids: 1 -> [0], 2 -> [-0.5s, +0.5s], 3 -> [-s, 0, +s], ...
    offsets = [(i - (count - 1) / 2.0) * spacing for i in range(count)]

    # For an isosceles triangle, the centroid lies 1/3 of the height from the base toward the tip.
    # If C is the centroid, then:
    #   tip = C + (2/3)*tri_height * u
    #   base_center = C - (1/3)*tri_height * u
    tip_offset = (2.0 / 3.0) * tri_height
    base_offset = (1.0 / 3.0) * tri_height

    for off in offsets:
        # Centroid position
        Cx = cx + int(round(ux * off))
        Cy = cy + int(round(uy * off))

        # Tip and base-center positions
        tip_x = int(round(Cx + ux * tip_offset))
        tip_y = int(round(Cy + uy * tip_offset))
        base_x = int(round(Cx - ux * base_offset))
        base_y = int(round(Cy - uy * base_offset))

        # Base vertices around base center along the perpendicular
        p1 = (tip_x, tip_y)
        p2 = (
            int(round(base_x + px * tri_half_base)),
            int(round(base_y + py * tri_half_base)),
        )
        p3 = (
            int(round(base_x - px * tri_half_base)),
            int(round(base_y - py * tri_half_base)),
        )

        draw.polygon([p1, p2, p3], fill=(255, 255, 255, 220), outline=(0, 0, 0, 220))

    return image

Trail and terminal checks

• Trail: record traversed positions between prev and current.

• Terminal: convenience validation and terminal checks.

grid_universe.utils.trail

Trail aggregation helpers.

Produces augmented trail maps merging current positions of specific entities with previously recorded traversed positions. Used by portal and potential AoE/effect systems to reason about paths taken rather than only endpoints.

get_augmented_trail

get_augmented_trail(state, entity_ids)

Return merged mapping of positions to entity sets (current + historic).

Parameters:

Name	Type	Description	Default
state	State	Current immutable world state containing both live entity positions and the accumulated historic trail mapping of prior positions.	required
entity_ids	PSet[EntityID]	Entity ids whose current position should be merged into the historic trail. Entities absent from state.position are ignored.	required

Returns:

Type	Description
PMap[Position, PSet[EntityID]]	PMap[Position, PSet[EntityID]]: Mapping from grid positions to the persistent set of entity ids that have either previously occupied (historic) or currently occupy that position among the provided tracked entities.

Source code in grid_universe/utils/trail.py

def get_augmented_trail(
    state: State, entity_ids: PSet[EntityID]
) -> PMap[Position, PSet[EntityID]]:
    """Return merged mapping of positions to entity sets (current + historic).

    Args:
        state (State): Current immutable world state containing both live entity positions
            and the accumulated historic ``trail`` mapping of prior positions.
        entity_ids (PSet[EntityID]): Entity ids whose current position should be merged
            into the historic trail. Entities absent from ``state.position`` are ignored.

    Returns:
        PMap[Position, PSet[EntityID]]: Mapping from grid positions to the persistent set of
            entity ids that have either previously occupied (historic) or currently occupy
            that position among the provided tracked entities.
    """
    pos_to_eids: DefaultDict[Position, Set[EntityID]] = defaultdict(set)
    for eid in entity_ids:
        if eid not in state.position:
            continue
        pos = state.position[eid]
        pos_to_eids[pos].add(eid)
    # Merge with existing trail:
    for pos, eid_set in state.trail.items():
        # ``eid_set`` is already a persistent set; its items are EntityID.
        pos_to_eids[pos].update(eid_set)
    # Convert to persistent structures:
    return pmap({pos: pset(eids) for pos, eids in pos_to_eids.items()})

add_trail_position

add_trail_position(state, entity_id, new_pos)

Return new state with entity_id recorded as having entered new_pos.

Idempotent for (entity, position) within an action: repeated additions of the same (entity, tile) pair are harmless due to set semantics.

Source code in grid_universe/utils/trail.py

def add_trail_position(state: State, entity_id: EntityID, new_pos: Position) -> State:
    """Return new state with ``entity_id`` recorded as having entered ``new_pos``.

    Idempotent for (entity, position) within an action: repeated additions of
    the same (entity, tile) pair are harmless due to set semantics.
    """
    return replace(
        state,
        trail=state.trail.set(new_pos, state.trail.get(new_pos, pset()).add(entity_id)),
    )

grid_universe.utils.terminal

Terminal condition helper predicates.

is_valid_state

is_valid_state(state, agent_id)

Return True if agent exists and has a position.

Source code in grid_universe/utils/terminal.py

def is_valid_state(state: State, agent_id: EntityID) -> bool:
    """Return True if agent exists and has a position."""
    return len(state.agent) > 0 and state.position.get(agent_id) is not None

is_terminal_state

is_terminal_state(state, agent_id)

Return True if state already satisfies win/lose or agent is dead.

Source code in grid_universe/utils/terminal.py

def is_terminal_state(state: State, agent_id: EntityID) -> bool:
    """Return True if state already satisfies win/lose or agent is dead."""
    return state.win or state.lose or agent_id in state.dead

GC (entity pruning)

• Prune entities not reachable from any live structure to keep State compact.

grid_universe.utils.gc

Garbage collection utilities.

Removes unreachable entity/component entries from state maps. Reachable entities include: * All IDs in the master entity map. * Effect entity IDs referenced by any Status component. * Item IDs referenced by any Inventory component.

The garbage collector prunes orphaned component entries (e.g., an effect map entry for an effect whose owning status no longer references it) which keeps state size bounded and avoids leaking stale objects during long simulations.

compute_alive_entities

compute_alive_entities(state)

Return the closure of entity IDs reachable from registries & references.

Source code in grid_universe/utils/gc.py

def compute_alive_entities(state: State) -> Set[EntityID]:
    """Return the closure of entity IDs reachable from registries & references."""
    alive: Set[EntityID] = set(state.position.keys())
    for stats in state.status.values():
        alive |= set(stats.effect_ids)
    for inv in state.inventory.values():
        alive |= set(inv.item_ids)
    return alive

run_garbage_collector

run_garbage_collector(state)

Prune component maps to only contain reachable entity IDs.

Source code in grid_universe/utils/gc.py

def run_garbage_collector(state: State) -> State:
    """Prune component maps to only contain reachable entity IDs."""
    alive = compute_alive_entities(state)
    new_fields: Dict[str, Any] = {}
    for field in state.__dataclass_fields__:
        value = getattr(state, field)
        if isinstance(value, type(pmap())):
            value_map = cast(PMap[EntityID, Any], value)
            filtered = pmap({k: v for k, v in value_map.items() if k in alive})
            new_fields[field] = filtered
    return replace(state, **new_fields)

Levels (Authoring)

Authoring model and factories

• Level: mutable grid of EntitySpec for authoring.

• EntitySpec: bag of components with authoring-only lists and wiring refs.

• Factories: ready-made objects (agent, floor, wall, key/door, portal, box, hazards, enemies, powerups).

grid_universe.levels.grid

Authoring grid representation (pre-immutable State).

Provides a simple editing API (add/remove/move) for building up a level prior to conversion. Use levels.factories helpers to create EntitySpec objects conveniently.

Level dataclass

Grid-centric, authoring-time level representation. - grid[y][x] is a list of EntityObject instances at that cell. - Level stores configuration like move_fn, objective_fn, seed, and simple meta (turn/score/etc.). - This module is State-agnostic. Use the converter (levels.convert.to_state / from_state) to bridge between Level and the immutable ECS State.

Source code in grid_universe/levels/grid.py

@dataclass
class Level:
    """
    Grid-centric, authoring-time level representation.
    - `grid[y][x]` is a list of `EntityObject` instances at that cell.
    - Level stores configuration like move_fn, objective_fn, seed, and simple meta (turn/score/etc.).
    - This module is State-agnostic. Use the converter (levels.convert.to_state / from_state)
      to bridge between Level and the immutable ECS State.
    """

    width: int
    height: int
    move_fn: MoveFn
    objective_fn: ObjectiveFn
    seed: Optional[int] = None

    # 2D array of cells: each cell holds a list of EntityObject
    grid: List[List[List[EntitySpec]]] = field(init=False)

    # Optional meta (carried through conversion)
    turn: int = 0
    score: int = 0
    win: bool = False
    lose: bool = False
    message: Optional[str] = None
    turn_limit: Optional[int] = None

    def __post_init__(self) -> None:
        # Initialize empty grid
        self.grid = [[[] for _ in range(self.width)] for _ in range(self.height)]

    # -------- Grid editing API (purely authoring-time) --------

    def add(self, pos: Position, obj: EntitySpec) -> None:
        """
        Place an EntityObject into the cell at pos (x, y).
        """
        x, y = pos
        self._check_bounds(x, y)
        self.grid[y][x].append(obj)

    def add_many(self, items: List[Tuple[Position, EntitySpec]]) -> None:
        """
        Place multiple EntityObject instances. Each entry is (pos, obj).
        """
        for pos, obj in items:
            self.add(pos, obj)

    def remove(self, pos: Position, obj: EntitySpec) -> bool:
        """
        Remove a specific EntityObject (by identity) from the cell at pos.
        Returns True if the object was found and removed, False otherwise.
        """
        x, y = pos
        self._check_bounds(x, y)
        cell = self.grid[y][x]
        for i, o in enumerate(cell):
            if o is obj:
                del cell[i]
                return True
        return False

    def remove_if(self, pos: Position, predicate: Callable[[EntitySpec], bool]) -> int:
        """
        Remove all objects in the cell at pos for which predicate(obj) is True.
        Returns the number of removed objects.
        """
        x, y = pos
        self._check_bounds(x, y)
        cell = self.grid[y][x]
        keep = [o for o in cell if not predicate(o)]
        removed = len(cell) - len(keep)
        self.grid[y][x] = keep
        return removed

    def move_obj(self, from_pos: Position, obj: EntitySpec, to_pos: Position) -> bool:
        """
        Move a specific EntityObject (by identity) from one cell to another.
        Returns True if moved (i.e., it was found in the source cell), False otherwise.
        """
        if not self.remove(from_pos, obj):
            return False
        self.add(to_pos, obj)
        return True

    def clear_cell(self, pos: Position) -> int:
        """
        Remove all objects from the cell at pos. Returns the number of removed objects.
        """
        x, y = pos
        self._check_bounds(x, y)
        n = len(self.grid[y][x])
        self.grid[y][x] = []
        return n

    def objects_at(self, pos: Position) -> List[EntitySpec]:
        """
        Return a shallow copy of the list of objects at pos.
        """
        x, y = pos
        self._check_bounds(x, y)
        return list(self.grid[y][x])

    # -------- Internal helpers --------

    def _check_bounds(self, x: int, y: int) -> None:
        if not (0 <= x < self.width and 0 <= y < self.height):
            raise IndexError(
                f"Out of bounds: {(x, y)} for grid {self.width}x{self.height}"
            )

add

add(pos, obj)

Place an EntityObject into the cell at pos (x, y).

Source code in grid_universe/levels/grid.py

def add(self, pos: Position, obj: EntitySpec) -> None:
    """
    Place an EntityObject into the cell at pos (x, y).
    """
    x, y = pos
    self._check_bounds(x, y)
    self.grid[y][x].append(obj)

add_many

add_many(items)

Place multiple EntityObject instances. Each entry is (pos, obj).

Source code in grid_universe/levels/grid.py

def add_many(self, items: List[Tuple[Position, EntitySpec]]) -> None:
    """
    Place multiple EntityObject instances. Each entry is (pos, obj).
    """
    for pos, obj in items:
        self.add(pos, obj)

remove

remove(pos, obj)

Remove a specific EntityObject (by identity) from the cell at pos. Returns True if the object was found and removed, False otherwise.

Source code in grid_universe/levels/grid.py

def remove(self, pos: Position, obj: EntitySpec) -> bool:
    """
    Remove a specific EntityObject (by identity) from the cell at pos.
    Returns True if the object was found and removed, False otherwise.
    """
    x, y = pos
    self._check_bounds(x, y)
    cell = self.grid[y][x]
    for i, o in enumerate(cell):
        if o is obj:
            del cell[i]
            return True
    return False

remove_if

remove_if(pos, predicate)

Remove all objects in the cell at pos for which predicate(obj) is True. Returns the number of removed objects.

Source code in grid_universe/levels/grid.py

def remove_if(self, pos: Position, predicate: Callable[[EntitySpec], bool]) -> int:
    """
    Remove all objects in the cell at pos for which predicate(obj) is True.
    Returns the number of removed objects.
    """
    x, y = pos
    self._check_bounds(x, y)
    cell = self.grid[y][x]
    keep = [o for o in cell if not predicate(o)]
    removed = len(cell) - len(keep)
    self.grid[y][x] = keep
    return removed

move_obj

move_obj(from_pos, obj, to_pos)

Move a specific EntityObject (by identity) from one cell to another. Returns True if moved (i.e., it was found in the source cell), False otherwise.

Source code in grid_universe/levels/grid.py

def move_obj(self, from_pos: Position, obj: EntitySpec, to_pos: Position) -> bool:
    """
    Move a specific EntityObject (by identity) from one cell to another.
    Returns True if moved (i.e., it was found in the source cell), False otherwise.
    """
    if not self.remove(from_pos, obj):
        return False
    self.add(to_pos, obj)
    return True

clear_cell

clear_cell(pos)

Remove all objects from the cell at pos. Returns the number of removed objects.

Source code in grid_universe/levels/grid.py

def clear_cell(self, pos: Position) -> int:
    """
    Remove all objects from the cell at pos. Returns the number of removed objects.
    """
    x, y = pos
    self._check_bounds(x, y)
    n = len(self.grid[y][x])
    self.grid[y][x] = []
    return n

objects_at

objects_at(pos)

Return a shallow copy of the list of objects at pos.

Source code in grid_universe/levels/grid.py

def objects_at(self, pos: Position) -> List[EntitySpec]:
    """
    Return a shallow copy of the list of objects at pos.
    """
    x, y = pos
    self._check_bounds(x, y)
    return list(self.grid[y][x])

grid_universe.levels.entity_spec

Authoring-time mutable entity specification.

EntitySpec instances gather optional component instances plus authoring metadata (inventory/status lists and wiring references). They are converted to immutable ECS entities by :mod:levels.convert.

EntitySpec dataclass

Mutable bag of ECS components for authoring (no Position here). Authoring-only wiring refs: - pathfind_target_ref: reference to another EntityObject to target - pathfinding_type: desired path type when wiring (if target ref set) - portal_pair_ref: reference to another EntityObject to pair with as a portal Authoring-only nested collections: - inventory: list of EntityObject (items carried; materialized as separate entities) - status: list of EntityObject (effects active; materialized as separate entities)

Source code in grid_universe/levels/entity_spec.py

@dataclass
class EntitySpec:
    """
    Mutable bag of ECS components for authoring (no Position here).
    Authoring-only wiring refs:
      - pathfind_target_ref: reference to another EntityObject to target
      - pathfinding_type: desired path type when wiring (if target ref set)
      - portal_pair_ref: reference to another EntityObject to pair with as a portal
    Authoring-only nested collections:
      - inventory: list of EntityObject (items carried; materialized as separate entities)
      - status: list of EntityObject (effects active; materialized as separate entities)
    """

    # Components
    agent: Optional[Agent] = None
    appearance: Optional[Appearance] = None
    blocking: Optional[Blocking] = None
    collectible: Optional[Collectible] = None
    collidable: Optional[Collidable] = None
    cost: Optional[Cost] = None
    damage: Optional[Damage] = None
    exit: Optional[Exit] = None
    health: Optional[Health] = None
    inventory: Optional[Inventory] = None
    key: Optional[Key] = None
    lethal_damage: Optional[LethalDamage] = None
    locked: Optional[Locked] = None
    moving: Optional[Moving] = None
    pathfinding: Optional[Pathfinding] = None
    portal: Optional[Portal] = None
    pushable: Optional[Pushable] = None
    required: Optional[Required] = None
    rewardable: Optional[Rewardable] = None
    status: Optional[Status] = None

    # Effects
    immunity: Optional[Immunity] = None
    phasing: Optional[Phasing] = None
    speed: Optional[Speed] = None
    time_limit: Optional[TimeLimit] = None
    usage_limit: Optional[UsageLimit] = None

    # Authoring-only nested objects (not State components)
    inventory_list: List["EntitySpec"] = field(default_factory=_empty_objs)
    status_list: List["EntitySpec"] = field(default_factory=_empty_objs)

    # Authoring-only wiring refs (resolved during conversion)
    pathfind_target_ref: Optional["EntitySpec"] = None
    pathfinding_type: Optional[PathfindingType] = None
    portal_pair_ref: Optional["EntitySpec"] = None

    def iter_components(self) -> List[Tuple[str, Any]]:
        """
        Yield (store_name, component) for non-None component fields that map to State stores.
        """
        out: List[Tuple[str, Any]] = []
        for _, store_name in COMPONENT_TO_FIELD.items():
            comp = getattr(self, store_name, None)
            if comp is not None:
                out.append((store_name, comp))
        return out

iter_components

iter_components()

Yield (store_name, component) for non-None component fields that map to State stores.

Source code in grid_universe/levels/entity_spec.py

def iter_components(self) -> List[Tuple[str, Any]]:
    """
    Yield (store_name, component) for non-None component fields that map to State stores.
    """
    out: List[Tuple[str, Any]] = []
    for _, store_name in COMPONENT_TO_FIELD.items():
        comp = getattr(self, store_name, None)
        if comp is not None:
            out.append((store_name, comp))
    return out

grid_universe.levels.factories

Convenience factory functions for authoring EntitySpec objects.

Each helper returns a preconfigured :class:EntitySpec with a common pattern (agent, floor, wall, coin, key, door, portal, hazards, effects, etc.). These are mutable authoring-time blueprints converted into immutable ECS entities by levels.convert.to_state.

create_agent

create_agent(health=5)

Player-controlled agent with health + inventory + empty status.

Source code in grid_universe/levels/factories.py

def create_agent(health: int = 5) -> EntitySpec:
    """Player-controlled agent with health + inventory + empty status."""
    return EntitySpec(
        agent=Agent(),
        appearance=Appearance(name=AppearanceName.HUMAN, priority=0),
        health=Health(health=health, max_health=health),
        collidable=Collidable(),
        inventory=Inventory(pset()),
        status=Status(pset()),
    )

create_floor

create_floor(cost_amount=1)

Background floor tile with movement cost.

Source code in grid_universe/levels/factories.py

def create_floor(cost_amount: int = 1) -> EntitySpec:
    """Background floor tile with movement cost."""
    return EntitySpec(
        appearance=Appearance(name=AppearanceName.FLOOR, background=True, priority=10),
        cost=Cost(amount=cost_amount),
    )

create_wall

create_wall()

Blocking wall tile.

Source code in grid_universe/levels/factories.py

def create_wall() -> EntitySpec:
    """Blocking wall tile."""
    return EntitySpec(
        appearance=Appearance(name=AppearanceName.WALL, background=True, priority=9),
        blocking=Blocking(),
    )

create_exit

create_exit()

Exit tile used in objectives.

Source code in grid_universe/levels/factories.py

def create_exit() -> EntitySpec:
    """Exit tile used in objectives."""
    return EntitySpec(
        appearance=Appearance(name=AppearanceName.EXIT, priority=9),
        exit=Exit(),
    )

create_coin

create_coin(reward=None)

Collectible coin awarding optional score when picked up.

Source code in grid_universe/levels/factories.py

def create_coin(reward: Optional[int] = None) -> EntitySpec:
    """Collectible coin awarding optional score when picked up."""
    return EntitySpec(
        appearance=Appearance(name=AppearanceName.COIN, icon=True, priority=4),
        collectible=Collectible(),
        rewardable=None if reward is None else Rewardable(amount=reward),
    )

create_core

create_core(reward=None, required=True)

Key objective collectible ("core") optionally giving reward.

Source code in grid_universe/levels/factories.py

def create_core(reward: Optional[int] = None, required: bool = True) -> EntitySpec:
    """Key objective collectible ("core") optionally giving reward."""
    return EntitySpec(
        appearance=Appearance(name=AppearanceName.CORE, icon=True, priority=4),
        collectible=Collectible(),
        rewardable=None if reward is None else Rewardable(amount=reward),
        required=Required() if required else None,
    )

create_key

create_key(key_id)

Key item unlocking doors with matching key_id.

Source code in grid_universe/levels/factories.py

def create_key(key_id: str) -> EntitySpec:
    """Key item unlocking doors with matching ``key_id``."""
    return EntitySpec(
        appearance=Appearance(name=AppearanceName.KEY, icon=True, priority=4),
        collectible=Collectible(),
        key=Key(key_id=key_id),
    )

create_door

create_door(key_id)

Locked door requiring a key with the same id.

Source code in grid_universe/levels/factories.py

def create_door(key_id: str) -> EntitySpec:
    """Locked door requiring a key with the same id."""
    return EntitySpec(
        appearance=Appearance(name=AppearanceName.DOOR, priority=6),
        blocking=Blocking(),
        locked=Locked(key_id=key_id),
    )

create_portal

create_portal(*, pair=None)

Portal endpoint (optionally auto-paired during authoring).

If pair is provided we set reciprocal refs so conversion wires the pair entities with each other's id.

Source code in grid_universe/levels/factories.py

def create_portal(*, pair: Optional[EntitySpec] = None) -> EntitySpec:
    """Portal endpoint (optionally auto-paired during authoring).

    If ``pair`` is provided we set reciprocal refs so conversion wires the
    pair entities with each other's id.
    """
    obj = EntitySpec(
        appearance=Appearance(name=AppearanceName.PORTAL, priority=7),
        portal=Portal(pair_entity=-1),
    )
    if pair is not None:
        obj.portal_pair_ref = pair
        if pair.portal_pair_ref is None:
            pair.portal_pair_ref = obj
    return obj

create_box

create_box(pushable=True, moving_axis=None, moving_direction=None, moving_bounce=True, moving_speed=1)

Pushable / blocking box (optionally not pushable).

Source code in grid_universe/levels/factories.py

def create_box(
    pushable: bool = True,
    moving_axis: Optional[MovingAxis] = None,
    moving_direction: Optional[int] = None,
    moving_bounce: bool = True,
    moving_speed: int = 1,
) -> EntitySpec:
    """Pushable / blocking box (optionally not pushable)."""
    return EntitySpec(
        appearance=Appearance(name=AppearanceName.BOX, priority=2),
        blocking=Blocking(),
        collidable=Collidable(),
        pushable=Pushable() if pushable else None,
        moving=None
        if moving_axis is None or moving_direction is None
        else Moving(
            axis=moving_axis,
            direction=moving_direction,
            bounce=moving_bounce,
            speed=moving_speed,
        ),
    )

create_monster

create_monster(damage=3, lethal=False, *, moving_axis=None, moving_direction=None, moving_bounce=True, moving_speed=1, pathfind_target=None, path_type=PathfindingType.PATH)

Basic enemy with damage and optional lethal + pathfinding target.

Source code in grid_universe/levels/factories.py

def create_monster(
    damage: int = 3,
    lethal: bool = False,
    *,
    moving_axis: Optional[MovingAxis] = None,
    moving_direction: Optional[int] = None,
    moving_bounce: bool = True,
    moving_speed: int = 1,
    pathfind_target: Optional[EntitySpec] = None,
    path_type: PathfindingType = PathfindingType.PATH,
) -> EntitySpec:
    """Basic enemy with damage and optional lethal + pathfinding target."""
    obj = EntitySpec(
        appearance=Appearance(name=AppearanceName.MONSTER, priority=1),
        collidable=Collidable(),
        damage=Damage(amount=damage),
        lethal_damage=LethalDamage() if lethal else None,
        moving=None
        if moving_axis is None or moving_direction is None
        else Moving(
            axis=moving_axis,
            direction=moving_direction,
            bounce=moving_bounce,
            speed=moving_speed,
        ),
    )
    if pathfind_target is not None:
        obj.pathfind_target_ref = pathfind_target
        obj.pathfinding_type = path_type
    return obj

create_hazard

create_hazard(appearance, damage, lethal=False, priority=7)

Static damaging (optionally lethal) tile-like hazard.

Source code in grid_universe/levels/factories.py

def create_hazard(
    appearance: AppearanceName,
    damage: int,
    lethal: bool = False,
    priority: int = 7,
) -> EntitySpec:
    """Static damaging (optionally lethal) tile-like hazard."""
    return EntitySpec(
        appearance=Appearance(name=appearance, priority=priority),
        collidable=Collidable(),
        damage=Damage(amount=damage),
        lethal_damage=LethalDamage() if lethal else None,
    )

create_speed_effect

create_speed_effect(multiplier, time=None, usage=None)

Collectible speed effect (optional time / usage limits).

Source code in grid_universe/levels/factories.py

def create_speed_effect(
    multiplier: int,
    time: Optional[int] = None,
    usage: Optional[int] = None,
) -> EntitySpec:
    """Collectible speed effect (optional time / usage limits)."""
    return EntitySpec(
        appearance=Appearance(name=AppearanceName.BOOTS, icon=True, priority=4),
        collectible=Collectible(),
        speed=Speed(multiplier=multiplier),
        time_limit=TimeLimit(amount=time) if time is not None else None,
        usage_limit=UsageLimit(amount=usage) if usage is not None else None,
    )

create_immunity_effect

create_immunity_effect(time=None, usage=None)

Collectible immunity effect (optional limits).

Source code in grid_universe/levels/factories.py

def create_immunity_effect(
    time: Optional[int] = None,
    usage: Optional[int] = None,
) -> EntitySpec:
    """Collectible immunity effect (optional limits)."""
    return EntitySpec(
        appearance=Appearance(name=AppearanceName.SHIELD, icon=True, priority=4),
        collectible=Collectible(),
        immunity=Immunity(),
        time_limit=TimeLimit(amount=time) if time is not None else None,
        usage_limit=UsageLimit(amount=usage) if usage is not None else None,
    )

create_phasing_effect

create_phasing_effect(time=None, usage=None)

Collectible phasing effect (optional limits).

Source code in grid_universe/levels/factories.py

def create_phasing_effect(
    time: Optional[int] = None,
    usage: Optional[int] = None,
) -> EntitySpec:
    """Collectible phasing effect (optional limits)."""
    return EntitySpec(
        appearance=Appearance(name=AppearanceName.GHOST, icon=True, priority=4),
        collectible=Collectible(),
        phasing=Phasing(),
        time_limit=TimeLimit(amount=time) if time is not None else None,
        usage_limit=UsageLimit(amount=usage) if usage is not None else None,
    )

Conversion between Level and State

• to_state: instantiate entities with Position; wire references; materialize nested lists.

• from_state: reconstruct authoring specs from positioned entities; restore lists and refs.

grid_universe.levels.convert

Conversion utilities between authoring Level and runtime State.

Two primary operations:

• to_state: Materialize immutable ECS world from a grid of EntitySpec.
• from_state: Reconstruct a mutable authoring representation from a live state.

Handles wiring of portals, pathfinding targets, inventory & status effect embedding (nested lists -> separate entities), and assigns deterministic EntityIDs.

to_state

to_state(level)

Convert a Level (grid of EntityObject) into an immutable State.

Semantics: - Copies all present ECS components from each EntityObject (including Inventory, Status) onto a new Entity. - Assigns Position for on-grid entities; nested inventory/effect entities have no Position. - Materializes authoring-only lists: * inventory_list: each item EntityObject becomes a new entity; its id is added to holder's Inventory.item_ids. If holder lacks an Inventory component, an empty one is created. * status_list: each effect EntityObject becomes a new entity; its id is added to holder's Status.effect_ids. If holder lacks a Status component, an empty one is created. - Wiring: * pathfind_target_ref: if set and the referenced object is placed, sets Pathfinding.target to that eid, creating a Pathfinding component if missing (type defaults to PathfindingType.PATH or uses obj.pathfinding_type). * portal_pair_ref: if set (on-grid for both ends), sets reciprocal Portal.pair_entity.

Source code in grid_universe/levels/convert.py

def to_state(level: Level) -> State:
    """
    Convert a Level (grid of EntityObject) into an immutable State.

    Semantics:
    - Copies all present ECS components from each EntityObject (including Inventory, Status) onto a new Entity.
    - Assigns Position for on-grid entities; nested inventory/effect entities have no Position.
    - Materializes authoring-only lists:
        * inventory_list: each item EntityObject becomes a new entity; its id is added to holder's Inventory.item_ids.
          If holder lacks an Inventory component, an empty one is created.
        * status_list: each effect EntityObject becomes a new entity; its id is added to holder's Status.effect_ids.
          If holder lacks a Status component, an empty one is created.
    - Wiring:
        * pathfind_target_ref: if set and the referenced object is placed, sets Pathfinding.target to that eid,
          creating a Pathfinding component if missing (type defaults to PathfindingType.PATH or uses obj.pathfinding_type).
        * portal_pair_ref: if set (on-grid for both ends), sets reciprocal Portal.pair_entity.
    """
    stores: Dict[str, Dict[EntityID, Any]] = _init_store_maps()
    next_eid_ref: List[int] = [0]

    # authoring-object -> eid for on-grid objects
    obj_to_eid: Dict[int, EntityID] = {}
    placed: List[Tuple[EntitySpec, EntityID]] = []

    for y in range(level.height):
        for x in range(level.width):
            for obj in level.grid[y][x]:
                eid = _alloc_from_obj(obj, stores, next_eid_ref, place_pos=(x, y))
                obj_to_eid[id(obj)] = eid
                placed.append((obj, eid))

                # Merge/ensure Inventory from component and/or authoring list
                if obj.inventory_list:
                    base_inv: Inventory = stores["inventory"].get(
                        eid,
                        obj.inventory
                        if obj.inventory is not None
                        else Inventory(pset()),
                    )
                    item_ids: List[EntityID] = [
                        _alloc_from_obj(item, stores, next_eid_ref, place_pos=None)
                        for item in obj.inventory_list
                    ]
                    stores["inventory"][eid] = Inventory(
                        item_ids=base_inv.item_ids.update(item_ids)
                    )
                elif obj.inventory is not None and eid not in stores["inventory"]:
                    stores["inventory"][eid] = obj.inventory

                # Merge/ensure Status from component and/or authoring list
                if obj.status_list:
                    base_status: Status = stores["status"].get(
                        eid, obj.status if obj.status is not None else Status(pset())
                    )
                    eff_ids: List[EntityID] = [
                        _alloc_from_obj(eff, stores, next_eid_ref, place_pos=None)
                        for eff in obj.status_list
                    ]
                    stores["status"][eid] = Status(
                        effect_ids=base_status.effect_ids.update(eff_ids)
                    )
                elif obj.status is not None and eid not in stores["status"]:
                    stores["status"][eid] = obj.status

    # Build immutable State before wiring
    state: State = _build_state(level, stores)

    # Wiring: pathfinding target references
    sp = state.pathfinding
    pf_changed = False
    for obj, eid in placed:
        tgt = obj.pathfind_target_ref
        if tgt is None:
            continue
        tgt_eid = obj_to_eid.get(id(tgt))
        if tgt_eid is None:
            continue
        desired_type: PathfindingType = obj.pathfinding_type or PathfindingType.PATH
        current = sp.get(eid)
        if current is None:
            sp = sp.set(eid, Pathfinding(target=tgt_eid, type=desired_type))
            pf_changed = True
        elif current.target is None:
            sp = sp.set(eid, Pathfinding(target=tgt_eid, type=current.type))
            pf_changed = True
    if pf_changed:
        state = replace(state, pathfinding=sp)

    # Wiring: portal pair references (bidirectional)
    spr = state.portal
    portal_changed = False
    for obj, eid in placed:
        mate = obj.portal_pair_ref
        if mate is None:
            continue
        mate_eid = obj_to_eid.get(id(mate))
        if mate_eid is None:
            continue
        spr = spr.set(eid, Portal(pair_entity=mate_eid))
        spr = spr.set(mate_eid, Portal(pair_entity=eid))
        portal_changed = True
    if portal_changed:
        state = replace(state, portal=spr)

    return state

from_state

from_state(state)

Convert an immutable State back into a mutable Level (grid of EntityObject).

Behavior: - Positioned entities are placed into Level.grid[y][x] in ascending eid order (deterministic). - EntityObject components (including Inventory/Status) are reconstructed for positioned entities. - Holder inventory_list/status_list are rebuilt from Inventory.item_ids / Status.effect_ids by reconstructing item/effect EntityObjects (not placed on the grid). - Authoring-time wiring refs (pathfind_target_ref, portal_pair_ref) are also restored for positioned entities when their targets/pairs are themselves positioned.

Source code in grid_universe/levels/convert.py

def from_state(state: State) -> Level:
    """
    Convert an immutable State back into a mutable Level (grid of EntityObject).

    Behavior:
    - Positioned entities are placed into `Level.grid[y][x]` in ascending eid order (deterministic).
    - EntityObject components (including Inventory/Status) are reconstructed for positioned entities.
    - Holder inventory_list/status_list are rebuilt from Inventory.item_ids / Status.effect_ids
      by reconstructing item/effect EntityObjects (not placed on the grid).
    - Authoring-time wiring refs (pathfind_target_ref, portal_pair_ref) are also restored for positioned
      entities when their targets/pairs are themselves positioned.
    """
    level = Level(
        width=state.width,
        height=state.height,
        move_fn=state.move_fn,
        objective_fn=state.objective_fn,
        seed=state.seed,
        turn=state.turn,
        score=state.score,
        turn_limit=state.turn_limit,
        win=state.win,
        lose=state.lose,
        message=state.message,
    )

    # eid -> positioned EntityObject
    placed_objs: Dict[EntityID, EntitySpec] = {}

    # Place entities on the grid
    for eid in sorted(state.position.keys()):
        pos = state.position.get(eid)
        if pos is None:
            continue
        x, y = pos.x, pos.y
        if not (0 <= x < level.width and 0 <= y < level.height):
            continue
        obj = _entity_object_from_state(state, eid)
        placed_objs[eid] = obj
        level.grid[y][x].append(obj)

    # Rebuild authoring lists from Inventory/Status sets
    for holder_eid, holder_obj in placed_objs.items():
        inv = state.inventory.get(holder_eid)
        if inv is not None and getattr(inv, "item_ids", None) is not None:
            for item_eid in inv.item_ids:
                holder_obj.inventory_list.append(
                    _entity_object_from_state(state, item_eid)
                )
            holder_obj.inventory = Inventory(pset())

        st = state.status.get(holder_eid)
        if st is not None and getattr(st, "effect_ids", None) is not None:
            for eff_eid in st.effect_ids:
                holder_obj.status_list.append(_entity_object_from_state(state, eff_eid))
            holder_obj.status = Status(pset())

    # Restore authoring-time wiring refs for positioned entities
    for eid, obj in placed_objs.items():
        # Pathfinding ref
        pf = state.pathfinding.get(eid)
        if pf is not None and pf.target is not None:
            tgt_obj = placed_objs.get(pf.target)
            if tgt_obj is not None:
                obj.pathfind_target_ref = tgt_obj
                obj.pathfinding_type = pf.type
            obj.pathfinding = None

        # Portal pair ref (bidirectional)
        pr = state.portal.get(eid)
        if pr is not None:
            mate_obj = placed_objs.get(pr.pair_entity)
            if mate_obj is not None:
                obj.portal_pair_ref = mate_obj
                # Set reciprocal if not already set
                if mate_obj.portal_pair_ref is None:
                    mate_obj.portal_pair_ref = obj
            obj.portal = Portal(pair_entity=-1)

    return level

Rendering

Texture renderer

• TextureRenderer: Compose background, main, corner icons; pick textures by (AppearanceName, properties); recolor groups; draw moving overlays.

grid_universe.renderer.texture

Texture-based renderer utilities.

Transforms a State into a composited RGBA image using per-entity Appearance metadata, property-derived texture variants, optional group recoloring and motion glyph overlays.

Rendering Model

1. Entities occupying the same cell are grouped into categories:
   • Background(s): appearance.background=True (e.g., floor, wall)
   • Main: highest-priority non-background entity
   • Corner Icons: up to four icon entities (appearance.icon=True) placed in tile corners (NW, NE, SW, SE)
   • Others: additional layered entities (drawn between background and main)
2. A texture path is chosen via an object + property signature lookup. If the path refers to a directory, a deterministic random selection occurs.
3. Group-based recoloring (e.g., matching keys and locks, portal pairs) applies a hue shift while preserving shading (value channel) and saturation rules.
4. Optional movement direction triangles are overlaid for moving entities.

Customization Hooks

• Provide a custom texture_map for alternative asset packs.
• Replace or extend DEFAULT_GROUP_RULES to recolor other sets of entities.
• Supply tex_lookup_fn to implement caching, animation frames, or atlas packing.

Performance Notes

• A lightweight cache key (path, size, group, movement vector, speed) helps reuse generated PIL images across frames.
• lru_cache on group_to_color ensures stable, deterministic colors without recomputing HSV conversions.

ObjectRendering dataclass

Lightweight container capturing render-relevant entity facets.

Attributes:

Name	Type	Description
appearance	Appearance	The entity's appearance component (or a default anonymous one).
properties	Tuple[str, ...]	Property component collection names (e.g. ('blocking', 'locked')) used to select texture variants.
group	str | None	Deterministic recolor group identifier.
move_dir	tuple[int, int] | None	(dx, dy) direction for movement glyph overlay.
move_speed	int	Movement speed (number of direction triangles to draw).

Source code in grid_universe/renderer/texture.py

@dataclass(frozen=True)
class ObjectRendering:
    """Lightweight container capturing render-relevant entity facets.

    Attributes:
        appearance (Appearance): The entity's appearance component (or a default anonymous one).
        properties (Tuple[str, ...]): Property component collection names (e.g. ``('blocking', 'locked')``)
            used to select texture variants.
        group (str | None): Deterministic recolor group identifier.
        move_dir (tuple[int, int] | None): (dx, dy) direction for movement glyph overlay.
        move_speed (int): Movement speed (number of direction triangles to draw).
    """

    appearance: Appearance
    properties: Tuple[str, ...]
    group: Optional[str] = None
    move_dir: Optional[Tuple[int, int]] = None
    move_speed: int = 0

    def asset(self) -> ObjectAsset:
        return (self.appearance.name, self.properties)

TextureRenderer

Source code in grid_universe/renderer/texture.py

class TextureRenderer:
    resolution: int
    subicon_percent: float
    texture_map: TextureMap
    asset_root: str
    tex_lookup_fn: Optional[TexLookupFn]

    def __init__(
        self,
        resolution: int = DEFAULT_RESOLUTION,
        subicon_percent: float = DEFAULT_SUBICON_PERCENT,
        texture_map: Optional[TextureMap] = None,
        asset_root: str = DEFAULT_ASSET_ROOT,
        tex_lookup_fn: Optional[TexLookupFn] = None,
    ):
        self.resolution = resolution
        self.subicon_percent = subicon_percent
        self.texture_map = texture_map or DEFAULT_TEXTURE_MAP
        self.asset_root = asset_root
        self.tex_lookup_fn = tex_lookup_fn

    def render(self, state: State) -> Image.Image:
        """Render convenience wrapper using stored configuration."""
        return render(
            state,
            resolution=self.resolution,
            subicon_percent=self.subicon_percent,
            texture_map=self.texture_map,
            asset_root=self.asset_root,
            tex_lookup_fn=self.tex_lookup_fn,
        )

render

render(state)

Render convenience wrapper using stored configuration.

Source code in grid_universe/renderer/texture.py

def render(self, state: State) -> Image.Image:
    """Render convenience wrapper using stored configuration."""
    return render(
        state,
        resolution=self.resolution,
        subicon_percent=self.subicon_percent,
        texture_map=self.texture_map,
        asset_root=self.asset_root,
        tex_lookup_fn=self.tex_lookup_fn,
    )

derive_groups

derive_groups(state, rules=DEFAULT_GROUP_RULES)

Apply grouping rules to each entity.

Later rendering stages may use groups to recolor related entities with a shared hue (e.g., all portals in a pair share the same color while still using the original texture shading).

Parameters:

Name	Type	Description	Default
state	State	Immutable simulation state.	required
rules	List[GroupRule]	Ordered list of functions; first non-None group returned is used.	DEFAULT_GROUP_RULES

Returns:

Type	Description
Dict[EntityID, Optional[str]]	Dict[EntityID, str | None]: Mapping of entity id to chosen group id (or None if ungrouped).

Source code in grid_universe/renderer/texture.py

def derive_groups(
    state: State, rules: List[GroupRule] = DEFAULT_GROUP_RULES
) -> Dict[EntityID, Optional[str]]:
    """Apply grouping rules to each entity.

    Later rendering stages may use groups to recolor related entities with a
    shared hue (e.g., all portals in a pair share the same color while still
    using the original texture shading).

    Args:
        state (State): Immutable simulation state.
        rules (List[GroupRule]): Ordered list of functions; first non-None group returned is used.

    Returns:
        Dict[EntityID, str | None]: Mapping of entity id to chosen group id (or ``None`` if ungrouped).
    """
    rule_groups: dict[str, set[str]] = defaultdict(set)
    out: Dict[EntityID, Optional[str]] = {}
    for eid, _ in state.position.items():
        group: Optional[str] = None
        for rule in rules:
            group = rule(state, eid)
            if group is not None:
                rule_groups[rule.__name__].add(group)
                break
        out[eid] = group
    for groups in rule_groups.values():
        if len(groups) > 1:
            continue
        # remove singleton groups to avoid unnecessary recoloring
        for eid, group in out.items():
            if group in groups:
                out[eid] = None
    return out

group_to_color cached

group_to_color(group_id)

Deterministically map a group string to an RGB color.

Uses the group id as a seed to generate stable but visually distinct HSV values, then converts them to RGB.

Source code in grid_universe/renderer/texture.py

@lru_cache(maxsize=2048)
def group_to_color(group_id: str) -> Tuple[int, int, int]:
    """Deterministically map a group string to an RGB color.

    Uses the group id as a seed to generate stable but visually distinct HSV
    values, then converts them to RGB.
    """
    rng = random.Random(group_id)
    h = rng.random()
    s = 0.6 + 0.3 * rng.random()
    v = 0.7 + 0.25 * rng.random()
    r, g, b = colorsys.hsv_to_rgb(h, s, v)
    return int(r * 255), int(g * 255), int(b * 255)

apply_recolor_if_group

apply_recolor_if_group(tex, group)

Recolor wrapper that sets hue to the group's color while preserving tone.

Delegates to :func:recolor_image_keep_tone; if no group is provided the texture is returned unchanged.

Source code in grid_universe/renderer/texture.py

def apply_recolor_if_group(
    tex: Image.Image,
    group: Optional[str],
) -> Image.Image:
    """Recolor wrapper that sets hue to the group's color while preserving tone.

    Delegates to :func:`recolor_image_keep_tone`; if no group is provided the
    texture is returned unchanged.
    """
    if group is None:
        return tex
    color = group_to_color(group)
    return recolor_image_keep_tone(tex, color)

load_texture

load_texture(path, size)

Load and resize a texture, returning None if inaccessible or invalid.

Source code in grid_universe/renderer/texture.py

def load_texture(path: str, size: int) -> Optional[Image.Image]:
    """Load and resize a texture, returning None if inaccessible or invalid."""
    try:
        return Image.open(path).convert("RGBA").resize((size, size))
    except Exception:
        return None

get_object_renderings

get_object_renderings(state, eids, groups)

Build rendering descriptors for entity IDs in a single cell.

Inspects component PMaps on the State to infer property labels, movement direction and speed, then packages them in ObjectRendering objects for subsequent texture lookup and layering decisions.

Source code in grid_universe/renderer/texture.py

def get_object_renderings(
    state: State, eids: List[EntityID], groups: Dict[EntityID, Optional[str]]
) -> List[ObjectRendering]:
    """Build rendering descriptors for entity IDs in a single cell.

    Inspects component PMaps on the ``State`` to infer property labels,
    movement direction and speed, then packages them in ``ObjectRendering``
    objects for subsequent texture lookup and layering decisions.
    """
    renderings: List[ObjectRendering] = []
    default_appearance: Appearance = Appearance(name=AppearanceName.NONE)
    for eid in eids:
        appearance = state.appearance.get(eid, default_appearance)
        properties = tuple(
            [
                component
                for component, value in state.__dict__.items()
                if isinstance(value, type(pmap())) and eid in value
            ]
        )

        move_dir: Optional[Tuple[int, int]] = None
        move_speed: int = 0
        if eid in state.moving:
            m = state.moving[eid]
            if m.axis.name == "HORIZONTAL":
                move_dir = (1 if m.direction > 0 else -1, 0)
            else:
                move_dir = (0, 1 if m.direction > 0 else -1)
            move_speed = m.speed

        renderings.append(
            ObjectRendering(
                appearance=appearance,
                properties=properties,
                group=groups.get(eid),
                move_dir=move_dir,
                move_speed=move_speed,
            )
        )
    return renderings

choose_background

choose_background(object_renderings)

Select the highest-priority background object.

Raises

ValueError If no candidate background exists in the cell.

Source code in grid_universe/renderer/texture.py

def choose_background(object_renderings: List[ObjectRendering]) -> ObjectRendering:
    """Select the highest-priority background object.

    Raises
    ------
    ValueError
        If no candidate background exists in the cell.
    """
    items = [
        object_rendering
        for object_rendering in object_renderings
        if object_rendering.appearance.background
    ]
    if len(items) == 0:
        raise ValueError(f"No matching background: {object_renderings}")
    return sorted(items, key=lambda x: x.appearance.priority)[
        -1
    ]  # take the lowest priority

choose_main

choose_main(object_renderings)

Select main (foreground) object: lowest appearance priority value.

Returns None if no non-background objects exist.

Source code in grid_universe/renderer/texture.py

def choose_main(object_renderings: List[ObjectRendering]) -> Optional[ObjectRendering]:
    """Select main (foreground) object: lowest appearance priority value.

    Returns ``None`` if no non-background objects exist.
    """
    items = [
        object_rendering
        for object_rendering in object_renderings
        if not object_rendering.appearance.background
    ]
    if len(items) == 0:
        return None
    return sorted(items, key=lambda x: x.appearance.priority)[
        0
    ]  # take the highest priority

choose_corner_icons

choose_corner_icons(object_renderings, main)

Return up to four icon objects (excluding main) sorted by priority.

Source code in grid_universe/renderer/texture.py

def choose_corner_icons(
    object_renderings: List[ObjectRendering], main: Optional[ObjectRendering]
) -> List[ObjectRendering]:
    """Return up to four icon objects (excluding main) sorted by priority."""
    items = set(
        [
            object_rendering
            for object_rendering in object_renderings
            if object_rendering.appearance.icon
        ]
    ) - set([main])
    return sorted(items, key=lambda x: x.appearance.priority)[
        :4
    ]  # take the highest priority

get_path

get_path(object_asset, texture_hmap)

Resolve a texture path for an object asset signature.

Attempts to find the nearest matching property tuple (maximizing shared properties, minimizing unmatched) to allow textures that only specify a subset of possible property labels.

Source code in grid_universe/renderer/texture.py

def get_path(
    object_asset: ObjectAsset, texture_hmap: ObjectPropertiesTextureMap
) -> str:
    """Resolve a texture path for an object asset signature.

    Attempts to find the nearest matching property tuple (maximizing shared
    properties, minimizing unmatched) to allow textures that only specify a
    subset of possible property labels.
    """
    object_name, object_properties = object_asset
    if object_name not in texture_hmap:
        raise ValueError(f"Object rendering {object_asset} is not found in texture map")
    nearest_object_properties = sorted(
        texture_hmap[object_name].keys(),
        key=lambda x: len(set(x).intersection(object_properties))
        - len(set(x) - set(object_properties)),
        reverse=True,
    )[0]
    return texture_hmap[object_name][nearest_object_properties]

select_texture_from_directory

select_texture_from_directory(dir, seed)

Choose a deterministic random texture file from a directory.

Source code in grid_universe/renderer/texture.py

def select_texture_from_directory(
    dir: str,
    seed: Optional[int],
) -> Optional[str]:
    """Choose a deterministic random texture file from a directory."""
    if not os.path.isdir(dir):
        return None

    try:
        entries = os.listdir(dir)
    except (FileNotFoundError, NotADirectoryError, PermissionError, OSError):
        return None

    files = sorted(
        f for f in entries if f.lower().endswith((".png", ".jpg", ".jpeg", ".gif"))
    )
    if not files:
        return None

    rng = random.Random(seed)
    chosen = rng.choice(files)
    return os.path.join(dir, chosen)

render

render(state, resolution=DEFAULT_RESOLUTION, subicon_percent=DEFAULT_SUBICON_PERCENT, texture_map=None, asset_root=DEFAULT_ASSET_ROOT, tex_lookup_fn=None, cache=None)

Render a State into a PIL Image.

Parameters:

Name	Type	Description	Default
state	State	Immutable game state to visualize.	required
resolution	int	Output image width in pixels (height derived from aspect ratio).	DEFAULT_RESOLUTION
subicon_percent	float	Relative size of corner icons compared to a cell's size.	DEFAULT_SUBICON_PERCENT
texture_map	TextureMap | None	Mapping from (appearance name, property tuple) to asset path.	None
asset_root	str	Root directory containing the asset hierarchy (e.g. "assets").	DEFAULT_ASSET_ROOT
tex_lookup_fn	TexLookupFn | None	Override for texture loading/recoloring/overlay logic.	None
cache	dict | None	Mutable memoization dict keyed by (path, size, group, move_dir, speed).	None

Returns:

Type	Description
Image	Image.Image: Composited RGBA image of the entire grid.

Source code in grid_universe/renderer/texture.py

def render(
    state: State,
    resolution: int = DEFAULT_RESOLUTION,
    subicon_percent: float = DEFAULT_SUBICON_PERCENT,
    texture_map: Optional[TextureMap] = None,
    asset_root: str = DEFAULT_ASSET_ROOT,
    tex_lookup_fn: Optional[TexLookupFn] = None,
    cache: Optional[
        Dict[
            Tuple[str, int, Optional[str], Optional[Tuple[int, int]], int],
            Optional[Image.Image],
        ]
    ] = None,
) -> Image.Image:
    """Render a ``State`` into a PIL Image.

    Args:
        state (State): Immutable game state to visualize.
        resolution (int): Output image width in pixels (height derived from aspect ratio).
        subicon_percent (float): Relative size of corner icons compared to a cell's size.
        texture_map (TextureMap | None): Mapping from ``(appearance name, property tuple)`` to asset path.
        asset_root (str): Root directory containing the asset hierarchy (e.g. ``"assets"``).
        tex_lookup_fn (TexLookupFn | None): Override for texture loading/recoloring/overlay logic.
        cache (dict | None): Mutable memoization dict keyed by ``(path, size, group, move_dir, speed)``.

    Returns:
        Image.Image: Composited RGBA image of the entire grid.
    """
    cell_size: int = resolution // state.width
    subicon_size: int = int(cell_size * subicon_percent)

    if texture_map is None:
        texture_map = DEFAULT_TEXTURE_MAP

    if cache is None:
        cache = {}

    texture_hmap: ObjectPropertiesTextureMap = defaultdict(dict)
    for (obj_name, obj_properties), value in texture_map.items():
        texture_hmap[obj_name][tuple(obj_properties)] = value

    width, height = state.width, state.height
    img = Image.new(
        "RGBA", (width * cell_size, height * cell_size), (128, 128, 128, 255)
    )

    state_rng = random.Random(state.seed)
    object_seeds = [state_rng.randint(0, 2**31) for _ in range(len(texture_map))]
    texture_map_values = list(texture_map.values())
    groups = derive_groups(state)

    def default_get_tex(
        object_rendering: ObjectRendering, size: int
    ) -> Optional[Image.Image]:
        path = get_path(object_rendering.asset(), texture_hmap)
        if not path:
            return None

        asset_path = f"{asset_root}/{path}"
        if os.path.isdir(asset_path):
            asset_index = texture_map_values.index(path)
            selected_asset_path = select_texture_from_directory(
                asset_path, object_seeds[asset_index]
            )
            if selected_asset_path is None:
                return None
            asset_path = selected_asset_path

        key = (
            asset_path,
            size,
            object_rendering.group,
            object_rendering.move_dir,
            object_rendering.move_speed,
        )
        if key in cache:
            return cache[key]

        texture = load_texture(asset_path, size)
        if texture is None:
            return None

        texture = apply_recolor_if_group(texture, object_rendering.group)
        if object_rendering.move_dir is not None and object_rendering.move_speed > 0:
            dx, dy = object_rendering.move_dir
            texture = draw_direction_triangles_on_image(
                texture.copy(), size, dx, dy, object_rendering.move_speed
            )

        cache[key] = texture
        return texture

    tex_lookup = tex_lookup_fn or default_get_tex

    grid_entities: Dict[Tuple[int, int], List[EntityID]] = {}
    for eid, pos in state.position.items():
        grid_entities.setdefault((pos.x, pos.y), []).append(eid)

    for (x, y), eids in grid_entities.items():
        x0, y0 = x * cell_size, y * cell_size

        object_renderings = get_object_renderings(state, eids, groups)

        background = choose_background(object_renderings)
        main = choose_main(object_renderings)
        corner_icons = choose_corner_icons(object_renderings, main)
        others = list(
            set(object_renderings) - set([main] + corner_icons + [background])
        )

        primary_renderings: List[ObjectRendering] = (
            [background] + others + ([main] if main is not None else [])
        )

        for object_rendering in primary_renderings:
            object_tex = tex_lookup(object_rendering, cell_size)
            if object_tex:
                img.alpha_composite(object_tex, (x0, y0))

        for idx, corner_icon in enumerate(corner_icons[:4]):
            dx = x0 + (cell_size - subicon_size if idx % 2 == 1 else 0)
            dy = y0 + (cell_size - subicon_size if idx // 2 == 1 else 0)
            tex = tex_lookup(corner_icon, subicon_size)
            if tex:
                img.alpha_composite(tex, (dx, dy))

    return img

Examples

Procedural maze

• Rich generator with walls/floors, agent/exit, keys/doors, portals, enemies, hazards, and powerups. Provides knobs for density and counts.

grid_universe.examples.maze

Procedural maze level generator example.

This module demonstrates authoring a parameterized maze-based level using the Level authoring API and factory helpers, then converting to an immutable State suitable for simulation or Gym-style environments.

Design Goals

• Showcase composition of factories (agent, walls, doors, portals, hazards, power-ups, enemies) with authoring-time references (e.g., portal pairing, enemy pathfinding target reference to the agent) that are resolved during to_state conversion.
• Provide tunable difficulty levers: wall density, counts of required objectives, rewards, hazards, enemies, doors, portals and power-ups.
• Illustrate how movement styles (static, directional patrol, straight-line pathfinding, full pathfinding) can be expressed via component choices.

Usage Example

from grid_universe.examples import maze
state = maze.generate(width=20, height=20, seed=123)

# Render / step the state using the engine's systems or gym wrapper.

Key Concepts Illustrated

Required Items: Use cores flagged as required=True which the default objective logic expects to be collected before reaching the exit. Power-Ups: Effects created with optional time or usage limits (speed, immunity, phasing) acting as pickups. Enemies: Configurable movement style and lethality; pathfinding enemies reference the agent to resolve target entity IDs later. Essential Path: Minimal union of shortest paths that touch required items and exit. Other entities (hazards, enemies, boxes) prefer non-essential cells.

generate

generate(width, height, num_required_items=1, num_rewardable_items=1, num_portals=1, num_doors=1, health=5, movement_cost=1, required_item_reward=10, rewardable_item_reward=10, boxes=DEFAULT_BOXES, powerups=DEFAULT_POWERUPS, hazards=DEFAULT_HAZARDS, enemies=DEFAULT_ENEMIES, wall_percentage=0.8, move_fn=default_move_fn, objective_fn=default_objective_fn, seed=None, turn_limit=None)

Generate a randomized maze game state.

This function orchestrates maze carving, tile classification, entity placement and authoring-time reference wiring before producing the immutable simulation State.

Parameters:

Name	Type	Description	Default
width	int	Width of the maze grid.	required
height	int	Height of the maze grid.	required
num_required_items	int	Number of required cores that must be collected before exit.	1
num_rewardable_items	int	Number of optional reward coins.	1
num_portals	int	Number of portal pairs to place (each pair consumes two open cells).	1
num_doors	int	Number of door/key pairs; each door is locked by its matching key.	1
health	int	Initial agent health points.	5
movement_cost	int	Per-tile movement cost encoded in floor components.	1
required_item_reward	int	Reward granted for collecting each required item.	10
rewardable_item_reward	int	Reward granted for each optional reward item (coin).	10
boxes	List[BoxSpec]	List defining (pushable?, speed) for box entities; speed > 0 creates moving boxes.	DEFAULT_BOXES
powerups	List[PowerupSpec]	Effect specifications converted into pickup entities.	DEFAULT_POWERUPS
hazards	List[HazardSpec]	Hazard specifications (appearance, damage, lethal).	DEFAULT_HAZARDS
enemies	List[EnemySpec]	Enemy specifications (damage, lethal, movement type, speed).	DEFAULT_ENEMIES
wall_percentage	float	Fraction of original maze walls to retain (0.0 => open field, 1.0 => perfect maze).	0.8
move_fn	MoveFn	Movement candidate function injected into the level.	default_move_fn
objective_fn	ObjectiveFn	Win condition predicate injected into the level.	default_objective_fn
seed	int | None	RNG seed for deterministic generation.	None

Returns:

Name	Type	Description
State	State	Fully wired immutable state ready for simulation.

Source code in grid_universe/examples/maze.py

def generate(
    width: int,
    height: int,
    num_required_items: int = 1,
    num_rewardable_items: int = 1,
    num_portals: int = 1,
    num_doors: int = 1,
    health: int = 5,
    movement_cost: int = 1,
    required_item_reward: int = 10,
    rewardable_item_reward: int = 10,
    boxes: List[BoxSpec] = DEFAULT_BOXES,
    powerups: List[PowerupSpec] = DEFAULT_POWERUPS,
    hazards: List[HazardSpec] = DEFAULT_HAZARDS,
    enemies: List[EnemySpec] = DEFAULT_ENEMIES,
    wall_percentage: float = 0.8,
    move_fn: MoveFn = default_move_fn,
    objective_fn: ObjectiveFn = default_objective_fn,
    seed: Optional[int] = None,
    turn_limit: Optional[int] = None,
) -> State:
    """Generate a randomized maze game state.

    This function orchestrates maze carving, tile classification, entity
    placement and authoring-time reference wiring before producing the
    immutable simulation ``State``.

    Args:
        width (int): Width of the maze grid.
        height (int): Height of the maze grid.
        num_required_items (int): Number of required cores that must be collected before exit.
        num_rewardable_items (int): Number of optional reward coins.
        num_portals (int): Number of portal pairs to place (each pair consumes two open cells).
        num_doors (int): Number of door/key pairs; each door is locked by its matching key.
        health (int): Initial agent health points.
        movement_cost (int): Per-tile movement cost encoded in floor components.
        required_item_reward (int): Reward granted for collecting each required item.
        rewardable_item_reward (int): Reward granted for each optional reward item (coin).
        boxes (List[BoxSpec]): List defining ``(pushable?, speed)`` for box entities; speed > 0 creates moving boxes.
        powerups (List[PowerupSpec]): Effect specifications converted into pickup entities.
        hazards (List[HazardSpec]): Hazard specifications ``(appearance, damage, lethal)``.
        enemies (List[EnemySpec]): Enemy specifications ``(damage, lethal, movement type, speed)``.
        wall_percentage (float): Fraction of original maze walls to retain (``0.0`` => open field, ``1.0`` => perfect maze).
        move_fn (MoveFn): Movement candidate function injected into the level.
        objective_fn (ObjectiveFn): Win condition predicate injected into the level.
        seed (int | None): RNG seed for deterministic generation.

    Returns:
        State: Fully wired immutable state ready for simulation.
    """
    rng = random.Random(seed)

    # 1) Base maze -> adjust walls
    maze_grid = generate_perfect_maze(width, height, rng)
    maze_grid = adjust_maze_wall_percentage(maze_grid, wall_percentage, rng)

    # 2) Level
    level = Level(
        width=width,
        height=height,
        move_fn=move_fn,
        objective_fn=objective_fn,
        seed=seed,
        turn_limit=turn_limit,
    )

    # 3) Collect positions
    open_positions: List[Position] = [
        pos for pos, is_open in maze_grid.items() if is_open
    ]
    wall_positions: List[Position] = [
        pos for pos, is_open in maze_grid.items() if not is_open
    ]
    rng.shuffle(open_positions)  # randomize for placement variety

    # 4) Floors on all open cells
    for pos in open_positions:
        level.add(pos, create_floor(cost_amount=movement_cost))

    # 5) Agent and exit
    start_pos: Position = _pop_or_fallback(open_positions, (0, 0))
    agent = create_agent(health=health)
    level.add(start_pos, agent)

    goal_pos: Position = _pop_or_fallback(open_positions, (width - 1, height - 1))
    level.add(goal_pos, create_exit())

    # 6) Required cores
    required_positions: List[Position] = []
    for _ in range(num_required_items):
        if not open_positions:
            break
        pos = open_positions.pop()
        level.add(pos, create_core(reward=required_item_reward, required=True))
        required_positions.append(pos)

    # Compute essential path set
    essential_path: Set[Position] = all_required_path_positions(
        maze_grid, start_pos, required_positions, goal_pos
    )

    # 7) Rewardable coins
    for _ in range(num_rewardable_items):
        if not open_positions:
            break
        level.add(open_positions.pop(), create_coin(reward=rewardable_item_reward))

    # 8) Portals (explicit pairing by reference)
    for _ in range(num_portals):
        if len(open_positions) < 2:
            break
        p1 = create_portal()
        p2 = create_portal(pair=p1)  # reciprocal authoring-time reference
        level.add(open_positions.pop(), p1)
        level.add(open_positions.pop(), p2)

    # 9) Doors/keys
    for i in range(num_doors):
        if len(open_positions) < 2:
            break
        key_pos = open_positions.pop()
        door_pos = open_positions.pop()
        key_id_str = f"key{i}"
        level.add(key_pos, create_key(key_id=key_id_str))
        level.add(door_pos, create_door(key_id=key_id_str))

    # 10) Powerups (as pickups)
    create_effect_fn_map: dict[EffectType, Callable[..., EntitySpec]] = {
        EffectType.SPEED: create_speed_effect,
        EffectType.IMMUNITY: create_immunity_effect,
        EffectType.PHASING: create_phasing_effect,
    }
    for type_, lim_type, lim_amount, extra in powerups:
        if not open_positions:
            break
        pos = open_positions.pop()
        create_effect_fn = create_effect_fn_map[type_]
        kwargs = {
            "time": lim_amount if lim_type == EffectLimit.TIME else None,
            "usage": lim_amount if lim_type == EffectLimit.USAGE else None,
        }
        level.add(pos, create_effect_fn(**extra, **kwargs))

    # 11) Non-essential positions (for enemies, hazards, moving boxes)
    open_non_essential: List[Position] = [
        p for p in open_positions if p not in essential_path
    ]
    rng.shuffle(open_non_essential)

    # 12) Boxes
    for pushable, speed in boxes:
        if not open_non_essential:
            break
        pos = open_non_essential.pop()
        axis, direction = _random_axis_and_dir(rng) if speed > 0 else (None, None)
        box = create_box(
            pushable=pushable,
            moving_axis=axis,
            moving_direction=direction,
            moving_speed=speed,
        )
        level.add(pos, box)

    # 13) Enemies (wire pathfinding to agent by reference if requested)
    for dmg, lethal, mtype, mspeed in enemies:
        if not open_non_essential:
            break
        pos = open_non_essential.pop()

        # Explicit pathfinding via reference to the agent (authoring-time)
        path_type: Optional[PathfindingType] = None
        if mtype == MovementType.PATHFINDING_LINE:
            path_type = PathfindingType.STRAIGHT_LINE
        elif mtype == MovementType.PATHFINDING_PATH:
            path_type = PathfindingType.PATH

        # If path_type is set, wire target to agent; otherwise directional/static
        if path_type is not None:
            enemy = create_monster(
                damage=dmg, lethal=lethal, pathfind_target=agent, path_type=path_type
            )
        else:
            maxis, mdirection = (
                _random_axis_and_dir(rng) if mspeed > 0 else (None, None)
            )
            enemy = create_monster(
                damage=dmg,
                lethal=lethal,
                moving_axis=maxis,
                moving_direction=mdirection,
                moving_speed=mspeed,
            )

        level.add(pos, enemy)

    # 14) Hazards
    for app_name, dmg, lethal in hazards:
        if not open_non_essential:
            break
        level.add(
            open_non_essential.pop(),
            create_hazard(app_name, damage=dmg, lethal=lethal, priority=7),
        )

    # 15) Walls
    for pos in wall_positions:
        level.add(pos, create_wall())

    # Convert to immutable State (wiring is resolved inside to_state)
    return to_state(level)

Authored gameplay progression levels

• Curated, hand-authored deterministic levels (L0–L13) that ramp mechanics: movement, maze turns, optional cost-reducing coin tiles, required cores, key–door, hazards, portals, pushable boxes, enemies, and power‑ups (Shield, Ghost, Boots), culminating in an integrated capstone puzzle.

• Each builder function accepts an optional seed (stored on the resulting State and used by rendering RNG). Use generate_task_suite(base_seed=...) or generate_task_suite(seed_list=[...]) to customize seeds while preserving structure.

Key functions:

from grid_universe.examples import gameplay_levels as gp

# Single level with custom seed
state = gp.build_level_power_boots(seed=9001)

# Full suite with deterministic shifted seeds
suite = gp.generate_task_suite(base_seed=5000)  # seeds = 5000..5013

# Explicit per-level seeds
suite = gp.generate_task_suite(seed_list=[10*i for i in range(14)])

grid_universe.examples.gameplay_levels

build_level_basic_movement

build_level_basic_movement(seed=100)

L0: Movement smoke test (Exit).

Parameters:

Name	Type	Description	Default
seed	int	Deterministic seed stored on resulting State (used by rendering / RNG subsystems).	100

Returns:

Name	Type	Description
State	State	Authored immutable state.

Source code in grid_universe/examples/gameplay_levels.py

def build_level_basic_movement(seed: int = 100) -> State:
    """L0: Movement smoke test (Exit).

    Args:
        seed (int): Deterministic seed stored on resulting ``State`` (used by rendering / RNG subsystems).

    Returns:
        State: Authored immutable state.
    """
    w, h = 7, 5
    lvl = Level(
        w,
        h,
        move_fn=default_move_fn,
        objective_fn=exit_objective_fn,
        seed=seed,
        turn_limit=TURN_LIMIT,
    )
    _floors(lvl)
    lvl.add((1, h // 2), create_agent(health=5))
    lvl.add((w - 2, h // 2), create_exit())
    # corridor wall
    for y in range(h):
        if y != h // 2:
            lvl.add((w // 2, y), create_wall())
    return to_state(lvl)

build_level_maze_turns

build_level_maze_turns(seed=101)

L1: Basic maze turns (Exit).

Parameters:

Name	Type	Description	Default
seed	int	Deterministic seed stored on resulting State.	101

Returns:

Name	Type	Description
State	State	Authored immutable state.

Source code in grid_universe/examples/gameplay_levels.py

def build_level_maze_turns(seed: int = 101) -> State:
    """L1: Basic maze turns (Exit).

    Args:
        seed (int): Deterministic seed stored on resulting ``State``.

    Returns:
        State: Authored immutable state.
    """
    w, h = 9, 7
    lvl = Level(
        w,
        h,
        move_fn=default_move_fn,
        objective_fn=exit_objective_fn,
        seed=seed,
        turn_limit=TURN_LIMIT,
    )
    _floors(lvl)
    _border(lvl)
    for x in range(2, w - 2):
        lvl.add((x, 2), create_wall())
    for x in range(2, w - 2):
        if x != w // 2:
            lvl.add((x, h - 3), create_wall())
    lvl.add((1, 1), create_agent(health=5))
    lvl.add((w - 2, h - 2), create_exit())
    return to_state(lvl)

build_level_optional_coin

build_level_optional_coin(seed=102)

L2: Optional coin path (Exit).

Coin reduce net cost along that route encouraging detour.

Parameters:

Name	Type	Description	Default
seed	int	Deterministic seed stored on resulting State.	102

Returns:

Name	Type	Description
State	State	Authored immutable state.

Source code in grid_universe/examples/gameplay_levels.py

def build_level_optional_coin(seed: int = 102) -> State:
    """L2: Optional coin path (Exit).

    Coin reduce net cost along that route encouraging detour.

    Args:
        seed (int): Deterministic seed stored on resulting ``State``.

    Returns:
        State: Authored immutable state.
    """
    w, h = 9, 7
    lvl = Level(
        w,
        h,
        move_fn=default_move_fn,
        objective_fn=exit_objective_fn,
        seed=seed,
        turn_limit=TURN_LIMIT,
    )
    _floors(lvl)
    _border(lvl)
    lvl.add((1, 2), create_wall())
    lvl.add((3, 3), create_wall())
    for x in range(3, w - 2):
        lvl.add((x, 2), create_wall())
    for x in range(2, w - 2):
        if x != w // 2:
            lvl.add((x, h - 3), create_wall())
    lvl.add((1, 1), create_agent(health=5))
    lvl.add((w - 2, h - 2), create_exit())
    for x in range(1, w - 2, 1):
        lvl.add((x, h - 2), create_coin(reward=COIN_REWARD))
    return to_state(lvl)

build_level_required_one

build_level_required_one(seed=103)

L3: One required core (Collect-then-Exit).

Parameters:

Name	Type	Description	Default
seed	int	Deterministic seed stored on resulting State.	103

Returns:

Name	Type	Description
State	State	Authored immutable state.

Source code in grid_universe/examples/gameplay_levels.py

def build_level_required_one(seed: int = 103) -> State:
    """L3: One required core (Collect-then-Exit).

    Args:
        seed (int): Deterministic seed stored on resulting ``State``.

    Returns:
        State: Authored immutable state.
    """
    w, h = 9, 7
    lvl = Level(
        w,
        h,
        move_fn=default_move_fn,
        objective_fn=default_objective_fn,
        seed=seed,
        turn_limit=TURN_LIMIT,
    )
    _floors(lvl)
    _border(lvl)
    for y in range(1, h - 1):
        if y != h // 2:
            lvl.add((w // 2, y), create_wall())
    lvl.add((1, h // 2), create_agent(health=5))
    lvl.add((w - 2, h // 2), create_exit())
    core = create_core(reward=CORE_REWARD, required=True)  # reward=0
    lvl.add((w // 2 - 1, h // 2 - 1), core)
    return to_state(lvl)

build_level_required_two

build_level_required_two(seed=104)

L4: Two required cores and backtracking (Collect-then-Exit).

Parameters:

Name	Type	Description	Default
seed	int	Deterministic seed stored on resulting State.	104

Returns:

Name	Type	Description
State	State	Authored immutable state.

Source code in grid_universe/examples/gameplay_levels.py

def build_level_required_two(seed: int = 104) -> State:
    """L4: Two required cores and backtracking (Collect-then-Exit).

    Args:
        seed (int): Deterministic seed stored on resulting ``State``.

    Returns:
        State: Authored immutable state.
    """
    w, h = 11, 9
    lvl = Level(
        w,
        h,
        move_fn=default_move_fn,
        objective_fn=default_objective_fn,
        seed=seed,
        turn_limit=TURN_LIMIT,
    )
    _floors(lvl)
    _border(lvl)
    midx, midy = w // 2, h // 2
    for x in range(1, w - 1):
        for y in range(1, h - 1):
            if x != midx and y != midy:
                lvl.add((x, y), create_wall())
    lvl.add((1, midy), create_agent(health=6))
    lvl.add((w - 2, midy), create_exit())
    lvl.add((midx, 1), create_core(reward=CORE_REWARD, required=True))  # reward=0
    lvl.add((midx, h - 2), create_core(reward=CORE_REWARD, required=True))  # reward=0
    return to_state(lvl)

build_level_key_door

build_level_key_door(seed=105)

L5: Key–Door gating (Exit).

Parameters:

Name	Type	Description	Default
seed	int	Deterministic seed stored on resulting State.	105

Returns:

Name	Type	Description
State	State	Authored immutable state.

Source code in grid_universe/examples/gameplay_levels.py

def build_level_key_door(seed: int = 105) -> State:
    """L5: Key–Door gating (Exit).

    Args:
        seed (int): Deterministic seed stored on resulting ``State``.

    Returns:
        State: Authored immutable state.
    """
    w, h = 11, 9
    lvl = Level(
        w,
        h,
        move_fn=default_move_fn,
        objective_fn=exit_objective_fn,
        seed=seed,
        turn_limit=TURN_LIMIT,
    )
    _floors(lvl)
    for y in range(h):
        if y != h // 2:
            lvl.add((w // 2, y), create_wall())
    lvl.add((1, h // 2), create_agent(health=5))
    lvl.add((w - 2, h // 2), create_exit())
    lvl.add((2, h // 2 - 1), create_key(key_id="alpha"))
    lvl.add((w // 2, h // 2), create_door(key_id="alpha"))
    return to_state(lvl)

build_level_hazard_detour

build_level_hazard_detour(seed=106)

L6: Hazard detour (damage=2) (Exit).

Hazard imposes only base step cost but reduces health on contact.

Parameters:

Name	Type	Description	Default
seed	int	Deterministic seed stored on resulting State.	106

Returns:

Name	Type	Description
State	State	Authored immutable state.

Source code in grid_universe/examples/gameplay_levels.py

def build_level_hazard_detour(seed: int = 106) -> State:
    """L6: Hazard detour (damage=2) (Exit).

    Hazard imposes only base step cost but reduces health on contact.

    Args:
        seed (int): Deterministic seed stored on resulting ``State``.

    Returns:
        State: Authored immutable state.
    """
    w, h = 11, 9
    lvl = Level(
        w,
        h,
        move_fn=default_move_fn,
        objective_fn=exit_objective_fn,
        seed=seed,
        turn_limit=TURN_LIMIT,
    )
    _floors(lvl)
    lvl.add((1, h // 2), create_agent(health=6))
    lvl.add((w - 2, h // 2), create_exit())
    # Central hazard (2 dmg); side wall encourages detour, but cost remains uniform except coin tiles
    lvl.add(
        (w // 2 - 1, h // 2),
        create_hazard(AppearanceName.SPIKE, damage=HAZARD_DAMAGE, lethal=False),
    )
    for y in range(1, h - 1):
        if y != h // 2:
            lvl.add((w // 2 - 1, y), create_wall())
    return to_state(lvl)

build_level_portal_shortcut

build_level_portal_shortcut(seed=107)

L7: Portal pair shortcut (Exit).

Parameters:

Name	Type	Description	Default
seed	int	Deterministic seed stored on resulting State.	107

Returns:

Name	Type	Description
State	State	Authored immutable state.

Source code in grid_universe/examples/gameplay_levels.py

def build_level_portal_shortcut(seed: int = 107) -> State:
    """L7: Portal pair shortcut (Exit).

    Args:
        seed (int): Deterministic seed stored on resulting ``State``.

    Returns:
        State: Authored immutable state.
    """
    w, h = 11, 9
    lvl = Level(
        w,
        h,
        move_fn=default_move_fn,
        objective_fn=exit_objective_fn,
        seed=seed,
        turn_limit=TURN_LIMIT,
    )
    _floors(lvl)
    lvl.add((1, h // 2), create_agent(health=5))
    lvl.add((w - 2, h // 2), create_exit())
    p1 = create_portal()
    p2 = create_portal(pair=p1)
    lvl.add((2, 1), p1)
    lvl.add((w - 1, h // 2), p2)
    for x in range(3, w - 3):
        lvl.add((x, h // 2 - 1), create_wall())
    return to_state(lvl)

build_level_pushable_box

build_level_pushable_box(seed=108)

L8: Pushable box in narrow corridor (Exit).

Parameters:

Name	Type	Description	Default
seed	int	Deterministic seed stored on resulting State.	108

Returns:

Name	Type	Description
State	State	Authored immutable state.

Source code in grid_universe/examples/gameplay_levels.py

def build_level_pushable_box(seed: int = 108) -> State:
    """L8: Pushable box in narrow corridor (Exit).

    Args:
        seed (int): Deterministic seed stored on resulting ``State``.

    Returns:
        State: Authored immutable state.
    """
    w, h = 11, 9
    lvl = Level(
        w,
        h,
        move_fn=default_move_fn,
        objective_fn=exit_objective_fn,
        seed=seed,
        turn_limit=TURN_LIMIT,
    )
    _floors(lvl)
    for y in range(h):
        if y != h // 2:
            lvl.add((w // 2, y), create_wall())
    lvl.add((1, h // 2), create_agent(health=5))
    lvl.add((w - 2, h // 2), create_exit())
    lvl.add((w // 2 - 1, h // 2), create_box(pushable=True))
    return to_state(lvl)

build_level_enemy_patrol

build_level_enemy_patrol(seed=109)

L9: Enemy patrol (damage=1) with safe avoidance (Exit).

Parameters:

Name	Type	Description	Default
seed	int	Deterministic seed stored on resulting State.	109

Returns:

Name	Type	Description
State	State	Authored immutable state.

Source code in grid_universe/examples/gameplay_levels.py

def build_level_enemy_patrol(seed: int = 109) -> State:
    """L9: Enemy patrol (damage=1) with safe avoidance (Exit).

    Args:
        seed (int): Deterministic seed stored on resulting ``State``.

    Returns:
        State: Authored immutable state.
    """
    w, h = 13, 9
    lvl = Level(
        w,
        h,
        move_fn=default_move_fn,
        objective_fn=exit_objective_fn,
        seed=seed,
        turn_limit=TURN_LIMIT,
    )
    _floors(lvl)
    lvl.add((2, h // 2), create_agent(health=1))
    lvl.add((w - 2, h // 2), create_exit())
    for y in range(h):
        if y not in [h // 2, h // 2 + 1]:
            lvl.add((w // 2, y), create_wall())
            lvl.add((w // 2 + 1, y), create_wall())
    enemy1 = create_monster(
        damage=ENEMY_DAMAGE,
        lethal=False,
        moving_axis=MovingAxis.VERTICAL,
        moving_direction=1,
        moving_bounce=True,
        moving_speed=1,
    )
    enemy2 = create_monster(
        damage=ENEMY_DAMAGE,
        lethal=False,
        moving_axis=MovingAxis.VERTICAL,
        moving_direction=1,
        moving_bounce=True,
        moving_speed=1,
    )
    lvl.add((w // 2, h // 2), enemy1)
    lvl.add((w // 2 + 1, h // 2), enemy2)
    return to_state(lvl)

build_level_power_shield

build_level_power_shield(seed=110)

L10: Shield (Immunity 5 uses) — necessary at a choke.

Unavoidable hazard (2 dmg) in a 1-wide corridor; agent has 2 HP. Without the shield effect the hazard would be lethal.

Parameters:

Name	Type	Description	Default
seed	int	Deterministic seed stored on resulting State.	110

Returns:

Name	Type	Description
State	State	Authored immutable state.

Source code in grid_universe/examples/gameplay_levels.py

def build_level_power_shield(seed: int = 110) -> State:
    """L10: Shield (Immunity 5 uses) — necessary at a choke.

    Unavoidable hazard (2 dmg) in a 1-wide corridor; agent has 2 HP. Without the
    shield effect the hazard would be lethal.

    Args:
        seed (int): Deterministic seed stored on resulting ``State``.

    Returns:
        State: Authored immutable state.
    """
    w, h = 11, 9
    lvl = Level(
        w,
        h,
        move_fn=default_move_fn,
        objective_fn=exit_objective_fn,
        seed=seed,
        turn_limit=TURN_LIMIT,
    )
    _floors(lvl)
    lvl.add((1, h // 2), create_agent(health=2))
    lvl.add((w - 2, h // 2), create_exit())
    for y in range(h):
        if y != h // 2:
            lvl.add((w // 2, y), create_wall())
    lvl.add((2, h // 2 - 3), create_immunity_effect(usage=5))  # Shield
    lvl.add(
        (w // 2, h // 2),
        create_hazard(AppearanceName.SPIKE, damage=HAZARD_DAMAGE, lethal=False),
    )
    return to_state(lvl)

build_level_power_ghost

build_level_power_ghost(seed=111)

L11: Ghost (Phasing 5 turns) — necessary to pass a door; no key provided.

Single corridor blocked by a locked door; phasing allows bypassing blocking.

Parameters:

Name	Type	Description	Default
seed	int	Deterministic seed stored on resulting State.	111

Returns:

Name	Type	Description
State	State	Authored immutable state.

Source code in grid_universe/examples/gameplay_levels.py

def build_level_power_ghost(seed: int = 111) -> State:
    """L11: Ghost (Phasing 5 turns) — necessary to pass a door; no key provided.

    Single corridor blocked by a locked door; phasing allows bypassing blocking.

    Args:
        seed (int): Deterministic seed stored on resulting ``State``.

    Returns:
        State: Authored immutable state.
    """
    w, h = 13, 9
    lvl = Level(
        w,
        h,
        move_fn=default_move_fn,
        objective_fn=exit_objective_fn,
        seed=seed,
        turn_limit=TURN_LIMIT,
    )
    _floors(lvl)
    lvl.add((1, h // 2), create_agent(health=5))
    lvl.add((w - 2, h // 2), create_exit())
    for y in range(h):
        if y != h // 2:
            lvl.add((w // 2, y), create_wall())
    lvl.add((2, h // 2 - 3), create_phasing_effect(time=5))  # Ghost
    lvl.add((w // 2, h // 2), create_door(key_id="alpha"))  # no key anywhere
    return to_state(lvl)

build_level_power_boots

build_level_power_boots(seed=112)

L12: Boots (Speed ×2, 5 turns) — useful to cross a 2-tile patrol window safely.

With 2× speed the agent traverses both tiles of a patrol gap in one action.

Parameters:

Name	Type	Description	Default
seed	int	Deterministic seed stored on resulting State.	112

Returns:

Name	Type	Description
State	State	Authored immutable state.

Source code in grid_universe/examples/gameplay_levels.py

def build_level_power_boots(seed: int = 112) -> State:
    """L12: Boots (Speed ×2, 5 turns) — useful to cross a 2-tile patrol window safely.

    With 2× speed the agent traverses both tiles of a patrol gap in one action.

    Args:
        seed (int): Deterministic seed stored on resulting ``State``.

    Returns:
        State: Authored immutable state.
    """
    w, h = 13, 9
    lvl = Level(
        w,
        h,
        move_fn=default_move_fn,
        objective_fn=exit_objective_fn,
        seed=seed,
        turn_limit=TURN_LIMIT,
    )
    _floors(lvl)
    lvl.add((1, h // 2), create_agent(health=1))
    lvl.add((w - 2, h // 2), create_exit())
    for y in range(h):
        if y not in [h // 2, h // 2 + 1]:
            lvl.add((w // 2, y), create_wall())
            lvl.add((w // 2 + 1, y), create_wall())
            lvl.add((w // 2 + 2, y), create_wall())
    lvl.add(
        (w // 2 - 1, h // 2 + 1), create_speed_effect(multiplier=2, time=5)
    )  # Boots
    enemy1 = create_monster(
        damage=ENEMY_DAMAGE,
        lethal=False,
        moving_axis=MovingAxis.VERTICAL,
        moving_direction=1,
        moving_bounce=True,
        moving_speed=1,
    )
    enemy2 = create_monster(
        damage=ENEMY_DAMAGE,
        lethal=False,
        moving_axis=MovingAxis.VERTICAL,
        moving_direction=1,
        moving_bounce=True,
        moving_speed=1,
    )
    enemy3 = create_monster(
        damage=ENEMY_DAMAGE,
        lethal=False,
        moving_axis=MovingAxis.VERTICAL,
        moving_direction=1,
        moving_bounce=True,
        moving_speed=1,
    )
    lvl.add((w // 2, h // 2), enemy1)
    lvl.add((w // 2 + 1, h // 2), enemy2)
    lvl.add((w // 2 + 2, h // 2), enemy3)
    return to_state(lvl)

generate_task_suite

generate_task_suite(base_seed=None, *, seed_list=None)

Return ordered suite of authored levels (L0..L13) with configurable seeds.

Seeding strategy precedence

1. seed_list if provided (must have length 14)
2. base_seed (seeds become base_seed + index)
3. Each builder's default seed constant (backwards compatible)

Parameters:

Name	Type	Description	Default
base_seed	int | None	Optional base; offsets determine per-level seeds.	None
seed_list	list[int] | None	Explicit seeds for each level (length must be 14).	None

Returns:

Type	Description
List[State]	list[State]: Immutable states for each level.

Source code in grid_universe/examples/gameplay_levels.py

def generate_task_suite(
    base_seed: int | None = None,
    *,
    seed_list: List[int] | None = None,
) -> List[State]:
    """Return ordered suite of authored levels (L0..L13) with configurable seeds.

    Seeding strategy precedence:
        1. ``seed_list`` if provided (must have length 14)
        2. ``base_seed`` (seeds become ``base_seed + index``)
        3. Each builder's default seed constant (backwards compatible)

    Args:
        base_seed (int | None): Optional base; offsets determine per-level seeds.
        seed_list (list[int] | None): Explicit seeds for each level (length must be 14).

    Returns:
        list[State]: Immutable states for each level.
    """
    builders = [
        build_level_basic_movement,  # L0
        build_level_maze_turns,  # L1
        build_level_optional_coin,  # L2
        build_level_required_one,  # L3
        build_level_required_two,  # L4
        build_level_key_door,  # L5
        build_level_hazard_detour,  # L6
        build_level_portal_shortcut,  # L7
        build_level_pushable_box,  # L8
        build_level_enemy_patrol,  # L9
        build_level_power_shield,  # L10 (Shield useful/necessary)
        build_level_power_ghost,  # L11 (Ghost necessary)
        build_level_power_boots,  # L12 (Boots strongly useful)
        build_level_capstone,  # L13 (capstone integration)
    ]

    if seed_list is not None:
        if len(seed_list) != len(builders):  # defensive check
            raise ValueError(
                f"seed_list must have length {len(builders)}; got {len(seed_list)}"
            )
        return [builder(seed) for builder, seed in zip(builders, seed_list)]

    if base_seed is not None:
        return [builder(base_seed + idx) for idx, builder in enumerate(builders)]

    # Default behavior keeps historical fixed seeds
    return [builder() for builder in builders]

Cipher levels (maze-based)

• Built on top of the procedural maze generator for structural consistency.
• API:
  • generate(...) – backward compatible convenience wrapper.
  • to_cipher_level(state, cipher_text_map, seed=None) – transform an existing state.
• Supports optional sampling of (cipher_text, objective_name) pairs (objective must be registered).

Usage examples:

from grid_universe.examples.cipher_objective_levels import generate, to_cipher_level
from grid_universe.examples import maze

# Direct generation (EXIT objective when num_required_items == 0)
exit_state = generate(width=9, height=7, num_required_items=0, cipher_objective_pairs=[("ABC","exit")], seed=123)

# Generation with required cores switches objective logic
collect_state = generate(width=11, height=9, num_required_items=2, cipher_objective_pairs=[("DATA","default")], seed=456)

# Transform an existing maze state with custom cipher/objective mapping
base = maze.generate(width=9, height=7, seed=999, num_required_items=1)
adapted = to_cipher_level(base, [("HELLO","exit")], seed=999)

grid_universe.examples.cipher_objective_levels

to_cipher_level

to_cipher_level(base_state, cipher_text_pairs, seed=None)

Transform an existing state into a cipher micro-level variant.

Parameters:

Name	Type	Description	Default
base_state	State	Source state (e.g. from maze.generate).	required
cipher_text_pairs	Iterable[CipherObjectivePair]	Iterable of (cipher_text, objective_name) pairs (required). At least one valid pair (objective registered) must be present.	required
seed	Optional[int]	Optional seed for deterministic sampling of the pair.	None

Returns:

Name	Type	Description
State	State	New immutable state with updated objective_fn and message.

Raises:

Type	Description
ValueError	If no valid pairs are supplied.

Source code in grid_universe/examples/cipher_objective_levels.py

def to_cipher_level(
    base_state: State,
    cipher_text_pairs: Iterable[CipherObjectivePair],
    seed: Optional[int] = None,
) -> State:
    """Transform an existing state into a cipher micro-level variant.

    Args:
        base_state: Source state (e.g. from ``maze.generate``).
        cipher_text_pairs: Iterable of (cipher_text, objective_name) pairs (required).
            At least one valid pair (objective registered) must be present.
        seed: Optional seed for deterministic sampling of the pair.

    Returns:
        State: New immutable state with updated ``objective_fn`` and ``message``.

    Raises:
        ValueError: If no valid pairs are supplied.
    """
    rng = random.Random(seed)
    cipher, obj_name = _sample_cipher_pair(rng, cipher_text_pairs)
    new_obj = OBJECTIVE_FN_REGISTRY[obj_name]
    return replace(base_state, objective_fn=new_obj, message=cipher)

generate

generate(width, height, num_required_items, cipher_objective_pairs, seed=None)

Generate a cipher micro-level using the maze generator and adapt it.

Parameters:

Name	Type	Description	Default
width	int	Grid width.	required
height	int	Grid height.	required
num_required_items	int	Number of required cores in the base maze.	required
cipher_objective_pairs	Iterable[CipherObjectivePair]	Iterable of (cipher, objective_name) pairs; required.	required
seed	Optional[int]	Deterministic seed for maze + cipher sampling.	None

Returns:

Name	Type	Description
State	State	Immutable cipher micro-level state.

Source code in grid_universe/examples/cipher_objective_levels.py

def generate(
    width: int,
    height: int,
    num_required_items: int,
    cipher_objective_pairs: Iterable[CipherObjectivePair],
    seed: Optional[int] = None,
) -> State:
    """Generate a cipher micro-level using the maze generator and adapt it.

    Args:
        width: Grid width.
        height: Grid height.
        num_required_items: Number of required cores in the base maze.
        cipher_objective_pairs: Iterable of (cipher, objective_name) pairs; required.
        seed: Deterministic seed for maze + cipher sampling.

    Returns:
        State: Immutable cipher micro-level state.
    """
    base = maze.generate(
        width=width,
        height=height,
        num_required_items=num_required_items,
        num_rewardable_items=0,
        num_portals=0,
        num_doors=0,
        health=5,
        movement_cost=3,
        required_item_reward=0,
        rewardable_item_reward=0,
        boxes=[],
        powerups=[],
        hazards=[],
        enemies=[],
        wall_percentage=0.8,
        seed=seed,
        turn_limit=TURN_LIMIT,
    )

    return to_cipher_level(base, cipher_objective_pairs, seed=seed)

Gym Environment

• GridUniverseEnv: Gymnasium-compatible Env returning image observations and structured info; reward is delta score.

grid_universe.gym_env

Gymnasium environment wrapper for Grid Universe.

Provides a structured observation that pairs a rendered RGBA image with rich info dictionaries (agent status / inventory / active effects, environment config). Reward is the delta of state.score per step. terminated is True on win, truncated on lose (mirrors many Gym environments that differentiate natural vs forced episode ends).

Observation schema (see docs for full details):

{ "image": np.ndarray(H,W,4), "info": { "agent": {...}, "status": {...}, "config": {...}, "message": str # empty string if None } }

or

Level # if observation_type="level"

Usage:

`` from grid_universe.gym_env import GridUniverseEnv from grid_universe.examples.maze import generate as maze_generate

env = GridUniverseEnv(initial_state_fn=maze_generate, width=9, height=9, seed=123) ``

Customization hooks

• initial_state_fn: Provide a callable that returns a fully built State.
• render_texture_map / resolution let you swap assets or resolution.

The environment is purposely not vectorized; wrap externally if needed.

EffectEntry

Bases: TypedDict

Single active effect entry.

Fields use sentinel defaults in the runtime observation

• Empty string ("") for absent text fields.
• -1 for numeric fields that are logically None / not applicable.

Source code in grid_universe/gym_env.py

class EffectEntry(TypedDict):
    """Single active effect entry.

    Fields use sentinel defaults in the runtime observation:
      * Empty string ("") for absent text fields.
      * -1 for numeric fields that are logically None / not applicable.
    """

    id: int  # Unique effect entity id
    type: str  # "" | "IMMUNITY" | "PHASING" | "SPEED"
    limit_type: str  # "" | "TIME" | "USAGE"
    limit_amount: int  # -1 if no limit
    multiplier: int  # Speed multiplier (or -1 if not SPEED)

InventoryItem

Bases: TypedDict

Inventory item entry (key / core / coin / generic item).

Source code in grid_universe/gym_env.py

class InventoryItem(TypedDict):
    """Inventory item entry (key / core / coin / generic item)."""

    id: int
    type: str  # "key" | "core" | "coin" | "item"
    key_id: str  # "" if not a key
    appearance_name: str  # "" if not known / provided

HealthInfo

Bases: TypedDict

Health block; -1 indicates missing (agent has no health component).

Source code in grid_universe/gym_env.py

class HealthInfo(TypedDict):
    """Health block; -1 indicates missing (agent has no health component)."""

    health: int
    max_health: int

AgentInfo

Bases: TypedDict

Agent sub‑observation grouping health, effects and inventory.

Source code in grid_universe/gym_env.py

class AgentInfo(TypedDict):
    """Agent sub‑observation grouping health, effects and inventory."""

    health: HealthInfo
    effects: List[EffectEntry]
    inventory: List[InventoryItem]

StatusInfo

Bases: TypedDict

Environment status (score, phase, current turn).

Source code in grid_universe/gym_env.py

class StatusInfo(TypedDict):
    """Environment status (score, phase, current turn)."""

    score: int
    phase: str  # "win" | "lose" | "ongoing"
    turn: int

ConfigInfo

Bases: TypedDict

Static / semi‑static config describing the active level & functions.

Source code in grid_universe/gym_env.py

class ConfigInfo(TypedDict):
    """Static / semi‑static config describing the active level & functions."""

    move_fn: str
    objective_fn: str
    seed: int  # -1 if None
    width: int
    height: int
    turn_limit: int  # -1 if unlimited

InfoDict

Bases: TypedDict

Full structured info payload accompanying every observation.

Source code in grid_universe/gym_env.py

class InfoDict(TypedDict):
    """Full structured info payload accompanying every observation."""

    agent: AgentInfo
    status: StatusInfo
    config: ConfigInfo
    message: str  # Narrative / status message ("" if none)

Observation

Bases: TypedDict

Top‑level observation returned by the environment.

image: RGBA image array (H x W x 4, dtype=uint8) info: Rich structured dictionaries (see :class:InfoDict).

Source code in grid_universe/gym_env.py

class Observation(TypedDict):
    """Top‑level observation returned by the environment.

    image: RGBA image array (H x W x 4, dtype=uint8)
    info:  Rich structured dictionaries (see :class:`InfoDict`).
    """

    image: ImageArray
    info: InfoDict

Action

Bases: IntEnum

Stable integer mapping for integration with Gymnasium Discrete spaces.

Source code in grid_universe/gym_env.py

class Action(IntEnum):
    """Stable integer mapping for integration with Gymnasium ``Discrete`` spaces."""

    UP = 0  # start at 0 for explicitness
    DOWN = auto()
    LEFT = auto()
    RIGHT = auto()
    USE_KEY = auto()
    PICK_UP = auto()
    WAIT = auto()

GridUniverseEnv

Bases: Env[Union[Observation, Level], integer]

Gymnasium Env implementation for the Grid Universe.

Parameters mirror the procedural level generator plus rendering knobs. The action space is Discrete(len(BaseAction)); see :mod:grid_universe.actions.

Source code in grid_universe/gym_env.py

class GridUniverseEnv(gym.Env[Union[Observation, Level], np.integer]):
    """Gymnasium ``Env`` implementation for the Grid Universe.

    Parameters mirror the procedural level generator plus rendering knobs. The
    action space is ``Discrete(len(BaseAction))``; see :mod:`grid_universe.actions`.
    """

    metadata = {"render_modes": ["human", "texture"]}

    def __init__(
        self,
        initial_state_fn: Callable[..., State],
        render_mode: str = "texture",
        render_resolution: int = DEFAULT_RESOLUTION,
        render_texture_map: TextureMap = DEFAULT_TEXTURE_MAP,
        render_asset_root: str = DEFAULT_ASSET_ROOT,
        observation_type: str = "image",
        **kwargs: Any,
    ):
        """Create a new environment instance.

        Args:
            render_mode (str): "texture" to return PIL image frames, "human" to open a window.
            render_resolution (int): Width (pixels) of rendered image (height derived).
            render_texture_map (TextureMap): Mapping of ``(AppearanceName, properties)`` to asset paths.
            initial_state_fn (Callable[..., State]): Callable returning an initial ``State``.
            **kwargs: Forwarded to ``initial_state_fn`` (e.g., size, densities, seed).
        """
        # Observation type: "image" (default behavior) or "level" (returns Level dataclass)
        if observation_type not in {"image", "level"}:
            raise ValueError(
                f"Unsupported observation_type '{observation_type}'. Expected 'image' or 'level'."
            )
        self._observation_type = observation_type

        # Generator/config kwargs for level creation
        self._initial_state_fn = initial_state_fn
        self._initial_state_kwargs = kwargs

        # Runtime state
        self.state: Optional[State] = None
        self.agent_id: Optional[EntityID] = None

        # Basic config
        self.width: int = int(kwargs.get("width", 9))
        self.height: int = int(kwargs.get("height", 9))
        self._render_resolution = render_resolution
        self._render_texture_map = render_texture_map
        self._render_asset_root = render_asset_root
        self._render_mode = render_mode

        # Rendering setup
        render_width: int = render_resolution
        render_height: int = int(self.height / self.width * render_width)
        self._texture_renderer: Optional[TextureRenderer] = None

        # Observation space helpers (Gymnasium has no Integer/Optional)
        base_chars = (
            string.ascii_lowercase + string.ascii_uppercase + string.digits + "_"
        )
        text_space_short = spaces.Text(
            max_length=32, min_length=0, charset=base_chars
        )  # enums
        text_space_medium = spaces.Text(
            max_length=128, min_length=0, charset=base_chars
        )  # fn names
        text_space_long = spaces.Text(
            max_length=512, min_length=0, charset=string.printable
        )

        def int_box(low: int, high: int) -> spaces.Box:
            return spaces.Box(
                low=np.array(low, dtype=np.int64),
                high=np.array(high, dtype=np.int64),
                shape=(),
                dtype=np.int64,
            )

        # Effect entry: use "" for absent strings, -1 for absent numbers
        effect_space = spaces.Dict(
            {
                "id": int_box(0, 1_000_000_000),
                "type": text_space_short,  # "", "IMMUNITY", "PHASING", "SPEED"
                "limit_type": text_space_short,  # "", "TIME", "USAGE"
                "limit_amount": int_box(-1, 1_000_000_000),  # -1 if none
                "multiplier": int_box(-1, 1_000_000),  # -1 if N/A (only SPEED)
            }
        )

        # Inventory item: type in {"key","core","coin","item"}; empty strings for optional text
        item_space = spaces.Dict(
            {
                "id": int_box(0, 1_000_000_000),
                "type": text_space_short,
                "key_id": text_space_medium,  # "" if not a key
                "appearance_name": text_space_short,  # "" if unknown
            }
        )

        # Health: -1 to indicate missing
        health_space = spaces.Dict(
            {
                "health": int_box(-1, 1_000_000),
                "max_health": int_box(-1, 1_000_000),
            }
        )

        if self._observation_type == "image":
            # Full observation space: image + structured info dict
            self.observation_space = cast(
                gym.Space[Observation],
                spaces.Dict(
                    {
                        "image": spaces.Box(
                            low=0,
                            high=255,
                            shape=(render_height, render_width, 4),
                            dtype=np.uint8,
                        ),
                        "info": spaces.Dict(
                            {
                                "agent": spaces.Dict(
                                    {
                                        "health": health_space,
                                        "effects": spaces.Sequence(effect_space),
                                        "inventory": spaces.Sequence(item_space),
                                    }
                                ),
                                "status": spaces.Dict(
                                    {
                                        "score": int_box(-1_000_000_000, 1_000_000_000),
                                        "phase": text_space_short,  # "win" / "lose" / "ongoing"
                                        "turn": int_box(0, 1_000_000_000),
                                    }
                                ),
                                "config": spaces.Dict(
                                    {
                                        "move_fn": text_space_medium,
                                        "objective_fn": text_space_medium,
                                        "seed": int_box(
                                            -1_000_000_000, 1_000_000_000
                                        ),  # use -1 to represent None if needed
                                        "width": int_box(1, 10_000),
                                        "height": int_box(1, 10_000),
                                        "turn_limit": int_box(-1, 1_000_000_000),
                                    }
                                ),
                                "message": text_space_long,
                            }
                        ),
                    }
                ),
            )
        else:
            # For Level observations we cannot define a strict Gym space (arbitrary Python object).
            # Provide a placeholder space (Discrete(1)) with documented contract that observations are Level.
            # Users leveraging RL libraries should stick to observation_type="image".
            self.observation_space = spaces.Discrete(1)  # type: ignore[assignment]

        # Actions
        self.action_space = spaces.Discrete(len(BaseAction))

        # Initialize first episode
        self.reset()

    def reset(
        self, *, seed: Optional[int] = None, options: Optional[Dict[str, object]] = None
    ) -> Tuple[Union[Observation, Level], Dict[str, object]]:
        """Start a new episode.

        Args:
            seed (int | None): Currently unused (procedural seed is passed via kwargs on construction).
            options (dict | None): Gymnasium options (unused).

        Returns:
            Tuple[Observation, dict]: Observation dict and empty info dict per Gymnasium API.
        """
        self.state = self._initial_state_fn(**self._initial_state_kwargs)
        self.agent_id = next(iter(self.state.agent.keys()))
        if self._observation_type == "image":
            self._setup_renderer()
        obs = self._get_obs()
        return obs, self._get_info()

    def step(
        self, action: np.integer | int | Action | BaseAction
    ) -> Tuple[Union[Observation, Level], float, bool, bool, Dict[str, object]]:
        """Apply one environment step.

        Args:
            action (int | np.integer | Action): Integer index (or ``Action`` enum
                member) selecting an action from the discrete action space.

        Returns:
            Tuple[Observation, float, bool, bool, dict]: ``(observation, reward, terminated, truncated, info)``.
        """
        assert self.state is not None and self.agent_id is not None

        step_action: BaseAction = BaseAction.WAIT  # default fallback

        if isinstance(action, BaseAction):
            step_action = action
        else:
            # Coerce provided action (Action / numpy integer / int) into index
            if isinstance(action, Action):
                action_index = int(action.value)
            else:
                # Try coercing to int (covers plain int and numpy integer). If this fails, raise.
                try:
                    action_index = int(action)
                except Exception as exc:
                    raise TypeError(
                        f"Action must be int-compatible or Action; got {type(action)!r}"
                    ) from exc

            if not 0 <= action_index < len(BaseAction):
                raise ValueError(
                    f"Invalid action index {action_index}; expected 0..{len(BaseAction) - 1}"
                )

            step_action = list(BaseAction)[action_index]

        prev_score = self.state.score
        self.state = step(self.state, step_action, agent_id=self.agent_id)
        reward = float(self.state.score - prev_score)
        obs = self._get_obs()
        terminated = self.state.win
        truncated = self.state.lose
        info = self._get_info()
        return obs, reward, terminated, truncated, info

    def render(self, mode: Optional[str] = None) -> Optional[PILImage]:  # type: ignore
        """Render the current state.

        Args:
            mode (str | None): "human" to display, "texture" to return PIL image. Defaults to
                the instance's configured render mode.
        """
        render_mode = mode or self._render_mode
        assert self.state is not None
        self._setup_renderer()
        assert self._texture_renderer is not None
        img = self._texture_renderer.render(self.state)
        if render_mode == "human":
            img.show()
            return None
        elif render_mode == "texture":
            return img
        else:
            raise NotImplementedError(f"Render mode '{render_mode}' not supported.")

    def state_info(self) -> InfoDict:
        """Return structured ``info`` sub-dict used in observations."""
        assert self.state is not None and self.agent_id is not None
        info_dict: InfoDict = {
            "agent": agent_observation_dict(self.state, self.agent_id),
            "status": env_status_observation_dict(self.state),
            "config": env_config_observation_dict(self.state),
            "message": self.state.message or "",
        }
        return info_dict

    def _get_obs(self) -> Union[Observation, Level]:
        """Internal helper constructing the observation per observation_type.

        observation_type="image": returns Observation (dict with image + info)
        observation_type="level": returns a freshly converted authoring-time Level object
            produced via levels.convert.from_state(state). This allows algorithms to
            reason over symbolic grid/entity structures directly. NOTE: This mode is
            not compatible with typical RL libraries expecting a numeric space.
        """
        assert self.state is not None and self.agent_id is not None
        if self._observation_type == "level":
            # Return authoring-time Level view (lossless reconstruction)
            return from_state(self.state)

        # Default image observation path
        self._setup_renderer()
        assert self._texture_renderer is not None
        img = self._texture_renderer.render(self.state)
        img_np: ImageArray = np.array(img)
        info_dict: InfoDict = self.state_info()
        return cast(Observation, {"image": img_np, "info": info_dict})

    def _get_info(self) -> Dict[str, object]:
        """Return the step info (empty placeholder for compatibility)."""
        return {}

    def _setup_renderer(self) -> None:
        """(Re)initialize the texture renderer if needed."""
        if self._texture_renderer is None:
            self._texture_renderer = TextureRenderer(
                resolution=self._render_resolution,
                texture_map=self._render_texture_map,
                asset_root=self._render_asset_root,
            )

    def close(self) -> None:
        """Release any renderer resources (no-op placeholder)."""
        pass

__init__

__init__(initial_state_fn, render_mode='texture', render_resolution=DEFAULT_RESOLUTION, render_texture_map=DEFAULT_TEXTURE_MAP, render_asset_root=DEFAULT_ASSET_ROOT, observation_type='image', **kwargs)

Create a new environment instance.

Parameters:

Name	Type	Description	Default
render_mode	str	"texture" to return PIL image frames, "human" to open a window.	'texture'
render_resolution	int	Width (pixels) of rendered image (height derived).	DEFAULT_RESOLUTION
render_texture_map	TextureMap	Mapping of (AppearanceName, properties) to asset paths.	DEFAULT_TEXTURE_MAP
initial_state_fn	Callable[..., State]	Callable returning an initial State.	required
**kwargs	Any	Forwarded to initial_state_fn (e.g., size, densities, seed).	{}

Source code in grid_universe/gym_env.py

def __init__(
    self,
    initial_state_fn: Callable[..., State],
    render_mode: str = "texture",
    render_resolution: int = DEFAULT_RESOLUTION,
    render_texture_map: TextureMap = DEFAULT_TEXTURE_MAP,
    render_asset_root: str = DEFAULT_ASSET_ROOT,
    observation_type: str = "image",
    **kwargs: Any,
):
    """Create a new environment instance.

    Args:
        render_mode (str): "texture" to return PIL image frames, "human" to open a window.
        render_resolution (int): Width (pixels) of rendered image (height derived).
        render_texture_map (TextureMap): Mapping of ``(AppearanceName, properties)`` to asset paths.
        initial_state_fn (Callable[..., State]): Callable returning an initial ``State``.
        **kwargs: Forwarded to ``initial_state_fn`` (e.g., size, densities, seed).
    """
    # Observation type: "image" (default behavior) or "level" (returns Level dataclass)
    if observation_type not in {"image", "level"}:
        raise ValueError(
            f"Unsupported observation_type '{observation_type}'. Expected 'image' or 'level'."
        )
    self._observation_type = observation_type

    # Generator/config kwargs for level creation
    self._initial_state_fn = initial_state_fn
    self._initial_state_kwargs = kwargs

    # Runtime state
    self.state: Optional[State] = None
    self.agent_id: Optional[EntityID] = None

    # Basic config
    self.width: int = int(kwargs.get("width", 9))
    self.height: int = int(kwargs.get("height", 9))
    self._render_resolution = render_resolution
    self._render_texture_map = render_texture_map
    self._render_asset_root = render_asset_root
    self._render_mode = render_mode

    # Rendering setup
    render_width: int = render_resolution
    render_height: int = int(self.height / self.width * render_width)
    self._texture_renderer: Optional[TextureRenderer] = None

    # Observation space helpers (Gymnasium has no Integer/Optional)
    base_chars = (
        string.ascii_lowercase + string.ascii_uppercase + string.digits + "_"
    )
    text_space_short = spaces.Text(
        max_length=32, min_length=0, charset=base_chars
    )  # enums
    text_space_medium = spaces.Text(
        max_length=128, min_length=0, charset=base_chars
    )  # fn names
    text_space_long = spaces.Text(
        max_length=512, min_length=0, charset=string.printable
    )

    def int_box(low: int, high: int) -> spaces.Box:
        return spaces.Box(
            low=np.array(low, dtype=np.int64),
            high=np.array(high, dtype=np.int64),
            shape=(),
            dtype=np.int64,
        )

    # Effect entry: use "" for absent strings, -1 for absent numbers
    effect_space = spaces.Dict(
        {
            "id": int_box(0, 1_000_000_000),
            "type": text_space_short,  # "", "IMMUNITY", "PHASING", "SPEED"
            "limit_type": text_space_short,  # "", "TIME", "USAGE"
            "limit_amount": int_box(-1, 1_000_000_000),  # -1 if none
            "multiplier": int_box(-1, 1_000_000),  # -1 if N/A (only SPEED)
        }
    )

    # Inventory item: type in {"key","core","coin","item"}; empty strings for optional text
    item_space = spaces.Dict(
        {
            "id": int_box(0, 1_000_000_000),
            "type": text_space_short,
            "key_id": text_space_medium,  # "" if not a key
            "appearance_name": text_space_short,  # "" if unknown
        }
    )

    # Health: -1 to indicate missing
    health_space = spaces.Dict(
        {
            "health": int_box(-1, 1_000_000),
            "max_health": int_box(-1, 1_000_000),
        }
    )

    if self._observation_type == "image":
        # Full observation space: image + structured info dict
        self.observation_space = cast(
            gym.Space[Observation],
            spaces.Dict(
                {
                    "image": spaces.Box(
                        low=0,
                        high=255,
                        shape=(render_height, render_width, 4),
                        dtype=np.uint8,
                    ),
                    "info": spaces.Dict(
                        {
                            "agent": spaces.Dict(
                                {
                                    "health": health_space,
                                    "effects": spaces.Sequence(effect_space),
                                    "inventory": spaces.Sequence(item_space),
                                }
                            ),
                            "status": spaces.Dict(
                                {
                                    "score": int_box(-1_000_000_000, 1_000_000_000),
                                    "phase": text_space_short,  # "win" / "lose" / "ongoing"
                                    "turn": int_box(0, 1_000_000_000),
                                }
                            ),
                            "config": spaces.Dict(
                                {
                                    "move_fn": text_space_medium,
                                    "objective_fn": text_space_medium,
                                    "seed": int_box(
                                        -1_000_000_000, 1_000_000_000
                                    ),  # use -1 to represent None if needed
                                    "width": int_box(1, 10_000),
                                    "height": int_box(1, 10_000),
                                    "turn_limit": int_box(-1, 1_000_000_000),
                                }
                            ),
                            "message": text_space_long,
                        }
                    ),
                }
            ),
        )
    else:
        # For Level observations we cannot define a strict Gym space (arbitrary Python object).
        # Provide a placeholder space (Discrete(1)) with documented contract that observations are Level.
        # Users leveraging RL libraries should stick to observation_type="image".
        self.observation_space = spaces.Discrete(1)  # type: ignore[assignment]

    # Actions
    self.action_space = spaces.Discrete(len(BaseAction))

    # Initialize first episode
    self.reset()

reset

reset(*, seed=None, options=None)

Start a new episode.

Parameters:

Name	Type	Description	Default
seed	int | None	Currently unused (procedural seed is passed via kwargs on construction).	None
options	dict | None	Gymnasium options (unused).	None

Returns:

Type	Description
Tuple[Union[Observation, Level], Dict[str, object]]	Tuple[Observation, dict]: Observation dict and empty info dict per Gymnasium API.

Source code in grid_universe/gym_env.py

def reset(
    self, *, seed: Optional[int] = None, options: Optional[Dict[str, object]] = None
) -> Tuple[Union[Observation, Level], Dict[str, object]]:
    """Start a new episode.

    Args:
        seed (int | None): Currently unused (procedural seed is passed via kwargs on construction).
        options (dict | None): Gymnasium options (unused).

    Returns:
        Tuple[Observation, dict]: Observation dict and empty info dict per Gymnasium API.
    """
    self.state = self._initial_state_fn(**self._initial_state_kwargs)
    self.agent_id = next(iter(self.state.agent.keys()))
    if self._observation_type == "image":
        self._setup_renderer()
    obs = self._get_obs()
    return obs, self._get_info()

step

step(action)

Apply one environment step.

Parameters:

Name	Type	Description	Default
action	int | integer | Action	Integer index (or Action enum member) selecting an action from the discrete action space.	required

Returns:

Type	Description
Tuple[Union[Observation, Level], float, bool, bool, Dict[str, object]]	Tuple[Observation, float, bool, bool, dict]: (observation, reward, terminated, truncated, info).

Source code in grid_universe/gym_env.py

def step(
    self, action: np.integer | int | Action | BaseAction
) -> Tuple[Union[Observation, Level], float, bool, bool, Dict[str, object]]:
    """Apply one environment step.

    Args:
        action (int | np.integer | Action): Integer index (or ``Action`` enum
            member) selecting an action from the discrete action space.

    Returns:
        Tuple[Observation, float, bool, bool, dict]: ``(observation, reward, terminated, truncated, info)``.
    """
    assert self.state is not None and self.agent_id is not None

    step_action: BaseAction = BaseAction.WAIT  # default fallback

    if isinstance(action, BaseAction):
        step_action = action
    else:
        # Coerce provided action (Action / numpy integer / int) into index
        if isinstance(action, Action):
            action_index = int(action.value)
        else:
            # Try coercing to int (covers plain int and numpy integer). If this fails, raise.
            try:
                action_index = int(action)
            except Exception as exc:
                raise TypeError(
                    f"Action must be int-compatible or Action; got {type(action)!r}"
                ) from exc

        if not 0 <= action_index < len(BaseAction):
            raise ValueError(
                f"Invalid action index {action_index}; expected 0..{len(BaseAction) - 1}"
            )

        step_action = list(BaseAction)[action_index]

    prev_score = self.state.score
    self.state = step(self.state, step_action, agent_id=self.agent_id)
    reward = float(self.state.score - prev_score)
    obs = self._get_obs()
    terminated = self.state.win
    truncated = self.state.lose
    info = self._get_info()
    return obs, reward, terminated, truncated, info

render

render(mode=None)

Render the current state.

Parameters:

Name	Type	Description	Default
mode	str | None	"human" to display, "texture" to return PIL image. Defaults to the instance's configured render mode.	None

Source code in grid_universe/gym_env.py

def render(self, mode: Optional[str] = None) -> Optional[PILImage]:  # type: ignore
    """Render the current state.

    Args:
        mode (str | None): "human" to display, "texture" to return PIL image. Defaults to
            the instance's configured render mode.
    """
    render_mode = mode or self._render_mode
    assert self.state is not None
    self._setup_renderer()
    assert self._texture_renderer is not None
    img = self._texture_renderer.render(self.state)
    if render_mode == "human":
        img.show()
        return None
    elif render_mode == "texture":
        return img
    else:
        raise NotImplementedError(f"Render mode '{render_mode}' not supported.")

state_info

state_info()

Return structured info sub-dict used in observations.

Source code in grid_universe/gym_env.py

def state_info(self) -> InfoDict:
    """Return structured ``info`` sub-dict used in observations."""
    assert self.state is not None and self.agent_id is not None
    info_dict: InfoDict = {
        "agent": agent_observation_dict(self.state, self.agent_id),
        "status": env_status_observation_dict(self.state),
        "config": env_config_observation_dict(self.state),
        "message": self.state.message or "",
    }
    return info_dict

close

close()

Release any renderer resources (no-op placeholder).

Source code in grid_universe/gym_env.py

def close(self) -> None:
    """Release any renderer resources (no-op placeholder)."""
    pass

agent_observation_dict

agent_observation_dict(state, agent_id)

Compose structured agent sub‑observation.

Includes health, list of active effect entries, and inventory items. Missing health is represented by None values which are later converted to sentinel numbers in the space definition (-1) when serialized to numpy arrays (Gym leaves them as ints here).

Source code in grid_universe/gym_env.py

def agent_observation_dict(state: State, agent_id: EntityID) -> AgentInfo:
    """Compose structured agent sub‑observation.

    Includes health, list of active effect entries, and inventory items.
    Missing health is represented by ``None`` values which are later converted
    to sentinel numbers in the space definition (-1) when serialized to numpy
    arrays (Gym leaves them as ints here).
    """
    # Health
    hp = state.health.get(agent_id)
    health_dict: Dict[str, Any] = {
        "health": int(hp.health) if hp else -1,
        "max_health": int(hp.max_health) if hp else -1,
    }

    # Active effects (status)
    effects: List[Dict[str, Any]] = []
    status = state.status.get(agent_id)
    if status is not None:
        for eff_id in status.effect_ids:
            effects.append(_serialize_effect(state, eff_id))

    # Inventory items
    inv_items: List[Dict[str, Any]] = []
    inv = state.inventory.get(agent_id)
    if inv:
        for item_eid in inv.item_ids:
            inv_items.append(_serialize_inventory_item(state, item_eid))

    return cast(
        AgentInfo,
        {
            "health": health_dict,
            "effects": tuple(effects),
            "inventory": tuple(inv_items),
        },
    )

env_status_observation_dict

env_status_observation_dict(state)

Status portion of observation (score, phase, turn).

Source code in grid_universe/gym_env.py

def env_status_observation_dict(state: State) -> StatusInfo:
    """Status portion of observation (score, phase, turn)."""
    # Derive phase for clarity
    phase = "ongoing"
    if state.win:
        phase = "win"
    elif state.lose:
        phase = "lose"
    return cast(
        StatusInfo,
        {
            "score": int(state.score),
            "phase": phase,
            "turn": int(state.turn),
        },
    )

env_config_observation_dict

env_config_observation_dict(state)

Config portion of observation (function names, seed, dimensions).

Source code in grid_universe/gym_env.py

def env_config_observation_dict(state: State) -> ConfigInfo:
    """Config portion of observation (function names, seed, dimensions)."""
    move_fn_name = getattr(state.move_fn, "__name__", str(state.move_fn))
    objective_fn_name = getattr(state.objective_fn, "__name__", str(state.objective_fn))
    return cast(
        ConfigInfo,
        {
            "move_fn": move_fn_name,
            "objective_fn": objective_fn_name,
            "seed": state.seed if state.seed is not None else -1,
            "width": state.width,
            "height": state.height,
            "turn_limit": state.turn_limit if state.turn_limit is not None else -1,
        },
    )

Registries and enums

• Movement registry: Name → MoveFn.

• Objective registry: Name → ObjectiveFn.

• Actions: string enum for core actions;

• EffectType and EffectLimit enums.

Reference snippets:

from grid_universe.moves import MOVE_FN_REGISTRY
from grid_universe.objectives import OBJECTIVE_FN_REGISTRY
from grid_universe.actions import Action

print(MOVE_FN_REGISTRY.keys())
print(OBJECTIVE_FN_REGISTRY.keys())
print(list(Action))

Schemas and Data Shapes

This page captures common data structures you may want to reference without digging into code: observation dicts from the Gym env, texture map keys/values, and group recoloring rules.

Contents

• Gym observation schema
• Texture map schema
• Grouping rules and color mapping

Gym observation schema

Observation (obs: Dict[str, Any]) returned by GridUniverseEnv:

• image

  • Type: numpy.ndarray

  • Shape: (H, W, 4), dtype=uint8 (RGBA)

• info

  • agent

    • health

      • health: int or -1

      • max_health: int or -1

    • effects: list of effect entries

      • id: int

      • type: "", "IMMUNITY", "PHASING", "SPEED"

      • limit_type: "", "TIME", "USAGE"

      • limit_amount: int or -1

      • multiplier: int (SPEED only; -1 otherwise)

    • inventory: list of item entries

      • id: int

      • type: "item" | "key" | "core" | "coin"

      • key_id: str ("" if not a key)

      • appearance_name: str ("" if unknown)

  • status

    • score: int

    • phase: "ongoing" | "win" | "lose"

    • turn: int

  • config

    • move_fn: str (function name)

    • objective_fn: str (function name)

    • seed: int (or -1)

    • width: int

    • height: int

  • message

    • str (empty string when absent); optional narrative/task hint text

Action space:

• Discrete(7), mapping to Action enum indices:

  • 0: UP

  • 1: DOWN

  • 2: LEFT

  • 3: RIGHT

  • 4: USE_KEY

  • 5: PICK_UP

  • 6: WAIT

Reward:

• Delta score between steps (float).

Texture map schema

TextureMap entry key/value:

• Key

  • Tuple[AppearanceName, Tuple[str, ...]]

  • Example: (AppearanceName.BOX, ()), (AppearanceName.BOX, ("pushable",))

• Value

  • str: path under asset_root to a file or directory

  • If directory: renderer picks a deterministic file per state.seed

Resolution:

• Asset path = f"{asset_root}/{value}"

• File types: .png, .jpg, .jpeg, .gif

Property matching:

• When rendering an entity, the renderer constructs the set of string “properties” for that entity based on which component maps contain its EID (e.g., "pushable", "pathfinding", "dead", "locked", "required").

• The best-matching texture is chosen by maximizing overlap and minimizing unmatched properties among the available keys for that AppearanceName.

Grouping rules and color mapping

Grouping rules (derive_groups) assign entities into color groups, e.g.:

• Keys/doors by key id:

  • key_door_group_rule → "key:<key_id>"

• Paired portals:

  • portal_pair_group_rule → "portal:<min_eid>-<max_eid>"

Color mapping:

• group_to_color(group_id) → (r, g, b)

  • Deterministic mapping using random.Random(group_id) to sample HSV; converted to RGB.

Recolor:

• apply_recolor_if_group(image, group) replaces hue while preserving per-pixel tone (value) and, by default, saturation.

Extensibility:

• Add new GroupRule functions to DEFAULT_GROUP_RULES (locally in your render wrapper) to recolor custom categories consistently.

from typing import Optional
from grid_universe.state import State
from grid_universe.types import EntityID

def my_group_rule(state: State, eid: EntityID) -> Optional[str]:
    if eid in state.pushable:
        return "pushable"
    return None