service.py

from __future__ import annotations

import random
import threading

from backend.payload_types import TopologySpecInput, TopologySpecPatch
from backend.rules import RuleRegistry
from backend.simulation.engine import SimulationEngine
from backend.simulation.models import CellMutationDelta, SimulationSnapshot, SimulationStateData
from backend.simulation.service_boards import (
    build_initial_state,
    clone_service_board,
    coerce_board_to_rule,
    empty_service_board,
    random_service_board,
    step_board,
    transfer_board,
)
from backend.simulation.service_cells import (
    set_cells_by_id,
    toggle_cells_by_id,
    validate_state_values,
)
from backend.simulation.service_snapshots import snapshot_state
from backend.simulation.service_transitions import (
    apply_config_transition,
    apply_reset_transition,
)
from backend.simulation.topology_builders import TopologyCellBudgetExceeded


class SimulationOperationError(ValueError):
    """Raised when a requested simulation operation is invalid for the current rule."""


class SimulationService:
    """Thread-safe state mutation service for the cellular automaton."""

    def __init__(
        self,
        rule_registry: RuleRegistry,
        engine: SimulationEngine | None = None,
        *,
        lock: threading.Lock | None = None,
        state: SimulationStateData | None = None,
    ) -> None:
        self.rule_registry = rule_registry
        self.engine = engine or SimulationEngine()
        self._lock = lock or threading.Lock()
        self._state = state or build_initial_state(rule_registry)

    @property
    def lock(self) -> threading.Lock:
        return self._lock

    @property
    def state(self) -> SimulationStateData:
        return self._state

    def runtime_plan(self) -> tuple[bool, float]:
        with self._lock:
            if self._state.running:
                return True, max(0.02, 1.0 / self._state.config.speed)
            return False, 0.2

    def get_state(self) -> SimulationSnapshot:
        with self._lock:
            return snapshot_state(self._state)

    def replace_state(self, next_state: SimulationStateData) -> None:
        with self._lock:
            self._state = SimulationStateData(
                config=next_state.config,
                running=bool(next_state.running),
                generation=int(next_state.generation),
                rule=next_state.rule,
                board=clone_service_board(next_state.board),
                state_revision=0,
            )

    def start(self) -> SimulationSnapshot:
        with self._lock:
            if not self._state.running:
                self._state.running = True
                self._state.state_revision += 1
            return snapshot_state(self._state)

    def pause(self) -> None:
        with self._lock:
            if self._state.running:
                self._state.running = False
                self._state.state_revision += 1

    def resume(self) -> SimulationSnapshot:
        return self.start()

    def step(self) -> None:
        with self._lock:
            self._state.running = False
            step_board(self.engine, self._state)
            self._state.state_revision += 1

    def step_if_running(self) -> bool:
        with self._lock:
            if not self._state.running:
                return False
            step_board(self.engine, self._state)
            self._state.state_revision += 1
            return True

    def reset(
        self,
        topology_spec: TopologySpecInput | None = None,
        rule_name: str | None = None,
        speed: float | None = None,
        randomize: bool = False,
    ) -> None:
        with self._lock:
            try:
                apply_reset_transition(
                    self._state,
                    self.rule_registry,
                    create_random_board=random_service_board,
                    create_empty_board=empty_service_board,
                    choice_fn=random.choices,
                    topology_spec=topology_spec,
                    rule_name=rule_name,
                    speed=speed,
                    randomize=randomize,
                )
                self._state.state_revision += 1
            except TopologyCellBudgetExceeded:
                raise
            except ValueError as exc:
                raise SimulationOperationError(str(exc)) from exc

    def update_config(
        self,
        topology_spec: TopologySpecPatch | None = None,
        speed: float | None = None,
        rule_name: str | None = None,
    ) -> None:
        with self._lock:
            previous_config = self._state.config
            previous_rule = self._state.rule
            try:
                apply_config_transition(
                    self._state,
                    self.rule_registry,
                    transfer_board=transfer_board,
                    coerce_board_to_rule=coerce_board_to_rule,
                    topology_spec=topology_spec,
                    speed=speed,
                    rule_name=rule_name,
                )
                if self._state.config != previous_config or self._state.rule is not previous_rule:
                    self._state.state_revision += 1
            except TopologyCellBudgetExceeded:
                raise
            except ValueError as exc:
                raise SimulationOperationError(str(exc)) from exc

    def _cell_mutation_delta(
        self, base_state_revision: int, updates: dict[str, int]
    ) -> CellMutationDelta:
        return CellMutationDelta(
            base_state_revision=base_state_revision,
            state_revision=self._state.state_revision,
            state_epoch=self._state.state_epoch,
            topology_revision=self._state.topology.topology_revision,
            generation=self._state.generation,
            cell_updates=tuple(updates.items()),
        )

    def toggle_cell_by_id(self, cell_id: str) -> CellMutationDelta:
        with self._lock:
            base_state_revision = self._state.state_revision
            updates = toggle_cells_by_id(self._state, [cell_id])
            if updates:
                self._state.state_revision += 1
            return self._cell_mutation_delta(base_state_revision, updates)

    def set_cell_state_by_id(self, cell_id: str, state: int) -> CellMutationDelta:
        with self._lock:
            base_state_revision = self._state.state_revision
            try:
                validate_state_values(self._state.rule, [state])
            except ValueError as exc:
                raise SimulationOperationError(str(exc)) from exc
            updates = set_cells_by_id(self._state, [(cell_id, state)])
            if updates:
                self._state.state_revision += 1
            return self._cell_mutation_delta(base_state_revision, updates)

    def set_cells_by_id(self, cells: list[tuple[str, int]]) -> CellMutationDelta:
        with self._lock:
            base_state_revision = self._state.state_revision
            try:
                validate_state_values(self._state.rule, [state for _, state in cells])
            except ValueError as exc:
                raise SimulationOperationError(str(exc)) from exc
            updates = set_cells_by_id(self._state, cells)
            if updates:
                self._state.state_revision += 1
            return self._cell_mutation_delta(base_state_revision, updates)