Python API Reference

This page exposes the current PHIDS Python API using mkdocstrings.

Schemas and ingress

Pydantic schemas for ECS components, configuration payloads, and REST API models.

This module defines payload and response models used by the REST API as well as schemata representing ECS components and species parameters.

AllOfConditionSchema

Bases: BaseModel

Boolean AND over nested activation predicates.

Source code in src/phids/api/schemas.py

class AllOfConditionSchema(BaseModel):
    """Boolean AND over nested activation predicates."""

    kind: Literal["all_of"] = "all_of"
    conditions: list[ConditionNode] = Field(
        ...,
        min_length=1,
        description="All child predicates must evaluate to true.",
    )

AnyOfConditionSchema

Bases: BaseModel

Boolean OR over nested activation predicates.

Source code in src/phids/api/schemas.py

class AnyOfConditionSchema(BaseModel):
    """Boolean OR over nested activation predicates."""

    kind: Literal["any_of"] = "any_of"
    conditions: list[ConditionNode] = Field(
        ...,
        min_length=1,
        description="At least one child predicate must evaluate to true.",
    )

DietCompatibilityMatrix

Bases: BaseModel

Boolean matrix [predator_species, flora_species] indicating edibility.

Source code in src/phids/api/schemas.py

class DietCompatibilityMatrix(BaseModel):
    """Boolean matrix [predator_species, flora_species] indicating edibility."""

    rows: list[list[bool]] = Field(
        ...,
        description=(
            "Outer index = predator species id, inner index = flora species id. "
            "True means the predator can consume that flora species."
        ),
    )

    @model_validator(mode="after")
    def _validate_shape(self) -> DietCompatibilityMatrix:
        n_pred = len(self.rows)
        if n_pred > MAX_PREDATOR_SPECIES:
            raise ValueError(
                f"DietCompatibilityMatrix has {n_pred} rows, max is {MAX_PREDATOR_SPECIES}."
            )
        for row in self.rows:
            if len(row) > MAX_FLORA_SPECIES:
                raise ValueError(
                    f"DietCompatibilityMatrix row length {len(row)} exceeds {MAX_FLORA_SPECIES}."
                )
        return self

EnemyPresenceConditionSchema

Bases: BaseModel

Leaf predicate requiring a predator species at the owner's cell.

Source code in src/phids/api/schemas.py

class EnemyPresenceConditionSchema(BaseModel):
    """Leaf predicate requiring a predator species at the owner's cell."""

    kind: Literal["enemy_presence"] = "enemy_presence"
    predator_species_id: PredatorId
    min_predator_population: int = Field(
        default=1,
        gt=0,
        description="Minimum co-located predator population required for this predicate.",
    )

FloraSpeciesParams

Bases: BaseModel

Per-species parameters for flora.

Source code in src/phids/api/schemas.py

class FloraSpeciesParams(BaseModel):
    """Per-species parameters for flora."""

    species_id: SpeciesId
    name: str
    base_energy: float = Field(..., gt=0.0)
    max_energy: float = Field(..., gt=0.0)
    growth_rate: float = Field(..., ge=0.0)
    survival_threshold: float = Field(..., ge=0.0)
    reproduction_interval: int = Field(..., gt=0)
    seed_min_dist: float = Field(default=1.0, ge=0.0)
    seed_max_dist: float = Field(default=3.0, gt=0.0)
    seed_energy_cost: float = Field(default=5.0, ge=0.0)
    camouflage: bool = False
    camouflage_factor: float = Field(default=1.0, ge=0.0, le=1.0)
    # Trigger matrix: list of trigger conditions associated with this species
    triggers: list[TriggerConditionSchema] = Field(default_factory=list)

InitialPlantPlacement

Bases: BaseModel

Single plant to place at simulation start.

Source code in src/phids/api/schemas.py

class InitialPlantPlacement(BaseModel):
    """Single plant to place at simulation start."""

    species_id: SpeciesId
    x: int = Field(..., ge=0)
    y: int = Field(..., ge=0)
    energy: float = Field(..., gt=0.0)

InitialSwarmPlacement

Bases: BaseModel

Single swarm to place at simulation start.

Source code in src/phids/api/schemas.py

class InitialSwarmPlacement(BaseModel):
    """Single swarm to place at simulation start."""

    species_id: PredatorId
    x: int = Field(..., ge=0)
    y: int = Field(..., ge=0)
    population: int = Field(..., gt=0)
    energy: float = Field(..., gt=0.0)

PlantComponentSchema

Bases: BaseModel

Pydantic schema for the Plant ECS component.

Source code in src/phids/api/schemas.py

class PlantComponentSchema(BaseModel):
    """Pydantic schema for the Plant ECS component."""

    entity_id: int = Field(..., description="Unique ECS entity identifier.")
    species_id: SpeciesId = Field(..., description="Flora species index [0, MAX_FLORA_SPECIES).")
    x: int = Field(..., ge=0, description="Grid x-coordinate.")
    y: int = Field(..., ge=0, description="Grid y-coordinate.")
    energy: float = Field(..., ge=0.0, description="Current energy reserve (E_i,j).")
    max_energy: float = Field(..., gt=0.0, description="Species-specific energy capacity (E_max).")
    base_energy: float = Field(..., gt=0.0, description="Initial energy E_i,j(0).")
    growth_rate: float = Field(..., ge=0.0, description="Per-tick growth rate r_i,j (%).")
    survival_threshold: float = Field(..., ge=0.0, description="Minimum energy B_i,j before death.")
    reproduction_interval: int = Field(
        ..., gt=0, description="Ticks between reproduction attempts (T_i)."
    )
    seed_min_dist: float = Field(..., ge=0.0, description="Minimum seed dispersal distance d_min.")
    seed_max_dist: float = Field(..., gt=0.0, description="Maximum seed dispersal distance d_max.")
    seed_energy_cost: float = Field(..., ge=0.0, description="Energy cost per reproduction event.")
    camouflage: bool = Field(default=False, description="Constitutive gradient attenuation flag.")
    camouflage_factor: float = Field(
        default=1.0, ge=0.0, le=1.0, description="Gradient multiplier when camouflaged."
    )
    last_reproduction_tick: int = Field(default=0, description="Last tick of reproduction.")

PredatorSpeciesParams

Bases: BaseModel

Per-species parameters for predators (herbivore swarms).

Source code in src/phids/api/schemas.py

class PredatorSpeciesParams(BaseModel):
    """Per-species parameters for predators (herbivore swarms)."""

    species_id: PredatorId
    name: str
    energy_min: float = Field(..., gt=0.0)
    velocity: int = Field(..., gt=0)
    consumption_rate: float = Field(..., gt=0.0)
    reproduction_energy_divisor: float = Field(
        default=1.0,
        gt=0.0,
        description="Denominator for φ(e_h,t) = floor(R(C_i,t) / E_min(e_h)).",
    )

SimulationConfig

Bases: BaseModel

Complete simulation configuration payload (REST /api/scenario/load body).

Source code in src/phids/api/schemas.py

class SimulationConfig(BaseModel):
    """Complete simulation configuration payload (REST /api/scenario/load body)."""

    grid_width: int = Field(default=40, ge=1, le=80)
    grid_height: int = Field(default=40, ge=1, le=80)
    max_ticks: int = Field(default=1000, gt=0)
    tick_rate_hz: float = Field(default=10.0, gt=0.0, description="WebSocket stream tick rate.")

    num_signals: int = Field(default=4, ge=1, le=MAX_SUBSTANCE_TYPES)
    num_toxins: int = Field(default=4, ge=1, le=MAX_SUBSTANCE_TYPES)

    wind_x: float = Field(default=0.0, description="Initial wind vector x-component.")
    wind_y: float = Field(default=0.0, description="Initial wind vector y-component.")

    flora_species: list[FloraSpeciesParams] = Field(..., min_length=1, max_length=MAX_FLORA_SPECIES)
    predator_species: list[PredatorSpeciesParams] = Field(
        ..., min_length=1, max_length=MAX_PREDATOR_SPECIES
    )
    diet_matrix: DietCompatibilityMatrix

    initial_plants: list[InitialPlantPlacement] = Field(default_factory=list)
    initial_swarms: list[InitialSwarmPlacement] = Field(default_factory=list)

    # Symbiotic network settings
    mycorrhizal_inter_species: bool = Field(
        default=False, description="Allow inter-species root connections."
    )
    mycorrhizal_connection_cost: float = Field(
        default=1.0, ge=0.0, description="Energy cost to establish a root link."
    )
    mycorrhizal_growth_interval_ticks: int = Field(
        default=8,
        ge=1,
        le=256,
        description=(
            "Ticks between mycorrhizal growth attempts. "
            "At most one new root link is formed per interval."
        ),
    )
    mycorrhizal_signal_velocity: int = Field(
        default=1, gt=0, description="Signal transfer speed t_g (ticks per hop)."
    )

    # Termination conditions
    z2_flora_species_extinction: int = Field(
        default=-1, description="Halt when this flora species id goes extinct (-1 = disabled)."
    )
    z4_predator_species_extinction: int = Field(
        default=-1,
        description="Halt when this predator species id goes extinct (-1 = disabled).",
    )
    z6_max_total_flora_energy: float = Field(
        default=-1.0, description="Halt when total flora energy exceeds this value (-1 = disabled)."
    )
    z7_max_total_predator_population: int = Field(
        default=-1,
        description="Halt when total predator population exceeds this value (-1 = disabled).",
    )

    @model_validator(mode="after")
    def _validate_species_ids(self) -> SimulationConfig:
        flora_ids = {s.species_id for s in self.flora_species}
        predator_ids = {s.species_id for s in self.predator_species}
        for placement in self.initial_plants:
            if placement.species_id not in flora_ids:
                raise ValueError(
                    f"InitialPlantPlacement references unknown "
                    f"flora species {placement.species_id}."
                )
        for placement in self.initial_swarms:
            if placement.species_id not in predator_ids:
                raise ValueError(
                    f"InitialSwarmPlacement references unknown predator species "
                    f"{placement.species_id}."
                )
        return self

SimulationStatusResponse

Bases: BaseModel

Response model for simulation state queries.

Source code in src/phids/api/schemas.py

class SimulationStatusResponse(BaseModel):
    """Response model for simulation state queries."""

    tick: int
    running: bool
    paused: bool
    terminated: bool
    termination_reason: str | None = None

SubstanceActiveConditionSchema

Bases: BaseModel

Leaf predicate requiring another substance to already be active.

Source code in src/phids/api/schemas.py

class SubstanceActiveConditionSchema(BaseModel):
    """Leaf predicate requiring another substance to already be active."""

    kind: Literal["substance_active"] = "substance_active"
    substance_id: SubstanceId = Field(
        ..., description="Active substance required for this predicate."
    )

SubstanceComponentSchema

Bases: BaseModel

Pydantic schema for a Substance (signal or toxin) ECS component.

Source code in src/phids/api/schemas.py

class SubstanceComponentSchema(BaseModel):
    """Pydantic schema for a Substance (signal or toxin) ECS component."""

    entity_id: int = Field(..., description="Unique ECS entity identifier.")
    substance_id: SubstanceId = Field(..., description="Substance layer index.")
    owner_plant_id: int = Field(..., description="ECS entity id of the producing plant.")
    is_toxin: bool = Field(default=False, description="True for toxins, False for signals.")
    synthesis_remaining: int = Field(
        default=0, ge=0, description="Ticks before substance becomes active."
    )
    active: bool = Field(default=False, description="Whether the substance is currently active.")
    aftereffect_ticks: int = Field(
        default=0, ge=0, description="Remaining aftereffect duration T_k."
    )
    lethal: bool = Field(default=False, description="Lethal toxin flag.")
    lethality_rate: float = Field(
        default=0.0, ge=0.0, description="Individuals eliminated per tick β(s_x, C_i)."
    )
    repellent: bool = Field(default=False, description="Repellent toxin flag.")
    repellent_walk_ticks: int = Field(
        default=0, ge=0, description="Random-walk duration k on repel trigger."
    )
    precursor_signal_id: int = Field(
        default=-1, description="Signal substance_id required before toxin activation (-1 = none)."
    )
    energy_cost_per_tick: float = Field(
        default=0.0, ge=0.0, description="Energy cost drained from the owner plant per active tick."
    )

SwarmComponentSchema

Bases: BaseModel

Pydantic schema for the Herbivore Swarm ECS component.

Source code in src/phids/api/schemas.py

class SwarmComponentSchema(BaseModel):
    """Pydantic schema for the Herbivore Swarm ECS component."""

    entity_id: int = Field(..., description="Unique ECS entity identifier.")
    species_id: PredatorId = Field(
        ..., description="Predator species index [0, MAX_PREDATOR_SPECIES)."
    )
    x: int = Field(..., ge=0, description="Grid x-coordinate.")
    y: int = Field(..., ge=0, description="Grid y-coordinate.")
    population: int = Field(..., gt=0, description="Current swarm head-count n(t).")
    initial_population: int = Field(..., gt=0, description="Initial population n(0) for mitosis.")
    energy: float = Field(..., ge=0.0, description="Current energy reserve.")
    energy_min: float = Field(..., gt=0.0, description="Minimum energy per individual E_min(e_h).")
    velocity: int = Field(..., gt=0, description="Movement period v_h (ticks between moves).")
    consumption_rate: float = Field(..., gt=0.0, description="Per-tick consumption scalar η(C_i).")
    starvation_ticks: int = Field(
        default=0, description="Consecutive ticks without adequate feeding."
    )
    repelled: bool = Field(default=False, description="Currently repelled by toxin.")
    repelled_ticks_remaining: int = Field(
        default=0, description="Ticks remaining in repelled random-walk."
    )
    target_plant_id: int = Field(default=-1, description="ECS id of targeted plant entity.")

TriggerConditionSchema

Bases: BaseModel

Trigger condition for substance synthesis (Interaction Matrix entry).

Maps a (plant species, predator species) pair to the substance that should be synthesised when the trigger conditions are met, together with all behavioural properties of the resulting substance.

Source code in src/phids/api/schemas.py

class TriggerConditionSchema(BaseModel):
    """Trigger condition for substance synthesis (Interaction Matrix entry).

    Maps a (plant species, predator species) pair to the substance that should
    be synthesised when the trigger conditions are met, together with all
    behavioural properties of the resulting substance.
    """

    predator_species_id: PredatorId
    min_predator_population: int = Field(
        ..., gt=0, description="Minimum swarm size n_i,min to trigger synthesis."
    )
    substance_id: SubstanceId = Field(..., description="Substance to synthesise.")
    synthesis_duration: int = Field(..., gt=0, description="Ticks to synthesise T(s_x).")
    is_toxin: bool = Field(default=False, description="True for toxins, False for signals.")
    lethal: bool = Field(default=False, description="Lethal toxin flag.")
    lethality_rate: float = Field(
        default=0.0, ge=0.0, description="Individuals eliminated per tick β(s_x, C_i)."
    )
    repellent: bool = Field(default=False, description="Repellent toxin flag.")
    repellent_walk_ticks: int = Field(
        default=0, ge=0, description="Random-walk duration k on repel trigger."
    )
    aftereffect_ticks: int = Field(
        default=0,
        ge=0,
        description="Aftereffect duration T_k (signals linger after emission ceases).",
    )
    precursor_signal_id: int = Field(
        default=-1,
        description="Single signal substance_id required before toxin activation (-1 = none). Legacy field.",
    )
    precursor_signal_ids: list[int] = Field(
        default_factory=list,
        description=(
            "All signal substance_ids that must ALL be active before this substance activates "
            "(AND logic). Empty list = no precursor required. Takes precedence over precursor_signal_id."
        ),
    )
    activation_condition: ConditionNode | None = Field(
        default=None,
        description=(
            "Optional nested predicate tree controlling whether the configured substance may activate. "
            "Supports explicit all_of/any_of composition over enemy_presence and substance_active leaves."
        ),
    )
    energy_cost_per_tick: float = Field(
        default=0.0, ge=0.0, description="Energy drained from the plant per tick while active."
    )

    @model_validator(mode="after")
    def _populate_activation_condition_from_legacy_precursors(self) -> TriggerConditionSchema:
        """Translate deprecated precursor fields into the richer condition tree."""
        if self.activation_condition is not None:
            return self

        signal_ids = self.precursor_signal_ids
        if not signal_ids and self.precursor_signal_id >= 0:
            signal_ids = [self.precursor_signal_id]

        if not signal_ids:
            return self

        leaves = [
            SubstanceActiveConditionSchema(substance_id=signal_id) for signal_id in signal_ids
        ]
        self.activation_condition = (
            leaves[0] if len(leaves) == 1 else AllOfConditionSchema(conditions=leaves)
        )
        return self

WindUpdatePayload

Bases: BaseModel

REST payload for dynamically updating wind vectors.

Source code in src/phids/api/schemas.py

class WindUpdatePayload(BaseModel):
    """REST payload for dynamically updating wind vectors."""

    wind_x: float
    wind_y: float

Scenario I/O helpers for loading and serialising SimulationConfig.

This module provides convenience functions to parse and validate simulation configurations from Python mappings or JSON files.

load_scenario_from_dict(data)

Parse and validate a simulation configuration from a mapping.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Raw configuration mapping (typically decoded from JSON).	required

Returns:

Name	Type	Description
SimulationConfig	SimulationConfig	Validated Pydantic configuration instance.

Raises:

Type	Description
ValidationError	If the configuration is invalid.

Source code in src/phids/io/scenario.py

def load_scenario_from_dict(data: dict[str, Any]) -> SimulationConfig:
    """Parse and validate a simulation configuration from a mapping.

    Args:
        data: Raw configuration mapping (typically decoded from JSON).

    Returns:
        SimulationConfig: Validated Pydantic configuration instance.

    Raises:
        pydantic.ValidationError: If the configuration is invalid.
    """
    config = SimulationConfig.model_validate(data)
    logger.debug(
        "Scenario validated from mapping (grid=%dx%d, flora=%d, predators=%d)",
        config.grid_width,
        config.grid_height,
        len(config.flora_species),
        len(config.predator_species),
    )
    return config

load_scenario_from_json(path)

Load and validate a simulation configuration from a JSON file.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the JSON scenario file.	required

Returns:

Name	Type	Description
SimulationConfig	SimulationConfig	Validated Pydantic configuration instance.

Source code in src/phids/io/scenario.py

def load_scenario_from_json(path: str | Path) -> SimulationConfig:
    """Load and validate a simulation configuration from a JSON file.

    Args:
        path: Path to the JSON scenario file.

    Returns:
        SimulationConfig: Validated Pydantic configuration instance.
    """
    source = Path(path)
    raw = source.read_text(encoding="utf-8")
    data: dict[str, Any] = json.loads(raw)
    config = load_scenario_from_dict(data)
    logger.info("Scenario loaded from %s", source)
    return config

scenario_to_json(config, path=None)

Serialise a SimulationConfig to a JSON string or file.

Parameters:

Name	Type	Description	Default
config	SimulationConfig	Configuration to serialise.	required
path	str | Path | None	Optional file path to write the JSON to. If None, only the JSON string is returned.	None

Returns:

Name	Type	Description
str	str	JSON representation of the configuration.

Source code in src/phids/io/scenario.py

def scenario_to_json(config: SimulationConfig, path: str | Path | None = None) -> str:
    """Serialise a SimulationConfig to a JSON string or file.

    Args:
        config: Configuration to serialise.
        path: Optional file path to write the JSON to. If ``None``, only the
            JSON string is returned.

    Returns:
        str: JSON representation of the configuration.
    """
    serialised = config.model_dump_json(indent=2)
    if path is not None:
        destination = Path(path)
        destination.write_text(serialised, encoding="utf-8")
        logger.info("Scenario written to %s", destination)
    return serialised

API and UI state surfaces

FastAPI application exposing REST endpoints and WebSocket streaming.

The application provides endpoints for loading scenarios, controlling the simulation lifecycle, updating environmental parameters and exporting telemetry. A WebSocket endpoint streams per-tick grid snapshots.

A second, browser-oriented surface is provided via Jinja2-rendered HTML served under / and /ui/*, together with HTMX-driven config endpoints (/api/config/*, /api/matrices/*) that mutate the server-side :mod:~phids.api.ui_state draft before the operator commits a scenario to the engine.

config_biotope(request, grid_width=40, grid_height=40, max_ticks=1000, tick_rate_hz=10.0, wind_x=0.0, wind_y=0.0, num_signals=4, num_toxins=4, mycorrhizal_inter_species='off', mycorrhizal_connection_cost=1.0, mycorrhizal_growth_interval_ticks=8, mycorrhizal_signal_velocity=1) async

Persist biotope parameters to the draft and return the updated partial.

Parameters:

Name	Type	Description	Default
request	Request	Starlette request.	required
grid_width	Annotated[int, Form()]	Grid width (10–80).	40
grid_height	Annotated[int, Form()]	Grid height (10–80).	40
max_ticks	Annotated[int, Form()]	Tick budget.	1000
tick_rate_hz	Annotated[float, Form()]	WebSocket stream rate.	10.0
wind_x	Annotated[float, Form()]	Uniform wind x-component.	0.0
wind_y	Annotated[float, Form()]	Uniform wind y-component.	0.0
num_signals	Annotated[int, Form()]	Signal layer count.	4
num_toxins	Annotated[int, Form()]	Toxin layer count.	4
mycorrhizal_inter_species	Annotated[str, Form()]	Checkbox value ("on" or "off").	'off'
mycorrhizal_connection_cost	Annotated[float, Form()]	Root link energy cost.	1.0
mycorrhizal_growth_interval_ticks	Annotated[int, Form()]	Ticks between new root-link attempts.	8
mycorrhizal_signal_velocity	Annotated[int, Form()]	Signal hops per tick.	1

Returns:

Name	Type	Description
TemplateResponse	Any	Updated biotope config form partial.

Source code in src/phids/api/main.py

@app.post("/api/config/biotope", response_class=HTMLResponse, summary="Update biotope draft config")
async def config_biotope(
    request: Request,
    grid_width: Annotated[int, Form()] = 40,
    grid_height: Annotated[int, Form()] = 40,
    max_ticks: Annotated[int, Form()] = 1000,
    tick_rate_hz: Annotated[float, Form()] = 10.0,
    wind_x: Annotated[float, Form()] = 0.0,
    wind_y: Annotated[float, Form()] = 0.0,
    num_signals: Annotated[int, Form()] = 4,
    num_toxins: Annotated[int, Form()] = 4,
    mycorrhizal_inter_species: Annotated[str, Form()] = "off",
    mycorrhizal_connection_cost: Annotated[float, Form()] = 1.0,
    mycorrhizal_growth_interval_ticks: Annotated[int, Form()] = 8,
    mycorrhizal_signal_velocity: Annotated[int, Form()] = 1,
) -> Any:
    """Persist biotope parameters to the draft and return the updated partial.

    Args:
        request: Starlette request.
        grid_width: Grid width (10–80).
        grid_height: Grid height (10–80).
        max_ticks: Tick budget.
        tick_rate_hz: WebSocket stream rate.
        wind_x: Uniform wind x-component.
        wind_y: Uniform wind y-component.
        num_signals: Signal layer count.
        num_toxins: Toxin layer count.
        mycorrhizal_inter_species: Checkbox value ("on" or "off").
        mycorrhizal_connection_cost: Root link energy cost.
        mycorrhizal_growth_interval_ticks: Ticks between new root-link attempts.
        mycorrhizal_signal_velocity: Signal hops per tick.

    Returns:
        TemplateResponse: Updated biotope config form partial.
    """
    draft = get_draft()
    clamped_grid_width = max(10, min(80, grid_width))
    clamped_grid_height = max(10, min(80, grid_height))
    clamped_max_ticks = max(1, max_ticks)
    clamped_tick_rate_hz = max(0.1, tick_rate_hz)
    clamped_num_signals = max(1, min(16, num_signals))
    clamped_num_toxins = max(1, min(16, num_toxins))
    clamped_connection_cost = max(0.0, mycorrhizal_connection_cost)
    clamped_growth_interval = max(1, min(256, mycorrhizal_growth_interval_ticks))
    clamped_signal_velocity = max(1, mycorrhizal_signal_velocity)
    draft.grid_width = clamped_grid_width
    draft.grid_height = clamped_grid_height
    draft.max_ticks = clamped_max_ticks
    draft.tick_rate_hz = clamped_tick_rate_hz
    draft.wind_x = wind_x
    draft.wind_y = wind_y
    draft.num_signals = clamped_num_signals
    draft.num_toxins = clamped_num_toxins
    draft.mycorrhizal_inter_species = mycorrhizal_inter_species == "on"
    draft.mycorrhizal_connection_cost = clamped_connection_cost
    draft.mycorrhizal_growth_interval_ticks = clamped_growth_interval
    draft.mycorrhizal_signal_velocity = clamped_signal_velocity
    logger.debug(
        "Draft biotope updated (grid=%dx%d, max_ticks=%d, tick_rate_hz=%.2f, wind=(%.3f, %.3f), signals=%d, toxins=%d, mycorrhiza_interval=%d)",
        draft.grid_width,
        draft.grid_height,
        draft.max_ticks,
        draft.tick_rate_hz,
        draft.wind_x,
        draft.wind_y,
        draft.num_signals,
        draft.num_toxins,
        draft.mycorrhizal_growth_interval_ticks,
    )
    if (
        clamped_grid_width != grid_width
        or clamped_grid_height != grid_height
        or clamped_max_ticks != max_ticks
        or clamped_tick_rate_hz != tick_rate_hz
        or clamped_num_signals != num_signals
        or clamped_num_toxins != num_toxins
        or clamped_connection_cost != mycorrhizal_connection_cost
        or clamped_growth_interval != mycorrhizal_growth_interval_ticks
        or clamped_signal_velocity != mycorrhizal_signal_velocity
    ):
        logger.warning(
            "Draft biotope values were clamped to valid ranges (requested_grid=%dx%d, applied_grid=%dx%d)",
            grid_width,
            grid_height,
            clamped_grid_width,
            clamped_grid_height,
        )
    return templates.TemplateResponse(
        request,
        "partials/biotope_config.html",
        {"draft": draft},
    )

config_flora_add(request, name='NewFlora', base_energy=10.0, max_energy=100.0, growth_rate=5.0, survival_threshold=1.0, reproduction_interval=10, seed_min_dist=1.0, seed_max_dist=3.0, seed_energy_cost=5.0, camouflage='off', camouflage_factor=1.0) async

Add a new flora species to the draft and return the updated table.

Parameters:

Name	Type	Description	Default
request	Request	Starlette request.	required
name	Annotated[str, Form()]	Species name.	'NewFlora'
base_energy	Annotated[float, Form()]	Starting energy.	10.0
max_energy	Annotated[float, Form()]	Energy capacity.	100.0
growth_rate	Annotated[float, Form()]	Per-tick growth percentage.	5.0
survival_threshold	Annotated[float, Form()]	Minimum energy before death.	1.0
reproduction_interval	Annotated[int, Form()]	Ticks between seeds.	10
seed_min_dist	Annotated[float, Form()]	Minimum dispersal distance.	1.0
seed_max_dist	Annotated[float, Form()]	Maximum dispersal distance.	3.0
seed_energy_cost	Annotated[float, Form()]	Energy cost per seed event.	5.0

Returns:

Name	Type	Description
TemplateResponse	Any	Updated flora config table partial.

Raises:

Type	Description
HTTPException	400 if Rule of 16 limit is already reached.

Source code in src/phids/api/main.py

@app.post("/api/config/flora", response_class=HTMLResponse, summary="Add flora species to draft")
async def config_flora_add(
    request: Request,
    name: Annotated[str, Form()] = "NewFlora",
    base_energy: Annotated[float, Form()] = 10.0,
    max_energy: Annotated[float, Form()] = 100.0,
    growth_rate: Annotated[float, Form()] = 5.0,
    survival_threshold: Annotated[float, Form()] = 1.0,
    reproduction_interval: Annotated[int, Form()] = 10,
    seed_min_dist: Annotated[float, Form()] = 1.0,
    seed_max_dist: Annotated[float, Form()] = 3.0,
    seed_energy_cost: Annotated[float, Form()] = 5.0,
    camouflage: Annotated[str, Form()] = "off",
    camouflage_factor: Annotated[float, Form()] = 1.0,
) -> Any:
    """Add a new flora species to the draft and return the updated table.

    Args:
        request: Starlette request.
        name: Species name.
        base_energy: Starting energy.
        max_energy: Energy capacity.
        growth_rate: Per-tick growth percentage.
        survival_threshold: Minimum energy before death.
        reproduction_interval: Ticks between seeds.
        seed_min_dist: Minimum dispersal distance.
        seed_max_dist: Maximum dispersal distance.
        seed_energy_cost: Energy cost per seed event.

    Returns:
        TemplateResponse: Updated flora config table partial.

    Raises:
        HTTPException: 400 if Rule of 16 limit is already reached.
    """
    draft = get_draft()
    if len(draft.flora_species) >= 16:
        logger.warning("Rule-of-16 rejected flora creation")
        raise HTTPException(status_code=400, detail="Rule of 16: maximum flora species reached.")
    new_id = len(draft.flora_species)
    params = FloraSpeciesParams(
        species_id=new_id,
        name=name,
        base_energy=base_energy,
        max_energy=max_energy,
        growth_rate=growth_rate,
        survival_threshold=survival_threshold,
        reproduction_interval=reproduction_interval,
        seed_min_dist=seed_min_dist,
        seed_max_dist=seed_max_dist,
        seed_energy_cost=seed_energy_cost,
        camouflage=camouflage == "on",
        camouflage_factor=max(0.0, min(1.0, camouflage_factor)),
        triggers=[],
    )
    draft.add_flora(params)
    logger.info("Flora species added via API (species_id=%d, name=%s)", new_id, name)
    return templates.TemplateResponse(
        request,
        "partials/flora_config.html",
        {"flora_species": draft.flora_species},
    )

config_flora_delete(species_id) async

Remove a flora species from the draft.

Parameters:

Name	Type	Description	Default
species_id	int	The species_id to remove.	required

Returns:

Name	Type	Description
HTMLResponse	HTMLResponse	Empty string (HTMX removes the row with outerHTML swap).

Raises:

Type	Description
HTTPException	404 if species_id not found.

Source code in src/phids/api/main.py

@app.delete(
    "/api/config/flora/{species_id}", response_class=HTMLResponse, summary="Delete flora species"
)
async def config_flora_delete(species_id: int) -> HTMLResponse:
    """Remove a flora species from the draft.

    Args:
        species_id: The ``species_id`` to remove.

    Returns:
        HTMLResponse: Empty string (HTMX removes the row with ``outerHTML`` swap).

    Raises:
        HTTPException: 404 if species_id not found.
    """
    draft = get_draft()
    try:
        draft.remove_flora(species_id)
    except ValueError as exc:
        logger.warning("Flora delete requested for unknown species_id=%d", species_id)
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    return HTMLResponse(content="")

config_flora_update(request, species_id, name=None, base_energy=None, max_energy=None, growth_rate=None, survival_threshold=None, reproduction_interval=None, seed_min_dist=None, seed_max_dist=None, seed_energy_cost=None, camouflage=None, camouflage_factor=None) async

Patch a single flora species in the draft and return the updated row.

Parameters:

Name	Type	Description	Default
request	Request	Starlette request.	required
species_id	int	Target species id.	required
name	Annotated[str | None, Form()]	Updated name (optional).	None
base_energy	Annotated[float | None, Form()]	Updated base energy (optional).	None
max_energy	Annotated[float | None, Form()]	Updated max energy (optional).	None
growth_rate	Annotated[float | None, Form()]	Updated growth rate (optional).	None
survival_threshold	Annotated[float | None, Form()]	Updated survival threshold (optional).	None
reproduction_interval	Annotated[int | None, Form()]	Updated reproduction interval (optional).	None
seed_min_dist	Annotated[float | None, Form()]	Updated minimum seed distance (optional).	None
seed_max_dist	Annotated[float | None, Form()]	Updated maximum seed distance (optional).	None
seed_energy_cost	Annotated[float | None, Form()]	Updated seed energy cost (optional).	None
camouflage	Annotated[str | None, Form()]	Checkbox value "on" or absent (optional).	None
camouflage_factor	Annotated[float | None, Form()]	Updated camouflage factor 0–1 (optional).	None

Returns:

Name	Type	Description
TemplateResponse	Any	Updated <tr> row partial (outerHTML swap).

Raises:

Type	Description
HTTPException	404 if species_id not found.

Source code in src/phids/api/main.py

@app.put(
    "/api/config/flora/{species_id}",
    response_class=HTMLResponse,
    summary="Update flora species row",
)
async def config_flora_update(
    request: Request,
    species_id: int,
    name: Annotated[str | None, Form()] = None,
    base_energy: Annotated[float | None, Form()] = None,
    max_energy: Annotated[float | None, Form()] = None,
    growth_rate: Annotated[float | None, Form()] = None,
    survival_threshold: Annotated[float | None, Form()] = None,
    reproduction_interval: Annotated[int | None, Form()] = None,
    seed_min_dist: Annotated[float | None, Form()] = None,
    seed_max_dist: Annotated[float | None, Form()] = None,
    seed_energy_cost: Annotated[float | None, Form()] = None,
    camouflage: Annotated[str | None, Form()] = None,
    camouflage_factor: Annotated[float | None, Form()] = None,
) -> Any:
    """Patch a single flora species in the draft and return the updated row.

    Args:
        request: Starlette request.
        species_id: Target species id.
        name: Updated name (optional).
        base_energy: Updated base energy (optional).
        max_energy: Updated max energy (optional).
        growth_rate: Updated growth rate (optional).
        survival_threshold: Updated survival threshold (optional).
        reproduction_interval: Updated reproduction interval (optional).
        seed_min_dist: Updated minimum seed distance (optional).
        seed_max_dist: Updated maximum seed distance (optional).
        seed_energy_cost: Updated seed energy cost (optional).
        camouflage: Checkbox value "on" or absent (optional).
        camouflage_factor: Updated camouflage factor 0–1 (optional).

    Returns:
        TemplateResponse: Updated ``<tr>`` row partial (outerHTML swap).

    Raises:
        HTTPException: 404 if species_id not found.
    """
    draft = get_draft()
    idx = next(
        (
            i
            for i, fp in enumerate(draft.flora_species)
            if hasattr(fp, "species_id") and fp.species_id == species_id  # type: ignore[union-attr]
        ),
        None,
    )
    if idx is None:
        logger.warning("Flora update requested for unknown species_id=%d", species_id)
        raise HTTPException(status_code=404, detail=f"Flora species {species_id} not found.")

    fp = draft.flora_species[idx]
    updates: dict[str, Any] = {}
    if name is not None:
        updates["name"] = name
    if base_energy is not None:
        updates["base_energy"] = base_energy
    if max_energy is not None:
        updates["max_energy"] = max_energy
    if growth_rate is not None:
        updates["growth_rate"] = growth_rate
    if survival_threshold is not None:
        updates["survival_threshold"] = survival_threshold
    if reproduction_interval is not None:
        updates["reproduction_interval"] = reproduction_interval
    if seed_min_dist is not None:
        updates["seed_min_dist"] = seed_min_dist
    if seed_max_dist is not None:
        updates["seed_max_dist"] = seed_max_dist
    if seed_energy_cost is not None:
        updates["seed_energy_cost"] = seed_energy_cost
    if camouflage is not None:
        updates["camouflage"] = camouflage == "on"
    if camouflage_factor is not None:
        updates["camouflage_factor"] = max(0.0, min(1.0, camouflage_factor))

    draft.flora_species[idx] = fp.model_copy(update=updates)  # type: ignore[union-attr]
    logger.debug(
        "Flora species updated via API (species_id=%d, fields=%s)", species_id, sorted(updates)
    )
    return templates.TemplateResponse(
        request,
        "partials/flora_config.html",
        {"flora_species": draft.flora_species},
    )

config_placement_plant_add(request, species_id, x, y, energy=10.0) async

Place a plant entity at the given grid position.

Parameters:

Name	Type	Description	Default
request	Request	Starlette request.	required
species_id	Annotated[int, Form()]	Flora species id.	required
x	Annotated[int, Form()]	Grid x-coordinate.	required
y	Annotated[int, Form()]	Grid y-coordinate.	required
energy	Annotated[float, Form()]	Starting energy.	10.0

Returns:

Name	Type	Description
TemplateResponse	Any	Updated placement list partial.

Source code in src/phids/api/main.py

@app.post(
    "/api/config/placements/plant", response_class=HTMLResponse, summary="Place a plant on the grid"
)
async def config_placement_plant_add(
    request: Request,
    species_id: Annotated[int, Form()],
    x: Annotated[int, Form()],
    y: Annotated[int, Form()],
    energy: Annotated[float, Form()] = 10.0,
) -> Any:
    """Place a plant entity at the given grid position.

    Args:
        request: Starlette request.
        species_id: Flora species id.
        x: Grid x-coordinate.
        y: Grid y-coordinate.
        energy: Starting energy.

    Returns:
        TemplateResponse: Updated placement list partial.
    """
    draft = get_draft()
    x = max(0, min(draft.grid_width - 1, x))
    y = max(0, min(draft.grid_height - 1, y))
    draft.add_plant_placement(species_id, x, y, max(0.1, energy))
    logger.info("Plant placement added via API (species_id=%d, x=%d, y=%d)", species_id, x, y)
    return templates.TemplateResponse(
        request,
        "partials/placement_list.html",
        {
            "flora_species": draft.flora_species,
            "predator_species": draft.predator_species,
            "initial_plants": draft.initial_plants,
            "initial_swarms": draft.initial_swarms,
        },
    )

config_placement_plant_delete(request, index) async

Remove a plant placement by list index.

Parameters:

Name	Type	Description	Default
request	Request	Starlette request.	required
index	int	Position in the initial_plants list.	required

Returns:

Name	Type	Description
TemplateResponse	Any	Updated placement list partial.

Source code in src/phids/api/main.py

@app.delete(
    "/api/config/placements/plant/{index}",
    response_class=HTMLResponse,
    summary="Remove a placed plant",
)
async def config_placement_plant_delete(request: Request, index: int) -> Any:
    """Remove a plant placement by list index.

    Args:
        request: Starlette request.
        index: Position in the initial_plants list.

    Returns:
        TemplateResponse: Updated placement list partial.
    """
    draft = get_draft()
    try:
        draft.remove_plant_placement(index)
    except IndexError as exc:
        logger.warning("Plant placement delete requested for unknown index=%d", index)
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    return templates.TemplateResponse(
        request,
        "partials/placement_list.html",
        {
            "flora_species": draft.flora_species,
            "predator_species": draft.predator_species,
            "initial_plants": draft.initial_plants,
            "initial_swarms": draft.initial_swarms,
        },
    )

config_placement_swarm_add(request, species_id, x, y, population=10, energy=50.0) async

Place a herbivore swarm at the given grid position.

Parameters:

Name	Type	Description	Default
request	Request	Starlette request.	required
species_id	Annotated[int, Form()]	Predator species id.	required
x	Annotated[int, Form()]	Grid x-coordinate.	required
y	Annotated[int, Form()]	Grid y-coordinate.	required
population	Annotated[int, Form()]	Initial swarm size.	10
energy	Annotated[float, Form()]	Starting energy.	50.0

Returns:

Name	Type	Description
TemplateResponse	Any	Updated placement list partial.

Source code in src/phids/api/main.py

@app.post(
    "/api/config/placements/swarm", response_class=HTMLResponse, summary="Place a swarm on the grid"
)
async def config_placement_swarm_add(
    request: Request,
    species_id: Annotated[int, Form()],
    x: Annotated[int, Form()],
    y: Annotated[int, Form()],
    population: Annotated[int, Form()] = 10,
    energy: Annotated[float, Form()] = 50.0,
) -> Any:
    """Place a herbivore swarm at the given grid position.

    Args:
        request: Starlette request.
        species_id: Predator species id.
        x: Grid x-coordinate.
        y: Grid y-coordinate.
        population: Initial swarm size.
        energy: Starting energy.

    Returns:
        TemplateResponse: Updated placement list partial.
    """
    draft = get_draft()
    x = max(0, min(draft.grid_width - 1, x))
    y = max(0, min(draft.grid_height - 1, y))
    draft.add_swarm_placement(species_id, x, y, max(1, population), max(0.1, energy))
    logger.info(
        "Swarm placement added via API (species_id=%d, x=%d, y=%d, population=%d)",
        species_id,
        x,
        y,
        max(1, population),
    )
    return templates.TemplateResponse(
        request,
        "partials/placement_list.html",
        {
            "flora_species": draft.flora_species,
            "predator_species": draft.predator_species,
            "initial_plants": draft.initial_plants,
            "initial_swarms": draft.initial_swarms,
        },
    )

config_placement_swarm_delete(request, index) async

Remove a swarm placement by list index.

Parameters:

Name	Type	Description	Default
request	Request	Starlette request.	required
index	int	Position in the initial_swarms list.	required

Returns:

Name	Type	Description
TemplateResponse	Any	Updated placement list partial.

Source code in src/phids/api/main.py

@app.delete(
    "/api/config/placements/swarm/{index}",
    response_class=HTMLResponse,
    summary="Remove a placed swarm",
)
async def config_placement_swarm_delete(request: Request, index: int) -> Any:
    """Remove a swarm placement by list index.

    Args:
        request: Starlette request.
        index: Position in the initial_swarms list.

    Returns:
        TemplateResponse: Updated placement list partial.
    """
    draft = get_draft()
    try:
        draft.remove_swarm_placement(index)
    except IndexError as exc:
        logger.warning("Swarm placement delete requested for unknown index=%d", index)
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    return templates.TemplateResponse(
        request,
        "partials/placement_list.html",
        {
            "flora_species": draft.flora_species,
            "predator_species": draft.predator_species,
            "initial_plants": draft.initial_plants,
            "initial_swarms": draft.initial_swarms,
        },
    )

config_placements_clear(request) async

Remove all plant and swarm placements from the draft.

Parameters:

Name	Type	Description	Default
request	Request	Starlette request.	required

Returns:

Name	Type	Description
TemplateResponse	Any	Updated placement list partial.

Source code in src/phids/api/main.py

@app.post(
    "/api/config/placements/clear", response_class=HTMLResponse, summary="Clear all placements"
)
async def config_placements_clear(request: Request) -> Any:
    """Remove all plant and swarm placements from the draft.

    Args:
        request: Starlette request.

    Returns:
        TemplateResponse: Updated placement list partial.
    """
    draft = get_draft()
    draft.clear_placements()
    logger.info("All draft placements cleared via API")
    return templates.TemplateResponse(
        request,
        "partials/placement_list.html",
        {
            "flora_species": draft.flora_species,
            "predator_species": draft.predator_species,
            "initial_plants": draft.initial_plants,
            "initial_swarms": draft.initial_swarms,
        },
    )

config_predator_add(request, name='NewPredator', energy_min=5.0, velocity=2, consumption_rate=10.0, reproduction_energy_divisor=1.0) async

Add a new predator species to the draft.

Parameters:

Name	Type	Description	Default
request	Request	Starlette request.	required
name	Annotated[str, Form()]	Species name.	'NewPredator'
energy_min	Annotated[float, Form()]	Minimum per-individual energy.	5.0
velocity	Annotated[int, Form()]	Movement period in ticks.	2
consumption_rate	Annotated[float, Form()]	Per-tick consumption scalar.	10.0
reproduction_energy_divisor	Annotated[float, Form()]	Divisor for reproduction formula.	1.0

Returns:

Name	Type	Description
TemplateResponse	Any	Updated predator config table partial.

Raises:

Type	Description
HTTPException	400 if Rule of 16 limit is reached.

Source code in src/phids/api/main.py

@app.post(
    "/api/config/predators", response_class=HTMLResponse, summary="Add predator species to draft"
)
async def config_predator_add(
    request: Request,
    name: Annotated[str, Form()] = "NewPredator",
    energy_min: Annotated[float, Form()] = 5.0,
    velocity: Annotated[int, Form()] = 2,
    consumption_rate: Annotated[float, Form()] = 10.0,
    reproduction_energy_divisor: Annotated[float, Form()] = 1.0,
) -> Any:
    """Add a new predator species to the draft.

    Args:
        request: Starlette request.
        name: Species name.
        energy_min: Minimum per-individual energy.
        velocity: Movement period in ticks.
        consumption_rate: Per-tick consumption scalar.
        reproduction_energy_divisor: Divisor for reproduction formula.

    Returns:
        TemplateResponse: Updated predator config table partial.

    Raises:
        HTTPException: 400 if Rule of 16 limit is reached.
    """
    draft = get_draft()
    if len(draft.predator_species) >= 16:
        logger.warning("Rule-of-16 rejected predator creation")
        raise HTTPException(status_code=400, detail="Rule of 16: maximum predator species reached.")
    new_id = len(draft.predator_species)
    params = PredatorSpeciesParams(
        species_id=new_id,
        name=name,
        energy_min=energy_min,
        velocity=velocity,
        consumption_rate=consumption_rate,
        reproduction_energy_divisor=reproduction_energy_divisor,
    )
    draft.add_predator(params)
    logger.info("Predator species added via API (species_id=%d, name=%s)", new_id, name)
    return templates.TemplateResponse(
        request,
        "partials/predator_config.html",
        {"predator_species": draft.predator_species},
    )

config_predator_delete(species_id) async

Remove a predator species from the draft.

Parameters:

Name	Type	Description	Default
species_id	int	The species_id to remove.	required

Returns:

Name	Type	Description
HTMLResponse	HTMLResponse	Empty string (HTMX removes the row).

Raises:

Type	Description
HTTPException	404 if species_id not found.

Source code in src/phids/api/main.py

@app.delete(
    "/api/config/predators/{species_id}",
    response_class=HTMLResponse,
    summary="Delete predator species",
)
async def config_predator_delete(species_id: int) -> HTMLResponse:
    """Remove a predator species from the draft.

    Args:
        species_id: The ``species_id`` to remove.

    Returns:
        HTMLResponse: Empty string (HTMX removes the row).

    Raises:
        HTTPException: 404 if species_id not found.
    """
    draft = get_draft()
    try:
        draft.remove_predator(species_id)
    except ValueError as exc:
        logger.warning("Predator delete requested for unknown species_id=%d", species_id)
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    return HTMLResponse(content="")

config_predator_update(request, species_id, name=None, energy_min=None, velocity=None, consumption_rate=None, reproduction_energy_divisor=None) async

Patch a single predator species in the draft.

Parameters:

Name	Type	Description	Default
request	Request	Starlette request.	required
species_id	int	Target species id.	required
name	Annotated[str | None, Form()]	Updated name (optional).	None
energy_min	Annotated[float | None, Form()]	Updated energy minimum (optional).	None
velocity	Annotated[int | None, Form()]	Updated velocity (optional).	None
consumption_rate	Annotated[float | None, Form()]	Updated consumption rate (optional).	None
reproduction_energy_divisor	Annotated[float | None, Form()]	Updated reproduction divisor (optional).	None

Returns:

Name	Type	Description
TemplateResponse	Any	Updated <tr> row partial.

Raises:

Type	Description
HTTPException	404 if species_id not found.

Source code in src/phids/api/main.py

@app.put(
    "/api/config/predators/{species_id}",
    response_class=HTMLResponse,
    summary="Update predator species row",
)
async def config_predator_update(
    request: Request,
    species_id: int,
    name: Annotated[str | None, Form()] = None,
    energy_min: Annotated[float | None, Form()] = None,
    velocity: Annotated[int | None, Form()] = None,
    consumption_rate: Annotated[float | None, Form()] = None,
    reproduction_energy_divisor: Annotated[float | None, Form()] = None,
) -> Any:
    """Patch a single predator species in the draft.

    Args:
        request: Starlette request.
        species_id: Target species id.
        name: Updated name (optional).
        energy_min: Updated energy minimum (optional).
        velocity: Updated velocity (optional).
        consumption_rate: Updated consumption rate (optional).
        reproduction_energy_divisor: Updated reproduction divisor (optional).

    Returns:
        TemplateResponse: Updated ``<tr>`` row partial.

    Raises:
        HTTPException: 404 if species_id not found.
    """
    draft = get_draft()
    idx = next(
        (
            i
            for i, pp in enumerate(draft.predator_species)
            if hasattr(pp, "species_id") and pp.species_id == species_id  # type: ignore[union-attr]
        ),
        None,
    )
    if idx is None:
        logger.warning("Predator update requested for unknown species_id=%d", species_id)
        raise HTTPException(status_code=404, detail=f"Predator species {species_id} not found.")

    pp = draft.predator_species[idx]
    updates: dict[str, Any] = {}
    if name is not None:
        updates["name"] = name
    if energy_min is not None:
        updates["energy_min"] = energy_min
    if velocity is not None:
        updates["velocity"] = velocity
    if consumption_rate is not None:
        updates["consumption_rate"] = consumption_rate
    if reproduction_energy_divisor is not None:
        updates["reproduction_energy_divisor"] = reproduction_energy_divisor

    draft.predator_species[idx] = pp.model_copy(update=updates)  # type: ignore[union-attr]
    logger.debug(
        "Predator species updated via API (species_id=%d, fields=%s)", species_id, sorted(updates)
    )
    return templates.TemplateResponse(
        request,
        "partials/predator_config.html",
        {"predator_species": draft.predator_species},
    )

config_substance_add(request, name='Signal', is_toxin='false', lethal='false', repellent='false', synthesis_duration=3, aftereffect_ticks=0, lethality_rate=0.0, repellent_walk_ticks=3, energy_cost_per_tick=1.0) async

Add a substance definition to the draft.

Parameters:

Name	Type	Description	Default
request	Request	Starlette request.	required
name	Annotated[str, Form()]	Substance name.	'Signal'
is_toxin	Annotated[str, Form()]	"true" for toxins.	'false'
lethal	Annotated[str, Form()]	"true" for lethal toxins.	'false'
repellent	Annotated[str, Form()]	"true" for repellent toxins.	'false'
synthesis_duration	Annotated[int, Form()]	Production ticks.	3
aftereffect_ticks	Annotated[int, Form()]	Linger duration.	0
lethality_rate	Annotated[float, Form()]	Population eliminated per tick.	0.0
repellent_walk_ticks	Annotated[int, Form()]	Random-walk duration.	3
energy_cost_per_tick	Annotated[float, Form()]	Energy drain per active tick.	1.0

Returns: TemplateResponse: Updated substance config table partial.

Raises:

Type	Description
HTTPException	400 if Rule of 16 limit is reached.

Source code in src/phids/api/main.py

@app.post(
    "/api/config/substances",
    response_class=HTMLResponse,
    summary="Add substance definition to draft",
)
async def config_substance_add(
    request: Request,
    name: Annotated[str, Form()] = "Signal",
    is_toxin: Annotated[str, Form()] = "false",
    lethal: Annotated[str, Form()] = "false",
    repellent: Annotated[str, Form()] = "false",
    synthesis_duration: Annotated[int, Form()] = 3,
    aftereffect_ticks: Annotated[int, Form()] = 0,
    lethality_rate: Annotated[float, Form()] = 0.0,
    repellent_walk_ticks: Annotated[int, Form()] = 3,
    energy_cost_per_tick: Annotated[float, Form()] = 1.0,
) -> Any:
    """Add a substance definition to the draft.

    Args:
        request: Starlette request.
        name: Substance name.
        is_toxin: ``"true"`` for toxins.
        lethal: ``"true"`` for lethal toxins.
        repellent: ``"true"`` for repellent toxins.
        synthesis_duration: Production ticks.
        aftereffect_ticks: Linger duration.
        lethality_rate: Population eliminated per tick.
        repellent_walk_ticks: Random-walk duration.
        energy_cost_per_tick: Energy drain per active tick.
    Returns:
        TemplateResponse: Updated substance config table partial.

    Raises:
        HTTPException: 400 if Rule of 16 limit is reached.
    """
    draft = get_draft()
    if len(draft.substance_definitions) >= 16:
        logger.warning("Rule-of-16 rejected substance creation")
        raise HTTPException(status_code=400, detail="Rule of 16: maximum substances reached.")
    new_id = len(draft.substance_definitions)
    toxin_flag = is_toxin.lower() in ("true", "1", "yes", "on")
    lethal_flag = lethal.lower() in ("true", "1", "yes", "on")
    repellent_flag = repellent.lower() in ("true", "1", "yes", "on")
    draft.substance_definitions.append(
        SubstanceDefinition(
            substance_id=new_id,
            name=name,
            is_toxin=toxin_flag,
            lethal=lethal_flag,
            repellent=repellent_flag,
            synthesis_duration=max(1, synthesis_duration),
            aftereffect_ticks=max(0, aftereffect_ticks),
            lethality_rate=max(0.0, lethality_rate),
            repellent_walk_ticks=max(0, repellent_walk_ticks),
            energy_cost_per_tick=max(0.0, energy_cost_per_tick),
        )
    )
    logger.info(
        "Substance added via API (substance_id=%d, name=%s, is_toxin=%s)", new_id, name, toxin_flag
    )
    return templates.TemplateResponse(
        request,
        "partials/substance_config.html",
        {"substances": draft.substance_definitions},
    )

config_substance_delete(substance_id) async

Remove a substance definition from the draft.

Parameters:

Name	Type	Description	Default
substance_id	int	The substance_id to remove.	required

Returns:

Name	Type	Description
HTMLResponse	HTMLResponse	Empty string (HTMX removes the row).

Raises:

Type	Description
HTTPException	404 if substance_id not found.

Source code in src/phids/api/main.py

@app.delete(
    "/api/config/substances/{substance_id}",
    response_class=HTMLResponse,
    summary="Delete substance definition",
)
async def config_substance_delete(substance_id: int) -> HTMLResponse:
    """Remove a substance definition from the draft.

    Args:
        substance_id: The ``substance_id`` to remove.

    Returns:
        HTMLResponse: Empty string (HTMX removes the row).

    Raises:
        HTTPException: 404 if substance_id not found.
    """
    draft = get_draft()
    idx = next(
        (i for i, s in enumerate(draft.substance_definitions) if s.substance_id == substance_id),
        None,
    )
    if idx is None:
        logger.warning("Substance delete requested for unknown substance_id=%d", substance_id)
        raise HTTPException(status_code=404, detail=f"Substance {substance_id} not found.")
    del draft.substance_definitions[idx]
    logger.info("Substance deleted via API (substance_id=%d)", substance_id)
    return HTMLResponse(content="")

config_substance_update(request, substance_id, name=None, type_label=None, synthesis_duration=None, aftereffect_ticks=None, lethality_rate=None, repellent_walk_ticks=None, energy_cost_per_tick=None) async

Patch a substance definition in the draft.

Parameters:

Name	Type	Description	Default
request	Request	Starlette request.	required
substance_id	int	Target substance id.	required
name	Annotated[str | None, Form()]	Updated name (optional).	None
type_label	Annotated[str | None, Form()]	One of "Signal", "Lethal Toxin", "Repellent Toxin", "Toxin" (optional).	None
synthesis_duration	Annotated[int | None, Form()]	Updated synthesis duration (optional).	None
aftereffect_ticks	Annotated[int | None, Form()]	Updated aftereffect ticks (optional).	None
lethality_rate	Annotated[float | None, Form()]	Updated lethality rate (optional).	None
repellent_walk_ticks	Annotated[int | None, Form()]	Updated random-walk duration (optional).	None
energy_cost_per_tick	Annotated[float | None, Form()]	Updated energy drain per tick (optional).	None

Returns:

Name	Type	Description
TemplateResponse	Any	Updated substance config table partial.

Raises:

Type	Description
HTTPException	404 if substance_id not found.

Source code in src/phids/api/main.py

@app.put(
    "/api/config/substances/{substance_id}",
    response_class=HTMLResponse,
    summary="Update substance definition row",
)
async def config_substance_update(
    request: Request,
    substance_id: int,
    name: Annotated[str | None, Form()] = None,
    type_label: Annotated[str | None, Form()] = None,
    synthesis_duration: Annotated[int | None, Form()] = None,
    aftereffect_ticks: Annotated[int | None, Form()] = None,
    lethality_rate: Annotated[float | None, Form()] = None,
    repellent_walk_ticks: Annotated[int | None, Form()] = None,
    energy_cost_per_tick: Annotated[float | None, Form()] = None,
) -> Any:
    """Patch a substance definition in the draft.

    Args:
        request: Starlette request.
        substance_id: Target substance id.
        name: Updated name (optional).
        type_label: One of ``"Signal"``, ``"Lethal Toxin"``,
            ``"Repellent Toxin"``, ``"Toxin"`` (optional).
        synthesis_duration: Updated synthesis duration (optional).
        aftereffect_ticks: Updated aftereffect ticks (optional).
        lethality_rate: Updated lethality rate (optional).
        repellent_walk_ticks: Updated random-walk duration (optional).
        energy_cost_per_tick: Updated energy drain per tick (optional).

    Returns:
        TemplateResponse: Updated substance config table partial.

    Raises:
        HTTPException: 404 if substance_id not found.
    """
    draft = get_draft()
    idx = next(
        (i for i, s in enumerate(draft.substance_definitions) if s.substance_id == substance_id),
        None,
    )
    if idx is None:
        logger.warning("Substance update requested for unknown substance_id=%d", substance_id)
        raise HTTPException(status_code=404, detail=f"Substance {substance_id} not found.")

    sd = draft.substance_definitions[idx]
    if name is not None:
        sd.name = name
    if type_label is not None:
        sd.is_toxin = type_label in ("Lethal Toxin", "Repellent Toxin", "Toxin")
        sd.lethal = type_label == "Lethal Toxin"
        sd.repellent = type_label == "Repellent Toxin"
    if synthesis_duration is not None:
        sd.synthesis_duration = max(1, synthesis_duration)
    if aftereffect_ticks is not None:
        sd.aftereffect_ticks = max(0, aftereffect_ticks)
    if lethality_rate is not None:
        sd.lethality_rate = max(0.0, lethality_rate)
    if repellent_walk_ticks is not None:
        sd.repellent_walk_ticks = max(0, repellent_walk_ticks)
    if energy_cost_per_tick is not None:
        sd.energy_cost_per_tick = max(0.0, energy_cost_per_tick)
    logger.debug("Substance updated via API (substance_id=%d, name=%s)", substance_id, sd.name)

    return templates.TemplateResponse(
        request,
        "partials/substance_config.html",
        {"substances": draft.substance_definitions},
    )

config_trigger_rule_add(request, flora_species_id, predator_species_id, substance_id, min_predator_population=5, activation_condition_json='') async

Add a new trigger rule to the draft.

Parameters:

Name	Type	Description	Default
request	Request	Starlette request.	required
flora_species_id	Annotated[int, Form()]	Flora species index.	required
predator_species_id	Annotated[int, Form()]	Predator species index.	required
substance_id	Annotated[int, Form()]	Substance to synthesise.	required
min_predator_population	Annotated[int, Form()]	Minimum swarm size to trigger.	5
activation_condition_json	Annotated[str, Form()]	Optional JSON condition tree controlling activation after synthesis completes.	''

Returns:

Name	Type	Description
TemplateResponse	Any	Updated trigger rules partial.

Source code in src/phids/api/main.py

@app.post("/api/config/trigger-rules", response_class=HTMLResponse, summary="Add a trigger rule")
async def config_trigger_rule_add(
    request: Request,
    flora_species_id: Annotated[int, Form()],
    predator_species_id: Annotated[int, Form()],
    substance_id: Annotated[int, Form()],
    min_predator_population: Annotated[int, Form()] = 5,
    activation_condition_json: Annotated[str, Form()] = "",
) -> Any:
    """Add a new trigger rule to the draft.

    Args:
        request: Starlette request.
        flora_species_id: Flora species index.
        predator_species_id: Predator species index.
        substance_id: Substance to synthesise.
        min_predator_population: Minimum swarm size to trigger.
        activation_condition_json: Optional JSON condition tree controlling
            activation after synthesis completes.

    Returns:
        TemplateResponse: Updated trigger rules partial.
    """
    draft = get_draft()
    draft.add_trigger_rule(
        flora_species_id=flora_species_id,
        predator_species_id=predator_species_id,
        substance_id=substance_id,
        min_predator_population=max(1, min_predator_population),
        activation_condition=_parse_activation_condition_json(activation_condition_json),
    )
    logger.info(
        "Trigger rule added via API (flora_species_id=%d, predator_species_id=%d, substance_id=%d)",
        flora_species_id,
        predator_species_id,
        substance_id,
    )
    return templates.TemplateResponse(
        request, "partials/trigger_rules.html", _trigger_rules_template_context(draft)
    )

config_trigger_rule_condition_child_add(request, index, node_kind, parent_path='') async

Append a new child node to an existing group condition.

Source code in src/phids/api/main.py

@app.post(
    "/api/config/trigger-rules/{index}/condition/child",
    response_class=HTMLResponse,
    summary="Append a child node to a trigger-rule condition group",
)
async def config_trigger_rule_condition_child_add(
    request: Request,
    index: int,
    node_kind: Annotated[str, Form()],
    parent_path: Annotated[str, Form()] = "",
) -> Any:
    """Append a new child node to an existing group condition."""
    draft = get_draft()
    rule = _trigger_rule_by_index(draft, index)
    try:
        draft.append_trigger_rule_condition_child(
            index,
            parent_path,
            _default_activation_condition_for_rule(draft, rule, node_kind),
        )
    except IndexError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    return templates.TemplateResponse(
        request, "partials/trigger_rules.html", _trigger_rules_template_context(draft)
    )

config_trigger_rule_condition_delete(request, index, path='') async

Delete one activation-condition node or clear the whole tree.

Source code in src/phids/api/main.py

@app.post(
    "/api/config/trigger-rules/{index}/condition/delete",
    response_class=HTMLResponse,
    summary="Delete a node from a trigger-rule condition tree",
)
async def config_trigger_rule_condition_delete(
    request: Request,
    index: int,
    path: Annotated[str, Form()] = "",
) -> Any:
    """Delete one activation-condition node or clear the whole tree."""
    draft = get_draft()
    _trigger_rule_by_index(draft, index)
    try:
        draft.delete_trigger_rule_condition_node(index, path)
    except IndexError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    return templates.TemplateResponse(
        request, "partials/trigger_rules.html", _trigger_rules_template_context(draft)
    )

config_trigger_rule_condition_node_update(request, index, path, kind=None, predator_species_id=None, min_predator_population=None, substance_id=None) async

Update or replace a node in a trigger-rule activation-condition tree.

Source code in src/phids/api/main.py

@app.put(
    "/api/config/trigger-rules/{index}/condition/node",
    response_class=HTMLResponse,
    summary="Update one node in a trigger-rule condition tree",
)
async def config_trigger_rule_condition_node_update(
    request: Request,
    index: int,
    path: Annotated[str, Form()],
    kind: Annotated[str | None, Form()] = None,
    predator_species_id: Annotated[int | None, Form()] = None,
    min_predator_population: Annotated[int | None, Form()] = None,
    substance_id: Annotated[int | None, Form()] = None,
) -> Any:
    """Update or replace a node in a trigger-rule activation-condition tree."""
    draft = get_draft()
    rule = _trigger_rule_by_index(draft, index)

    if rule.activation_condition is None:
        if kind is None or path:
            raise HTTPException(
                status_code=400, detail="Trigger rule has no activation condition to update."
            )
        draft.set_trigger_rule_activation_condition(
            index,
            _default_activation_condition_for_rule(draft, rule, kind),
        )
        return templates.TemplateResponse(
            request, "partials/trigger_rules.html", _trigger_rules_template_context(draft)
        )

    try:
        current_node: dict[str, Any] = rule.activation_condition
        if path:
            path_indices = [int(part) for part in path.split(".") if part != ""]
            for child_index in path_indices:
                children = current_node.get("conditions")
                if current_node.get("kind") not in {"all_of", "any_of"} or not isinstance(
                    children, list
                ):
                    raise IndexError("Condition path traversed into a non-group node.")
                next_node = children[child_index]
                if not isinstance(next_node, dict):
                    raise IndexError("Condition path resolved to an invalid child node.")
                current_node = next_node

        if kind is not None and current_node.get("kind") != kind:
            replacement = _default_activation_condition_for_rule(draft, rule, kind)
            draft.replace_trigger_rule_condition_node(index, path, replacement)
        else:
            updates: dict[str, object] = {}
            if current_node.get("kind") == "enemy_presence":
                if predator_species_id is not None:
                    updates["predator_species_id"] = predator_species_id
                if min_predator_population is not None:
                    updates["min_predator_population"] = max(1, min_predator_population)
            elif current_node.get("kind") == "substance_active":
                if substance_id is not None:
                    updates["substance_id"] = substance_id
            elif current_node.get("kind") in {"all_of", "any_of"} and kind is not None:
                updates["kind"] = kind

            if updates:
                draft.update_trigger_rule_condition_node(index, path, **updates)
    except IndexError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc

    return templates.TemplateResponse(
        request, "partials/trigger_rules.html", _trigger_rules_template_context(draft)
    )

config_trigger_rule_condition_root(request, index, node_kind) async

Create the root activation condition for one trigger rule.

Source code in src/phids/api/main.py

@app.post(
    "/api/config/trigger-rules/{index}/condition/root",
    response_class=HTMLResponse,
    summary="Create or replace a trigger rule root condition",
)
async def config_trigger_rule_condition_root(
    request: Request,
    index: int,
    node_kind: Annotated[str, Form()],
) -> Any:
    """Create the root activation condition for one trigger rule."""
    draft = get_draft()
    rule = _trigger_rule_by_index(draft, index)
    draft.set_trigger_rule_activation_condition(
        index,
        _default_activation_condition_for_rule(draft, rule, node_kind),
    )
    return templates.TemplateResponse(
        request, "partials/trigger_rules.html", _trigger_rules_template_context(draft)
    )

config_trigger_rule_delete(request, index) async

Remove a trigger rule from the draft.

Parameters:

Name	Type	Description	Default
request	Request	Starlette request.	required
index	int	Rule position in the list.	required

Returns:

Name	Type	Description
TemplateResponse	Any	Updated trigger rules partial.

Raises:

Type	Description
HTTPException	404 if index out of range.

Source code in src/phids/api/main.py

@app.delete(
    "/api/config/trigger-rules/{index}",
    response_class=HTMLResponse,
    summary="Delete a trigger rule",
)
async def config_trigger_rule_delete(request: Request, index: int) -> Any:
    """Remove a trigger rule from the draft.

    Args:
        request: Starlette request.
        index: Rule position in the list.

    Returns:
        TemplateResponse: Updated trigger rules partial.

    Raises:
        HTTPException: 404 if index out of range.
    """
    draft = get_draft()
    try:
        draft.remove_trigger_rule(index)
    except IndexError as exc:
        logger.warning("Trigger rule delete requested for unknown index=%d", index)
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    return templates.TemplateResponse(
        request, "partials/trigger_rules.html", _trigger_rules_template_context(draft)
    )

config_trigger_rule_update(request, index, flora_species_id=None, predator_species_id=None, substance_id=None, min_predator_population=None, activation_condition_json=None) async

Update an existing trigger rule in the draft.

Parameters:

Name	Type	Description	Default
request	Request	Starlette request.	required
index	int	Rule position in the list.	required
flora_species_id	Annotated[int | None, Form()]	New flora species id (optional).	None
predator_species_id	Annotated[int | None, Form()]	New predator species id (optional).	None
substance_id	Annotated[int | None, Form()]	New substance id (optional).	None
min_predator_population	Annotated[int | None, Form()]	New minimum population (optional).	None
activation_condition_json	Annotated[str | None, Form()]	Optional JSON condition tree.	None

Returns:

Name	Type	Description
TemplateResponse	Any	Updated trigger rules partial.

Raises:

Type	Description
HTTPException	404 if index out of range.

Source code in src/phids/api/main.py

@app.put(
    "/api/config/trigger-rules/{index}",
    response_class=HTMLResponse,
    summary="Update a trigger rule",
)
async def config_trigger_rule_update(
    request: Request,
    index: int,
    flora_species_id: Annotated[int | None, Form()] = None,
    predator_species_id: Annotated[int | None, Form()] = None,
    substance_id: Annotated[int | None, Form()] = None,
    min_predator_population: Annotated[int | None, Form()] = None,
    activation_condition_json: Annotated[str | None, Form()] = None,
) -> Any:
    """Update an existing trigger rule in the draft.

    Args:
        request: Starlette request.
        index: Rule position in the list.
        flora_species_id: New flora species id (optional).
        predator_species_id: New predator species id (optional).
        substance_id: New substance id (optional).
        min_predator_population: New minimum population (optional).
        activation_condition_json: Optional JSON condition tree.

    Returns:
        TemplateResponse: Updated trigger rules partial.

    Raises:
        HTTPException: 404 if index out of range.
    """
    draft = get_draft()
    if index < 0 or index >= len(draft.trigger_rules):
        logger.warning("Trigger rule update requested for unknown index=%d", index)
        raise HTTPException(status_code=404, detail=f"Trigger rule {index} not found.")

    draft.update_trigger_rule(
        index,
        flora_species_id=flora_species_id,
        predator_species_id=predator_species_id,
        substance_id=substance_id,
        min_predator_population=min_predator_population,
        activation_condition=(
            _parse_activation_condition_json(activation_condition_json)
            if activation_condition_json is not None
            else None
        ),
    )
    logger.debug("Trigger rule updated via API (index=%d)", index)
    return templates.TemplateResponse(
        request, "partials/trigger_rules.html", _trigger_rules_template_context(draft)
    )

export_telemetry_csv() async

Stream Lotka-Volterra analytics as a downloadable CSV file.

Returns:

Name	Type	Description
Response	Response	FastAPI response containing CSV bytes and headers.

Source code in src/phids/api/main.py

@app.get("/api/telemetry/export/csv", summary="Export telemetry as CSV")
async def export_telemetry_csv() -> Response:
    """Stream Lotka-Volterra analytics as a downloadable CSV file.

    Returns:
        Response: FastAPI response containing CSV bytes and headers.
    """
    loop = _get_loop()
    data = export_bytes_csv(loop.telemetry.dataframe)
    logger.info("Telemetry exported as CSV (%d rows)", loop.telemetry.dataframe.height)
    return Response(
        content=data,
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=telemetry.csv"},
    )

export_telemetry_json() async

Stream Lotka-Volterra analytics as a downloadable NDJSON file.

Returns:

Name	Type	Description
Response	Response	FastAPI response containing NDJSON bytes and headers.

Source code in src/phids/api/main.py

@app.get("/api/telemetry/export/json", summary="Export telemetry as JSON")
async def export_telemetry_json() -> Response:
    """Stream Lotka-Volterra analytics as a downloadable NDJSON file.

    Returns:
        Response: FastAPI response containing NDJSON bytes and headers.
    """
    loop = _get_loop()
    data = export_bytes_json(loop.telemetry.dataframe)
    logger.info("Telemetry exported as NDJSON (%d rows)", loop.telemetry.dataframe.height)
    return Response(
        content=data,
        media_type="application/x-ndjson",
        headers={"Content-Disposition": "attachment; filename=telemetry.ndjson"},
    )

load_scenario(config) async

Initialise the simulation loop with the provided configuration.

Parameters:

Name	Type	Description	Default
config	SimulationConfig	Validated :class:~phids.api.schemas.SimulationConfig.	required

Returns:

Name	Type	Description
dict	dict[str, Any]	Confirmation message including grid dimensions.

Source code in src/phids/api/main.py

@app.post("/api/scenario/load", summary="Load simulation scenario")
async def load_scenario(config: SimulationConfig) -> dict[str, Any]:
    """Initialise the simulation loop with the provided configuration.

    Args:
        config: Validated :class:`~phids.api.schemas.SimulationConfig`.

    Returns:
        dict: Confirmation message including grid dimensions.
    """
    global _sim_loop, _sim_task  # noqa: PLW0603

    # Cancel any running simulation
    if _sim_task is not None and not _sim_task.done():
        logger.info("Cancelling existing background simulation task before loading a new scenario")
        _sim_task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await _sim_task

    _sim_loop = SimulationLoop(config)
    _set_simulation_substance_names(config)
    logger.info(
        "Scenario loaded: %dx%d grid, %d flora species, %d predator species",
        config.grid_width,
        config.grid_height,
        len(config.flora_species),
        len(config.predator_species),
    )
    return {
        "message": "Scenario loaded.",
        "grid_width": config.grid_width,
        "grid_height": config.grid_height,
    }

log_http_requests(request, call_next) async

Log API/UI request timing with low overhead.

DEBUG logging is emitted for successful API, HTMX, and UI requests. WARNING logging is emitted for client/server error responses.

Source code in src/phids/api/main.py

@app.middleware("http")
async def log_http_requests(request: Request, call_next: Any) -> Response:
    """Log API/UI request timing with low overhead.

    DEBUG logging is emitted for successful API, HTMX, and UI requests.
    WARNING logging is emitted for client/server error responses.
    """
    started = time.perf_counter()
    response = await call_next(request)
    duration_ms = (time.perf_counter() - started) * 1000.0

    is_interactive_path = request.url.path.startswith(("/api/", "/ui/")) or request.url.path == "/"
    if response.status_code >= 400 and is_interactive_path:
        logger.warning(
            "HTTP %s %s -> %d in %.2fms",
            request.method,
            request.url.path,
            response.status_code,
            duration_ms,
        )
    elif is_interactive_path and logger.isEnabledFor(logging.DEBUG):
        logger.debug(
            "HTTP %s %s -> %d in %.2fms%s",
            request.method,
            request.url.path,
            response.status_code,
            duration_ms,
            " [HTMX]" if _is_htmx_request(request) else "",
        )

    return response

matrix_diet(request, predator_idx, flora_idx, compatible='toggle') async

Toggle or set a diet compatibility matrix cell.

Parameters:

Name	Type	Description	Default
request	Request	Starlette request.	required
predator_idx	Annotated[int, Form()]	Row index (predator species list position).	required
flora_idx	Annotated[int, Form()]	Column index (flora species list position).	required
compatible	Annotated[str, Form()]	"toggle" to flip or "true"/"false" to set.	'toggle'

Returns:

Name	Type	Description
TemplateResponse	Any	Updated diet matrix partial.

Source code in src/phids/api/main.py

@app.post("/api/matrices/diet", response_class=HTMLResponse, summary="Toggle diet matrix cell")
async def matrix_diet(
    request: Request,
    predator_idx: Annotated[int, Form()],
    flora_idx: Annotated[int, Form()],
    compatible: Annotated[str, Form()] = "toggle",
) -> Any:
    """Toggle or set a diet compatibility matrix cell.

    Args:
        request: Starlette request.
        predator_idx: Row index (predator species list position).
        flora_idx: Column index (flora species list position).
        compatible: ``"toggle"`` to flip or ``"true"``/``"false"`` to set.

    Returns:
        TemplateResponse: Updated diet matrix partial.
    """
    draft = get_draft()
    if predator_idx < len(draft.diet_matrix) and flora_idx < len(draft.diet_matrix[predator_idx]):
        if compatible == "toggle":
            draft.diet_matrix[predator_idx][flora_idx] = not draft.diet_matrix[predator_idx][
                flora_idx
            ]
        else:
            draft.diet_matrix[predator_idx][flora_idx] = compatible.lower() in ("true", "1", "on")
        logger.debug(
            "Diet matrix updated (predator_idx=%d, flora_idx=%d, compatible=%s)",
            predator_idx,
            flora_idx,
            draft.diet_matrix[predator_idx][flora_idx],
        )
    else:
        logger.warning(
            "Diet matrix update ignored for out-of-range indices (predator_idx=%d, flora_idx=%d)",
            predator_idx,
            flora_idx,
        )
    return templates.TemplateResponse(
        request,
        "partials/diet_matrix.html",
        {
            "flora_species": draft.flora_species,
            "predator_species": draft.predator_species,
            "diet_matrix": draft.diet_matrix,
        },
    )

pause_simulation(request) async

Toggle pause state of the running simulation.

Returns:

Name	Type	Description
dict	Any	Message indicating current paused/resumed state.

Source code in src/phids/api/main.py

@app.post("/api/simulation/pause", summary="Pause or resume simulation")
async def pause_simulation(request: Request) -> Any:
    """Toggle pause state of the running simulation.

    Returns:
        dict: Message indicating current paused/resumed state.
    """
    loop = _get_loop()
    loop.pause()
    state = "paused" if loop.paused else "resumed"
    logger.info("Simulation %s via API", state)
    if _is_htmx_request(request):
        return HTMLResponse(content=_render_status_badge_html())
    return {"message": f"Simulation {state}."}

placement_data() async

Return current plant and swarm placements as JSON for canvas rendering.

Returns:

Name	Type	Description
JSONResponse	JSONResponse	JSON with plants, swarms, grid_width, grid_height, flora_species, predator_species.

Source code in src/phids/api/main.py

@app.get("/api/config/placements/data", summary="Get placement data as JSON")
async def placement_data() -> JSONResponse:
    """Return current plant and swarm placements as JSON for canvas rendering.

    Returns:
        JSONResponse: JSON with ``plants``, ``swarms``, ``grid_width``,
            ``grid_height``, ``flora_species``, ``predator_species``.
    """
    draft = get_draft()
    plants = [
        {"idx": i, "species_id": p.species_id, "x": p.x, "y": p.y, "energy": p.energy}
        for i, p in enumerate(draft.initial_plants)
    ]
    swarms = [
        {
            "idx": i,
            "species_id": s.species_id,
            "x": s.x,
            "y": s.y,
            "population": s.population,
            "energy": s.energy,
        }
        for i, s in enumerate(draft.initial_swarms)
    ]
    flora = [
        {"species_id": getattr(fp, "species_id", i), "name": getattr(fp, "name", f"Flora {i}")}
        for i, fp in enumerate(draft.flora_species)
    ]
    predators = [
        {"species_id": getattr(pp, "species_id", i), "name": getattr(pp, "name", f"Pred {i}")}
        for i, pp in enumerate(draft.predator_species)
    ]
    mycorrhizal_links = _build_draft_mycorrhizal_links(draft)
    return JSONResponse(
        content={
            "plants": plants,
            "swarms": swarms,
            "grid_width": draft.grid_width,
            "grid_height": draft.grid_height,
            "flora_species": flora,
            "predator_species": predators,
            "mycorrhizal_links": mycorrhizal_links,
        }
    )

reset_simulation(request) async

Recreate the live simulation loop from the currently loaded config.

Returns:

Type	Description
Any	dict[str, Any]: Confirmation and reset tick.

Source code in src/phids/api/main.py

@app.post("/api/simulation/reset", summary="Reset simulation to the loaded scenario")
async def reset_simulation(request: Request) -> Any:
    """Recreate the live simulation loop from the currently loaded config.

    Returns:
        dict[str, Any]: Confirmation and reset tick.
    """
    global _sim_loop, _sim_task  # noqa: PLW0603

    loop = _get_loop()

    if _sim_task is not None and not _sim_task.done():
        logger.info("Cancelling existing background simulation task before reset")
        _sim_task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await _sim_task

    _sim_loop = SimulationLoop(loop.config)
    _sim_task = None
    _set_simulation_substance_names(loop.config)
    logger.info("Simulation reset to the loaded scenario")
    if _is_htmx_request(request):
        return HTMLResponse(content=_render_status_badge_html())
    return {"message": "Simulation reset.", "tick": 0}

root(request) async

Render the PHIDS Control Centre root page.

Parameters:

Name	Type	Description	Default
request	Request	FastAPI/Starlette request object.	required

Returns:

Name	Type	Description
TemplateResponse	Any	Rendered index.html with dashboard content.

Source code in src/phids/api/main.py

@app.get("/", response_class=HTMLResponse, summary="Main UI")
async def root(request: Request) -> Any:
    """Render the PHIDS Control Centre root page.

    Args:
        request: FastAPI/Starlette request object.

    Returns:
        TemplateResponse: Rendered ``index.html`` with dashboard content.
    """
    draft = get_draft()
    return templates.TemplateResponse(
        request,
        "index.html",
        {"scenario_name": draft.scenario_name},
    )

scenario_export() async

Serialise the draft state as a downloadable JSON file.

Returns:

Name	Type	Description
Response	Response	JSON file download response.

Source code in src/phids/api/main.py

@app.get("/api/scenario/export", summary="Export draft as JSON")
async def scenario_export() -> Response:
    """Serialise the draft state as a downloadable JSON file.

    Returns:
        Response: JSON file download response.
    """
    draft = get_draft()
    try:
        config = draft.build_sim_config()
        data = json.dumps(config.model_dump(), indent=2)  # type: ignore[union-attr]
    except (ValueError, AttributeError) as exc:
        logger.warning("Scenario export failed: %s", exc)
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    logger.info("Draft scenario exported (scenario_name=%s)", draft.scenario_name)
    return Response(
        content=data,
        media_type="application/json",
        headers={
            "Content-Disposition": (
                f'attachment; filename="{draft.scenario_name.replace(" ", "_")}.json"'
            )
        },
    )

scenario_import(file=File(...)) async

Parse an uploaded JSON scenario and replace the draft.

Parameters:

Name	Type	Description	Default
file	UploadFile	Uploaded .json file containing a SimulationConfig serialisation.	File(...)

Returns:

Name	Type	Description
JSONResponse	JSONResponse	Confirmation with imported scenario grid dimensions.

Raises:

Type	Description
HTTPException	422 if the JSON does not validate against :class:~phids.api.schemas.SimulationConfig.

Source code in src/phids/api/main.py

@app.post("/api/scenario/import", summary="Import scenario from JSON file")
async def scenario_import(file: UploadFile = File(...)) -> JSONResponse:
    """Parse an uploaded JSON scenario and replace the draft.

    Args:
        file: Uploaded ``.json`` file containing a ``SimulationConfig``
            serialisation.

    Returns:
        JSONResponse: Confirmation with imported scenario grid dimensions.

    Raises:
        HTTPException: 422 if the JSON does not validate against
            :class:`~phids.api.schemas.SimulationConfig`.
    """
    raw = await file.read()
    try:
        payload = json.loads(raw)
        config = SimulationConfig.model_validate(payload)
    except Exception as exc:
        logger.warning("Scenario import failed for file %s: %s", file.filename, exc)
        raise HTTPException(status_code=422, detail=f"Invalid scenario JSON: {exc}") from exc

    # Reconstruct trigger rules from the imported flora triggers
    imported_trigger_rules: list[TriggerRule] = []
    imported_substances: list[SubstanceDefinition] = []
    seen_substance_ids: set[int] = set()
    for flora_spec in config.flora_species:
        for trig in flora_spec.triggers:
            imported_trigger_rules.append(
                TriggerRule(
                    flora_species_id=flora_spec.species_id,
                    predator_species_id=trig.predator_species_id,
                    substance_id=trig.substance_id,
                    min_predator_population=trig.min_predator_population,
                    activation_condition=(
                        trig.activation_condition.model_dump(mode="json")
                        if trig.activation_condition is not None
                        else None
                    ),
                )
            )
            if trig.substance_id not in seen_substance_ids:
                seen_substance_ids.add(trig.substance_id)
                imported_substances.append(
                    SubstanceDefinition(
                        substance_id=trig.substance_id,
                        name=f"Substance {trig.substance_id}",
                        is_toxin=trig.is_toxin,
                        lethal=trig.lethal,
                        repellent=trig.repellent,
                        synthesis_duration=trig.synthesis_duration,
                        aftereffect_ticks=trig.aftereffect_ticks,
                        lethality_rate=trig.lethality_rate,
                        repellent_walk_ticks=trig.repellent_walk_ticks,
                        energy_cost_per_tick=trig.energy_cost_per_tick,
                        precursor_signal_id=(
                            trig.precursor_signal_ids[0]
                            if len(trig.precursor_signal_ids) == 1
                            else trig.precursor_signal_id
                        ),
                        min_predator_population=trig.min_predator_population,
                    )
                )

    new_draft = DraftState(
        scenario_name=(file.filename or "imported").replace(".json", ""),
        grid_width=config.grid_width,
        grid_height=config.grid_height,
        max_ticks=config.max_ticks,
        tick_rate_hz=config.tick_rate_hz,
        wind_x=config.wind_x,
        wind_y=config.wind_y,
        num_signals=config.num_signals,
        num_toxins=config.num_toxins,
        mycorrhizal_inter_species=config.mycorrhizal_inter_species,
        mycorrhizal_connection_cost=config.mycorrhizal_connection_cost,
        mycorrhizal_growth_interval_ticks=config.mycorrhizal_growth_interval_ticks,
        mycorrhizal_signal_velocity=config.mycorrhizal_signal_velocity,
        flora_species=list(config.flora_species),
        predator_species=list(config.predator_species),
        diet_matrix=[list(row) for row in config.diet_matrix.rows],
        trigger_rules=imported_trigger_rules,
        substance_definitions=imported_substances,
        initial_plants=[
            PlacedPlant(species_id=p.species_id, x=p.x, y=p.y, energy=p.energy)
            for p in config.initial_plants
        ],
        initial_swarms=[
            PlacedSwarm(
                species_id=s.species_id,
                x=s.x,
                y=s.y,
                population=s.population,
                energy=s.energy,
            )
            for s in config.initial_swarms
        ],
    )
    set_draft(new_draft)
    logger.info(
        "Scenario imported into draft (file=%s, grid=%dx%d, flora=%d, predators=%d)",
        file.filename,
        config.grid_width,
        config.grid_height,
        len(config.flora_species),
        len(config.predator_species),
    )
    return JSONResponse(
        content={
            "message": "Scenario imported.",
            "grid_width": config.grid_width,
            "grid_height": config.grid_height,
        }
    )

scenario_load_draft(request) async

Commit the current draft to the simulation engine.

Equivalent to POST /api/scenario/load but uses the server-side draft rather than a request body.

Returns:

Name	Type	Description
HTMLResponse	Any	Updated status badge HTML fragment.

Raises:

Type	Description
HTTPException	400 if draft is invalid or missing required species.

Source code in src/phids/api/main.py

@app.post(
    "/api/scenario/load-draft",
    response_class=HTMLResponse,
    summary="Load draft config into simulation engine",
)
async def scenario_load_draft(request: Request) -> Any:
    """Commit the current draft to the simulation engine.

    Equivalent to ``POST /api/scenario/load`` but uses the server-side
    draft rather than a request body.

    Returns:
        HTMLResponse: Updated status badge HTML fragment.

    Raises:
        HTTPException: 400 if draft is invalid or missing required species.
    """
    global _sim_loop, _sim_task  # noqa: PLW0603

    draft = get_draft()
    try:
        config = draft.build_sim_config()
    except (ValueError, Exception) as exc:
        logger.warning("Draft load into simulation failed: %s", exc)
        raise HTTPException(status_code=400, detail=str(exc)) from exc

    if _sim_task is not None and not _sim_task.done():
        logger.info("Cancelling existing background simulation task before loading draft")
        _sim_task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await _sim_task

    _sim_loop = SimulationLoop(config)  # type: ignore[arg-type]
    _sim_task = None
    _set_simulation_substance_names(config, draft=draft)  # type: ignore[arg-type]
    logger.info(
        "Draft loaded: %dx%d grid, %d flora, %d predators",
        config.grid_width,
        config.grid_height,  # type: ignore[union-attr]
        len(config.flora_species),
        len(config.predator_species),  # type: ignore[union-attr]
    )
    return HTMLResponse(content=_render_status_badge_html())

simulation_status() async

Return the current tick and simulation state flags.

Returns:

Name	Type	Description
SimulationStatusResponse	SimulationStatusResponse	Pydantic response model with status.

Source code in src/phids/api/main.py

@app.get(
    "/api/simulation/status",
    response_model=SimulationStatusResponse,
    summary="Get simulation status",
)
async def simulation_status() -> SimulationStatusResponse:
    """Return the current tick and simulation state flags.

    Returns:
        SimulationStatusResponse: Pydantic response model with status.
    """
    loop = _get_loop()
    return SimulationStatusResponse(
        tick=loop.tick,
        running=loop.running,
        paused=loop.paused,
        terminated=loop.terminated,
        termination_reason=loop.termination_reason,
    )

simulation_stream(websocket) async

Stream grid state snapshots over WebSocket at each simulation tick.

The state is serialised with :mod:msgpack and compressed with :mod:zlib for compact binary transport. If no scenario is loaded the connection is closed with code 1008 (Policy Violation).

Source code in src/phids/api/main.py

@app.websocket("/ws/simulation/stream")
async def simulation_stream(websocket: WebSocket) -> None:
    """Stream grid state snapshots over WebSocket at each simulation tick.

    The state is serialised with :mod:`msgpack` and compressed with
    :mod:`zlib` for compact binary transport. If no scenario is loaded the
    connection is closed with code 1008 (Policy Violation).
    """
    await websocket.accept()
    logger.debug("WebSocket connected: /ws/simulation/stream")

    if _sim_loop is None:
        logger.warning("Closing /ws/simulation/stream because no scenario is loaded")
        await websocket.close(code=1008, reason="No scenario loaded.")
        return

    loop = _sim_loop
    last_tick = -1

    try:
        while True:
            if loop.terminated:
                # Send final state and close
                snapshot = loop.get_state_snapshot()
                packed = msgpack.packb(snapshot, use_bin_type=True)
                await websocket.send_bytes(zlib.compress(packed))
                break

            if loop.tick != last_tick:
                snapshot = loop.get_state_snapshot()
                packed = msgpack.packb(snapshot, use_bin_type=True)
                await websocket.send_bytes(zlib.compress(packed))
                last_tick = loop.tick

            await asyncio.sleep(1.0 / max(1.0, loop.config.tick_rate_hz))

    except WebSocketDisconnect:
        logger.info("WebSocket client disconnected from /ws/simulation/stream")
    finally:
        await websocket.close()

start_simulation(request) async

Begin background execution of the simulation loop.

Returns:

Name	Type	Description
dict	Any	Message confirming the simulation was started.

Source code in src/phids/api/main.py

@app.post("/api/simulation/start", summary="Start or resume simulation")
async def start_simulation(request: Request) -> Any:
    """Begin background execution of the simulation loop.

    Returns:
        dict: Message confirming the simulation was started.
    """
    global _sim_task  # noqa: PLW0603
    loop = _get_loop()

    if loop.running and not loop.paused:
        logger.info("Start requested while simulation was already running")
        if _is_htmx_request(request):
            return HTMLResponse(content=_render_status_badge_html())
        return {"message": "Simulation already running."}

    if loop.terminated:
        logger.warning(
            "Start requested for a terminated simulation (reason=%s)", loop.termination_reason
        )
        raise HTTPException(status_code=400, detail="Simulation has terminated.")

    loop.start()

    async def _bg() -> None:
        await loop.run()

    _sim_task = asyncio.create_task(_bg())
    logger.info("Background simulation task created")
    if _is_htmx_request(request):
        return HTMLResponse(content=_render_status_badge_html())
    return {"message": "Simulation started."}

step_simulation(request) async

Execute a single deterministic simulation tick.

Returns:

Type	Description
Any	dict[str, Any]: Updated simulation status after the step.

Raises:

Type	Description
HTTPException	If the simulation is running in the background or has terminated.

Source code in src/phids/api/main.py

@app.post("/api/simulation/step", summary="Advance simulation by one tick")
async def step_simulation(request: Request) -> Any:
    """Execute a single deterministic simulation tick.

    Returns:
        dict[str, Any]: Updated simulation status after the step.

    Raises:
        HTTPException: If the simulation is running in the background or has terminated.
    """
    loop = _get_loop()

    if _sim_task is not None and not _sim_task.done() and loop.running and not loop.paused:
        logger.warning("Single-step requested while simulation is already running")
        raise HTTPException(status_code=400, detail="Pause the simulation before stepping.")

    if loop.terminated:
        logger.warning(
            "Single-step requested for a terminated simulation (reason=%s)", loop.termination_reason
        )
        raise HTTPException(status_code=400, detail="Simulation has terminated.")

    result = await loop.step()
    logger.info(
        "Simulation advanced by one tick via API (tick=%d, terminated=%s)",
        loop.tick,
        result.terminated,
    )
    if _is_htmx_request(request):
        return HTMLResponse(content=_render_status_badge_html())
    return {
        "message": "Simulation advanced by one tick.",
        "tick": loop.tick,
        "terminated": loop.terminated,
        "termination_reason": loop.termination_reason,
    }

telemetry_chart(request) async

Return an SVG chart fragment for the HTMX polling target.

Parameters:

Name	Type	Description	Default
request	Request	FastAPI/Starlette request object.	required

Returns:

Name	Type	Description
TemplateResponse	Any	Rendered partials/telemetry_chart.html.

Source code in src/phids/api/main.py

@app.get("/api/telemetry", summary="Telemetry SVG chart partial")
async def telemetry_chart(request: Request) -> Any:
    """Return an SVG chart fragment for the HTMX polling target.

    Args:
        request: FastAPI/Starlette request object.

    Returns:
        TemplateResponse: Rendered ``partials/telemetry_chart.html``.
    """
    if _sim_loop is None:
        svg = _build_telemetry_svg(None)
        legend = False
        latest_metrics = None
        live_summary = None
    else:
        svg = _build_telemetry_svg(_sim_loop.telemetry.dataframe)
        legend = True
        latest_metrics = _sim_loop.telemetry.get_latest_metrics()
        live_summary = _build_live_summary()

    return templates.TemplateResponse(
        request,
        "partials/telemetry_chart.html",
        {
            "svg_content": svg,
            "legend": legend,
            "latest_metrics": latest_metrics,
            "live_summary": live_summary,
        },
    )

ui_biotope(request) async

Return the biotope configuration form partial.

Parameters:

Name	Type	Description	Default
request	Request	FastAPI/Starlette request object.	required

Returns:

Name	Type	Description
TemplateResponse	Any	Rendered partials/biotope_config.html.

Source code in src/phids/api/main.py

@app.get("/ui/biotope", response_class=HTMLResponse, summary="Biotope config partial")
async def ui_biotope(request: Request) -> Any:
    """Return the biotope configuration form partial.

    Args:
        request: FastAPI/Starlette request object.

    Returns:
        TemplateResponse: Rendered ``partials/biotope_config.html``.
    """
    return templates.TemplateResponse(
        request,
        "partials/biotope_config.html",
        {"draft": get_draft()},
    )

ui_cell_details(x, y, expected_tick=None) async

Return rich grid-cell details for dashboard tooltips.

When a live simulation exists, data is sourced from the current ECS world and environment layers. Otherwise the draft placement preview is returned.

Source code in src/phids/api/main.py

@app.get("/api/ui/cell-details", summary="Detailed tooltip payload for one grid cell")
async def ui_cell_details(x: int, y: int, expected_tick: int | None = None) -> JSONResponse:
    """Return rich grid-cell details for dashboard tooltips.

    When a live simulation exists, data is sourced from the current ECS world
    and environment layers. Otherwise the draft placement preview is returned.
    """
    if _sim_loop is not None and expected_tick is not None and expected_tick != _sim_loop.tick:
        return JSONResponse(
            status_code=409,
            content={
                "detail": "Live simulation advanced before tooltip details were fetched.",
                "expected_tick": expected_tick,
                "tick": _sim_loop.tick,
            },
        )

    payload = (
        _build_live_cell_details(_sim_loop, x, y)
        if _sim_loop is not None
        else _build_preview_cell_details(x, y)
    )
    return JSONResponse(content=payload)

ui_dashboard(request) async

Return the dashboard HTMX partial.

Parameters:

Name	Type	Description	Default
request	Request	FastAPI/Starlette request object.	required

Returns:

Name	Type	Description
TemplateResponse	Any	Rendered partials/dashboard.html.

Source code in src/phids/api/main.py

@app.get("/ui/dashboard", response_class=HTMLResponse, summary="Dashboard partial")
async def ui_dashboard(request: Request) -> Any:
    """Return the dashboard HTMX partial.

    Args:
        request: FastAPI/Starlette request object.

    Returns:
        TemplateResponse: Rendered ``partials/dashboard.html``.
    """
    return templates.TemplateResponse(request, "partials/dashboard.html")

ui_diagnostics_backend(request) async

Render structured recent backend logs for the diagnostics rail.

Source code in src/phids/api/main.py

@app.get("/ui/diagnostics/backend", response_class=HTMLResponse, summary="Diagnostics backend tab")
async def ui_diagnostics_backend(request: Request) -> Any:
    """Render structured recent backend logs for the diagnostics rail."""
    return templates.TemplateResponse(
        request,
        "partials/diagnostics_backend.html",
        {"recent_logs": get_recent_logs(limit=120)},
    )

ui_diagnostics_frontend(request) async

Render the client-side diagnostics tab shell.

Source code in src/phids/api/main.py

@app.get(
    "/ui/diagnostics/frontend", response_class=HTMLResponse, summary="Diagnostics frontend tab"
)
async def ui_diagnostics_frontend(request: Request) -> Any:
    """Render the client-side diagnostics tab shell."""
    return templates.TemplateResponse(request, "partials/diagnostics_frontend.html")

ui_diagnostics_model(request) async

Render live model counters, telemetry, and starvation watch.

Source code in src/phids/api/main.py

@app.get("/ui/diagnostics/model", response_class=HTMLResponse, summary="Diagnostics model tab")
async def ui_diagnostics_model(request: Request) -> Any:
    """Render live model counters, telemetry, and starvation watch."""
    return templates.TemplateResponse(
        request,
        "partials/diagnostics_model.html",
        {
            "draft": get_draft(),
            "live_summary": _build_live_summary(),
            "latest_metrics": _sim_loop.telemetry.get_latest_metrics()
            if _sim_loop is not None
            else None,
            "starving_swarms": _build_starving_swarms(),
        },
    )

ui_diet_matrix(request) async

Return the diet compatibility matrix partial.

Parameters:

Name	Type	Description	Default
request	Request	FastAPI/Starlette request object.	required

Returns:

Name	Type	Description
TemplateResponse	Any	Rendered partials/diet_matrix.html.

Source code in src/phids/api/main.py

@app.get("/ui/diet-matrix", response_class=HTMLResponse, summary="Diet matrix partial")
async def ui_diet_matrix(request: Request) -> Any:
    """Return the diet compatibility matrix partial.

    Args:
        request: FastAPI/Starlette request object.

    Returns:
        TemplateResponse: Rendered ``partials/diet_matrix.html``.
    """
    draft = get_draft()
    return templates.TemplateResponse(
        request,
        "partials/diet_matrix.html",
        {
            "flora_species": draft.flora_species,
            "predator_species": draft.predator_species,
            "diet_matrix": draft.diet_matrix,
        },
    )

ui_flora(request) async

Return the flora species table partial.

Parameters:

Name	Type	Description	Default
request	Request	FastAPI/Starlette request object.	required

Returns:

Name	Type	Description
TemplateResponse	Any	Rendered partials/flora_config.html.

Source code in src/phids/api/main.py

@app.get("/ui/flora", response_class=HTMLResponse, summary="Flora config partial")
async def ui_flora(request: Request) -> Any:
    """Return the flora species table partial.

    Args:
        request: FastAPI/Starlette request object.

    Returns:
        TemplateResponse: Rendered ``partials/flora_config.html``.
    """
    draft = get_draft()
    return templates.TemplateResponse(
        request,
        "partials/flora_config.html",
        {"flora_species": draft.flora_species},
    )

ui_placements(request) async

Return the placement editor partial with interactive grid canvas.

Parameters:

Name	Type	Description	Default
request	Request	FastAPI/Starlette request object.	required

Returns:

Name	Type	Description
TemplateResponse	Any	Rendered partials/placement_editor.html.

Source code in src/phids/api/main.py

@app.get("/ui/placements", response_class=HTMLResponse, summary="Placement editor partial")
async def ui_placements(request: Request) -> Any:
    """Return the placement editor partial with interactive grid canvas.

    Args:
        request: FastAPI/Starlette request object.

    Returns:
        TemplateResponse: Rendered ``partials/placement_editor.html``.
    """
    draft = get_draft()
    return templates.TemplateResponse(
        request,
        "partials/placement_editor.html",
        {
            "draft": draft,
            "flora_species": draft.flora_species,
            "predator_species": draft.predator_species,
            "initial_plants": draft.initial_plants,
            "initial_swarms": draft.initial_swarms,
        },
    )

ui_predators(request) async

Return the predator species table partial.

Parameters:

Name	Type	Description	Default
request	Request	FastAPI/Starlette request object.	required

Returns:

Name	Type	Description
TemplateResponse	Any	Rendered partials/predator_config.html.

Source code in src/phids/api/main.py

@app.get("/ui/predators", response_class=HTMLResponse, summary="Predator config partial")
async def ui_predators(request: Request) -> Any:
    """Return the predator species table partial.

    Args:
        request: FastAPI/Starlette request object.

    Returns:
        TemplateResponse: Rendered ``partials/predator_config.html``.
    """
    draft = get_draft()
    return templates.TemplateResponse(
        request,
        "partials/predator_config.html",
        {"predator_species": draft.predator_species},
    )

ui_status_badge() async

Return a small status <span> for HTMX outerHTML swap.

Returns:

Name	Type	Description
HTMLResponse	HTMLResponse	Styled <span id="sim-status"> fragment.

Source code in src/phids/api/main.py

@app.get("/api/ui/status-badge", summary="Simulation status badge HTML")
async def ui_status_badge() -> HTMLResponse:
    """Return a small status ``<span>`` for HTMX outerHTML swap.

    Returns:
        HTMLResponse: Styled ``<span id="sim-status">`` fragment.
    """
    return HTMLResponse(content=_render_status_badge_html())

ui_stream(websocket) async

Stream lightweight JSON grid snapshots for the browser canvas.

Each message contains plant_energy (2-D list), swarms (list of {x, y, population}), tick, and max_energy. Reconnects are handled client-side with an exponential back-off.

Source code in src/phids/api/main.py

@app.websocket("/ws/ui/stream")
async def ui_stream(websocket: WebSocket) -> None:
    """Stream lightweight JSON grid snapshots for the browser canvas.

    Each message contains ``plant_energy`` (2-D list), ``swarms``
    (list of ``{x, y, population}``), ``tick``, and ``max_energy``.
    Reconnects are handled client-side with an exponential back-off.
    """
    await websocket.accept()
    logger.debug("WebSocket connected: /ws/ui/stream")
    last_state_signature: tuple[int, int, bool, bool, bool] | None = None
    try:
        while True:
            loop = _sim_loop
            if loop is None:
                await asyncio.sleep(0.5)
                continue

            loop_id = id(loop)
            state_signature = (loop_id, loop.tick, loop.running, loop.paused, loop.terminated)
            # Send whenever the rendered state changes, including pause/resume toggles.
            if state_signature != last_state_signature:
                payload = _build_live_dashboard_payload(loop)
                await websocket.send_text(json.dumps(payload))
                last_state_signature = state_signature

            await asyncio.sleep(1.0 / max(1.0, loop.config.tick_rate_hz))

    except WebSocketDisconnect:
        logger.info("WebSocket client disconnected from /ws/ui/stream")
    finally:
        await websocket.close()

ui_substances(request) async

Return the substance definitions table partial.

Parameters:

Name	Type	Description	Default
request	Request	FastAPI/Starlette request object.	required

Returns:

Name	Type	Description
TemplateResponse	Any	Rendered partials/substance_config.html.

Source code in src/phids/api/main.py

@app.get("/ui/substances", response_class=HTMLResponse, summary="Substance config partial")
async def ui_substances(request: Request) -> Any:
    """Return the substance definitions table partial.

    Args:
        request: FastAPI/Starlette request object.

    Returns:
        TemplateResponse: Rendered ``partials/substance_config.html``.
    """
    draft = get_draft()
    return templates.TemplateResponse(
        request,
        "partials/substance_config.html",
        {"substances": draft.substance_definitions},
    )

ui_tick() async

Return the current tick as plain text for HTMX innerHTML swap.

Returns:

Name	Type	Description
Response	Response	Tick count as plain-text body.

Source code in src/phids/api/main.py

@app.get("/api/ui/tick", summary="Current simulation tick (plain text)")
async def ui_tick() -> Response:
    """Return the current tick as plain text for HTMX innerHTML swap.

    Returns:
        Response: Tick count as plain-text body.
    """
    tick = _sim_loop.tick if _sim_loop is not None else 0
    return Response(content=str(tick), media_type="text/plain")

ui_trigger_matrix_legacy(request) async

Redirect-compatible alias for the trigger rules page.

Source code in src/phids/api/main.py

@app.get("/ui/trigger-matrix", response_class=HTMLResponse, summary="Trigger rules (legacy URL)")
async def ui_trigger_matrix_legacy(request: Request) -> Any:
    """Redirect-compatible alias for the trigger rules page."""
    return await ui_trigger_rules(request)

ui_trigger_rules(request) async

Return the trigger rules editor partial.

Parameters:

Name	Type	Description	Default
request	Request	FastAPI/Starlette request object.	required

Returns:

Name	Type	Description
TemplateResponse	Any	Rendered partials/trigger_rules.html.

Source code in src/phids/api/main.py

@app.get("/ui/trigger-rules", response_class=HTMLResponse, summary="Trigger rules partial")
async def ui_trigger_rules(request: Request) -> Any:
    """Return the trigger rules editor partial.

    Args:
        request: FastAPI/Starlette request object.

    Returns:
        TemplateResponse: Rendered ``partials/trigger_rules.html``.
    """
    draft = get_draft()
    return templates.TemplateResponse(
        request, "partials/trigger_rules.html", _trigger_rules_template_context(draft)
    )

update_wind(payload) async

Dynamically update the simulation wind vector.

Parameters:

Name	Type	Description	Default
payload	WindUpdatePayload	:class:~phids.api.schemas.WindUpdatePayload.	required

Returns:

Name	Type	Description
dict	dict[str, Any]	Confirmation and the applied wind vector.

Source code in src/phids/api/main.py

@app.put("/api/simulation/wind", summary="Update wind vector")
async def update_wind(payload: WindUpdatePayload) -> dict[str, Any]:
    """Dynamically update the simulation wind vector.

    Args:
        payload: :class:`~phids.api.schemas.WindUpdatePayload`.

    Returns:
        dict: Confirmation and the applied wind vector.
    """
    loop = _get_loop()
    loop.update_wind(payload.wind_x, payload.wind_y)
    logger.info("Wind updated via API to (vx=%.3f, vy=%.3f)", payload.wind_x, payload.wind_y)
    return {"message": "Wind updated.", "wind_x": payload.wind_x, "wind_y": payload.wind_y}

Server-side mutable draft state for the HTMX scenario-builder UI.

The :class:DraftState accumulates all configuration choices made by the operator through the web interface before they are committed to the simulation engine via POST /api/scenario/load-draft.

The module exposes a single global :class:DraftState instance accessed through :func:get_draft and :func:reset_draft. All route handlers mutate this instance directly; no concurrency-safe locking is applied because the server is expected to serve a single-operator workbench.

DraftState dataclass

Mutable server-side configuration accumulator for the builder UI.

Attributes:

Name	Type	Description
scenario_name	str	Human-readable label used in the UI header.
grid_width	int	Biotope width in cells (1–80).
grid_height	int	Biotope height in cells (1–80).
max_ticks	int	Simulation tick budget.
tick_rate_hz	float	WebSocket streaming rate in ticks per second.
wind_x	float	Initial uniform wind x-component.
wind_y	float	Initial uniform wind y-component.
num_signals	int	Number of airborne signal layers.
num_toxins	int	Number of toxin layers.
mycorrhizal_inter_species	bool	Allow root connections across species.
mycorrhizal_connection_cost	float	Energy to establish a root link.
mycorrhizal_growth_interval_ticks	int	Ticks between root-growth attempts.
mycorrhizal_signal_velocity	int	Signal hops per tick through roots.
flora_species	list[object]	Flora species parameter list (species_id == index).
predator_species	list[object]	Predator species parameter list (species_id == index).
diet_matrix	list[list[bool]]	Boolean matrix [pred_idx][flora_idx] for edibility.
trigger_rules	list[TriggerRule]	List of explicit chemical-defense trigger rules. Multiple rules per (flora, predator) pair are allowed.
substance_definitions	list[SubstanceDefinition]	Named substance registry indexed by substance_id.
initial_plants	list[PlacedPlant]	Plants placed on the grid before simulation start.
initial_swarms	list[PlacedSwarm]	Swarms placed on the grid before simulation start.

Source code in src/phids/api/ui_state.py

@dataclasses.dataclass
class DraftState:
    """Mutable server-side configuration accumulator for the builder UI.

    Attributes:
        scenario_name: Human-readable label used in the UI header.
        grid_width: Biotope width in cells (1–80).
        grid_height: Biotope height in cells (1–80).
        max_ticks: Simulation tick budget.
        tick_rate_hz: WebSocket streaming rate in ticks per second.
        wind_x: Initial uniform wind x-component.
        wind_y: Initial uniform wind y-component.
        num_signals: Number of airborne signal layers.
        num_toxins: Number of toxin layers.
        mycorrhizal_inter_species: Allow root connections across species.
        mycorrhizal_connection_cost: Energy to establish a root link.
        mycorrhizal_growth_interval_ticks: Ticks between root-growth attempts.
        mycorrhizal_signal_velocity: Signal hops per tick through roots.
        flora_species: Flora species parameter list (species_id == index).
        predator_species: Predator species parameter list (species_id == index).
        diet_matrix: Boolean matrix ``[pred_idx][flora_idx]`` for edibility.
        trigger_rules: List of explicit chemical-defense trigger rules.
            Multiple rules per (flora, predator) pair are allowed.
        substance_definitions: Named substance registry indexed by substance_id.
        initial_plants: Plants placed on the grid before simulation start.
        initial_swarms: Swarms placed on the grid before simulation start.
    """

    scenario_name: str = _DEFAULT_SCENARIO_NAME
    grid_width: int = 40
    grid_height: int = 40
    max_ticks: int = 1000
    tick_rate_hz: float = 10.0
    wind_x: float = 0.0
    wind_y: float = 0.0
    num_signals: int = 4
    num_toxins: int = 4
    mycorrhizal_inter_species: bool = False
    mycorrhizal_connection_cost: float = 1.0
    mycorrhizal_growth_interval_ticks: int = 8
    mycorrhizal_signal_velocity: int = 1
    flora_species: list[object] = dataclasses.field(default_factory=list)
    predator_species: list[object] = dataclasses.field(default_factory=list)
    diet_matrix: list[list[bool]] = dataclasses.field(default_factory=list)
    trigger_rules: list[TriggerRule] = dataclasses.field(default_factory=list)
    substance_definitions: list[SubstanceDefinition] = dataclasses.field(default_factory=list)
    initial_plants: list[PlacedPlant] = dataclasses.field(default_factory=list)
    initial_swarms: list[PlacedSwarm] = dataclasses.field(default_factory=list)

    # ------------------------------------------------------------------
    # Matrix resize helpers (diet_matrix only now)
    # ------------------------------------------------------------------

    def _resize_diet_matrix(self) -> None:
        """Resize diet matrix to match species list lengths."""
        n_pred = len(self.predator_species)
        n_flora = len(self.flora_species)

        while len(self.diet_matrix) < n_pred:
            self.diet_matrix.append([False] * n_flora)
        self.diet_matrix = self.diet_matrix[:n_pred]
        for row in self.diet_matrix:
            while len(row) < n_flora:
                row.append(False)
            del row[n_flora:]

    def rebuild_species_ids(self) -> None:
        """Re-assign sequential ``species_id`` values after list mutation."""
        from phids.api.schemas import FloraSpeciesParams, PredatorSpeciesParams

        self.flora_species = [
            fp.model_copy(update={"species_id": i})  # type: ignore[union-attr]
            for i, fp in enumerate(self.flora_species)
            if isinstance(fp, FloraSpeciesParams)
        ]
        self.predator_species = [
            pp.model_copy(update={"species_id": i})  # type: ignore[union-attr]
            for i, pp in enumerate(self.predator_species)
            if isinstance(pp, PredatorSpeciesParams)
        ]

    # ------------------------------------------------------------------
    # Species mutation helpers
    # ------------------------------------------------------------------

    def add_flora(self, params: object) -> None:
        """Append a flora species and extend the diet matrix.

        Args:
            params: :class:`~phids.api.schemas.FloraSpeciesParams`.
        """
        self.flora_species.append(params)
        self.rebuild_species_ids()
        self._resize_diet_matrix()
        logger.debug(
            "Draft flora added (species_id=%s, total_flora=%d)",
            getattr(params, "species_id", "?"),
            len(self.flora_species),
        )

    def remove_flora(self, species_id: int) -> None:
        """Remove a flora species by id and clean up dependent data.

        Args:
            species_id: The ``species_id`` of the species to remove.

        Raises:
            ValueError: If no species with the given id exists.
        """
        from phids.api.schemas import FloraSpeciesParams

        idx = next(
            (
                i
                for i, fp in enumerate(self.flora_species)
                if isinstance(fp, FloraSpeciesParams) and fp.species_id == species_id
            ),
            None,
        )
        if idx is None:
            raise ValueError(f"Flora species_id {species_id} not found.")

        del self.flora_species[idx]
        # Remove diet_matrix column for this flora species
        for row in self.diet_matrix:
            if idx < len(row):
                del row[idx]

        # Remove trigger rules referencing this flora species; compact higher IDs
        new_rules: list[TriggerRule] = []
        for rule in self.trigger_rules:
            if rule.flora_species_id == species_id:
                continue  # orphaned rule
            new_rule = dataclasses.replace(rule)
            if new_rule.flora_species_id > species_id:
                new_rule.flora_species_id -= 1  # compact after removal
            new_rules.append(new_rule)
        self.trigger_rules = new_rules

        # Remove placements
        self.initial_plants = [p for p in self.initial_plants if p.species_id != species_id]
        self.rebuild_species_ids()
        self._resize_diet_matrix()
        logger.debug(
            "Draft flora removed (species_id=%d, total_flora=%d, remaining_trigger_rules=%d)",
            species_id,
            len(self.flora_species),
            len(self.trigger_rules),
        )

    def add_predator(self, params: object) -> None:
        """Append a predator species and extend the diet matrix.

        Args:
            params: :class:`~phids.api.schemas.PredatorSpeciesParams`.
        """
        self.predator_species.append(params)
        self.rebuild_species_ids()
        self._resize_diet_matrix()
        logger.debug(
            "Draft predator added (species_id=%s, total_predators=%d)",
            getattr(params, "species_id", "?"),
            len(self.predator_species),
        )

    def remove_predator(self, species_id: int) -> None:
        """Remove a predator species by id and clean up dependent data.

        Args:
            species_id: The ``species_id`` of the species to remove.

        Raises:
            ValueError: If no species with the given id exists.
        """
        from phids.api.schemas import PredatorSpeciesParams

        idx = next(
            (
                i
                for i, pp in enumerate(self.predator_species)
                if isinstance(pp, PredatorSpeciesParams) and pp.species_id == species_id
            ),
            None,
        )
        if idx is None:
            raise ValueError(f"Predator species_id {species_id} not found.")

        del self.predator_species[idx]
        # Remove diet_matrix row for this predator
        if idx < len(self.diet_matrix):
            del self.diet_matrix[idx]

        # Remove trigger rules referencing this predator; compact higher IDs
        new_rules: list[TriggerRule] = []
        for rule in self.trigger_rules:
            if rule.predator_species_id == species_id:
                continue  # orphaned rule
            new_rule = dataclasses.replace(rule)
            if new_rule.predator_species_id > species_id:
                new_rule.predator_species_id -= 1
            new_rule.activation_condition = _remap_condition_references(
                deepcopy(new_rule.activation_condition),
                removed_predator_id=species_id,
            )
            new_rules.append(new_rule)
        self.trigger_rules = new_rules

        # Remove placements
        self.initial_swarms = [s for s in self.initial_swarms if s.species_id != species_id]
        self.rebuild_species_ids()
        self._resize_diet_matrix()
        logger.debug(
            "Draft predator removed (species_id=%d, total_predators=%d, remaining_trigger_rules=%d)",
            species_id,
            len(self.predator_species),
            len(self.trigger_rules),
        )

    # ------------------------------------------------------------------
    # Trigger rule helpers
    # ------------------------------------------------------------------

    def add_trigger_rule(
        self,
        flora_species_id: int,
        predator_species_id: int,
        substance_id: int,
        min_predator_population: int = 5,
        activation_condition: dict[str, object] | None = None,
        required_signal_ids: list[int] | None = None,
    ) -> None:
        """Append a new trigger rule.

        Args:
            flora_species_id: Flora species index.
            predator_species_id: Predator species index.
            substance_id: Substance to synthesise.
            min_predator_population: Minimum swarm size threshold.
            activation_condition: Optional nested predicate tree.
            required_signal_ids: Deprecated compatibility shorthand for
                signal-only AND gates.
        """
        self.trigger_rules.append(
            TriggerRule(
                flora_species_id=flora_species_id,
                predator_species_id=predator_species_id,
                substance_id=substance_id,
                min_predator_population=min_predator_population,
                activation_condition=deepcopy(
                    activation_condition
                    if activation_condition is not None
                    else _legacy_signal_ids_to_activation_condition(required_signal_ids)
                ),
            )
        )
        logger.debug(
            "Draft trigger rule added (flora_species_id=%d, predator_species_id=%d, substance_id=%d, total_rules=%d)",
            flora_species_id,
            predator_species_id,
            substance_id,
            len(self.trigger_rules),
        )

    def remove_trigger_rule(self, index: int) -> None:
        """Remove a trigger rule by list index.

        Args:
            index: Position in the trigger_rules list.

        Raises:
            IndexError: If index is out of range.
        """
        removed = self.trigger_rules[index]
        del self.trigger_rules[index]
        logger.debug(
            "Draft trigger rule removed (index=%d, flora_species_id=%d, predator_species_id=%d, substance_id=%d, total_rules=%d)",
            index,
            removed.flora_species_id,
            removed.predator_species_id,
            removed.substance_id,
            len(self.trigger_rules),
        )

    def update_trigger_rule(
        self,
        index: int,
        *,
        flora_species_id: int | None = None,
        predator_species_id: int | None = None,
        substance_id: int | None = None,
        min_predator_population: int | None = None,
        activation_condition: dict[str, object] | None = None,
        required_signal_ids: list[int] | None = None,
    ) -> None:
        """Update fields of a trigger rule in-place.

        Args:
            index: Position in the trigger_rules list.
            flora_species_id: New flora species id (optional).
            predator_species_id: New predator species id (optional).
            substance_id: New substance id (optional).
            min_predator_population: New minimum population (optional).
            activation_condition: New condition tree (optional).
            required_signal_ids: Deprecated compatibility shorthand for
                signal-only AND gates.

        Raises:
            IndexError: If index is out of range.
        """
        rule = self.trigger_rules[index]
        if flora_species_id is not None:
            rule.flora_species_id = flora_species_id
        if predator_species_id is not None:
            rule.predator_species_id = predator_species_id
        if substance_id is not None:
            rule.substance_id = substance_id
        if min_predator_population is not None:
            rule.min_predator_population = min_predator_population
        if activation_condition is not None:
            rule.activation_condition = deepcopy(activation_condition)
        elif required_signal_ids is not None:
            rule.activation_condition = _legacy_signal_ids_to_activation_condition(
                required_signal_ids
            )
        logger.debug(
            "Draft trigger rule updated (index=%d, flora_species_id=%d, predator_species_id=%d, substance_id=%d)",
            index,
            rule.flora_species_id,
            rule.predator_species_id,
            rule.substance_id,
        )

    def set_trigger_rule_activation_condition(
        self,
        index: int,
        condition: dict[str, object] | None,
    ) -> None:
        """Replace the full activation-condition tree for one trigger rule."""
        self.trigger_rules[index].activation_condition = deepcopy(condition)

    def replace_trigger_rule_condition_node(
        self,
        index: int,
        path: str,
        condition: dict[str, object],
    ) -> None:
        """Replace the node at ``path`` inside one trigger rule's condition tree."""
        rule = self.trigger_rules[index]
        if not path:
            rule.activation_condition = deepcopy(condition)
            return
        if rule.activation_condition is None:
            raise IndexError("Trigger rule has no activation condition to replace.")
        root = deepcopy(rule.activation_condition)
        path_indices = _parse_condition_path(path)
        parent = _condition_node_at_path(root, path_indices[:-1])
        if parent.get("kind") not in {"all_of", "any_of"}:
            raise IndexError("Condition parent is not a group node.")
        children = parent.get("conditions")
        if not isinstance(children, list):
            raise IndexError("Condition parent has no child list.")
        child_index = path_indices[-1]
        if child_index < 0 or child_index >= len(children):
            raise IndexError("Condition node index is out of range.")
        children[child_index] = deepcopy(condition)
        rule.activation_condition = root

    def append_trigger_rule_condition_child(
        self,
        index: int,
        parent_path: str,
        condition: dict[str, object],
    ) -> None:
        """Append a child node to a group within one trigger rule's condition tree."""
        rule = self.trigger_rules[index]
        if rule.activation_condition is None:
            raise IndexError("Trigger rule has no activation condition to append to.")
        root = deepcopy(rule.activation_condition)
        parent = _condition_node_at_path(root, _parse_condition_path(parent_path))
        if parent.get("kind") not in {"all_of", "any_of"}:
            raise IndexError("Condition parent is not a group node.")
        children = parent.setdefault("conditions", [])
        if not isinstance(children, list):
            raise IndexError("Condition parent has an invalid child list.")
        children.append(deepcopy(condition))
        rule.activation_condition = root

    def delete_trigger_rule_condition_node(self, index: int, path: str) -> None:
        """Delete the node at ``path`` from one trigger rule's condition tree."""
        rule = self.trigger_rules[index]
        if rule.activation_condition is None:
            return
        if not path:
            rule.activation_condition = None
            return
        root = deepcopy(rule.activation_condition)
        path_indices = _parse_condition_path(path)
        parent = _condition_node_at_path(root, path_indices[:-1])
        if parent.get("kind") not in {"all_of", "any_of"}:
            raise IndexError("Condition parent is not a group node.")
        children = parent.get("conditions")
        if not isinstance(children, list):
            raise IndexError("Condition parent has no child list.")
        child_index = path_indices[-1]
        if child_index < 0 or child_index >= len(children):
            raise IndexError("Condition node index is out of range.")
        del children[child_index]
        rule.activation_condition = _prune_empty_condition_groups(root)

    def update_trigger_rule_condition_node(
        self,
        index: int,
        path: str,
        **fields: object,
    ) -> None:
        """Patch selected fields on the condition node at ``path``."""
        rule = self.trigger_rules[index]
        if rule.activation_condition is None:
            raise IndexError("Trigger rule has no activation condition to update.")
        root = deepcopy(rule.activation_condition)
        node = _condition_node_at_path(root, _parse_condition_path(path))
        node.update(fields)
        rule.activation_condition = root

    # ------------------------------------------------------------------
    # Placement helpers
    # ------------------------------------------------------------------

    def add_plant_placement(self, species_id: int, x: int, y: int, energy: float) -> None:
        """Add a plant placement to the draft."""
        self.initial_plants.append(PlacedPlant(species_id=species_id, x=x, y=y, energy=energy))
        logger.debug(
            "Draft plant placement added (species_id=%d, x=%d, y=%d, total_plants=%d)",
            species_id,
            x,
            y,
            len(self.initial_plants),
        )

    def add_swarm_placement(
        self, species_id: int, x: int, y: int, population: int, energy: float
    ) -> None:
        """Add a swarm placement to the draft."""
        self.initial_swarms.append(
            PlacedSwarm(
                species_id=species_id,
                x=x,
                y=y,
                population=population,
                energy=energy,
            )
        )
        logger.debug(
            "Draft swarm placement added (species_id=%d, x=%d, y=%d, population=%d, total_swarms=%d)",
            species_id,
            x,
            y,
            population,
            len(self.initial_swarms),
        )

    def remove_plant_placement(self, index: int) -> None:
        """Remove a plant placement by list index."""
        removed = self.initial_plants[index]
        del self.initial_plants[index]
        logger.debug(
            "Draft plant placement removed (index=%d, species_id=%d, x=%d, y=%d, total_plants=%d)",
            index,
            removed.species_id,
            removed.x,
            removed.y,
            len(self.initial_plants),
        )

    def remove_swarm_placement(self, index: int) -> None:
        """Remove a swarm placement by list index."""
        removed = self.initial_swarms[index]
        del self.initial_swarms[index]
        logger.debug(
            "Draft swarm placement removed (index=%d, species_id=%d, x=%d, y=%d, total_swarms=%d)",
            index,
            removed.species_id,
            removed.x,
            removed.y,
            len(self.initial_swarms),
        )

    def clear_placements(self) -> None:
        """Remove all plant and swarm placements."""
        cleared_plants = len(self.initial_plants)
        cleared_swarms = len(self.initial_swarms)
        self.initial_plants.clear()
        self.initial_swarms.clear()
        logger.debug(
            "Draft placements cleared (plants=%d, swarms=%d)",
            cleared_plants,
            cleared_swarms,
        )

    # ------------------------------------------------------------------
    # Config export
    # ------------------------------------------------------------------

    def build_sim_config(self) -> object:
        """Assemble a :class:`~phids.api.schemas.SimulationConfig`.

        Returns:
            SimulationConfig: Validated simulation configuration.

        Raises:
            ValueError: If no flora or predator species defined.
        """
        from phids.api.schemas import (
            DietCompatibilityMatrix,
            FloraSpeciesParams,
            InitialPlantPlacement,
            InitialSwarmPlacement,
            SimulationConfig,
            TriggerConditionSchema,
        )

        if not self.flora_species or not self.predator_species:
            logger.warning(
                "Draft build rejected because required species are missing (flora=%d, predators=%d)",
                len(self.flora_species),
                len(self.predator_species),
            )
            raise ValueError("At least one flora and one predator species are required.")

        subs_by_id: dict[int, SubstanceDefinition] = {
            sd.substance_id: sd for sd in self.substance_definitions
        }

        # Group trigger rules by flora_species_id
        triggers_by_flora: dict[int, list[TriggerConditionSchema]] = {}
        for rule in self.trigger_rules:
            sd = subs_by_id.get(rule.substance_id)
            if sd is None:
                logger.warning(
                    "Skipping trigger rule with missing substance definition (flora_species_id=%d, predator_species_id=%d, substance_id=%d)",
                    rule.flora_species_id,
                    rule.predator_species_id,
                    rule.substance_id,
                )
                continue
            triggers_by_flora.setdefault(rule.flora_species_id, []).append(
                TriggerConditionSchema(
                    predator_species_id=rule.predator_species_id,
                    min_predator_population=rule.min_predator_population,
                    substance_id=rule.substance_id,
                    synthesis_duration=sd.synthesis_duration,
                    is_toxin=sd.is_toxin,
                    lethal=sd.lethal,
                    lethality_rate=sd.lethality_rate,
                    repellent=sd.repellent,
                    repellent_walk_ticks=sd.repellent_walk_ticks,
                    aftereffect_ticks=sd.aftereffect_ticks,
                    activation_condition=deepcopy(rule.activation_condition),
                    energy_cost_per_tick=sd.energy_cost_per_tick,
                )
            )

        flora_with_triggers: list[FloraSpeciesParams] = []
        for fp in self.flora_species:
            if not isinstance(fp, FloraSpeciesParams):
                continue
            triggers = triggers_by_flora.get(fp.species_id, [])
            flora_with_triggers.append(fp.model_copy(update={"triggers": triggers}))

        n_pred = len(self.predator_species)
        n_flora = len(flora_with_triggers)
        diet_rows = [
            (self.diet_matrix[pi][:n_flora] if pi < len(self.diet_matrix) else [False] * n_flora)
            for pi in range(n_pred)
        ]

        plant_placements = [
            InitialPlantPlacement(species_id=p.species_id, x=p.x, y=p.y, energy=p.energy)
            for p in self.initial_plants
        ]
        swarm_placements = [
            InitialSwarmPlacement(
                species_id=s.species_id,
                x=s.x,
                y=s.y,
                population=s.population,
                energy=s.energy,
            )
            for s in self.initial_swarms
        ]

        config = SimulationConfig(
            grid_width=self.grid_width,
            grid_height=self.grid_height,
            max_ticks=self.max_ticks,
            tick_rate_hz=self.tick_rate_hz,
            num_signals=self.num_signals,
            num_toxins=self.num_toxins,
            wind_x=self.wind_x,
            wind_y=self.wind_y,
            flora_species=flora_with_triggers,
            predator_species=list(self.predator_species),  # type: ignore[arg-type]
            diet_matrix=DietCompatibilityMatrix(rows=diet_rows),
            initial_plants=plant_placements,
            initial_swarms=swarm_placements,
            mycorrhizal_inter_species=self.mycorrhizal_inter_species,
            mycorrhizal_connection_cost=self.mycorrhizal_connection_cost,
            mycorrhizal_growth_interval_ticks=self.mycorrhizal_growth_interval_ticks,
            mycorrhizal_signal_velocity=self.mycorrhizal_signal_velocity,
        )
        logger.info(
            "Draft converted to SimulationConfig (grid=%dx%d, flora=%d, predators=%d, trigger_rules=%d, plants=%d, swarms=%d)",
            self.grid_width,
            self.grid_height,
            len(flora_with_triggers),
            len(self.predator_species),
            len(self.trigger_rules),
            len(self.initial_plants),
            len(self.initial_swarms),
        )
        return config

    @classmethod
    def default(cls) -> DraftState:
        """Create the built-in default draft state."""
        from phids.api.schemas import FloraSpeciesParams, PredatorSpeciesParams

        state = cls(
            flora_species=[
                FloraSpeciesParams(
                    species_id=0,
                    name="Grass",
                    base_energy=10.0,
                    max_energy=100.0,
                    growth_rate=5.0,
                    survival_threshold=1.0,
                    reproduction_interval=10,
                    seed_min_dist=1.0,
                    seed_max_dist=3.0,
                    seed_energy_cost=5.0,
                    triggers=[],
                )
            ],
            predator_species=[
                PredatorSpeciesParams(
                    species_id=0,
                    name="Herbivore",
                    energy_min=5.0,
                    velocity=2,
                    consumption_rate=10.0,
                )
            ],
            diet_matrix=[[True]],
            trigger_rules=[],
            substance_definitions=[],
            initial_plants=[],
            initial_swarms=[],
        )
        return state

add_flora(params)

Append a flora species and extend the diet matrix.

Parameters:

Name	Type	Description	Default
params	object	:class:~phids.api.schemas.FloraSpeciesParams.	required

Source code in src/phids/api/ui_state.py

def add_flora(self, params: object) -> None:
    """Append a flora species and extend the diet matrix.

    Args:
        params: :class:`~phids.api.schemas.FloraSpeciesParams`.
    """
    self.flora_species.append(params)
    self.rebuild_species_ids()
    self._resize_diet_matrix()
    logger.debug(
        "Draft flora added (species_id=%s, total_flora=%d)",
        getattr(params, "species_id", "?"),
        len(self.flora_species),
    )

add_plant_placement(species_id, x, y, energy)

Add a plant placement to the draft.

Source code in src/phids/api/ui_state.py

def add_plant_placement(self, species_id: int, x: int, y: int, energy: float) -> None:
    """Add a plant placement to the draft."""
    self.initial_plants.append(PlacedPlant(species_id=species_id, x=x, y=y, energy=energy))
    logger.debug(
        "Draft plant placement added (species_id=%d, x=%d, y=%d, total_plants=%d)",
        species_id,
        x,
        y,
        len(self.initial_plants),
    )

add_predator(params)

Append a predator species and extend the diet matrix.

Parameters:

Name	Type	Description	Default
params	object	:class:~phids.api.schemas.PredatorSpeciesParams.	required

Source code in src/phids/api/ui_state.py

def add_predator(self, params: object) -> None:
    """Append a predator species and extend the diet matrix.

    Args:
        params: :class:`~phids.api.schemas.PredatorSpeciesParams`.
    """
    self.predator_species.append(params)
    self.rebuild_species_ids()
    self._resize_diet_matrix()
    logger.debug(
        "Draft predator added (species_id=%s, total_predators=%d)",
        getattr(params, "species_id", "?"),
        len(self.predator_species),
    )

add_swarm_placement(species_id, x, y, population, energy)

Add a swarm placement to the draft.

Source code in src/phids/api/ui_state.py

def add_swarm_placement(
    self, species_id: int, x: int, y: int, population: int, energy: float
) -> None:
    """Add a swarm placement to the draft."""
    self.initial_swarms.append(
        PlacedSwarm(
            species_id=species_id,
            x=x,
            y=y,
            population=population,
            energy=energy,
        )
    )
    logger.debug(
        "Draft swarm placement added (species_id=%d, x=%d, y=%d, population=%d, total_swarms=%d)",
        species_id,
        x,
        y,
        population,
        len(self.initial_swarms),
    )

add_trigger_rule(flora_species_id, predator_species_id, substance_id, min_predator_population=5, activation_condition=None, required_signal_ids=None)

Append a new trigger rule.

Parameters:

Name	Type	Description	Default
flora_species_id	int	Flora species index.	required
predator_species_id	int	Predator species index.	required
substance_id	int	Substance to synthesise.	required
min_predator_population	int	Minimum swarm size threshold.	5
activation_condition	dict[str, object] | None	Optional nested predicate tree.	None
required_signal_ids	list[int] | None	Deprecated compatibility shorthand for signal-only AND gates.	None

Source code in src/phids/api/ui_state.py

def add_trigger_rule(
    self,
    flora_species_id: int,
    predator_species_id: int,
    substance_id: int,
    min_predator_population: int = 5,
    activation_condition: dict[str, object] | None = None,
    required_signal_ids: list[int] | None = None,
) -> None:
    """Append a new trigger rule.

    Args:
        flora_species_id: Flora species index.
        predator_species_id: Predator species index.
        substance_id: Substance to synthesise.
        min_predator_population: Minimum swarm size threshold.
        activation_condition: Optional nested predicate tree.
        required_signal_ids: Deprecated compatibility shorthand for
            signal-only AND gates.
    """
    self.trigger_rules.append(
        TriggerRule(
            flora_species_id=flora_species_id,
            predator_species_id=predator_species_id,
            substance_id=substance_id,
            min_predator_population=min_predator_population,
            activation_condition=deepcopy(
                activation_condition
                if activation_condition is not None
                else _legacy_signal_ids_to_activation_condition(required_signal_ids)
            ),
        )
    )
    logger.debug(
        "Draft trigger rule added (flora_species_id=%d, predator_species_id=%d, substance_id=%d, total_rules=%d)",
        flora_species_id,
        predator_species_id,
        substance_id,
        len(self.trigger_rules),
    )

append_trigger_rule_condition_child(index, parent_path, condition)

Append a child node to a group within one trigger rule's condition tree.

Source code in src/phids/api/ui_state.py

def append_trigger_rule_condition_child(
    self,
    index: int,
    parent_path: str,
    condition: dict[str, object],
) -> None:
    """Append a child node to a group within one trigger rule's condition tree."""
    rule = self.trigger_rules[index]
    if rule.activation_condition is None:
        raise IndexError("Trigger rule has no activation condition to append to.")
    root = deepcopy(rule.activation_condition)
    parent = _condition_node_at_path(root, _parse_condition_path(parent_path))
    if parent.get("kind") not in {"all_of", "any_of"}:
        raise IndexError("Condition parent is not a group node.")
    children = parent.setdefault("conditions", [])
    if not isinstance(children, list):
        raise IndexError("Condition parent has an invalid child list.")
    children.append(deepcopy(condition))
    rule.activation_condition = root

build_sim_config()

Assemble a :class:~phids.api.schemas.SimulationConfig.

Returns:

Name	Type	Description
SimulationConfig	object	Validated simulation configuration.

Raises:

Type	Description
ValueError	If no flora or predator species defined.

Source code in src/phids/api/ui_state.py

def build_sim_config(self) -> object:
    """Assemble a :class:`~phids.api.schemas.SimulationConfig`.

    Returns:
        SimulationConfig: Validated simulation configuration.

    Raises:
        ValueError: If no flora or predator species defined.
    """
    from phids.api.schemas import (
        DietCompatibilityMatrix,
        FloraSpeciesParams,
        InitialPlantPlacement,
        InitialSwarmPlacement,
        SimulationConfig,
        TriggerConditionSchema,
    )

    if not self.flora_species or not self.predator_species:
        logger.warning(
            "Draft build rejected because required species are missing (flora=%d, predators=%d)",
            len(self.flora_species),
            len(self.predator_species),
        )
        raise ValueError("At least one flora and one predator species are required.")

    subs_by_id: dict[int, SubstanceDefinition] = {
        sd.substance_id: sd for sd in self.substance_definitions
    }

    # Group trigger rules by flora_species_id
    triggers_by_flora: dict[int, list[TriggerConditionSchema]] = {}
    for rule in self.trigger_rules:
        sd = subs_by_id.get(rule.substance_id)
        if sd is None:
            logger.warning(
                "Skipping trigger rule with missing substance definition (flora_species_id=%d, predator_species_id=%d, substance_id=%d)",
                rule.flora_species_id,
                rule.predator_species_id,
                rule.substance_id,
            )
            continue
        triggers_by_flora.setdefault(rule.flora_species_id, []).append(
            TriggerConditionSchema(
                predator_species_id=rule.predator_species_id,
                min_predator_population=rule.min_predator_population,
                substance_id=rule.substance_id,
                synthesis_duration=sd.synthesis_duration,
                is_toxin=sd.is_toxin,
                lethal=sd.lethal,
                lethality_rate=sd.lethality_rate,
                repellent=sd.repellent,
                repellent_walk_ticks=sd.repellent_walk_ticks,
                aftereffect_ticks=sd.aftereffect_ticks,
                activation_condition=deepcopy(rule.activation_condition),
                energy_cost_per_tick=sd.energy_cost_per_tick,
            )
        )

    flora_with_triggers: list[FloraSpeciesParams] = []
    for fp in self.flora_species:
        if not isinstance(fp, FloraSpeciesParams):
            continue
        triggers = triggers_by_flora.get(fp.species_id, [])
        flora_with_triggers.append(fp.model_copy(update={"triggers": triggers}))

    n_pred = len(self.predator_species)
    n_flora = len(flora_with_triggers)
    diet_rows = [
        (self.diet_matrix[pi][:n_flora] if pi < len(self.diet_matrix) else [False] * n_flora)
        for pi in range(n_pred)
    ]

    plant_placements = [
        InitialPlantPlacement(species_id=p.species_id, x=p.x, y=p.y, energy=p.energy)
        for p in self.initial_plants
    ]
    swarm_placements = [
        InitialSwarmPlacement(
            species_id=s.species_id,
            x=s.x,
            y=s.y,
            population=s.population,
            energy=s.energy,
        )
        for s in self.initial_swarms
    ]

    config = SimulationConfig(
        grid_width=self.grid_width,
        grid_height=self.grid_height,
        max_ticks=self.max_ticks,
        tick_rate_hz=self.tick_rate_hz,
        num_signals=self.num_signals,
        num_toxins=self.num_toxins,
        wind_x=self.wind_x,
        wind_y=self.wind_y,
        flora_species=flora_with_triggers,
        predator_species=list(self.predator_species),  # type: ignore[arg-type]
        diet_matrix=DietCompatibilityMatrix(rows=diet_rows),
        initial_plants=plant_placements,
        initial_swarms=swarm_placements,
        mycorrhizal_inter_species=self.mycorrhizal_inter_species,
        mycorrhizal_connection_cost=self.mycorrhizal_connection_cost,
        mycorrhizal_growth_interval_ticks=self.mycorrhizal_growth_interval_ticks,
        mycorrhizal_signal_velocity=self.mycorrhizal_signal_velocity,
    )
    logger.info(
        "Draft converted to SimulationConfig (grid=%dx%d, flora=%d, predators=%d, trigger_rules=%d, plants=%d, swarms=%d)",
        self.grid_width,
        self.grid_height,
        len(flora_with_triggers),
        len(self.predator_species),
        len(self.trigger_rules),
        len(self.initial_plants),
        len(self.initial_swarms),
    )
    return config

clear_placements()

Remove all plant and swarm placements.

Source code in src/phids/api/ui_state.py

def clear_placements(self) -> None:
    """Remove all plant and swarm placements."""
    cleared_plants = len(self.initial_plants)
    cleared_swarms = len(self.initial_swarms)
    self.initial_plants.clear()
    self.initial_swarms.clear()
    logger.debug(
        "Draft placements cleared (plants=%d, swarms=%d)",
        cleared_plants,
        cleared_swarms,
    )

default() classmethod

Create the built-in default draft state.

Source code in src/phids/api/ui_state.py

@classmethod
def default(cls) -> DraftState:
    """Create the built-in default draft state."""
    from phids.api.schemas import FloraSpeciesParams, PredatorSpeciesParams

    state = cls(
        flora_species=[
            FloraSpeciesParams(
                species_id=0,
                name="Grass",
                base_energy=10.0,
                max_energy=100.0,
                growth_rate=5.0,
                survival_threshold=1.0,
                reproduction_interval=10,
                seed_min_dist=1.0,
                seed_max_dist=3.0,
                seed_energy_cost=5.0,
                triggers=[],
            )
        ],
        predator_species=[
            PredatorSpeciesParams(
                species_id=0,
                name="Herbivore",
                energy_min=5.0,
                velocity=2,
                consumption_rate=10.0,
            )
        ],
        diet_matrix=[[True]],
        trigger_rules=[],
        substance_definitions=[],
        initial_plants=[],
        initial_swarms=[],
    )
    return state

delete_trigger_rule_condition_node(index, path)

Delete the node at path from one trigger rule's condition tree.

Source code in src/phids/api/ui_state.py

def delete_trigger_rule_condition_node(self, index: int, path: str) -> None:
    """Delete the node at ``path`` from one trigger rule's condition tree."""
    rule = self.trigger_rules[index]
    if rule.activation_condition is None:
        return
    if not path:
        rule.activation_condition = None
        return
    root = deepcopy(rule.activation_condition)
    path_indices = _parse_condition_path(path)
    parent = _condition_node_at_path(root, path_indices[:-1])
    if parent.get("kind") not in {"all_of", "any_of"}:
        raise IndexError("Condition parent is not a group node.")
    children = parent.get("conditions")
    if not isinstance(children, list):
        raise IndexError("Condition parent has no child list.")
    child_index = path_indices[-1]
    if child_index < 0 or child_index >= len(children):
        raise IndexError("Condition node index is out of range.")
    del children[child_index]
    rule.activation_condition = _prune_empty_condition_groups(root)

rebuild_species_ids()

Re-assign sequential species_id values after list mutation.

Source code in src/phids/api/ui_state.py

def rebuild_species_ids(self) -> None:
    """Re-assign sequential ``species_id`` values after list mutation."""
    from phids.api.schemas import FloraSpeciesParams, PredatorSpeciesParams

    self.flora_species = [
        fp.model_copy(update={"species_id": i})  # type: ignore[union-attr]
        for i, fp in enumerate(self.flora_species)
        if isinstance(fp, FloraSpeciesParams)
    ]
    self.predator_species = [
        pp.model_copy(update={"species_id": i})  # type: ignore[union-attr]
        for i, pp in enumerate(self.predator_species)
        if isinstance(pp, PredatorSpeciesParams)
    ]

remove_flora(species_id)

Remove a flora species by id and clean up dependent data.

Parameters:

Name	Type	Description	Default
species_id	int	The species_id of the species to remove.	required

Raises:

Type	Description
ValueError	If no species with the given id exists.

Source code in src/phids/api/ui_state.py

def remove_flora(self, species_id: int) -> None:
    """Remove a flora species by id and clean up dependent data.

    Args:
        species_id: The ``species_id`` of the species to remove.

    Raises:
        ValueError: If no species with the given id exists.
    """
    from phids.api.schemas import FloraSpeciesParams

    idx = next(
        (
            i
            for i, fp in enumerate(self.flora_species)
            if isinstance(fp, FloraSpeciesParams) and fp.species_id == species_id
        ),
        None,
    )
    if idx is None:
        raise ValueError(f"Flora species_id {species_id} not found.")

    del self.flora_species[idx]
    # Remove diet_matrix column for this flora species
    for row in self.diet_matrix:
        if idx < len(row):
            del row[idx]

    # Remove trigger rules referencing this flora species; compact higher IDs
    new_rules: list[TriggerRule] = []
    for rule in self.trigger_rules:
        if rule.flora_species_id == species_id:
            continue  # orphaned rule
        new_rule = dataclasses.replace(rule)
        if new_rule.flora_species_id > species_id:
            new_rule.flora_species_id -= 1  # compact after removal
        new_rules.append(new_rule)
    self.trigger_rules = new_rules

    # Remove placements
    self.initial_plants = [p for p in self.initial_plants if p.species_id != species_id]
    self.rebuild_species_ids()
    self._resize_diet_matrix()
    logger.debug(
        "Draft flora removed (species_id=%d, total_flora=%d, remaining_trigger_rules=%d)",
        species_id,
        len(self.flora_species),
        len(self.trigger_rules),
    )

remove_plant_placement(index)

Remove a plant placement by list index.

Source code in src/phids/api/ui_state.py

def remove_plant_placement(self, index: int) -> None:
    """Remove a plant placement by list index."""
    removed = self.initial_plants[index]
    del self.initial_plants[index]
    logger.debug(
        "Draft plant placement removed (index=%d, species_id=%d, x=%d, y=%d, total_plants=%d)",
        index,
        removed.species_id,
        removed.x,
        removed.y,
        len(self.initial_plants),
    )

remove_predator(species_id)

Remove a predator species by id and clean up dependent data.

Parameters:

Name	Type	Description	Default
species_id	int	The species_id of the species to remove.	required

Raises:

Type	Description
ValueError	If no species with the given id exists.

Source code in src/phids/api/ui_state.py

def remove_predator(self, species_id: int) -> None:
    """Remove a predator species by id and clean up dependent data.

    Args:
        species_id: The ``species_id`` of the species to remove.

    Raises:
        ValueError: If no species with the given id exists.
    """
    from phids.api.schemas import PredatorSpeciesParams

    idx = next(
        (
            i
            for i, pp in enumerate(self.predator_species)
            if isinstance(pp, PredatorSpeciesParams) and pp.species_id == species_id
        ),
        None,
    )
    if idx is None:
        raise ValueError(f"Predator species_id {species_id} not found.")

    del self.predator_species[idx]
    # Remove diet_matrix row for this predator
    if idx < len(self.diet_matrix):
        del self.diet_matrix[idx]

    # Remove trigger rules referencing this predator; compact higher IDs
    new_rules: list[TriggerRule] = []
    for rule in self.trigger_rules:
        if rule.predator_species_id == species_id:
            continue  # orphaned rule
        new_rule = dataclasses.replace(rule)
        if new_rule.predator_species_id > species_id:
            new_rule.predator_species_id -= 1
        new_rule.activation_condition = _remap_condition_references(
            deepcopy(new_rule.activation_condition),
            removed_predator_id=species_id,
        )
        new_rules.append(new_rule)
    self.trigger_rules = new_rules

    # Remove placements
    self.initial_swarms = [s for s in self.initial_swarms if s.species_id != species_id]
    self.rebuild_species_ids()
    self._resize_diet_matrix()
    logger.debug(
        "Draft predator removed (species_id=%d, total_predators=%d, remaining_trigger_rules=%d)",
        species_id,
        len(self.predator_species),
        len(self.trigger_rules),
    )

remove_swarm_placement(index)

Remove a swarm placement by list index.

Source code in src/phids/api/ui_state.py

def remove_swarm_placement(self, index: int) -> None:
    """Remove a swarm placement by list index."""
    removed = self.initial_swarms[index]
    del self.initial_swarms[index]
    logger.debug(
        "Draft swarm placement removed (index=%d, species_id=%d, x=%d, y=%d, total_swarms=%d)",
        index,
        removed.species_id,
        removed.x,
        removed.y,
        len(self.initial_swarms),
    )

remove_trigger_rule(index)

Remove a trigger rule by list index.

Parameters:

Name	Type	Description	Default
index	int	Position in the trigger_rules list.	required

Raises:

Type	Description
IndexError	If index is out of range.

Source code in src/phids/api/ui_state.py

def remove_trigger_rule(self, index: int) -> None:
    """Remove a trigger rule by list index.

    Args:
        index: Position in the trigger_rules list.

    Raises:
        IndexError: If index is out of range.
    """
    removed = self.trigger_rules[index]
    del self.trigger_rules[index]
    logger.debug(
        "Draft trigger rule removed (index=%d, flora_species_id=%d, predator_species_id=%d, substance_id=%d, total_rules=%d)",
        index,
        removed.flora_species_id,
        removed.predator_species_id,
        removed.substance_id,
        len(self.trigger_rules),
    )

replace_trigger_rule_condition_node(index, path, condition)

Replace the node at path inside one trigger rule's condition tree.

Source code in src/phids/api/ui_state.py

def replace_trigger_rule_condition_node(
    self,
    index: int,
    path: str,
    condition: dict[str, object],
) -> None:
    """Replace the node at ``path`` inside one trigger rule's condition tree."""
    rule = self.trigger_rules[index]
    if not path:
        rule.activation_condition = deepcopy(condition)
        return
    if rule.activation_condition is None:
        raise IndexError("Trigger rule has no activation condition to replace.")
    root = deepcopy(rule.activation_condition)
    path_indices = _parse_condition_path(path)
    parent = _condition_node_at_path(root, path_indices[:-1])
    if parent.get("kind") not in {"all_of", "any_of"}:
        raise IndexError("Condition parent is not a group node.")
    children = parent.get("conditions")
    if not isinstance(children, list):
        raise IndexError("Condition parent has no child list.")
    child_index = path_indices[-1]
    if child_index < 0 or child_index >= len(children):
        raise IndexError("Condition node index is out of range.")
    children[child_index] = deepcopy(condition)
    rule.activation_condition = root

set_trigger_rule_activation_condition(index, condition)

Replace the full activation-condition tree for one trigger rule.

Source code in src/phids/api/ui_state.py

def set_trigger_rule_activation_condition(
    self,
    index: int,
    condition: dict[str, object] | None,
) -> None:
    """Replace the full activation-condition tree for one trigger rule."""
    self.trigger_rules[index].activation_condition = deepcopy(condition)

update_trigger_rule(index, *, flora_species_id=None, predator_species_id=None, substance_id=None, min_predator_population=None, activation_condition=None, required_signal_ids=None)

Update fields of a trigger rule in-place.

Parameters:

Name	Type	Description	Default
index	int	Position in the trigger_rules list.	required
flora_species_id	int | None	New flora species id (optional).	None
predator_species_id	int | None	New predator species id (optional).	None
substance_id	int | None	New substance id (optional).	None
min_predator_population	int | None	New minimum population (optional).	None
activation_condition	dict[str, object] | None	New condition tree (optional).	None
required_signal_ids	list[int] | None	Deprecated compatibility shorthand for signal-only AND gates.	None

Raises:

Type	Description
IndexError	If index is out of range.

Source code in src/phids/api/ui_state.py

def update_trigger_rule(
    self,
    index: int,
    *,
    flora_species_id: int | None = None,
    predator_species_id: int | None = None,
    substance_id: int | None = None,
    min_predator_population: int | None = None,
    activation_condition: dict[str, object] | None = None,
    required_signal_ids: list[int] | None = None,
) -> None:
    """Update fields of a trigger rule in-place.

    Args:
        index: Position in the trigger_rules list.
        flora_species_id: New flora species id (optional).
        predator_species_id: New predator species id (optional).
        substance_id: New substance id (optional).
        min_predator_population: New minimum population (optional).
        activation_condition: New condition tree (optional).
        required_signal_ids: Deprecated compatibility shorthand for
            signal-only AND gates.

    Raises:
        IndexError: If index is out of range.
    """
    rule = self.trigger_rules[index]
    if flora_species_id is not None:
        rule.flora_species_id = flora_species_id
    if predator_species_id is not None:
        rule.predator_species_id = predator_species_id
    if substance_id is not None:
        rule.substance_id = substance_id
    if min_predator_population is not None:
        rule.min_predator_population = min_predator_population
    if activation_condition is not None:
        rule.activation_condition = deepcopy(activation_condition)
    elif required_signal_ids is not None:
        rule.activation_condition = _legacy_signal_ids_to_activation_condition(
            required_signal_ids
        )
    logger.debug(
        "Draft trigger rule updated (index=%d, flora_species_id=%d, predator_species_id=%d, substance_id=%d)",
        index,
        rule.flora_species_id,
        rule.predator_species_id,
        rule.substance_id,
    )

update_trigger_rule_condition_node(index, path, **fields)

Patch selected fields on the condition node at path.

Source code in src/phids/api/ui_state.py

def update_trigger_rule_condition_node(
    self,
    index: int,
    path: str,
    **fields: object,
) -> None:
    """Patch selected fields on the condition node at ``path``."""
    rule = self.trigger_rules[index]
    if rule.activation_condition is None:
        raise IndexError("Trigger rule has no activation condition to update.")
    root = deepcopy(rule.activation_condition)
    node = _condition_node_at_path(root, _parse_condition_path(path))
    node.update(fields)
    rule.activation_condition = root

PlacedPlant dataclass

A plant placed on the grid before the simulation starts.

Parameters:

Name	Type	Description	Default
species_id	int	Flora species index.	required
x	int	Grid x-coordinate.	required
y	int	Grid y-coordinate.	required
energy	float	Initial energy reserve.	required

Source code in src/phids/api/ui_state.py

@dataclasses.dataclass
class PlacedPlant:
    """A plant placed on the grid before the simulation starts.

    Args:
        species_id: Flora species index.
        x: Grid x-coordinate.
        y: Grid y-coordinate.
        energy: Initial energy reserve.
    """

    species_id: int
    x: int
    y: int
    energy: float

PlacedSwarm dataclass

A herbivore swarm placed on the grid before the simulation starts.

Parameters:

Name	Type	Description	Default
species_id	int	Predator species index.	required
x	int	Grid x-coordinate.	required
y	int	Grid y-coordinate.	required
population	int	Initial swarm population.	required
energy	float	Initial energy reserve.	required

Source code in src/phids/api/ui_state.py

@dataclasses.dataclass
class PlacedSwarm:
    """A herbivore swarm placed on the grid before the simulation starts.

    Args:
        species_id: Predator species index.
        x: Grid x-coordinate.
        y: Grid y-coordinate.
        population: Initial swarm population.
        energy: Initial energy reserve.
    """

    species_id: int
    x: int
    y: int
    population: int
    energy: float

SubstanceDefinition dataclass

Named substance with physical/biological properties.

A substance definition captures how a chemical behaves once produced. The trigger matrix separately records which (flora, predator) pair activates synthesis.

Parameters:

Name	Type	Description	Default
substance_id	int	Layer index in GridEnvironment.signal_layers or toxin_layers (0 ≤ id < MAX_SUBSTANCE_TYPES).	required
name	str	Human-readable label shown in the UI.	'Signal'
is_toxin	bool	True for toxins; False for airborne signals.	False
lethal	bool	Lethal-toxin flag (ignored if is_toxin is False).	False
repellent	bool	Repellent-toxin flag.	False
synthesis_duration	int	Ticks to complete synthesis (production time).	3
aftereffect_ticks	int	Ticks the substance lingers after emission ceases.	0
lethality_rate	float	Population units eliminated per tick (β).	0.0
repellent_walk_ticks	int	Random-walk duration on repel trigger.	3
energy_cost_per_tick	float	Energy drained from the plant per active tick.	1.0
precursor_signal_id	int	Signal id required before activation (−1 = none).	-1
min_predator_population	int	Minimum swarm size to trigger synthesis.	5

Source code in src/phids/api/ui_state.py

@dataclasses.dataclass
class SubstanceDefinition:
    """Named substance with physical/biological properties.

    A substance definition captures how a chemical behaves once produced.
    The trigger matrix separately records *which* (flora, predator) pair
    activates synthesis.

    Args:
        substance_id: Layer index in ``GridEnvironment.signal_layers`` or
            ``toxin_layers`` (0 ≤ id < MAX_SUBSTANCE_TYPES).
        name: Human-readable label shown in the UI.
        is_toxin: ``True`` for toxins; ``False`` for airborne signals.
        lethal: Lethal-toxin flag (ignored if ``is_toxin`` is ``False``).
        repellent: Repellent-toxin flag.
        synthesis_duration: Ticks to complete synthesis (production time).
        aftereffect_ticks: Ticks the substance lingers after emission ceases.
        lethality_rate: Population units eliminated per tick (β).
        repellent_walk_ticks: Random-walk duration on repel trigger.
        energy_cost_per_tick: Energy drained from the plant per active tick.
        precursor_signal_id: Signal id required before activation (−1 = none).
        min_predator_population: Minimum swarm size to trigger synthesis.
    """

    substance_id: int
    name: str = "Signal"
    is_toxin: bool = False
    lethal: bool = False
    repellent: bool = False
    synthesis_duration: int = 3
    aftereffect_ticks: int = 0
    lethality_rate: float = 0.0
    repellent_walk_ticks: int = 3
    energy_cost_per_tick: float = 1.0
    precursor_signal_id: int = -1
    min_predator_population: int = 5

    @property
    def type_label(self) -> str:
        """Human-readable substance type.

        Returns:
            str: One of ``"Signal"``, ``"Lethal Toxin"``,
                ``"Repellent Toxin"``, or ``"Toxin"``.
        """
        if not self.is_toxin:
            return "Signal"
        if self.lethal:
            return "Lethal Toxin"
        if self.repellent:
            return "Repellent Toxin"
        return "Toxin"

type_label property

Human-readable substance type.

Returns:

Name	Type	Description
str	str	One of "Signal", "Lethal Toxin", "Repellent Toxin", or "Toxin".

TriggerRule dataclass

One explicit chemical-defense trigger rule.

A rule says: "when flora species flora_species_id is attacked by predator species predator_species_id with at least min_predator_population individuals, synthesise substance substance_id. Optional nested activation conditions can additionally require active substances and/or other enemy presences via explicit all_of / any_of predicate trees. None = unconditional."

Multiple rules may share the same (flora, predator) pair to express production of different substances simultaneously.

Parameters:

Name	Type	Description	Default
flora_species_id	int	Flora species index (0-based).	required
predator_species_id	int	Predator species index (0-based).	required
substance_id	int	Substance layer index to synthesise.	required
min_predator_population	int	Minimum swarm size to trigger this rule.	5
activation_condition	dict[str, object] | None	Optional JSON-serialisable predicate tree.	None

Source code in src/phids/api/ui_state.py

@dataclasses.dataclass
class TriggerRule:
    """One explicit chemical-defense trigger rule.

    A rule says: "when flora species *flora_species_id* is attacked by
    predator species *predator_species_id* with at least
    *min_predator_population* individuals, synthesise substance
    *substance_id*. Optional nested activation conditions can additionally
    require active substances and/or other enemy presences via explicit
    ``all_of`` / ``any_of`` predicate trees. ``None`` = unconditional."

    Multiple rules may share the same (flora, predator) pair to express
    production of different substances simultaneously.

    Args:
        flora_species_id: Flora species index (0-based).
        predator_species_id: Predator species index (0-based).
        substance_id: Substance layer index to synthesise.
        min_predator_population: Minimum swarm size to trigger this rule.
        activation_condition: Optional JSON-serialisable predicate tree.
    """

    flora_species_id: int
    predator_species_id: int
    substance_id: int
    min_predator_population: int = 5
    activation_condition: dict[str, object] | None = None

get_draft()

Return the current draft state, initialising a default if needed.

Returns:

Name	Type	Description
DraftState	DraftState	The active draft configuration.

Source code in src/phids/api/ui_state.py

def get_draft() -> DraftState:
    """Return the current draft state, initialising a default if needed.

    Returns:
        DraftState: The active draft configuration.
    """
    global _draft  # noqa: PLW0603
    if _draft is None:
        _draft = DraftState.default()
        logger.info("Draft state initialised with built-in default scenario")
    return _draft

reset_draft()

Reset the draft state to the built-in default.

Source code in src/phids/api/ui_state.py

def reset_draft() -> None:
    """Reset the draft state to the built-in default."""
    global _draft  # noqa: PLW0603
    _draft = DraftState.default()
    logger.info("Draft state reset to built-in default scenario")

set_draft(state)

Replace the active draft state.

Parameters:

Name	Type	Description	Default
state	DraftState	New :class:DraftState to activate.	required

Source code in src/phids/api/ui_state.py

def set_draft(state: DraftState) -> None:
    """Replace the active draft state.

    Args:
        state: New :class:`DraftState` to activate.
    """
    global _draft  # noqa: PLW0603
    _draft = state
    logger.info(
        "Draft state replaced (scenario_name=%s, flora=%d, predators=%d, substances=%d)",
        state.scenario_name,
        len(state.flora_species),
        len(state.predator_species),
        len(state.substance_definitions),
    )

Engine orchestration

Simulation loop orchestration with deterministic ticks.

This module implements the main simulation driver which advances the grid environment and ECS world through ordered systems. It captures per-tick snapshots for replay and telemetry and enforces deterministic update ordering using an asyncio lock.

SimulationLoop

Orchestrate deterministic, double-buffered simulation ticks.

Double-buffering is achieved by keeping a read-copy of the grid state while writing results to live objects; these writes become the read state for the next tick. Concurrent access is protected by asyncio.Lock.

Parameters:

Name	Type	Description	Default
config	SimulationConfig	Validated :class:~phids.api.schemas.SimulationConfig.	required

Source code in src/phids/engine/loop.py

class SimulationLoop:
    """Orchestrate deterministic, double-buffered simulation ticks.

    Double-buffering is achieved by keeping a read-copy of the grid state
    while writing results to live objects; these writes become the read
    state for the next tick. Concurrent access is protected by
    ``asyncio.Lock``.

    Args:
        config: Validated :class:`~phids.api.schemas.SimulationConfig`.
    """

    def __init__(self, config: SimulationConfig) -> None:
        """Initialise the SimulationLoop with the provided configuration.

        Args:
            config: Validated SimulationConfig instance from the API payload.
        """
        self.config = config
        self.tick: int = 0
        self.running: bool = False
        self.paused: bool = False
        self.terminated: bool = False
        self.termination_reason: str | None = None
        self._lock: asyncio.Lock = asyncio.Lock()
        self._debug_tick_interval: int = get_simulation_debug_interval()

        # Build environment
        self.env = GridEnvironment(
            width=config.grid_width,
            height=config.grid_height,
            num_signals=config.num_signals,
            num_toxins=config.num_toxins,
        )
        self.env.set_uniform_wind(config.wind_x, config.wind_y)

        # Build ECS
        self.world = ECSWorld()

        # Telemetry
        self.telemetry = TelemetryRecorder()
        # Deterministic replay state frames (msgpack serialisation per tick)
        self.replay = ReplayBuffer()

        # Pre-compute species parameter lookups
        self._flora_params: dict[int, Any] = {sp.species_id: sp for sp in config.flora_species}
        self._trigger_conditions: dict[int, list[Any]] = {
            sp.species_id: list(sp.triggers) for sp in config.flora_species
        }
        self._diet_matrix: list[list[bool]] = config.diet_matrix.rows

        # Spawn initial entities
        self._spawn_initial_entities()
        logger.info(
            "SimulationLoop initialised (grid=%dx%d, flora_species=%d, predator_species=%d, signals=%d, toxins=%d, tick_rate_hz=%.2f)",
            config.grid_width,
            config.grid_height,
            len(config.flora_species),
            len(config.predator_species),
            config.num_signals,
            config.num_toxins,
            config.tick_rate_hz,
        )

    # ------------------------------------------------------------------
    # Initialisation helpers
    # ------------------------------------------------------------------

    def _spawn_initial_entities(self) -> None:
        """Place initial plants and swarms from the configuration.

        The method creates entity instances, attaches components, registers
        spatial positions in the :class:`ECSWorld`, and populates the
        environment's plant energy buffers.
        """
        spawned_plants = 0
        spawned_swarms = 0

        for placement in self.config.initial_plants:
            params = self._flora_params.get(placement.species_id)
            if params is None:
                logger.warning(
                    "Skipping initial plant placement with unknown flora species_id=%d at (%d, %d)",
                    placement.species_id,
                    placement.x,
                    placement.y,
                )
                continue
            entity = self.world.create_entity()
            plant = PlantComponent(
                entity_id=entity.entity_id,
                species_id=placement.species_id,
                x=placement.x,
                y=placement.y,
                energy=placement.energy,
                max_energy=params.max_energy,
                base_energy=params.base_energy,
                growth_rate=params.growth_rate,
                survival_threshold=params.survival_threshold,
                reproduction_interval=params.reproduction_interval,
                seed_min_dist=params.seed_min_dist,
                seed_max_dist=params.seed_max_dist,
                seed_energy_cost=params.seed_energy_cost,
                camouflage=params.camouflage,
                camouflage_factor=params.camouflage_factor,
            )
            self.world.add_component(entity.entity_id, plant)
            self.world.register_position(entity.entity_id, placement.x, placement.y)
            self.env.set_plant_energy(
                placement.x, placement.y, placement.species_id, placement.energy
            )
            spawned_plants += 1

        for placement in self.config.initial_swarms:
            entity = self.world.create_entity()
            swarm = SwarmComponent(
                entity_id=entity.entity_id,
                species_id=placement.species_id,
                x=placement.x,
                y=placement.y,
                population=placement.population,
                initial_population=placement.population,
                energy=placement.energy,
                energy_min=self._get_predator_energy_min(placement.species_id),
                velocity=self._get_predator_velocity(placement.species_id),
                consumption_rate=self._get_predator_consumption_rate(placement.species_id),
                reproduction_energy_divisor=self._get_predator_reproduction_divisor(
                    placement.species_id
                ),
            )
            self.world.add_component(entity.entity_id, swarm)
            self.world.register_position(entity.entity_id, placement.x, placement.y)
            spawned_swarms += 1

        self.env.rebuild_energy_layer()
        logger.info(
            "Initial entities spawned (plants=%d, swarms=%d)",
            spawned_plants,
            spawned_swarms,
        )

    def _get_predator_energy_min(self, species_id: int) -> float:
        """Return the configured minimum energy for a predator species.

        Args:
            species_id: Predator species identifier to look up.

        Returns:
            float: Configured minimum energy if found, otherwise a sensible
            default of 1.0.
        """
        for sp in self.config.predator_species:
            if sp.species_id == species_id:
                return sp.energy_min
        return 1.0

    def _get_predator_velocity(self, species_id: int) -> int:
        """Return the configured movement period (velocity) for a predator.

        Args:
            species_id: Predator species identifier to look up.

        Returns:
            int: Movement period in ticks; defaults to 1 when not found.
        """
        for sp in self.config.predator_species:
            if sp.species_id == species_id:
                return sp.velocity
        return 1

    def _get_predator_consumption_rate(self, species_id: int) -> float:
        """Return the per-tick consumption rate for a predator species.

        Args:
            species_id: Predator species identifier to look up.

        Returns:
            float: Consumption rate if present, otherwise 1.0 by default.
        """
        for sp in self.config.predator_species:
            if sp.species_id == species_id:
                return sp.consumption_rate
        return 1.0

    def _get_predator_reproduction_divisor(self, species_id: int) -> float:
        """Return the configured reproduction divisor for a predator species.

        Args:
            species_id: Predator species identifier to look up.

        Returns:
            float: Reproduction divisor if present, otherwise 1.0.
        """
        for sp in self.config.predator_species:
            if sp.species_id == species_id:
                return sp.reproduction_energy_divisor
        return 1.0

    # ------------------------------------------------------------------
    # Simulation control
    # ------------------------------------------------------------------

    def start(self) -> None:
        """Mark the simulation as running.

        Sets running state to True and clears the paused flag.
        """
        self.running = True
        self.paused = False
        logger.info("Simulation loop started/resumed at tick %d", self.tick)

    def pause(self) -> None:
        """Toggle the paused state.

        Flips the ``paused`` boolean.
        """
        self.paused = not self.paused
        logger.info(
            "Simulation loop %s at tick %d", "paused" if self.paused else "resumed", self.tick
        )

    def stop(self) -> None:
        """Halt the simulation by clearing the running flag."""
        self.running = False
        logger.info("Simulation loop stopped at tick %d", self.tick)

    def _should_log_debug_summary(self) -> bool:
        """Return whether the current tick should emit a DEBUG summary."""
        return (
            logger.isEnabledFor(logging.DEBUG)
            and self._debug_tick_interval > 0
            and self.tick % self._debug_tick_interval == 0
        )

    def _log_debug_tick_summary(
        self,
        *,
        latest_metrics: dict[str, Any] | None,
        phase_timings_ms: dict[str, float],
    ) -> None:
        """Emit a coarse DEBUG snapshot for the current tick."""
        swarm_population = 0
        for entity in self.world.query(SwarmComponent):
            swarm_population += entity.get_component(SwarmComponent).population

        logger.debug(
            (
                "Tick summary (tick=%d, flora_energy=%.3f, flora_population=%d, "
                "predator_clusters=%d, predator_population=%d, replay_frames=%d, "
                "phase_timings_ms=%s)"
            ),
            self.tick,
            float(latest_metrics.get("total_flora_energy", 0.0)) if latest_metrics else 0.0,
            int(latest_metrics.get("flora_population", 0)) if latest_metrics else 0,
            int(latest_metrics.get("predator_clusters", 0)) if latest_metrics else 0,
            swarm_population,
            len(self.replay),
            phase_timings_ms,
        )

    # ------------------------------------------------------------------
    # Core tick
    # ------------------------------------------------------------------

    async def step(self) -> TerminationResult:
        """Execute one deterministic simulation tick.

        The method performs the ordered phases of the simulation (flow-field
        update, lifecycle, interaction, signaling, telemetry) while holding
        an asyncio lock to ensure async-safety. After processing it
        evaluates termination conditions.

        Returns:
            TerminationResult: Termination state after the tick.
        """
        async with self._lock:
            if self.terminated:
                logger.debug(
                    "Simulation step skipped because loop is already terminated at tick %d",
                    self.tick,
                )
                return TerminationResult(terminated=True, reason=self.termination_reason or "")

            debug_summary = self._should_log_debug_summary()
            phase_timings_ms: dict[str, float] = {}
            phase_started = time.perf_counter()

            # --------------------------------------------------------
            # Phase 1: Flow-field update (uses current read state)
            # --------------------------------------------------------
            self.env.flow_field = compute_flow_field(
                self.env.plant_energy_layer,
                self.env.toxin_layers,
                self.env.width,
                self.env.height,
            )
            if debug_summary:
                phase_timings_ms["flow_field"] = (time.perf_counter() - phase_started) * 1000.0
                phase_started = time.perf_counter()

            # Apply camouflage attenuations
            for entity in self.world.query(PlantComponent):
                plant: PlantComponent = entity.get_component(PlantComponent)
                if plant.camouflage:
                    apply_camouflage(self.env.flow_field, plant.x, plant.y, plant.camouflage_factor)

            # --------------------------------------------------------
            # Phase 2: Lifecycle (grow, connect, reproduce, cull)
            # --------------------------------------------------------
            run_lifecycle(
                self.world,
                self.env,
                self.tick,
                self._flora_params,
                mycorrhizal_connection_cost=self.config.mycorrhizal_connection_cost,
                mycorrhizal_growth_interval_ticks=self.config.mycorrhizal_growth_interval_ticks,
                mycorrhizal_inter_species=self.config.mycorrhizal_inter_species,
            )
            if debug_summary:
                phase_timings_ms["lifecycle"] = (time.perf_counter() - phase_started) * 1000.0
                phase_started = time.perf_counter()

            # --------------------------------------------------------
            # Phase 3: Interaction (movement, feeding, starvation, mitosis)
            # --------------------------------------------------------
            run_interaction(self.world, self.env, self._diet_matrix, self.tick)
            if debug_summary:
                phase_timings_ms["interaction"] = (time.perf_counter() - phase_started) * 1000.0
                phase_started = time.perf_counter()

            # --------------------------------------------------------
            # Phase 4: Signaling (substance synthesis, diffusion, toxins)
            # --------------------------------------------------------
            run_signaling(
                self.world,
                self.env,
                self._trigger_conditions,
                self.config.mycorrhizal_inter_species,
                self.config.mycorrhizal_signal_velocity,
                self.tick,
            )
            if debug_summary:
                phase_timings_ms["signaling"] = (time.perf_counter() - phase_started) * 1000.0
                phase_started = time.perf_counter()

            # --------------------------------------------------------
            # Phase 5: Telemetry
            # --------------------------------------------------------
            self.telemetry.record(self.world, self.tick)
            self.replay.append(self.get_state_snapshot())
            latest_metrics = self.telemetry.get_latest_metrics()
            if debug_summary:
                phase_timings_ms["telemetry_replay"] = (
                    time.perf_counter() - phase_started
                ) * 1000.0
                phase_started = time.perf_counter()

            # --------------------------------------------------------
            # Phase 6: Termination check (double-buffer swap happens here
            #          implicitly – all writes committed before check)
            # --------------------------------------------------------
            result = check_termination(
                self.world,
                self.tick,
                max_ticks=self.config.max_ticks,
                z2_flora_species=self.config.z2_flora_species_extinction,
                z4_predator_species=self.config.z4_predator_species_extinction,
                z6_max_flora_energy=self.config.z6_max_total_flora_energy,
                z7_max_predator_population=self.config.z7_max_total_predator_population,
            )
            if debug_summary:
                phase_timings_ms["termination"] = (time.perf_counter() - phase_started) * 1000.0

            self.tick += 1
            if debug_summary:
                self._log_debug_tick_summary(
                    latest_metrics=latest_metrics,
                    phase_timings_ms=phase_timings_ms,
                )

            if result.terminated:
                self.terminated = True
                self.running = False
                self.termination_reason = result.reason
                logger.info("Simulation terminated at tick %d: %s", self.tick, result.reason)

            return result

    async def run(self) -> None:
        """Run the simulation loop until termination at configured tick rate.

        The loop respects ``paused`` and sleeps to maintain ``tick_rate_hz``.
        """
        tick_interval = 1.0 / self.config.tick_rate_hz
        self.start()
        logger.info("Simulation run loop entering background execution")

        while self.running and not self.terminated:
            if self.paused:
                await asyncio.sleep(tick_interval)
                continue

            t0 = time.monotonic()
            result = await self.step()
            if result.terminated:
                break
            elapsed = time.monotonic() - t0
            sleep_time = max(0.0, tick_interval - elapsed)
            await asyncio.sleep(sleep_time)

        logger.info(
            "Simulation run loop exited (tick=%d, terminated=%s, reason=%s)",
            self.tick,
            self.terminated,
            self.termination_reason,
        )

    # ------------------------------------------------------------------
    # Wind update (REST API integration point)
    # ------------------------------------------------------------------

    def update_wind(self, vx: float, vy: float) -> None:
        """Update the environment uniform wind vector.

        Args:
            vx: Wind X component.
            vy: Wind Y component.
        """
        self.env.set_uniform_wind(vx, vy)
        logger.info("Simulation wind updated to (vx=%.3f, vy=%.3f)", vx, vy)

    # ------------------------------------------------------------------
    # State snapshot for WebSocket streaming
    # ------------------------------------------------------------------

    def get_state_snapshot(self) -> dict[str, Any]:
        """Return a serialisable snapshot of the current grid state.

        Returns:
            dict[str, Any]: Snapshot containing tick, termination state and
            environment dictionary (from :meth:`GridEnvironment.to_dict`).
        """
        return {
            "tick": self.tick,
            "terminated": self.terminated,
            "termination_reason": self.termination_reason,
            **self.env.to_dict(),
        }

__init__(config)

Initialise the SimulationLoop with the provided configuration.

Parameters:

Name	Type	Description	Default
config	SimulationConfig	Validated SimulationConfig instance from the API payload.	required

Source code in src/phids/engine/loop.py

def __init__(self, config: SimulationConfig) -> None:
    """Initialise the SimulationLoop with the provided configuration.

    Args:
        config: Validated SimulationConfig instance from the API payload.
    """
    self.config = config
    self.tick: int = 0
    self.running: bool = False
    self.paused: bool = False
    self.terminated: bool = False
    self.termination_reason: str | None = None
    self._lock: asyncio.Lock = asyncio.Lock()
    self._debug_tick_interval: int = get_simulation_debug_interval()

    # Build environment
    self.env = GridEnvironment(
        width=config.grid_width,
        height=config.grid_height,
        num_signals=config.num_signals,
        num_toxins=config.num_toxins,
    )
    self.env.set_uniform_wind(config.wind_x, config.wind_y)

    # Build ECS
    self.world = ECSWorld()

    # Telemetry
    self.telemetry = TelemetryRecorder()
    # Deterministic replay state frames (msgpack serialisation per tick)
    self.replay = ReplayBuffer()

    # Pre-compute species parameter lookups
    self._flora_params: dict[int, Any] = {sp.species_id: sp for sp in config.flora_species}
    self._trigger_conditions: dict[int, list[Any]] = {
        sp.species_id: list(sp.triggers) for sp in config.flora_species
    }
    self._diet_matrix: list[list[bool]] = config.diet_matrix.rows

    # Spawn initial entities
    self._spawn_initial_entities()
    logger.info(
        "SimulationLoop initialised (grid=%dx%d, flora_species=%d, predator_species=%d, signals=%d, toxins=%d, tick_rate_hz=%.2f)",
        config.grid_width,
        config.grid_height,
        len(config.flora_species),
        len(config.predator_species),
        config.num_signals,
        config.num_toxins,
        config.tick_rate_hz,
    )

get_state_snapshot()

Return a serialisable snapshot of the current grid state.

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Snapshot containing tick, termination state and
dict[str, Any]	environment dictionary (from :meth:GridEnvironment.to_dict).

Source code in src/phids/engine/loop.py

def get_state_snapshot(self) -> dict[str, Any]:
    """Return a serialisable snapshot of the current grid state.

    Returns:
        dict[str, Any]: Snapshot containing tick, termination state and
        environment dictionary (from :meth:`GridEnvironment.to_dict`).
    """
    return {
        "tick": self.tick,
        "terminated": self.terminated,
        "termination_reason": self.termination_reason,
        **self.env.to_dict(),
    }

pause()

Toggle the paused state.

Flips the paused boolean.

Source code in src/phids/engine/loop.py

def pause(self) -> None:
    """Toggle the paused state.

    Flips the ``paused`` boolean.
    """
    self.paused = not self.paused
    logger.info(
        "Simulation loop %s at tick %d", "paused" if self.paused else "resumed", self.tick
    )

run() async

Run the simulation loop until termination at configured tick rate.

The loop respects paused and sleeps to maintain tick_rate_hz.

Source code in src/phids/engine/loop.py

async def run(self) -> None:
    """Run the simulation loop until termination at configured tick rate.

    The loop respects ``paused`` and sleeps to maintain ``tick_rate_hz``.
    """
    tick_interval = 1.0 / self.config.tick_rate_hz
    self.start()
    logger.info("Simulation run loop entering background execution")

    while self.running and not self.terminated:
        if self.paused:
            await asyncio.sleep(tick_interval)
            continue

        t0 = time.monotonic()
        result = await self.step()
        if result.terminated:
            break
        elapsed = time.monotonic() - t0
        sleep_time = max(0.0, tick_interval - elapsed)
        await asyncio.sleep(sleep_time)

    logger.info(
        "Simulation run loop exited (tick=%d, terminated=%s, reason=%s)",
        self.tick,
        self.terminated,
        self.termination_reason,
    )

start()

Mark the simulation as running.

Sets running state to True and clears the paused flag.

Source code in src/phids/engine/loop.py

def start(self) -> None:
    """Mark the simulation as running.

    Sets running state to True and clears the paused flag.
    """
    self.running = True
    self.paused = False
    logger.info("Simulation loop started/resumed at tick %d", self.tick)

step() async

Execute one deterministic simulation tick.

The method performs the ordered phases of the simulation (flow-field update, lifecycle, interaction, signaling, telemetry) while holding an asyncio lock to ensure async-safety. After processing it evaluates termination conditions.

Returns:

Name	Type	Description
TerminationResult	TerminationResult	Termination state after the tick.

Source code in src/phids/engine/loop.py

async def step(self) -> TerminationResult:
    """Execute one deterministic simulation tick.

    The method performs the ordered phases of the simulation (flow-field
    update, lifecycle, interaction, signaling, telemetry) while holding
    an asyncio lock to ensure async-safety. After processing it
    evaluates termination conditions.

    Returns:
        TerminationResult: Termination state after the tick.
    """
    async with self._lock:
        if self.terminated:
            logger.debug(
                "Simulation step skipped because loop is already terminated at tick %d",
                self.tick,
            )
            return TerminationResult(terminated=True, reason=self.termination_reason or "")

        debug_summary = self._should_log_debug_summary()
        phase_timings_ms: dict[str, float] = {}
        phase_started = time.perf_counter()

        # --------------------------------------------------------
        # Phase 1: Flow-field update (uses current read state)
        # --------------------------------------------------------
        self.env.flow_field = compute_flow_field(
            self.env.plant_energy_layer,
            self.env.toxin_layers,
            self.env.width,
            self.env.height,
        )
        if debug_summary:
            phase_timings_ms["flow_field"] = (time.perf_counter() - phase_started) * 1000.0
            phase_started = time.perf_counter()

        # Apply camouflage attenuations
        for entity in self.world.query(PlantComponent):
            plant: PlantComponent = entity.get_component(PlantComponent)
            if plant.camouflage:
                apply_camouflage(self.env.flow_field, plant.x, plant.y, plant.camouflage_factor)

        # --------------------------------------------------------
        # Phase 2: Lifecycle (grow, connect, reproduce, cull)
        # --------------------------------------------------------
        run_lifecycle(
            self.world,
            self.env,
            self.tick,
            self._flora_params,
            mycorrhizal_connection_cost=self.config.mycorrhizal_connection_cost,
            mycorrhizal_growth_interval_ticks=self.config.mycorrhizal_growth_interval_ticks,
            mycorrhizal_inter_species=self.config.mycorrhizal_inter_species,
        )
        if debug_summary:
            phase_timings_ms["lifecycle"] = (time.perf_counter() - phase_started) * 1000.0
            phase_started = time.perf_counter()

        # --------------------------------------------------------
        # Phase 3: Interaction (movement, feeding, starvation, mitosis)
        # --------------------------------------------------------
        run_interaction(self.world, self.env, self._diet_matrix, self.tick)
        if debug_summary:
            phase_timings_ms["interaction"] = (time.perf_counter() - phase_started) * 1000.0
            phase_started = time.perf_counter()

        # --------------------------------------------------------
        # Phase 4: Signaling (substance synthesis, diffusion, toxins)
        # --------------------------------------------------------
        run_signaling(
            self.world,
            self.env,
            self._trigger_conditions,
            self.config.mycorrhizal_inter_species,
            self.config.mycorrhizal_signal_velocity,
            self.tick,
        )
        if debug_summary:
            phase_timings_ms["signaling"] = (time.perf_counter() - phase_started) * 1000.0
            phase_started = time.perf_counter()

        # --------------------------------------------------------
        # Phase 5: Telemetry
        # --------------------------------------------------------
        self.telemetry.record(self.world, self.tick)
        self.replay.append(self.get_state_snapshot())
        latest_metrics = self.telemetry.get_latest_metrics()
        if debug_summary:
            phase_timings_ms["telemetry_replay"] = (
                time.perf_counter() - phase_started
            ) * 1000.0
            phase_started = time.perf_counter()

        # --------------------------------------------------------
        # Phase 6: Termination check (double-buffer swap happens here
        #          implicitly – all writes committed before check)
        # --------------------------------------------------------
        result = check_termination(
            self.world,
            self.tick,
            max_ticks=self.config.max_ticks,
            z2_flora_species=self.config.z2_flora_species_extinction,
            z4_predator_species=self.config.z4_predator_species_extinction,
            z6_max_flora_energy=self.config.z6_max_total_flora_energy,
            z7_max_predator_population=self.config.z7_max_total_predator_population,
        )
        if debug_summary:
            phase_timings_ms["termination"] = (time.perf_counter() - phase_started) * 1000.0

        self.tick += 1
        if debug_summary:
            self._log_debug_tick_summary(
                latest_metrics=latest_metrics,
                phase_timings_ms=phase_timings_ms,
            )

        if result.terminated:
            self.terminated = True
            self.running = False
            self.termination_reason = result.reason
            logger.info("Simulation terminated at tick %d: %s", self.tick, result.reason)

        return result

stop()

Halt the simulation by clearing the running flag.

Source code in src/phids/engine/loop.py

def stop(self) -> None:
    """Halt the simulation by clearing the running flag."""
    self.running = False
    logger.info("Simulation loop stopped at tick %d", self.tick)

update_wind(vx, vy)

Update the environment uniform wind vector.

Parameters:

Name	Type	Description	Default
vx	float	Wind X component.	required
vy	float	Wind Y component.	required

Source code in src/phids/engine/loop.py

def update_wind(self, vx: float, vy: float) -> None:
    """Update the environment uniform wind vector.

    Args:
        vx: Wind X component.
        vy: Wind Y component.
    """
    self.env.set_uniform_wind(vx, vy)
    logger.info("Simulation wind updated to (vx=%.3f, vy=%.3f)", vx, vy)

ECS components

Plant ECS component dataclass.

This module defines the :class:PlantComponent dataclass which stores the runtime state for a plant entity used by lifecycle and signaling systems.

PlantComponent dataclass

Holds runtime state for a single plant entity.

Attributes:

Name	Type	Description
entity_id	int	ECS entity identifier.
species_id	int	Flora species index.
x,	y	Current grid coordinates.
energy	float	Current energy reserve E_i,j(t).
max_energy	float	Species-specific energy capacity E_max.
base_energy	float	Initial energy used by growth formula.
growth_rate	float	Per-tick growth rate in percent.
survival_threshold	float	Energy threshold below which the plant dies.
reproduction_interval	int	Ticks between reproduction attempts.
seed_min_dist	float	Minimum seed dispersal distance.
seed_max_dist	float	Maximum seed dispersal distance.
seed_energy_cost	float	Energy cost paid for reproduction.
camouflage	bool	Whether constitutive camouflage is active.
camouflage_factor	float	Gradient multiplier when camouflaged.
last_reproduction_tick	int	Tick of the most recent reproduction.
mycorrhizal_connections	set[int]	Set of connected plant entity ids.

Source code in src/phids/engine/components/plant.py

@dataclass(slots=True)
class PlantComponent:
    """Holds runtime state for a single plant entity.

    Attributes:
        entity_id: ECS entity identifier.
        species_id: Flora species index.
        x, y: Current grid coordinates.
        energy: Current energy reserve E_i,j(t).
        max_energy: Species-specific energy capacity E_max.
        base_energy: Initial energy used by growth formula.
        growth_rate: Per-tick growth rate in percent.
        survival_threshold: Energy threshold below which the plant dies.
        reproduction_interval: Ticks between reproduction attempts.
        seed_min_dist: Minimum seed dispersal distance.
        seed_max_dist: Maximum seed dispersal distance.
        seed_energy_cost: Energy cost paid for reproduction.
        camouflage: Whether constitutive camouflage is active.
        camouflage_factor: Gradient multiplier when camouflaged.
        last_reproduction_tick: Tick of the most recent reproduction.
        mycorrhizal_connections: Set of connected plant entity ids.
    """

    entity_id: int
    species_id: int
    x: int
    y: int
    energy: float
    max_energy: float
    base_energy: float
    growth_rate: float
    survival_threshold: float
    reproduction_interval: int
    seed_min_dist: float
    seed_max_dist: float
    seed_energy_cost: float
    camouflage: bool = False
    camouflage_factor: float = 1.0
    last_reproduction_tick: int = 0
    mycorrhizal_connections: set[int] = field(default_factory=set)

Herbivore Swarm ECS component dataclass.

Defines :class:SwarmComponent storing runtime state for a herbivore swarm entity used by interaction and telemetry subsystems.

SwarmComponent dataclass

Holds runtime state for a single herbivore swarm entity.

Attributes:

Name	Type	Description
entity_id	int	ECS entity identifier.
species_id	int	Predator species index.
x,	y	Current grid coordinates.
population	int	Current swarm head-count.
initial_population	int	Head-count at spawn; used for mitosis checks.
energy	float	Current energy reserve.
energy_min	float	Minimum energy per individual.
velocity	int	Movement period in ticks between moves.
consumption_rate	float	Per-tick consumption scalar.
reproduction_energy_divisor	float	Species-level growth throttle.
starvation_ticks	int	Consecutive ticks without feeding.
repelled	bool	Whether the swarm is currently repelled by toxin.
repelled_ticks_remaining	int	Remaining ticks of repelled behavior.
target_plant_id	int	Entity id of the targeted plant (-1 = none).
move_cooldown	int	Ticks remaining until the next movement.

Source code in src/phids/engine/components/swarm.py

@dataclass(slots=True)
class SwarmComponent:
    """Holds runtime state for a single herbivore swarm entity.

    Attributes:
        entity_id: ECS entity identifier.
        species_id: Predator species index.
        x, y: Current grid coordinates.
        population: Current swarm head-count.
        initial_population: Head-count at spawn; used for mitosis checks.
        energy: Current energy reserve.
        energy_min: Minimum energy per individual.
        velocity: Movement period in ticks between moves.
        consumption_rate: Per-tick consumption scalar.
        reproduction_energy_divisor: Species-level growth throttle.
        starvation_ticks: Consecutive ticks without feeding.
        repelled: Whether the swarm is currently repelled by toxin.
        repelled_ticks_remaining: Remaining ticks of repelled behavior.
        target_plant_id: Entity id of the targeted plant (-1 = none).
        move_cooldown: Ticks remaining until the next movement.
    """

    entity_id: int
    species_id: int
    x: int
    y: int
    population: int
    initial_population: int
    energy: float
    energy_min: float
    velocity: int
    consumption_rate: float
    reproduction_energy_divisor: float = 1.0
    starvation_ticks: int = 0
    repelled: bool = False
    repelled_ticks_remaining: int = 0
    target_plant_id: int = -1
    move_cooldown: int = 0

Substance ECS component dataclass.

Defines :class:SubstanceComponent for volatile signals and toxins emitted by plants in response to herbivore presence.

SubstanceComponent dataclass

Holds runtime state for a single substance entity.

A substance represents either a volatile signal (VOC) or a toxin.

Attributes:

Name	Type	Description
entity_id	int	ECS entity identifier.
substance_id	int	Layer index into signal or toxin layers.
owner_plant_id	int	Entity id of the producing plant.
is_toxin	bool	True for toxins, False for signals.
synthesis_duration	int	Configured synthesis duration in ticks.
synthesis_remaining	int	Ticks remaining before activation.
active	bool	Whether the substance is currently active.
aftereffect_ticks	int	Configured aftereffect duration after trigger removal.
aftereffect_remaining_ticks	int	Remaining aftereffect duration at runtime.
lethal	bool	Whether the toxin is lethal.
lethality_rate	float	Individuals eliminated per tick when lethal.
repellent	bool	Whether the toxin repels swarms.
repellent_walk_ticks	int	Duration of repelled random-walk in ticks.
precursor_signal_id	int	Single required precursor signal id (-1 = none). Legacy.
precursor_signal_ids	tuple[int, ...]	All signal ids that must ALL be active before this substance activates (AND logic). Empty tuple = no precursor required.
activation_condition	dict[str, object] | None	Optional nested activation predicate tree stored in JSON-serialisable form for runtime evaluation and tooltip display.
energy_cost_per_tick	float	Energy drained from the owner plant per active tick.
triggered_this_tick	bool	Whether the trigger condition was satisfied in the current signaling pass.

Source code in src/phids/engine/components/substances.py

@dataclass(slots=True)
class SubstanceComponent:
    """Holds runtime state for a single substance entity.

    A substance represents either a volatile signal (VOC) or a toxin.

    Attributes:
        entity_id: ECS entity identifier.
        substance_id: Layer index into signal or toxin layers.
        owner_plant_id: Entity id of the producing plant.
        is_toxin: True for toxins, False for signals.
        synthesis_duration: Configured synthesis duration in ticks.
        synthesis_remaining: Ticks remaining before activation.
        active: Whether the substance is currently active.
        aftereffect_ticks: Configured aftereffect duration after trigger removal.
        aftereffect_remaining_ticks: Remaining aftereffect duration at runtime.
        lethal: Whether the toxin is lethal.
        lethality_rate: Individuals eliminated per tick when lethal.
        repellent: Whether the toxin repels swarms.
        repellent_walk_ticks: Duration of repelled random-walk in ticks.
        precursor_signal_id: Single required precursor signal id (-1 = none). Legacy.
        precursor_signal_ids: All signal ids that must ALL be active before this
            substance activates (AND logic).  Empty tuple = no precursor required.
        activation_condition: Optional nested activation predicate tree stored
            in JSON-serialisable form for runtime evaluation and tooltip display.
        energy_cost_per_tick: Energy drained from the owner plant per active tick.
        triggered_this_tick: Whether the trigger condition was satisfied in the
            current signaling pass.
    """

    entity_id: int
    substance_id: int
    owner_plant_id: int
    is_toxin: bool = False
    synthesis_duration: int = 0
    synthesis_remaining: int = 0
    active: bool = False
    aftereffect_ticks: int = 0
    aftereffect_remaining_ticks: int = 0
    lethal: bool = False
    lethality_rate: float = 0.0
    repellent: bool = False
    repellent_walk_ticks: int = 0
    precursor_signal_id: int = -1
    precursor_signal_ids: tuple[int, ...] = ()
    activation_condition: dict[str, object] | None = None
    energy_cost_per_tick: float = 0.0
    trigger_predator_species_id: int = -1
    trigger_min_predator_population: int = 0
    triggered_this_tick: bool = False

Core runtime structures

GridEnvironment: NumPy-backed biotope with 2-D convolution diffusion.

All cellular automata layers are pre-allocated according to the Rule of 16 and use explicit read/write double-buffering to avoid race conditions when performing per-tick writes.

GridEnvironment

Manage vectorised biotope layers and diffusion helpers.

Parameters:

Name	Type	Description	Default
width	int	Grid width W (1 ≤ W ≤ GRID_W_MAX).	40
height	int	Grid height H (1 ≤ H ≤ GRID_H_MAX).	40
num_signals	int	Number of signal substance layers (1 ≤ n ≤ MAX_SUBSTANCE_TYPES).	4
num_toxins	int	Number of toxin substance layers (1 ≤ n ≤ MAX_SUBSTANCE_TYPES).	4

Source code in src/phids/engine/core/biotope.py

class GridEnvironment:
    """Manage vectorised biotope layers and diffusion helpers.

    Args:
        width: Grid width W (1 ≤ W ≤ GRID_W_MAX).
        height: Grid height H (1 ≤ H ≤ GRID_H_MAX).
        num_signals: Number of signal substance layers
            (1 ≤ n ≤ MAX_SUBSTANCE_TYPES).
        num_toxins: Number of toxin substance layers
            (1 ≤ n ≤ MAX_SUBSTANCE_TYPES).
    """

    def __init__(
        self,
        width: int = 40,
        height: int = 40,
        num_signals: int = 4,
        num_toxins: int = 4,
    ) -> None:
        """Initialise grid layers and double-buffered storage.

        Args:
            width: Grid width in cells.
            height: Grid height in cells.
            num_signals: Number of airborne signal layers.
            num_toxins: Number of toxin layers.
        """
        if not (1 <= width <= GRID_W_MAX):
            raise ValueError(f"width {width} out of range [1, {GRID_W_MAX}].")
        if not (1 <= height <= GRID_H_MAX):
            raise ValueError(f"height {height} out of range [1, {GRID_H_MAX}].")
        if not (1 <= num_signals <= MAX_SUBSTANCE_TYPES):
            raise ValueError(f"num_signals {num_signals} out of range [1, {MAX_SUBSTANCE_TYPES}].")
        if not (1 <= num_toxins <= MAX_SUBSTANCE_TYPES):
            raise ValueError(f"num_toxins {num_toxins} out of range [1, {MAX_SUBSTANCE_TYPES}].")

        self.width = width
        self.height = height
        self.num_signals = num_signals
        self.num_toxins = num_toxins

        shape: tuple[int, int] = (width, height)

        # ------------------------------------------------------------------
        # Plant energy layers (read/write buffers)
        # ------------------------------------------------------------------
        self.plant_energy_layer: npt.NDArray[np.float64] = np.zeros(shape, dtype=np.float64)
        self._plant_energy_layer_write: npt.NDArray[np.float64] = np.zeros(shape, dtype=np.float64)

        # Per-species energy layers (Rule of 16 pre-allocation)
        self.plant_energy_by_species: npt.NDArray[np.float64] = np.zeros(
            (MAX_FLORA_SPECIES, width, height), dtype=np.float64
        )
        self._plant_energy_by_species_write: npt.NDArray[np.float64] = np.zeros_like(
            self.plant_energy_by_species
        )

        # ------------------------------------------------------------------
        # Wind layers (dynamic, updated via REST API)
        # ------------------------------------------------------------------
        self.wind_vector_x: npt.NDArray[np.float64] = np.zeros(shape, dtype=np.float64)
        self.wind_vector_y: npt.NDArray[np.float64] = np.zeros(shape, dtype=np.float64)

        # ------------------------------------------------------------------
        # Signal layers  [num_signals, W, H] – read buffer
        # ------------------------------------------------------------------
        self.signal_layers: npt.NDArray[np.float64] = np.zeros(
            (num_signals, width, height), dtype=np.float64
        )
        # Write buffer for double-buffering
        self._signal_layers_write: npt.NDArray[np.float64] = np.zeros_like(self.signal_layers)

        # ------------------------------------------------------------------
        # Toxin layers  [num_toxins, W, H]
        # ------------------------------------------------------------------
        self.toxin_layers: npt.NDArray[np.float64] = np.zeros(
            (num_toxins, width, height), dtype=np.float64
        )
        self._toxin_layers_write: npt.NDArray[np.float64] = np.zeros_like(self.toxin_layers)

        # ------------------------------------------------------------------
        # Flow-field gradient (scalar attraction field, W×H)
        # ------------------------------------------------------------------
        self.flow_field: npt.NDArray[np.float64] = np.zeros(shape, dtype=np.float64)

    # ------------------------------------------------------------------
    # Wind helpers
    # ------------------------------------------------------------------

    def set_uniform_wind(self, vx: float, vy: float) -> None:
        """Fill wind layers with a spatially uniform vector.

        Args:
            vx: X component of the wind.
            vy: Y component of the wind.
        """
        self.wind_vector_x[:] = vx
        self.wind_vector_y[:] = vy

    def update_wind_at(self, x: int, y: int, vx: float, vy: float) -> None:
        """Update the wind vector at a single grid cell.

        Args:
            x: X coordinate.
            y: Y coordinate.
            vx: X component of the wind.
            vy: Y component of the wind.
        """
        self.wind_vector_x[x, y] = vx
        self.wind_vector_y[x, y] = vy

    # ------------------------------------------------------------------
    # Diffusion
    # ------------------------------------------------------------------

    def diffuse_signals(self) -> None:
        """Compute one diffusion tick for all signal layers.

        This applies a 2-D convolution with a pre-computed Gaussian kernel,
        advects the result by the mean wind vector using integer pixel
        rolls, and applies a sparsity threshold to zero small values.
        """
        # Compute mean wind shift (integer pixel shift for np.roll)
        mean_vx: int = int(round(float(self.wind_vector_x.mean())))
        mean_vy: int = int(round(float(self.wind_vector_y.mean())))

        for s in range(self.num_signals):
            layer: npt.NDArray[np.float64] = self.signal_layers[s]
            convolved: npt.NDArray[np.float64] = convolve2d(
                layer, DIFFUSION_KERNEL, mode="same", boundary="fill", fillvalue=0.0
            )
            # Wind advection via cyclic roll
            shifted: npt.NDArray[np.float64] = np.roll(
                np.roll(convolved, mean_vx, axis=0), mean_vy, axis=1
            )
            shifted *= SIGNAL_DECAY_FACTOR
            # Zero sub-threshold values to preserve matrix sparsity
            shifted[shifted < SIGNAL_EPSILON] = 0.0
            self._signal_layers_write[s] = shifted

        # Swap buffers
        self.signal_layers, self._signal_layers_write = (
            self._signal_layers_write,
            self.signal_layers,
        )

    def diffuse_toxins(self) -> None:
        """Compute one diffusion tick for all toxin layers.

        Toxins diffuse using the same kernel as signals. Small values are
        zeroed using the same epsilon threshold.
        """
        mean_vx: int = int(round(float(self.wind_vector_x.mean())))
        mean_vy: int = int(round(float(self.wind_vector_y.mean())))

        for t in range(self.num_toxins):
            layer: npt.NDArray[np.float64] = self.toxin_layers[t]
            convolved: npt.NDArray[np.float64] = convolve2d(
                layer, DIFFUSION_KERNEL, mode="same", boundary="fill", fillvalue=0.0
            )
            shifted: npt.NDArray[np.float64] = np.roll(
                np.roll(convolved, mean_vx, axis=0), mean_vy, axis=1
            )
            shifted[shifted < SIGNAL_EPSILON] = 0.0
            self._toxin_layers_write[t] = shifted

        self.toxin_layers, self._toxin_layers_write = (
            self._toxin_layers_write,
            self.toxin_layers,
        )

    # ------------------------------------------------------------------
    # Plant energy helpers
    # ------------------------------------------------------------------

    def rebuild_energy_layer(self) -> None:
        """Recompute aggregate plant energy layer and swap buffers.

        Aggregates per-species write buffers into the global write buffer,
        then swaps read/write buffers so that subsequent reads observe the
        newly-written values.
        """
        self._plant_energy_layer_write[:] = self._plant_energy_by_species_write.sum(axis=0)
        self.plant_energy_by_species, self._plant_energy_by_species_write = (
            self._plant_energy_by_species_write,
            self.plant_energy_by_species,
        )
        self.plant_energy_layer, self._plant_energy_layer_write = (
            self._plant_energy_layer_write,
            self.plant_energy_layer,
        )
        self._plant_energy_by_species_write[:] = self.plant_energy_by_species
        self._plant_energy_layer_write[:] = self.plant_energy_layer

    def set_plant_energy(self, x: int, y: int, species_id: int, value: float) -> None:
        """Set a species-specific energy contribution in the write buffer.

        Args:
            x: X coordinate.
            y: Y coordinate.
            species_id: Species index.
            value: Energy contribution (clamped to >= 0).
        """
        self._plant_energy_by_species_write[species_id, x, y] = max(0.0, value)

    def clear_plant_energy(self, x: int, y: int, species_id: int) -> None:
        """Clear a species-specific energy contribution in the write buffer.

        Args:
            x: X coordinate.
            y: Y coordinate.
            species_id: Species index.
        """
        self._plant_energy_by_species_write[species_id, x, y] = 0.0

    # ------------------------------------------------------------------
    # State snapshot (for serialisation / streaming)
    # ------------------------------------------------------------------

    def to_dict(self) -> dict[str, object]:
        """Return a lightweight snapshot dict suitable for msgpack serialisation.

        Returns:
            dict: Mapping containing numpy arrays converted to nested lists.
        """

        return {
            "plant_energy_layer": self.plant_energy_layer.tolist(),
            "signal_layers": self.signal_layers.tolist(),
            "toxin_layers": self.toxin_layers.tolist(),
            "flow_field": self.flow_field.tolist(),
            "wind_vector_x": self.wind_vector_x.tolist(),
            "wind_vector_y": self.wind_vector_y.tolist(),
        }

__init__(width=40, height=40, num_signals=4, num_toxins=4)

Initialise grid layers and double-buffered storage.

Parameters:

Name	Type	Description	Default
width	int	Grid width in cells.	40
height	int	Grid height in cells.	40
num_signals	int	Number of airborne signal layers.	4
num_toxins	int	Number of toxin layers.	4

Source code in src/phids/engine/core/biotope.py

def __init__(
    self,
    width: int = 40,
    height: int = 40,
    num_signals: int = 4,
    num_toxins: int = 4,
) -> None:
    """Initialise grid layers and double-buffered storage.

    Args:
        width: Grid width in cells.
        height: Grid height in cells.
        num_signals: Number of airborne signal layers.
        num_toxins: Number of toxin layers.
    """
    if not (1 <= width <= GRID_W_MAX):
        raise ValueError(f"width {width} out of range [1, {GRID_W_MAX}].")
    if not (1 <= height <= GRID_H_MAX):
        raise ValueError(f"height {height} out of range [1, {GRID_H_MAX}].")
    if not (1 <= num_signals <= MAX_SUBSTANCE_TYPES):
        raise ValueError(f"num_signals {num_signals} out of range [1, {MAX_SUBSTANCE_TYPES}].")
    if not (1 <= num_toxins <= MAX_SUBSTANCE_TYPES):
        raise ValueError(f"num_toxins {num_toxins} out of range [1, {MAX_SUBSTANCE_TYPES}].")

    self.width = width
    self.height = height
    self.num_signals = num_signals
    self.num_toxins = num_toxins

    shape: tuple[int, int] = (width, height)

    # ------------------------------------------------------------------
    # Plant energy layers (read/write buffers)
    # ------------------------------------------------------------------
    self.plant_energy_layer: npt.NDArray[np.float64] = np.zeros(shape, dtype=np.float64)
    self._plant_energy_layer_write: npt.NDArray[np.float64] = np.zeros(shape, dtype=np.float64)

    # Per-species energy layers (Rule of 16 pre-allocation)
    self.plant_energy_by_species: npt.NDArray[np.float64] = np.zeros(
        (MAX_FLORA_SPECIES, width, height), dtype=np.float64
    )
    self._plant_energy_by_species_write: npt.NDArray[np.float64] = np.zeros_like(
        self.plant_energy_by_species
    )

    # ------------------------------------------------------------------
    # Wind layers (dynamic, updated via REST API)
    # ------------------------------------------------------------------
    self.wind_vector_x: npt.NDArray[np.float64] = np.zeros(shape, dtype=np.float64)
    self.wind_vector_y: npt.NDArray[np.float64] = np.zeros(shape, dtype=np.float64)

    # ------------------------------------------------------------------
    # Signal layers  [num_signals, W, H] – read buffer
    # ------------------------------------------------------------------
    self.signal_layers: npt.NDArray[np.float64] = np.zeros(
        (num_signals, width, height), dtype=np.float64
    )
    # Write buffer for double-buffering
    self._signal_layers_write: npt.NDArray[np.float64] = np.zeros_like(self.signal_layers)

    # ------------------------------------------------------------------
    # Toxin layers  [num_toxins, W, H]
    # ------------------------------------------------------------------
    self.toxin_layers: npt.NDArray[np.float64] = np.zeros(
        (num_toxins, width, height), dtype=np.float64
    )
    self._toxin_layers_write: npt.NDArray[np.float64] = np.zeros_like(self.toxin_layers)

    # ------------------------------------------------------------------
    # Flow-field gradient (scalar attraction field, W×H)
    # ------------------------------------------------------------------
    self.flow_field: npt.NDArray[np.float64] = np.zeros(shape, dtype=np.float64)

clear_plant_energy(x, y, species_id)

Clear a species-specific energy contribution in the write buffer.

Parameters:

Name	Type	Description	Default
x	int	X coordinate.	required
y	int	Y coordinate.	required
species_id	int	Species index.	required

Source code in src/phids/engine/core/biotope.py

def clear_plant_energy(self, x: int, y: int, species_id: int) -> None:
    """Clear a species-specific energy contribution in the write buffer.

    Args:
        x: X coordinate.
        y: Y coordinate.
        species_id: Species index.
    """
    self._plant_energy_by_species_write[species_id, x, y] = 0.0

diffuse_signals()

Compute one diffusion tick for all signal layers.

This applies a 2-D convolution with a pre-computed Gaussian kernel, advects the result by the mean wind vector using integer pixel rolls, and applies a sparsity threshold to zero small values.

Source code in src/phids/engine/core/biotope.py

def diffuse_signals(self) -> None:
    """Compute one diffusion tick for all signal layers.

    This applies a 2-D convolution with a pre-computed Gaussian kernel,
    advects the result by the mean wind vector using integer pixel
    rolls, and applies a sparsity threshold to zero small values.
    """
    # Compute mean wind shift (integer pixel shift for np.roll)
    mean_vx: int = int(round(float(self.wind_vector_x.mean())))
    mean_vy: int = int(round(float(self.wind_vector_y.mean())))

    for s in range(self.num_signals):
        layer: npt.NDArray[np.float64] = self.signal_layers[s]
        convolved: npt.NDArray[np.float64] = convolve2d(
            layer, DIFFUSION_KERNEL, mode="same", boundary="fill", fillvalue=0.0
        )
        # Wind advection via cyclic roll
        shifted: npt.NDArray[np.float64] = np.roll(
            np.roll(convolved, mean_vx, axis=0), mean_vy, axis=1
        )
        shifted *= SIGNAL_DECAY_FACTOR
        # Zero sub-threshold values to preserve matrix sparsity
        shifted[shifted < SIGNAL_EPSILON] = 0.0
        self._signal_layers_write[s] = shifted

    # Swap buffers
    self.signal_layers, self._signal_layers_write = (
        self._signal_layers_write,
        self.signal_layers,
    )

diffuse_toxins()

Compute one diffusion tick for all toxin layers.

Toxins diffuse using the same kernel as signals. Small values are zeroed using the same epsilon threshold.

Source code in src/phids/engine/core/biotope.py

def diffuse_toxins(self) -> None:
    """Compute one diffusion tick for all toxin layers.

    Toxins diffuse using the same kernel as signals. Small values are
    zeroed using the same epsilon threshold.
    """
    mean_vx: int = int(round(float(self.wind_vector_x.mean())))
    mean_vy: int = int(round(float(self.wind_vector_y.mean())))

    for t in range(self.num_toxins):
        layer: npt.NDArray[np.float64] = self.toxin_layers[t]
        convolved: npt.NDArray[np.float64] = convolve2d(
            layer, DIFFUSION_KERNEL, mode="same", boundary="fill", fillvalue=0.0
        )
        shifted: npt.NDArray[np.float64] = np.roll(
            np.roll(convolved, mean_vx, axis=0), mean_vy, axis=1
        )
        shifted[shifted < SIGNAL_EPSILON] = 0.0
        self._toxin_layers_write[t] = shifted

    self.toxin_layers, self._toxin_layers_write = (
        self._toxin_layers_write,
        self.toxin_layers,
    )

rebuild_energy_layer()

Recompute aggregate plant energy layer and swap buffers.

Aggregates per-species write buffers into the global write buffer, then swaps read/write buffers so that subsequent reads observe the newly-written values.

Source code in src/phids/engine/core/biotope.py

def rebuild_energy_layer(self) -> None:
    """Recompute aggregate plant energy layer and swap buffers.

    Aggregates per-species write buffers into the global write buffer,
    then swaps read/write buffers so that subsequent reads observe the
    newly-written values.
    """
    self._plant_energy_layer_write[:] = self._plant_energy_by_species_write.sum(axis=0)
    self.plant_energy_by_species, self._plant_energy_by_species_write = (
        self._plant_energy_by_species_write,
        self.plant_energy_by_species,
    )
    self.plant_energy_layer, self._plant_energy_layer_write = (
        self._plant_energy_layer_write,
        self.plant_energy_layer,
    )
    self._plant_energy_by_species_write[:] = self.plant_energy_by_species
    self._plant_energy_layer_write[:] = self.plant_energy_layer

set_plant_energy(x, y, species_id, value)

Set a species-specific energy contribution in the write buffer.

Parameters:

Name	Type	Description	Default
x	int	X coordinate.	required
y	int	Y coordinate.	required
species_id	int	Species index.	required
value	float	Energy contribution (clamped to >= 0).	required

Source code in src/phids/engine/core/biotope.py

def set_plant_energy(self, x: int, y: int, species_id: int, value: float) -> None:
    """Set a species-specific energy contribution in the write buffer.

    Args:
        x: X coordinate.
        y: Y coordinate.
        species_id: Species index.
        value: Energy contribution (clamped to >= 0).
    """
    self._plant_energy_by_species_write[species_id, x, y] = max(0.0, value)

set_uniform_wind(vx, vy)

Fill wind layers with a spatially uniform vector.

Parameters:

Name	Type	Description	Default
vx	float	X component of the wind.	required
vy	float	Y component of the wind.	required

Source code in src/phids/engine/core/biotope.py

def set_uniform_wind(self, vx: float, vy: float) -> None:
    """Fill wind layers with a spatially uniform vector.

    Args:
        vx: X component of the wind.
        vy: Y component of the wind.
    """
    self.wind_vector_x[:] = vx
    self.wind_vector_y[:] = vy

to_dict()

Return a lightweight snapshot dict suitable for msgpack serialisation.

Returns:

Name	Type	Description
dict	dict[str, object]	Mapping containing numpy arrays converted to nested lists.

Source code in src/phids/engine/core/biotope.py

def to_dict(self) -> dict[str, object]:
    """Return a lightweight snapshot dict suitable for msgpack serialisation.

    Returns:
        dict: Mapping containing numpy arrays converted to nested lists.
    """

    return {
        "plant_energy_layer": self.plant_energy_layer.tolist(),
        "signal_layers": self.signal_layers.tolist(),
        "toxin_layers": self.toxin_layers.tolist(),
        "flow_field": self.flow_field.tolist(),
        "wind_vector_x": self.wind_vector_x.tolist(),
        "wind_vector_y": self.wind_vector_y.tolist(),
    }

update_wind_at(x, y, vx, vy)

Update the wind vector at a single grid cell.

Parameters:

Name	Type	Description	Default
x	int	X coordinate.	required
y	int	Y coordinate.	required
vx	float	X component of the wind.	required
vy	float	Y component of the wind.	required

Source code in src/phids/engine/core/biotope.py

def update_wind_at(self, x: int, y: int, vx: float, vy: float) -> None:
    """Update the wind vector at a single grid cell.

    Args:
        x: X coordinate.
        y: Y coordinate.
        vx: X component of the wind.
        vy: Y component of the wind.
    """
    self.wind_vector_x[x, y] = vx
    self.wind_vector_y[x, y] = vy

Entity-Component-System (ECS) registry with O(1) spatial hash support.

The ECS maintains a flat entity registry, a per-component index for fast queries, and a spatial hash (grid cell roster) enabling O(1) membership lookups for entities occupying a cell.

ECSWorld

Central ECS registry managing entities, components and spatial hash.

The world provides helpers for entity lifecycle, component indexing and a spatial hash grid for efficient cell membership queries.

Source code in src/phids/engine/core/ecs.py

class ECSWorld:
    """Central ECS registry managing entities, components and spatial hash.

    The world provides helpers for entity lifecycle, component indexing and
    a spatial hash grid for efficient cell membership queries.
    """

    def __init__(self) -> None:
        """Initialise the ECS world and its internal indices.

        Attributes initialized:
            _next_id: Next entity id to allocate.
            _entities: Mapping of entity id to Entity.
            _component_index: Index mapping component types to entity id sets.
            _spatial_hash: Grid cell roster mapping (x, y) to entity id sets.
        """
        self._next_id: int = 0
        self._entities: dict[int, Entity] = {}
        # component_type -> set of entity ids
        self._component_index: dict[type[Any], set[int]] = defaultdict(set)
        # (x, y) -> set of entity ids (Spatial Hash / Grid Cell Roster)
        self._spatial_hash: dict[tuple[int, int], set[int]] = defaultdict(set)

    # ------------------------------------------------------------------
    # Entity lifecycle
    # ------------------------------------------------------------------

    def create_entity(self) -> Entity:
        """Allocate and register a new entity.

        Returns:
            Entity: Newly created entity object.
        """
        eid = self._next_id
        self._next_id += 1
        entity = Entity(entity_id=eid)
        self._entities[eid] = entity
        return entity

    def destroy_entity(self, entity_id: int) -> None:
        """Remove an entity and clean up index and spatial hash references.

        Args:
            entity_id: Identifier of the entity to destroy.
        """
        entity = self._entities.pop(entity_id, None)
        if entity is None:
            return
        # Clean component index
        for ctype in list(entity._components.keys()):
            self._component_index[ctype].discard(entity_id)
        # Clean spatial hash (search all cells – acceptable for sparse grids)
        for cell_set in self._spatial_hash.values():
            cell_set.discard(entity_id)

    def has_entity(self, entity_id: int) -> bool:
        """Return True if the entity exists.

        Args:
            entity_id: Entity identifier.

        Returns:
            bool: True if present.
        """
        return entity_id in self._entities

    def get_entity(self, entity_id: int) -> Entity:
        """Return the entity instance for the given id.

        Args:
            entity_id: Entity identifier.

        Returns:
            Entity: Matching entity.
        """
        return self._entities[entity_id]

    # ------------------------------------------------------------------
    # Component helpers
    # ------------------------------------------------------------------

    def add_component(self, entity_id: int, component: Any) -> None:
        """Attach a component to an entity and update the component index.

        Args:
            entity_id: Target entity id.
            component: Component instance to attach.
        """
        entity = self._entities[entity_id]
        entity.add_component(component)
        self._component_index[type(component)].add(entity_id)

    def remove_component(self, entity_id: int, component_type: type[Any]) -> None:
        """Detach a component of the specified type from an entity.

        Args:
            entity_id: Target entity id.
            component_type: Component class/type to remove.
        """
        entity = self._entities[entity_id]
        entity.remove_component(component_type)
        self._component_index[component_type].discard(entity_id)

    def query(self, *component_types: type[Any]) -> Iterator[Entity]:
        """Yield all entities that possess all listed component types.

        Args:
            *component_types: Component classes/types to require.

        Yields:
            Entity: Entities matching the component set.
        """
        if not component_types:
            yield from self._entities.values()
            return
        # Start from the smallest set for efficiency
        sets = [self._component_index.get(ct, set()) for ct in component_types]
        smallest = min(sets, key=len)
        for eid in list(smallest):
            entity = self._entities.get(eid)
            if entity is None:
                continue
            if all(entity.has_component(ct) for ct in component_types):
                yield entity

    # ------------------------------------------------------------------
    # Spatial Hash
    # ------------------------------------------------------------------

    def register_position(self, entity_id: int, x: int, y: int) -> None:
        """Register an entity at grid cell (x, y).

        Args:
            entity_id: Entity identifier.
            x: X coordinate of the cell.
            y: Y coordinate of the cell.
        """
        self._spatial_hash[(x, y)].add(entity_id)

    def unregister_position(self, entity_id: int, x: int, y: int) -> None:
        """Remove an entity from a grid cell.

        Args:
            entity_id: Entity identifier.
            x: X coordinate of the cell.
            y: Y coordinate of the cell.
        """
        cell = self._spatial_hash.get((x, y))
        if cell is not None:
            cell.discard(entity_id)

    def move_entity(self, entity_id: int, old_x: int, old_y: int, new_x: int, new_y: int) -> None:
        """Atomically update spatial hash when an entity moves.

        Args:
            entity_id: Entity identifier.
            old_x: Previous X coordinate.
            old_y: Previous Y coordinate.
            new_x: New X coordinate.
            new_y: New Y coordinate.
        """
        self.unregister_position(entity_id, old_x, old_y)
        self.register_position(entity_id, new_x, new_y)

    def entities_at(self, x: int, y: int) -> set[int]:
        """Return the set of entity ids occupying a cell.

        Args:
            x: X coordinate.
            y: Y coordinate.

        Returns:
            set[int]: Entity ids occupying the cell.
        """
        return self._spatial_hash.get((x, y), set())

    # ------------------------------------------------------------------
    # Garbage collection
    # ------------------------------------------------------------------

    def collect_garbage(self, dead_entity_ids: list[int]) -> None:
        """Bulk destroy a list of dead entities.

        Args:
            dead_entity_ids: List of entity ids to remove.
        """
        for eid in dead_entity_ids:
            self.destroy_entity(eid)

__init__()

Initialise the ECS world and its internal indices.

Attributes initialized

_next_id: Next entity id to allocate. _entities: Mapping of entity id to Entity. _component_index: Index mapping component types to entity id sets. _spatial_hash: Grid cell roster mapping (x, y) to entity id sets.

Source code in src/phids/engine/core/ecs.py

def __init__(self) -> None:
    """Initialise the ECS world and its internal indices.

    Attributes initialized:
        _next_id: Next entity id to allocate.
        _entities: Mapping of entity id to Entity.
        _component_index: Index mapping component types to entity id sets.
        _spatial_hash: Grid cell roster mapping (x, y) to entity id sets.
    """
    self._next_id: int = 0
    self._entities: dict[int, Entity] = {}
    # component_type -> set of entity ids
    self._component_index: dict[type[Any], set[int]] = defaultdict(set)
    # (x, y) -> set of entity ids (Spatial Hash / Grid Cell Roster)
    self._spatial_hash: dict[tuple[int, int], set[int]] = defaultdict(set)

add_component(entity_id, component)

Attach a component to an entity and update the component index.

Parameters:

Name	Type	Description	Default
entity_id	int	Target entity id.	required
component	Any	Component instance to attach.	required

Source code in src/phids/engine/core/ecs.py

def add_component(self, entity_id: int, component: Any) -> None:
    """Attach a component to an entity and update the component index.

    Args:
        entity_id: Target entity id.
        component: Component instance to attach.
    """
    entity = self._entities[entity_id]
    entity.add_component(component)
    self._component_index[type(component)].add(entity_id)

collect_garbage(dead_entity_ids)

Bulk destroy a list of dead entities.

Parameters:

Name	Type	Description	Default
dead_entity_ids	list[int]	List of entity ids to remove.	required

Source code in src/phids/engine/core/ecs.py

def collect_garbage(self, dead_entity_ids: list[int]) -> None:
    """Bulk destroy a list of dead entities.

    Args:
        dead_entity_ids: List of entity ids to remove.
    """
    for eid in dead_entity_ids:
        self.destroy_entity(eid)

create_entity()

Allocate and register a new entity.

Returns:

Name	Type	Description
Entity	Entity	Newly created entity object.

Source code in src/phids/engine/core/ecs.py

def create_entity(self) -> Entity:
    """Allocate and register a new entity.

    Returns:
        Entity: Newly created entity object.
    """
    eid = self._next_id
    self._next_id += 1
    entity = Entity(entity_id=eid)
    self._entities[eid] = entity
    return entity

destroy_entity(entity_id)

Remove an entity and clean up index and spatial hash references.

Parameters:

Name	Type	Description	Default
entity_id	int	Identifier of the entity to destroy.	required

Source code in src/phids/engine/core/ecs.py

def destroy_entity(self, entity_id: int) -> None:
    """Remove an entity and clean up index and spatial hash references.

    Args:
        entity_id: Identifier of the entity to destroy.
    """
    entity = self._entities.pop(entity_id, None)
    if entity is None:
        return
    # Clean component index
    for ctype in list(entity._components.keys()):
        self._component_index[ctype].discard(entity_id)
    # Clean spatial hash (search all cells – acceptable for sparse grids)
    for cell_set in self._spatial_hash.values():
        cell_set.discard(entity_id)

entities_at(x, y)

Return the set of entity ids occupying a cell.

Parameters:

Name	Type	Description	Default
x	int	X coordinate.	required
y	int	Y coordinate.	required

Returns:

Type	Description
set[int]	set[int]: Entity ids occupying the cell.

Source code in src/phids/engine/core/ecs.py

def entities_at(self, x: int, y: int) -> set[int]:
    """Return the set of entity ids occupying a cell.

    Args:
        x: X coordinate.
        y: Y coordinate.

    Returns:
        set[int]: Entity ids occupying the cell.
    """
    return self._spatial_hash.get((x, y), set())

get_entity(entity_id)

Return the entity instance for the given id.

Parameters:

Name	Type	Description	Default
entity_id	int	Entity identifier.	required

Returns:

Name	Type	Description
Entity	Entity	Matching entity.

Source code in src/phids/engine/core/ecs.py

def get_entity(self, entity_id: int) -> Entity:
    """Return the entity instance for the given id.

    Args:
        entity_id: Entity identifier.

    Returns:
        Entity: Matching entity.
    """
    return self._entities[entity_id]

has_entity(entity_id)

Return True if the entity exists.

Parameters:

Name	Type	Description	Default
entity_id	int	Entity identifier.	required

Returns:

Name	Type	Description
bool	bool	True if present.

Source code in src/phids/engine/core/ecs.py

def has_entity(self, entity_id: int) -> bool:
    """Return True if the entity exists.

    Args:
        entity_id: Entity identifier.

    Returns:
        bool: True if present.
    """
    return entity_id in self._entities

move_entity(entity_id, old_x, old_y, new_x, new_y)

Atomically update spatial hash when an entity moves.

Parameters:

Name	Type	Description	Default
entity_id	int	Entity identifier.	required
old_x	int	Previous X coordinate.	required
old_y	int	Previous Y coordinate.	required
new_x	int	New X coordinate.	required
new_y	int	New Y coordinate.	required

Source code in src/phids/engine/core/ecs.py

def move_entity(self, entity_id: int, old_x: int, old_y: int, new_x: int, new_y: int) -> None:
    """Atomically update spatial hash when an entity moves.

    Args:
        entity_id: Entity identifier.
        old_x: Previous X coordinate.
        old_y: Previous Y coordinate.
        new_x: New X coordinate.
        new_y: New Y coordinate.
    """
    self.unregister_position(entity_id, old_x, old_y)
    self.register_position(entity_id, new_x, new_y)

query(*component_types)

Yield all entities that possess all listed component types.

Parameters:

Name	Type	Description	Default
*component_types	type[Any]	Component classes/types to require.	()

Yields:

Name	Type	Description
Entity	Entity	Entities matching the component set.

Source code in src/phids/engine/core/ecs.py

def query(self, *component_types: type[Any]) -> Iterator[Entity]:
    """Yield all entities that possess all listed component types.

    Args:
        *component_types: Component classes/types to require.

    Yields:
        Entity: Entities matching the component set.
    """
    if not component_types:
        yield from self._entities.values()
        return
    # Start from the smallest set for efficiency
    sets = [self._component_index.get(ct, set()) for ct in component_types]
    smallest = min(sets, key=len)
    for eid in list(smallest):
        entity = self._entities.get(eid)
        if entity is None:
            continue
        if all(entity.has_component(ct) for ct in component_types):
            yield entity

register_position(entity_id, x, y)

Register an entity at grid cell (x, y).

Parameters:

Name	Type	Description	Default
entity_id	int	Entity identifier.	required
x	int	X coordinate of the cell.	required
y	int	Y coordinate of the cell.	required

Source code in src/phids/engine/core/ecs.py

def register_position(self, entity_id: int, x: int, y: int) -> None:
    """Register an entity at grid cell (x, y).

    Args:
        entity_id: Entity identifier.
        x: X coordinate of the cell.
        y: Y coordinate of the cell.
    """
    self._spatial_hash[(x, y)].add(entity_id)

remove_component(entity_id, component_type)

Detach a component of the specified type from an entity.

Parameters:

Name	Type	Description	Default
entity_id	int	Target entity id.	required
component_type	type[Any]	Component class/type to remove.	required

Source code in src/phids/engine/core/ecs.py

def remove_component(self, entity_id: int, component_type: type[Any]) -> None:
    """Detach a component of the specified type from an entity.

    Args:
        entity_id: Target entity id.
        component_type: Component class/type to remove.
    """
    entity = self._entities[entity_id]
    entity.remove_component(component_type)
    self._component_index[component_type].discard(entity_id)

unregister_position(entity_id, x, y)

Remove an entity from a grid cell.

Parameters:

Name	Type	Description	Default
entity_id	int	Entity identifier.	required
x	int	X coordinate of the cell.	required
y	int	Y coordinate of the cell.	required

Source code in src/phids/engine/core/ecs.py

def unregister_position(self, entity_id: int, x: int, y: int) -> None:
    """Remove an entity from a grid cell.

    Args:
        entity_id: Entity identifier.
        x: X coordinate of the cell.
        y: Y coordinate of the cell.
    """
    cell = self._spatial_hash.get((x, y))
    if cell is not None:
        cell.discard(entity_id)

Entity dataclass

Lightweight wrapper holding an entity id and attached components.

Source code in src/phids/engine/core/ecs.py

@dataclass(slots=True)
class Entity:
    """Lightweight wrapper holding an entity id and attached components."""

    entity_id: int
    _components: dict[type[Any], Any] = field(default_factory=dict, repr=False)

    def add_component(self, component: Any) -> None:
        """Attach a component instance keyed by its type.

        Args:
            component: Component instance to attach.
        """
        self._components[type(component)] = component

    def get_component(self, component_type: type[C]) -> C:
        """Return attached component of the given type.

        Args:
            component_type: The component class/type to retrieve.

        Returns:
            The component instance for the entity.
        """
        return self._components[component_type]  # type: ignore[return-value]

    def has_component(self, component_type: type[Any]) -> bool:
        """Return True if the entity has a component of the given type.

        Args:
            component_type: Component class/type to check for.

        Returns:
            bool: True if present, False otherwise.
        """
        return component_type in self._components

    def remove_component(self, component_type: type[Any]) -> None:
        """Detach a component of the given type (no-op if absent).

        Args:
            component_type: Component class/type to remove.
        """
        self._components.pop(component_type, None)

add_component(component)

Attach a component instance keyed by its type.

Parameters:

Name	Type	Description	Default
component	Any	Component instance to attach.	required

Source code in src/phids/engine/core/ecs.py

def add_component(self, component: Any) -> None:
    """Attach a component instance keyed by its type.

    Args:
        component: Component instance to attach.
    """
    self._components[type(component)] = component

get_component(component_type)

Return attached component of the given type.

Parameters:

Name	Type	Description	Default
component_type	type[C]	The component class/type to retrieve.	required

Returns:

Type	Description
C	The component instance for the entity.

Source code in src/phids/engine/core/ecs.py

def get_component(self, component_type: type[C]) -> C:
    """Return attached component of the given type.

    Args:
        component_type: The component class/type to retrieve.

    Returns:
        The component instance for the entity.
    """
    return self._components[component_type]  # type: ignore[return-value]

has_component(component_type)

Return True if the entity has a component of the given type.

Parameters:

Name	Type	Description	Default
component_type	type[Any]	Component class/type to check for.	required

Returns:

Name	Type	Description
bool	bool	True if present, False otherwise.

Source code in src/phids/engine/core/ecs.py

def has_component(self, component_type: type[Any]) -> bool:
    """Return True if the entity has a component of the given type.

    Args:
        component_type: Component class/type to check for.

    Returns:
        bool: True if present, False otherwise.
    """
    return component_type in self._components

remove_component(component_type)

Detach a component of the given type (no-op if absent).

Parameters:

Name	Type	Description	Default
component_type	type[Any]	Component class/type to remove.	required

Source code in src/phids/engine/core/ecs.py

def remove_component(self, component_type: type[Any]) -> None:
    """Detach a component of the given type (no-op if absent).

    Args:
        component_type: Component class/type to remove.
    """
    self._components.pop(component_type, None)

Flow-field gradient generation accelerated with Numba @njit.

The global attraction gradient is computed by combining plant attraction and toxin repulsion, then propagating values to neighbours. The scalar field is intended to populate :class:GridEnvironment.flow_field.

apply_camouflage(flow_field, x, y, factor)

Attenuate the flow-field gradient at cell (x, y) in-place.

Parameters:

Name	Type	Description	Default
flow_field	NDArray[float64]	Mutable gradient array (W, H).	required
x	int	X coordinate.	required
y	int	Y coordinate.	required
factor	float	Multiplier in [0, 1]; 0 = invisible, 1 = no attenuation.	required

Source code in src/phids/engine/core/flow_field.py

def apply_camouflage(
    flow_field: npt.NDArray[np.float64],
    x: int,
    y: int,
    factor: float,
) -> None:
    """Attenuate the flow-field gradient at cell (x, y) in-place.

    Args:
        flow_field: Mutable gradient array ``(W, H)``.
        x: X coordinate.
        y: Y coordinate.
        factor: Multiplier in [0, 1]; 0 = invisible, 1 = no attenuation.
    """
    flow_field[x, y] *= factor

compute_flow_field(plant_energy, toxin_layers, width, height)

Public wrapper: sum toxin layers and delegate to the Numba kernel.

Parameters:

Name	Type	Description	Default
plant_energy	NDArray[float64]	Shape (W, H) aggregate plant energy.	required
toxin_layers	NDArray[float64]	Shape (num_toxins, W, H) toxin concentration layers.	required
width	int	Grid width.	required
height	int	Grid height.	required

Returns:

Type	Description
NDArray[float64]	npt.NDArray[np.float64]: Flow-field gradient of shape (W, H).

Source code in src/phids/engine/core/flow_field.py

def compute_flow_field(
    plant_energy: npt.NDArray[np.float64],
    toxin_layers: npt.NDArray[np.float64],
    width: int,
    height: int,
) -> npt.NDArray[np.float64]:
    """Public wrapper: sum toxin layers and delegate to the Numba kernel.

    Args:
        plant_energy: Shape ``(W, H)`` aggregate plant energy.
        toxin_layers: Shape ``(num_toxins, W, H)`` toxin concentration layers.
        width: Grid width.
        height: Grid height.

    Returns:
        npt.NDArray[np.float64]: Flow-field gradient of shape ``(W, H)``.
    """
    toxin_sum: npt.NDArray[np.float64] = toxin_layers.sum(axis=0)
    return _compute_flow_field(plant_energy, toxin_sum, width, height)

Engine systems

Lifecycle system: plant growth, reproduction, mycorrhizal networking and death.

This module implements per-tick plant updates including growth according to species parameters, reproduction attempts, establishment of symbiotic root connections, and culling of dead plants. It should run before interaction and signaling phases.

run_lifecycle(world, env, tick, flora_species_params, mycorrhizal_connection_cost=1.0, mycorrhizal_growth_interval_ticks=8, mycorrhizal_inter_species=False)

Execute one lifecycle tick: grow, connect, reproduce, and cull.

Parameters:

Name	Type	Description	Default
world	ECSWorld	The ECS world registry.	required
env	GridEnvironment	The GridEnvironment instance.	required
tick	int	Current simulation tick index.	required
flora_species_params	dict[int, object]	Mapping of species_id to species parameters.	required
mycorrhizal_connection_cost	float	Energy cost per new root connection.	1.0
mycorrhizal_growth_interval_ticks	int	Ticks between new root-growth attempts. At most one new link is created per attempt.	8
mycorrhizal_inter_species	bool	Allow inter-species root connections.	False

Source code in src/phids/engine/systems/lifecycle.py

def run_lifecycle(
    world: ECSWorld,
    env: GridEnvironment,
    tick: int,
    flora_species_params: dict[int, object],
    mycorrhizal_connection_cost: float = 1.0,
    mycorrhizal_growth_interval_ticks: int = 8,
    mycorrhizal_inter_species: bool = False,
) -> None:
    """Execute one lifecycle tick: grow, connect, reproduce, and cull.

    Args:
        world: The ECS world registry.
        env: The GridEnvironment instance.
        tick: Current simulation tick index.
        flora_species_params: Mapping of species_id to species parameters.
        mycorrhizal_connection_cost: Energy cost per new root connection.
        mycorrhizal_growth_interval_ticks: Ticks between new root-growth
            attempts. At most one new link is created per attempt.
        mycorrhizal_inter_species: Allow inter-species root connections.
    """
    dead: list[int] = []

    for entity in list(world.query(PlantComponent)):
        plant: PlantComponent = entity.get_component(PlantComponent)

        # Growth
        _grow(plant, tick)

        # Reproduction
        _attempt_reproduction(plant, tick, world, env, flora_species_params)

        # Update biotope energy
        env.set_plant_energy(plant.x, plant.y, plant.species_id, plant.energy)

        # Prune dead mycorrhizal links
        plant.mycorrhizal_connections = {
            eid for eid in plant.mycorrhizal_connections if world.has_entity(eid)
        }

        # Survival check
        if plant.energy < plant.survival_threshold:
            env.clear_plant_energy(plant.x, plant.y, plant.species_id)
            world.unregister_position(entity.entity_id, plant.x, plant.y)
            dead.append(entity.entity_id)

    # Establish new mycorrhizal root connections between adjacent plants
    if _should_attempt_mycorrhizal_growth(tick, mycorrhizal_growth_interval_ticks):
        _establish_mycorrhizal_connections(
            world,
            env,
            mycorrhizal_connection_cost,
            mycorrhizal_inter_species,
            excluded_entity_ids=set(dead),
        )

    world.collect_garbage(dead)
    env.rebuild_energy_layer()

Interaction system: swarm movement, feeding, mitosis and toxin effects.

This module implements swarm behaviour including movement (gradient navigation or random walk), feeding using the diet compatibility matrix, starvation attrition, mitosis and application of toxin effects. Spatial hash lookups provide O(1) co-occupancy checks.

run_interaction(world, env, diet_matrix, tick)

Execute one interaction tick for all swarm entities.

The routine performs movement, feeding using the diet matrix, toxin damage, starvation attrition, and mitosis, then collects dead swarms and rebuilds the plant energy layer.

Parameters:

Name	Type	Description	Default
world	ECSWorld	ECSWorld registry.	required
env	GridEnvironment	GridEnvironment instance (provides flow_field and toxin layers).	required
diet_matrix	list[list[bool]]	Compatibility matrix indexed by predator_id then flora_id.	required
tick	int	Current simulation tick (reserved for future use).	required

Source code in src/phids/engine/systems/interaction.py

def run_interaction(
    world: ECSWorld,
    env: GridEnvironment,
    diet_matrix: list[list[bool]],
    tick: int,
) -> None:
    """Execute one interaction tick for all swarm entities.

    The routine performs movement, feeding using the diet matrix, toxin
    damage, starvation attrition, and mitosis, then collects dead swarms
    and rebuilds the plant energy layer.

    Args:
        world: ECSWorld registry.
        env: GridEnvironment instance (provides flow_field and toxin layers).
        diet_matrix: Compatibility matrix indexed by predator_id then flora_id.
        tick: Current simulation tick (reserved for future use).
    """
    dead_swarms: list[int] = []
    for entity in list(world.query(SwarmComponent)):
        swarm: SwarmComponent = entity.get_component(SwarmComponent)

        # ----------------------------------------------------------------
        # 1. Movement cooldown
        # ----------------------------------------------------------------
        if swarm.move_cooldown > 0:
            swarm.move_cooldown -= 1
        else:
            # ----------------------------------------------------------------
            # 2. Navigate
            # ----------------------------------------------------------------
            old_x, old_y = swarm.x, swarm.y

            if swarm.repelled and swarm.repelled_ticks_remaining > 0:
                nx, ny = _random_walk_step(swarm.x, swarm.y, env.width, env.height)
                swarm.repelled_ticks_remaining -= 1
                if swarm.repelled_ticks_remaining <= 0:
                    swarm.repelled = False
                    swarm.target_plant_id = -1
            else:
                nx, ny = _choose_neighbour_by_flow_probability(
                    swarm.x,
                    swarm.y,
                    env.flow_field,
                    env.width,
                    env.height,
                )

            if (nx, ny) != (old_x, old_y):
                world.move_entity(entity.entity_id, old_x, old_y, nx, ny)
                swarm.x, swarm.y = nx, ny

            swarm.move_cooldown = swarm.velocity - 1

        # ----------------------------------------------------------------
        # 3. Feeding – check co-located plants via spatial hash
        # ----------------------------------------------------------------
        fed = False
        for co_eid in list(world.entities_at(swarm.x, swarm.y)):
            co_entity = world.get_entity(co_eid)
            if not co_entity.has_component(PlantComponent):
                continue
            plant: PlantComponent = co_entity.get_component(PlantComponent)

            # Diet compatibility check
            pred_row = diet_matrix[swarm.species_id] if swarm.species_id < len(diet_matrix) else []
            if not (plant.species_id < len(pred_row) and pred_row[plant.species_id]):
                continue

            consumed = min(swarm.consumption_rate * swarm.population, plant.energy)
            plant.energy -= consumed
            env.set_plant_energy(plant.x, plant.y, plant.species_id, plant.energy)
            swarm.energy += consumed
            swarm.starvation_ticks = 0
            fed = True

            # Kill plant if energy below threshold
            if plant.energy < plant.survival_threshold:
                env.clear_plant_energy(plant.x, plant.y, plant.species_id)
                world.unregister_position(co_eid, plant.x, plant.y)
                world.collect_garbage([co_eid])

        if not fed:
            swarm.starvation_ticks += 1

        # ----------------------------------------------------------------
        # 4. Lethal toxin damage at current cell (from toxin_layers)
        # ----------------------------------------------------------------
        for t_idx in range(env.num_toxins):
            toxin_val = float(env.toxin_layers[t_idx, swarm.x, swarm.y])
            if toxin_val > 0.0:
                # Each toxin layer can cause casualties; handled by signaling system
                # which writes the lethality_rate into toxin_layers scaled values.
                # Here we apply a generic casualty proportional to concentration.
                casualties = int(toxin_val * swarm.population * TOXIN_CASUALTY_FACTOR)
                swarm.population = max(0, swarm.population - casualties)

        # ----------------------------------------------------------------
        # 5. Starvation attrition
        # ----------------------------------------------------------------
        if swarm.starvation_ticks > 1:
            attrition = max(1, int(swarm.population * 0.05 * swarm.starvation_ticks))
            swarm.population = max(0, swarm.population - attrition)

        # ----------------------------------------------------------------
        # 6. Death check
        # ----------------------------------------------------------------
        if swarm.population <= 0:
            world.unregister_position(entity.entity_id, swarm.x, swarm.y)
            dead_swarms.append(entity.entity_id)
            continue

        # ----------------------------------------------------------------
        # 7. Reproduction: convert only swarm-scale surplus energy into growth
        # ----------------------------------------------------------------
        reproduction_threshold = max(
            swarm.energy_min,
            swarm.population * swarm.energy_min * swarm.reproduction_energy_divisor,
        )
        new_individuals = int(swarm.energy // reproduction_threshold)
        if new_individuals > 0:
            swarm.population += new_individuals
            swarm.energy -= new_individuals * reproduction_threshold

        # ----------------------------------------------------------------
        # 8. Mitosis
        # ----------------------------------------------------------------
        if swarm.population >= 2 * swarm.initial_population:
            _perform_mitosis(swarm, world)

    world.collect_garbage(dead_swarms)
    env.rebuild_energy_layer()

Signaling system: substance synthesis, activation and toxin effects.

This module manages the lifecycle of VOC signals and defensive toxins, including trigger evaluation via the spatial hash, synthesis countdowns, delegation of airborne diffusion to :class:GridEnvironment, mycorrhizal signal relays and application of toxin effects to swarms.

run_signaling(world, env, trigger_conditions, mycorrhizal_inter_species, signal_velocity, tick)

Execute one signaling tick, handling synthesis, emission and diffusion.

Parameters:

Name	Type	Description	Default
world	ECSWorld	ECS world registry.	required
env	GridEnvironment	Grid environment holding signal/toxin layers.	required
trigger_conditions	dict[int, list[object]]	Mapping of flora species_id to trigger schemas.	required
mycorrhizal_inter_species	bool	Whether inter-species mycorrhizal signaling is permitted.	required
signal_velocity	int	Ticks per hop for root-network relays.	required
tick	int	Current simulation tick.	required

Source code in src/phids/engine/systems/signaling.py

def run_signaling(
    world: ECSWorld,
    env: GridEnvironment,
    trigger_conditions: dict[int, list[object]],
    mycorrhizal_inter_species: bool,
    signal_velocity: int,
    tick: int,
) -> None:
    """Execute one signaling tick, handling synthesis, emission and diffusion.

    Args:
        world: ECS world registry.
        env: Grid environment holding signal/toxin layers.
        trigger_conditions: Mapping of flora species_id to trigger schemas.
        mycorrhizal_inter_species: Whether inter-species mycorrhizal signaling
            is permitted.
        signal_velocity: Ticks per hop for root-network relays.
        tick: Current simulation tick.
    """
    from phids.api.schemas import TriggerConditionSchema  # avoid circular at module level

    dead_substances: list[int] = []

    # Toxins are local, per-tick defenses: rebuild their field from currently
    # active emitters each signaling pass so concentrations vanish immediately
    # once triggers cease.
    env.toxin_layers[:] = 0.0
    env._toxin_layers_write[:] = 0.0

    for entity in world.query(SubstanceComponent):
        sub = entity.get_component(SubstanceComponent)
        sub.triggered_this_tick = False

    # ------------------------------------------------------------------
    # 1. Evaluate trigger conditions for all plants
    # ------------------------------------------------------------------
    for entity in list(world.query(PlantComponent)):
        plant: PlantComponent = entity.get_component(PlantComponent)
        triggers = trigger_conditions.get(plant.species_id, [])

        for trig_raw in triggers:
            if not isinstance(trig_raw, TriggerConditionSchema):
                continue
            trig: TriggerConditionSchema = trig_raw

            # Check for predator presence at this cell via spatial hash
            triggered = (
                _co_located_swarm_population(
                    world,
                    plant.x,
                    plant.y,
                    trig.predator_species_id,
                )
                >= trig.min_predator_population
            )

            if not triggered:
                continue

            # Ensure a substance entity exists for this (plant, substance_id) pair
            existing_sub = None
            for sub_entity in world.query(SubstanceComponent):
                sub: SubstanceComponent = sub_entity.get_component(SubstanceComponent)
                if sub.owner_plant_id == plant.entity_id and sub.substance_id == trig.substance_id:
                    existing_sub = sub
                    break

            if existing_sub is None:
                # Spawn new substance entity with full properties from trigger
                new_entity = world.create_entity()
                existing_sub = SubstanceComponent(
                    entity_id=new_entity.entity_id,
                    substance_id=trig.substance_id,
                    owner_plant_id=plant.entity_id,
                    is_toxin=trig.is_toxin,
                    synthesis_duration=trig.synthesis_duration,
                    synthesis_remaining=trig.synthesis_duration,
                    lethal=trig.lethal,
                    lethality_rate=trig.lethality_rate,
                    repellent=trig.repellent,
                    repellent_walk_ticks=trig.repellent_walk_ticks,
                    aftereffect_ticks=trig.aftereffect_ticks,
                    aftereffect_remaining_ticks=trig.aftereffect_ticks,
                    precursor_signal_id=trig.precursor_signal_id,
                    precursor_signal_ids=tuple(trig.precursor_signal_ids),
                    activation_condition=(
                        trig.activation_condition.model_dump(mode="json")
                        if trig.activation_condition is not None
                        else None
                    ),
                    energy_cost_per_tick=trig.energy_cost_per_tick,
                    trigger_predator_species_id=trig.predator_species_id,
                    trigger_min_predator_population=trig.min_predator_population,
                )
                world.add_component(new_entity.entity_id, existing_sub)
            else:
                if (
                    not existing_sub.active
                    and existing_sub.synthesis_remaining <= 0
                    and existing_sub.aftereffect_remaining_ticks <= 0
                ):
                    existing_sub.synthesis_remaining = existing_sub.synthesis_duration

            existing_sub.triggered_this_tick = True

    # ------------------------------------------------------------------
    # 2. Advance synthesis timers & activate substances
    # ------------------------------------------------------------------
    for entity in list(world.query(SubstanceComponent)):
        sub = entity.get_component(SubstanceComponent)
        if sub.active:
            continue
        if not sub.triggered_this_tick:
            continue
        owner_entity = (
            world.get_entity(sub.owner_plant_id) if world.has_entity(sub.owner_plant_id) else None
        )
        if owner_entity is None:
            dead_substances.append(entity.entity_id)
            continue
        plant = owner_entity.get_component(PlantComponent)
        if sub.synthesis_remaining > 0:
            sub.synthesis_remaining -= 1
        if sub.synthesis_remaining <= 0:
            if not _check_activation_condition(
                plant,
                sub.owner_plant_id,
                sub.activation_condition,
                world,
            ):
                continue
            sub.active = True
            sub.aftereffect_remaining_ticks = sub.aftereffect_ticks

    # ------------------------------------------------------------------
    # 3. Emit active signals / toxins into environment layers
    # ------------------------------------------------------------------
    for entity in list(world.query(SubstanceComponent)):
        sub = entity.get_component(SubstanceComponent)
        if not sub.active:
            continue

        if not sub.triggered_this_tick:
            if sub.is_toxin or sub.aftereffect_remaining_ticks <= 0:
                sub.active = False
                continue

        owner_entity = (
            world.get_entity(sub.owner_plant_id) if world.has_entity(sub.owner_plant_id) else None
        )
        if owner_entity is None:
            dead_substances.append(entity.entity_id)
            continue

        plant = owner_entity.get_component(PlantComponent)

        # --- Energy maintenance cost (Section 4: continuous depletion) ---
        if sub.energy_cost_per_tick > 0.0:
            plant.energy -= sub.energy_cost_per_tick
            env.set_plant_energy(plant.x, plant.y, plant.species_id, plant.energy)

        if sub.is_toxin:
            if sub.substance_id < env.num_toxins:
                env.toxin_layers[sub.substance_id, plant.x, plant.y] = min(
                    1.0,
                    float(env.toxin_layers[sub.substance_id, plant.x, plant.y])
                    + SUBSTANCE_EMIT_RATE,
                )
            # Apply toxin effects to nearby swarms
            _apply_toxin_to_swarms(sub, env, world)
        else:
            if sub.substance_id < env.num_signals:
                env.signal_layers[sub.substance_id, plant.x, plant.y] = min(
                    1.0,
                    float(env.signal_layers[sub.substance_id, plant.x, plant.y])
                    + SUBSTANCE_EMIT_RATE,
                )
            # Relay via mycorrhizal network
            _relay_signal_via_mycorrhizal(
                plant,
                sub.substance_id,
                SUBSTANCE_EMIT_RATE,
                env,
                world,
                mycorrhizal_inter_species,
                signal_velocity,
                tick,
            )

    # ------------------------------------------------------------------
    # 4. Check aftereffects and deactivate expired substances
    # ------------------------------------------------------------------
    for entity in list(world.query(SubstanceComponent)):
        sub = entity.get_component(SubstanceComponent)
        if not sub.active:
            continue

        # Verify owner still exists
        if not world.has_entity(sub.owner_plant_id):
            dead_substances.append(entity.entity_id)
            continue

        owner_entity = world.get_entity(sub.owner_plant_id)
        plant = owner_entity.get_component(PlantComponent)

        if sub.triggered_this_tick:
            sub.aftereffect_remaining_ticks = sub.aftereffect_ticks
            continue

        if sub.is_toxin:
            sub.active = False
            if sub.substance_id < env.num_toxins:
                env.toxin_layers[sub.substance_id, plant.x, plant.y] = 0.0
            continue

        if sub.aftereffect_remaining_ticks > 0:
            sub.aftereffect_remaining_ticks -= 1
            if sub.aftereffect_remaining_ticks <= 0:
                sub.active = False
        else:
            sub.active = False

    # ------------------------------------------------------------------
    # 5. Diffusion (delegated to GridEnvironment)
    # ------------------------------------------------------------------
    env.diffuse_signals()
    env.diffuse_toxins()

    # ------------------------------------------------------------------
    # 6. Garbage collect expired substance entities
    # ------------------------------------------------------------------
    world.collect_garbage(dead_substances)

Telemetry and replay

Telemetry analytics: accumulate Lotka-Volterra metrics into a DataFrame.

The :class:TelemetryRecorder accumulates per-tick metrics into an in-memory buffer and exposes a lazily-built :class:polars.DataFrame for export.

TelemetryRecorder

Accumulate per-tick Lotka-Volterra metrics into a Polars DataFrame.

The recorder stores rows containing the following fields per tick: tick, total_flora_energy, flora_population, predator_clusters, predator_population.

Source code in src/phids/telemetry/analytics.py

class TelemetryRecorder:
    """Accumulate per-tick Lotka-Volterra metrics into a Polars DataFrame.

    The recorder stores rows containing the following fields per tick:
    ``tick``, ``total_flora_energy``, ``flora_population``,
    ``predator_clusters``, ``predator_population``.
    """

    def __init__(self) -> None:
        """Create a TelemetryRecorder with empty in-memory buffers."""
        self._rows: list[dict[str, Any]] = []
        self._df: pl.DataFrame | None = None

    def record(self, world: ECSWorld, tick: int) -> None:
        """Snapshot current metrics and append to the internal buffer.

        Args:
            world: The ECS world to sample entity components from.
            tick: Current simulation tick index.
        """
        total_flora_energy = 0.0
        flora_population = 0
        for entity in world.query(PlantComponent):
            plant: PlantComponent = entity.get_component(PlantComponent)
            total_flora_energy += plant.energy
            flora_population += 1

        predator_clusters = 0
        predator_population = 0
        for entity in world.query(SwarmComponent):
            swarm: SwarmComponent = entity.get_component(SwarmComponent)
            predator_clusters += 1
            predator_population += swarm.population

        self._rows.append(
            {
                "tick": tick,
                "total_flora_energy": total_flora_energy,
                "flora_population": flora_population,
                "predator_clusters": predator_clusters,
                "predator_population": predator_population,
            }
        )
        self._df = None  # invalidate cache

    def get_latest_metrics(self) -> dict[str, Any] | None:
        """Return the latest recorded telemetry row, if available.

        Returns:
            dict[str, Any] | None: Most recent metrics row or ``None``.
        """
        if not self._rows:
            return None
        return self._rows[-1]

    @property
    def dataframe(self) -> pl.DataFrame:
        """Return recorded metrics as a Polars DataFrame (lazily built).

        Returns:
            pl.DataFrame: DataFrame containing accumulated telemetry rows.
        """
        if self._df is None:
            logger.debug("Materialising telemetry dataframe from %d rows", len(self._rows))
            if self._rows:
                self._df = pl.DataFrame(self._rows)
            else:
                self._df = pl.DataFrame(
                    {
                        "tick": pl.Series([], dtype=pl.Int64),
                        "total_flora_energy": pl.Series([], dtype=pl.Float64),
                        "flora_population": pl.Series([], dtype=pl.Int64),
                        "predator_clusters": pl.Series([], dtype=pl.Int64),
                        "predator_population": pl.Series([], dtype=pl.Int64),
                    }
                )
        return self._df

    def reset(self) -> None:
        """Clear accumulated telemetry and reset internal cache."""
        logger.info("Resetting telemetry recorder with %d buffered rows", len(self._rows))
        self._rows = []
        self._df = None

dataframe property

Return recorded metrics as a Polars DataFrame (lazily built).

Returns:

Type	Description
DataFrame	pl.DataFrame: DataFrame containing accumulated telemetry rows.

__init__()

Create a TelemetryRecorder with empty in-memory buffers.

Source code in src/phids/telemetry/analytics.py

def __init__(self) -> None:
    """Create a TelemetryRecorder with empty in-memory buffers."""
    self._rows: list[dict[str, Any]] = []
    self._df: pl.DataFrame | None = None

get_latest_metrics()

Return the latest recorded telemetry row, if available.

Returns:

Type	Description
dict[str, Any] | None	dict[str, Any] | None: Most recent metrics row or None.

Source code in src/phids/telemetry/analytics.py

def get_latest_metrics(self) -> dict[str, Any] | None:
    """Return the latest recorded telemetry row, if available.

    Returns:
        dict[str, Any] | None: Most recent metrics row or ``None``.
    """
    if not self._rows:
        return None
    return self._rows[-1]

record(world, tick)

Snapshot current metrics and append to the internal buffer.

Parameters:

Name	Type	Description	Default
world	ECSWorld	The ECS world to sample entity components from.	required
tick	int	Current simulation tick index.	required

Source code in src/phids/telemetry/analytics.py

def record(self, world: ECSWorld, tick: int) -> None:
    """Snapshot current metrics and append to the internal buffer.

    Args:
        world: The ECS world to sample entity components from.
        tick: Current simulation tick index.
    """
    total_flora_energy = 0.0
    flora_population = 0
    for entity in world.query(PlantComponent):
        plant: PlantComponent = entity.get_component(PlantComponent)
        total_flora_energy += plant.energy
        flora_population += 1

    predator_clusters = 0
    predator_population = 0
    for entity in world.query(SwarmComponent):
        swarm: SwarmComponent = entity.get_component(SwarmComponent)
        predator_clusters += 1
        predator_population += swarm.population

    self._rows.append(
        {
            "tick": tick,
            "total_flora_energy": total_flora_energy,
            "flora_population": flora_population,
            "predator_clusters": predator_clusters,
            "predator_population": predator_population,
        }
    )
    self._df = None  # invalidate cache

reset()

Clear accumulated telemetry and reset internal cache.

Source code in src/phids/telemetry/analytics.py

def reset(self) -> None:
    """Clear accumulated telemetry and reset internal cache."""
    logger.info("Resetting telemetry recorder with %d buffered rows", len(self._rows))
    self._rows = []
    self._df = None

Termination condition evaluators (Z1–Z7).

Provide checks for simulation termination conditions such as maximum ticks, species extinction and aggregate population/energy thresholds.

TerminationResult dataclass

Result returned by :func:check_termination.

Attributes:

Name	Type	Description
terminated	bool	True when a termination condition has been met.
reason	str	Human-readable explanation for termination.

Source code in src/phids/telemetry/conditions.py

@dataclass(slots=True)
class TerminationResult:
    """Result returned by :func:`check_termination`.

    Attributes:
        terminated: True when a termination condition has been met.
        reason: Human-readable explanation for termination.
    """

    terminated: bool
    reason: str

check_termination(world, tick, max_ticks, z2_flora_species=-1, z3_check_all_flora=True, z4_predator_species=-1, z5_check_all_predators=True, z6_max_flora_energy=-1.0, z7_max_predator_population=-1)

Evaluate termination conditions and return the first triggered one.

Parameters:

Name	Type	Description	Default
world	ECSWorld	ECS world registry.	required
tick	int	Current simulation tick.	required
max_ticks	int	Z1 – maximum allowed ticks (halt when reached).	required
z2_flora_species	int	Species id that triggers Z2 on extinction (-1 disables).	-1
z3_check_all_flora	bool	If True, halt when all flora are extinct (Z3).	True
z4_predator_species	int	Species id that triggers Z4 on extinction (-1 disables).	-1
z5_check_all_predators	bool	If True, halt when all predators are extinct (Z5).	True
z6_max_flora_energy	float	Aggregate flora energy threshold for Z6 (-1 disables).	-1.0
z7_max_predator_population	int	Aggregate predator population threshold for Z7 (-1 disables).	-1

Returns:

Name	Type	Description
TerminationResult	TerminationResult	Object indicating whether termination occurred and why.

Source code in src/phids/telemetry/conditions.py

def check_termination(
    world: ECSWorld,
    tick: int,
    max_ticks: int,
    z2_flora_species: int = -1,
    z3_check_all_flora: bool = True,
    z4_predator_species: int = -1,
    z5_check_all_predators: bool = True,
    z6_max_flora_energy: float = -1.0,
    z7_max_predator_population: int = -1,
) -> TerminationResult:
    """Evaluate termination conditions and return the first triggered one.

    Args:
        world: ECS world registry.
        tick: Current simulation tick.
        max_ticks: Z1 – maximum allowed ticks (halt when reached).
        z2_flora_species: Species id that triggers Z2 on extinction (-1 disables).
        z3_check_all_flora: If True, halt when all flora are extinct (Z3).
        z4_predator_species: Species id that triggers Z4 on extinction (-1 disables).
        z5_check_all_predators: If True, halt when all predators are extinct (Z5).
        z6_max_flora_energy: Aggregate flora energy threshold for Z6 (-1 disables).
        z7_max_predator_population: Aggregate predator population threshold for Z7 (-1 disables).

    Returns:
        TerminationResult: Object indicating whether termination occurred and why.
    """
    # Z1 – maximum tick count
    if tick >= max_ticks:
        return TerminationResult(terminated=True, reason=f"Z1: reached max_ticks={max_ticks}")

    # Gather live flora
    flora_species_alive: set[int] = set()
    total_flora_energy = 0.0
    flora_alive = False
    for entity in world.query(PlantComponent):
        plant: PlantComponent = entity.get_component(PlantComponent)
        flora_species_alive.add(plant.species_id)
        total_flora_energy += plant.energy
        flora_alive = True

    # Z2 – specific flora species extinction
    if z2_flora_species >= 0 and z2_flora_species not in flora_species_alive:
        return TerminationResult(
            terminated=True, reason=f"Z2: flora species {z2_flora_species} extinct"
        )

    # Z3 – all flora extinct
    if z3_check_all_flora and not flora_alive:
        return TerminationResult(terminated=True, reason="Z3: all flora extinct")

    # Z6 – aggregate flora energy exceeds upper bound
    if z6_max_flora_energy > 0.0 and total_flora_energy > z6_max_flora_energy:
        return TerminationResult(
            terminated=True,
            reason=f"Z6: total flora energy {total_flora_energy:.1f} > {z6_max_flora_energy}",
        )

    # Gather live predators
    predator_species_alive: set[int] = set()
    total_predator_population = 0
    predators_alive = False
    for entity in world.query(SwarmComponent):
        swarm: SwarmComponent = entity.get_component(SwarmComponent)
        predator_species_alive.add(swarm.species_id)
        total_predator_population += swarm.population
        predators_alive = True

    # Z4 – specific predator species extinction
    if z4_predator_species >= 0 and z4_predator_species not in predator_species_alive:
        return TerminationResult(
            terminated=True, reason=f"Z4: predator species {z4_predator_species} extinct"
        )

    # Z5 – all predators extinct
    if z5_check_all_predators and not predators_alive:
        return TerminationResult(terminated=True, reason="Z5: all predators extinct")

    # Z7 – aggregate predator population exceeds upper bound
    if z7_max_predator_population > 0 and total_predator_population > z7_max_predator_population:
        return TerminationResult(
            terminated=True,
            reason=(
                f"Z7: total predator population {total_predator_population} "
                f"> {z7_max_predator_population}"
            ),
        )

    return TerminationResult(terminated=False, reason="")

Telemetry export utilities: CSV and NDJSON export helpers.

Helpers to persist or stream Polars DataFrames produced by the telemetry recorder as CSV or newline-delimited JSON (NDJSON).

export_bytes_csv(df)

Return the telemetry DataFrame serialized as CSV bytes.

Parameters:

Name	Type	Description	Default
df	DataFrame	Polars DataFrame to serialize.	required

Returns:

Name	Type	Description
bytes	bytes	CSV-encoded bytes.

Source code in src/phids/telemetry/export.py

def export_bytes_csv(df: pl.DataFrame) -> bytes:
    """Return the telemetry DataFrame serialized as CSV bytes.

    Args:
        df: Polars DataFrame to serialize.

    Returns:
        bytes: CSV-encoded bytes.
    """
    return df.write_csv().encode()

export_bytes_json(df)

Return the telemetry DataFrame serialized as NDJSON bytes.

Parameters:

Name	Type	Description	Default
df	DataFrame	Polars DataFrame to serialize.	required

Returns:

Name	Type	Description
bytes	bytes	NDJSON-encoded bytes.

Source code in src/phids/telemetry/export.py

def export_bytes_json(df: pl.DataFrame) -> bytes:
    """Return the telemetry DataFrame serialized as NDJSON bytes.

    Args:
        df: Polars DataFrame to serialize.

    Returns:
        bytes: NDJSON-encoded bytes.
    """
    return df.write_ndjson().encode()

export_csv(df, path)

Write the telemetry DataFrame to a CSV file.

Parameters:

Name	Type	Description	Default
df	DataFrame	Polars DataFrame produced by the telemetry recorder.	required
path	str | Path	Destination file path.	required

Source code in src/phids/telemetry/export.py

def export_csv(df: pl.DataFrame, path: str | Path) -> None:
    """Write the telemetry DataFrame to a CSV file.

    Args:
        df: Polars DataFrame produced by the telemetry recorder.
        path: Destination file path.
    """
    df.write_csv(str(path))

export_json(df, path)

Write the telemetry DataFrame to a newline-delimited JSON file.

Parameters:

Name	Type	Description	Default
df	DataFrame	Polars DataFrame produced by the telemetry recorder.	required
path	str | Path	Destination file path.	required

Source code in src/phids/telemetry/export.py

def export_json(df: pl.DataFrame, path: str | Path) -> None:
    """Write the telemetry DataFrame to a newline-delimited JSON file.

    Args:
        df: Polars DataFrame produced by the telemetry recorder.
        path: Destination file path.
    """
    df.write_ndjson(str(path))

Replay I/O for deterministic simulations.

This module provides compact binary serialisation of per-tick state snapshots using :mod:msgpack. Each tick is stored as a discrete frame to support deterministic re-simulation and on-disk replay files.

ReplayBuffer

Append-only buffer of binary-serialised tick frames.

Supports writing to and reading from a binary replay file for deterministic re-simulation.

Source code in src/phids/io/replay.py

class ReplayBuffer:
    """Append-only buffer of binary-serialised tick frames.

    Supports writing to and reading from a binary replay file for
    deterministic re-simulation.
    """

    def __init__(self) -> None:
        """Create an empty replay buffer.

        The buffer stores msgpack-serialised frames in an append-only list.
        """
        self._frames: list[bytes] = []

    def append(self, state: dict[str, Any]) -> None:
        """Serialize and append a tick state to the buffer.

        Args:
            state: Tick state mapping to serialize and store.
        """
        self._frames.append(serialise_state(state))

    def __len__(self) -> int:
        """Return number of stored frames."""
        return len(self._frames)

    def get_frame(self, tick: int) -> dict[str, Any]:
        """Return the deserialised state for the specified tick index.

        Args:
            tick: Index of the frame to retrieve (0-based).

        Returns:
            dict[str, Any]: Decoded state mapping for the requested tick.
        """
        return deserialise_state(self._frames[tick])

    def save(self, path: str | Path) -> None:
        """Write all frames to a binary replay file.

        The file format is an append of records where each record begins with a
        4-byte little-endian unsigned integer describing the following frame
        length, immediately followed by the frame bytes.

        Args:
            path: Destination file path.
        """
        destination = Path(path)
        with destination.open("wb") as fp:
            for frame in self._frames:
                length = len(frame).to_bytes(4, "little")
                fp.write(length)
                fp.write(frame)
        logger.info("Replay saved to %s (%d frames)", destination, len(self._frames))

    @classmethod
    def load(cls, path: str | Path) -> ReplayBuffer:
        """Load a replay file produced by :meth:`save`.

        Args:
            path: Path to the binary replay file.

        Returns:
            ReplayBuffer: Populated buffer ready for re-simulation.
        """
        source = Path(path)
        buf = cls()
        with source.open("rb") as fp:
            while True:
                length_bytes = fp.read(4)
                if len(length_bytes) < 4:
                    break
                length = int.from_bytes(length_bytes, "little")
                frame = fp.read(length)
                if len(frame) < length:
                    logger.warning(
                        "Replay file %s ended mid-frame (expected=%d bytes, got=%d)",
                        source,
                        length,
                        len(frame),
                    )
                    break
                buf._frames.append(frame)
        logger.info("Replay loaded from %s (%d frames)", source, len(buf._frames))
        return buf

__init__()

Create an empty replay buffer.

The buffer stores msgpack-serialised frames in an append-only list.

Source code in src/phids/io/replay.py

def __init__(self) -> None:
    """Create an empty replay buffer.

    The buffer stores msgpack-serialised frames in an append-only list.
    """
    self._frames: list[bytes] = []

__len__()

Return number of stored frames.

Source code in src/phids/io/replay.py

def __len__(self) -> int:
    """Return number of stored frames."""
    return len(self._frames)

append(state)

Serialize and append a tick state to the buffer.

Parameters:

Name	Type	Description	Default
state	dict[str, Any]	Tick state mapping to serialize and store.	required

Source code in src/phids/io/replay.py

def append(self, state: dict[str, Any]) -> None:
    """Serialize and append a tick state to the buffer.

    Args:
        state: Tick state mapping to serialize and store.
    """
    self._frames.append(serialise_state(state))

get_frame(tick)

Return the deserialised state for the specified tick index.

Parameters:

Name	Type	Description	Default
tick	int	Index of the frame to retrieve (0-based).	required

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Decoded state mapping for the requested tick.

Source code in src/phids/io/replay.py

def get_frame(self, tick: int) -> dict[str, Any]:
    """Return the deserialised state for the specified tick index.

    Args:
        tick: Index of the frame to retrieve (0-based).

    Returns:
        dict[str, Any]: Decoded state mapping for the requested tick.
    """
    return deserialise_state(self._frames[tick])

load(path) classmethod

Load a replay file produced by :meth:save.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the binary replay file.	required

Returns:

Name	Type	Description
ReplayBuffer	ReplayBuffer	Populated buffer ready for re-simulation.

Source code in src/phids/io/replay.py

@classmethod
def load(cls, path: str | Path) -> ReplayBuffer:
    """Load a replay file produced by :meth:`save`.

    Args:
        path: Path to the binary replay file.

    Returns:
        ReplayBuffer: Populated buffer ready for re-simulation.
    """
    source = Path(path)
    buf = cls()
    with source.open("rb") as fp:
        while True:
            length_bytes = fp.read(4)
            if len(length_bytes) < 4:
                break
            length = int.from_bytes(length_bytes, "little")
            frame = fp.read(length)
            if len(frame) < length:
                logger.warning(
                    "Replay file %s ended mid-frame (expected=%d bytes, got=%d)",
                    source,
                    length,
                    len(frame),
                )
                break
            buf._frames.append(frame)
    logger.info("Replay loaded from %s (%d frames)", source, len(buf._frames))
    return buf

save(path)

Write all frames to a binary replay file.

The file format is an append of records where each record begins with a 4-byte little-endian unsigned integer describing the following frame length, immediately followed by the frame bytes.

Parameters:

Name	Type	Description	Default
path	str | Path	Destination file path.	required

Source code in src/phids/io/replay.py

def save(self, path: str | Path) -> None:
    """Write all frames to a binary replay file.

    The file format is an append of records where each record begins with a
    4-byte little-endian unsigned integer describing the following frame
    length, immediately followed by the frame bytes.

    Args:
        path: Destination file path.
    """
    destination = Path(path)
    with destination.open("wb") as fp:
        for frame in self._frames:
            length = len(frame).to_bytes(4, "little")
            fp.write(length)
            fp.write(frame)
    logger.info("Replay saved to %s (%d frames)", destination, len(self._frames))

deserialise_state(data)

Deserialize a msgpack frame into a state mapping.

Parameters:

Name	Type	Description	Default
data	bytes	msgpack-encoded bytes produced by :func:serialise_state.	required

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Decoded state mapping.

Source code in src/phids/io/replay.py

def deserialise_state(data: bytes) -> dict[str, Any]:
    """Deserialize a msgpack frame into a state mapping.

    Args:
        data: msgpack-encoded bytes produced by :func:`serialise_state`.

    Returns:
        dict[str, Any]: Decoded state mapping.
    """
    result: dict[str, Any] = msgpack.unpackb(data, raw=False)
    return result

serialise_state(state)

Serialise a state snapshot to a msgpack frame.

Parameters:

Name	Type	Description	Default
state	dict[str, Any]	Tick state mapping (for example, the output of SimulationLoop.get_state_snapshot()).	required

Returns:

Name	Type	Description
bytes	bytes	msgpack-encoded frame.

Source code in src/phids/io/replay.py

def serialise_state(state: dict[str, Any]) -> bytes:
    """Serialise a state snapshot to a msgpack frame.

    Args:
        state: Tick state mapping (for example, the output of
            ``SimulationLoop.get_state_snapshot()``).

    Returns:
        bytes: msgpack-encoded frame.
    """
    return msgpack.packb(state, use_bin_type=True)  # type: ignore[no-any-return]

Shared utilities

Shared constants for PHIDS.

This module defines compile-time limits (Rule of 16), grid bounds and other numeric sentinel values used across the codebase.

Central logging configuration for PHIDS.

The package uses a single idempotent logging bootstrap so API, UI, engine, telemetry, and I/O modules share consistent formatting and levels. Configuration is environment-driven to keep detailed debugging available without paying for verbose logs by default.

InMemoryLogHandler

Bases: Handler

Capture recent structured log entries for the diagnostics UI.

Source code in src/phids/shared/logging_config.py

class InMemoryLogHandler(logging.Handler):
    """Capture recent structured log entries for the diagnostics UI."""

    def emit(self, record: logging.LogRecord) -> None:
        """Append one formatted record to the in-memory diagnostics buffer."""
        try:
            message = record.getMessage()
            if record.exc_info:
                exc_text = self.formatException(record.exc_info)
                if exc_text:
                    message = f"{message}\n{exc_text}"
            entry = {
                "timestamp": datetime.fromtimestamp(record.created).strftime("%H:%M:%S"),
                "level": record.levelname,
                "logger": record.name,
                "module": record.module,
                "message": message,
            }
            with _RECENT_LOGS_LOCK:
                _RECENT_LOGS.append(entry)
        except Exception:  # pragma: no cover - logging must never fail app code
            self.handleError(record)

emit(record)

Append one formatted record to the in-memory diagnostics buffer.

Source code in src/phids/shared/logging_config.py

def emit(self, record: logging.LogRecord) -> None:
    """Append one formatted record to the in-memory diagnostics buffer."""
    try:
        message = record.getMessage()
        if record.exc_info:
            exc_text = self.formatException(record.exc_info)
            if exc_text:
                message = f"{message}\n{exc_text}"
        entry = {
            "timestamp": datetime.fromtimestamp(record.created).strftime("%H:%M:%S"),
            "level": record.levelname,
            "logger": record.name,
            "module": record.module,
            "message": message,
        }
        with _RECENT_LOGS_LOCK:
            _RECENT_LOGS.append(entry)
    except Exception:  # pragma: no cover - logging must never fail app code
        self.handleError(record)

configure_logging(*, force=False)

Configure PHIDS package logging.

Environment variables

PHIDS_LOG_LEVEL: Package log level (default INFO). PHIDS_LOG_FILE: Optional file path for a rotating debug log. PHIDS_LOG_FILE_LEVEL: File handler level (default DEBUG). PHIDS_LOG_SIM_DEBUG_INTERVAL: Tick interval for engine summaries.

Parameters:

Name	Type	Description	Default
force	bool	Reconfigure logging even if already configured.	False

Source code in src/phids/shared/logging_config.py

def configure_logging(*, force: bool = False) -> None:
    """Configure PHIDS package logging.

    Environment variables:
        ``PHIDS_LOG_LEVEL``: Package log level (default ``INFO``).
        ``PHIDS_LOG_FILE``: Optional file path for a rotating debug log.
        ``PHIDS_LOG_FILE_LEVEL``: File handler level (default ``DEBUG``).
        ``PHIDS_LOG_SIM_DEBUG_INTERVAL``: Tick interval for engine summaries.

    Args:
        force: Reconfigure logging even if already configured.
    """
    global _CONFIGURED  # noqa: PLW0603
    if _CONFIGURED and not force:
        return

    package_level = _coerce_log_level(os.getenv("PHIDS_LOG_LEVEL"))
    file_level = _coerce_log_level(os.getenv("PHIDS_LOG_FILE_LEVEL"), default="DEBUG")
    log_file = os.getenv("PHIDS_LOG_FILE")

    handlers: dict[str, dict[str, object]] = {
        "console": {
            "class": "logging.StreamHandler",
            "level": package_level,
            "formatter": "standard",
        },
        "memory": {
            "class": "phids.shared.logging_config.InMemoryLogHandler",
            "level": package_level,
        },
    }
    root_handlers = ["console", "memory"]

    if force:
        with _RECENT_LOGS_LOCK:
            _RECENT_LOGS.clear()

    if log_file:
        log_path = Path(log_file)
        log_path.parent.mkdir(parents=True, exist_ok=True)
        handlers["file"] = {
            "class": "logging.handlers.RotatingFileHandler",
            "level": file_level,
            "formatter": "standard",
            "filename": str(log_path),
            "maxBytes": 2_000_000,
            "backupCount": 3,
            "encoding": "utf-8",
        }
        root_handlers.append("file")

    logging.config.dictConfig(
        {
            "version": 1,
            "disable_existing_loggers": False,
            "formatters": {
                "standard": {"format": ("%(asctime)s | %(levelname)-8s | %(name)s | %(message)s")}
            },
            "handlers": handlers,
            "root": {
                "level": "WARNING",
                "handlers": root_handlers,
            },
            "loggers": {
                logger_name: {
                    "level": package_level,
                    "handlers": [],
                    "propagate": True,
                }
                for logger_name in _LOGGER_NAMES
            },
        }
    )

    _CONFIGURED = True
    logging.getLogger(__name__).debug(
        "Logging configured (package_level=%s, file_logging=%s, sim_debug_interval=%d)",
        package_level,
        bool(log_file),
        get_simulation_debug_interval(),
    )

get_recent_logs(*, limit=80)

Return the newest structured PHIDS log entries first.

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of entries to return.	80

Returns:

Type	Description
list[dict[str, str]]	list[dict[str, str]]: Structured log entries for diagnostics panels.

Source code in src/phids/shared/logging_config.py

def get_recent_logs(*, limit: int = 80) -> list[dict[str, str]]:
    """Return the newest structured PHIDS log entries first.

    Args:
        limit: Maximum number of entries to return.

    Returns:
        list[dict[str, str]]: Structured log entries for diagnostics panels.
    """
    clamped_limit = max(1, limit)
    with _RECENT_LOGS_LOCK:
        return list(reversed(list(_RECENT_LOGS)[-clamped_limit:]))

get_simulation_debug_interval()

Return the interval used for periodic simulation debug summaries.

Returns:

Name	Type	Description
int	int	Tick interval for DEBUG summaries.

Source code in src/phids/shared/logging_config.py

def get_simulation_debug_interval() -> int:
    """Return the interval used for periodic simulation debug summaries.

    Returns:
        int: Tick interval for DEBUG summaries.
    """
    return _coerce_positive_int(
        os.getenv("PHIDS_LOG_SIM_DEBUG_INTERVAL"),
        default=_DEFAULT_SIM_DEBUG_INTERVAL,
    )