Skip to content

Nagel-Schreckenberg

The Nagel-Schreckenberg traffic model on a ring road, where jams emerge above a critical density.

  • Name: nagel_schreckenberg
  • Field: engineering/traffic
  • Formalism: lattice dynamics
  • References: Nagel and Schreckenberg 1992; Charbonneau 2017

Factory

@register_model(
    field='engineering/traffic',
    tags=('lattice', 'traffic', 'jams', 'stochastic'),
    references=('Nagel and Schreckenberg 1992', 'Charbonneau 2017'),
    defaults={'until': 200, 'dt': 1},
)
def nagel_schreckenberg(
    length: int = 1000,
    *,
    density: float = 0.2,
    slowdown: float = 0.3,
    vmax: int = 5,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return the Nagel-Schreckenberg traffic model on a ring road, where jams emerge above a critical density.

    Args:
        length:
            The number of cells of the road.

        density:
            The fraction of cells with a car.

        slowdown:
            The probability of random braking.

        vmax:
            The largest velocity.

        random_state:
            The seed or generator of the initial positions of the cars.

    Returns:
        system:
            The model, with the velocities `road.velocity`, -1 on empty cells.

    Examples:
        >>> from skcomplex.models.engineering.traffic import nagel_schreckenberg
        >>> int((nagel_schreckenberg(100, density=0.1).elements[0].values['velocity'] >= 0).sum())
        10
    """
    rng = np.random.default_rng(random_state)
    velocity = np.full(length, -1.0)
    velocity[rng.choice(length, size=round(density * length), replace=False)] = 0.0
    road = LatticeElement(
        'road',
        element=RoadElement,
        space=LatticeSpace(length),
        values={'velocity': velocity, 'slowdown': slowdown},
    )
    ahead = LatticeNeighborhood(tuple((distance,) for distance in range(vmax + 1)))
    behind = LatticeNeighborhood(tuple((-distance,) for distance in range(vmax + 1)))
    return System(
        elements=[road],
        interactions=[
            LocalRuleInteraction(
                'acceleration',
                target='road',
                state='velocity',
                rule=_accelerate,
                neighborhood=ahead,
                stochastic=True,
            ),
            LocalRuleInteraction('movement', target='road', state='velocity', rule=_move, neighborhood=behind),
        ],
    )

Components

class RoadElement(Element):
    """Cell of the Nagel-Schreckenberg road: the velocity of the car on it, or -1 when it is empty.

    The local rules of the road make cars accelerate by one up to the largest velocity, brake to the gap ahead, slow
    down by one with probability `slowdown`, and move forward by their velocity, every step.

    Args:
        label:
            The label of the cell or the lattice group.

        visual:
            How the element is drawn, or `None` for `default_visual_`.

        velocity:
            The initial velocity, or -1 for an empty cell.

        slowdown:
            The probability of random braking.

    Examples:
        >>> from skcomplex.models.engineering.traffic import RoadElement
        >>> RoadElement('cell').slowdown
        0.3
    """

    def __init__(
        self,
        label: str,
        visual: VisualSpec | None = None,
        *,
        velocity: Annotated[float, StateSpec(domain=ZZ)] = -1,
        slowdown: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.3,
    ) -> None:
        super().__init__(label, visual=visual)
        self.velocity = velocity
        self.slowdown = slowdown

class LocalRuleInteraction(Interaction):
    """Local rule of a cellular automaton, which sets the next value of every cell from its ordered neighbourhood.

    The rule is a table of the next value of every configuration of the neighbourhood, `TableSpec`, a rule of the sum
    of the neighbourhood, `TotalisticSpec`, a rule of the cell and the sum of its neighbours, `OuterTotalisticSpec`, an
    object whose `table()` returns a `TableSpec`, or a function. A function receives one array per neighbour of every
    read state, in neighbourhood order, with one value per cell, the uniform array `random` when the rule is stochastic,
    and the parameters of the bound group it names, and returns the next values of the states, one array per state.

    Args:
        label:
            The label of the interaction, unique among the parts of its system.

        target:
            The path of the group on the cells of a lattice, the nodes of a network, or a Cayley graph.

        state:
            The name of the state the rule sets, or a tuple of the names of the states it sets together.

        rule:
            The table, the totalistic rule, or the function.

        neighborhood:
            The ordered neighbourhood of every cell, such as `VonNeumannNeighborhood(center=True)`.

        reads:
            The names of the states whose neighbourhoods a function reads, or `None` for the states it sets.

        schedule:
            `'synchronous'`, `'random-sequential'`, `'random-site'`, `'checkerboard'`, or `'colored'`.

        stochastic:
            Whether a function receives the uniform random array `random`.

    Raises:
        ValidationError:
            The state or the read states are not names, or the neighbourhood is not a neighbourhood.

    Examples:
        >>> from skcomplex.core.interactions import LocalRuleInteraction
        >>> from skcomplex.core.spaces import VonNeumannNeighborhood
        >>> from skcomplex.specs import TotalisticSpec
        >>> parity = TotalisticSpec({1: 1, 3: 1})
        >>> around = VonNeumannNeighborhood(n_dim=1, center=True)
        >>> rule = LocalRuleInteraction('parity', target='cells', state='cell', rule=parity, neighborhood=around)
        >>> rule.check().schedule
        'synchronous'
    """

    def __init__(
        self,
        label: str,
        *,
        target: Annotated[str, RoleSpec()],
        state: str | tuple[str, ...],
        rule: object,
        neighborhood: BaseNeighborhood,
        reads: Sequence[str] | None = None,
        schedule: str = 'synchronous',
        stochastic: bool = False,
    ) -> None:
        super().__init__(label)
        self.target = target
        self.state = state
        self.rule = rule
        self.neighborhood = neighborhood
        self.reads = reads
        self.schedule = schedule
        self.stochastic = stochastic

    def _get_names(self) -> tuple[str, ...]:
        """Return the names of the states the rule sets."""
        return (self.state,) if isinstance(self.state, str) else tuple(self.state)

    def _check(self) -> None:
        """Raise an error for states that are not names, or a neighbourhood that is not one.

        Raises:
            ValidationError:
                The state or the read states are not names, or the neighbourhood is not a neighbourhood.
        """
        super()._check()
        names = [*self._get_names(), *(self.reads or ())]
        if not names or not all(isinstance(name, str) and name.isidentifier() for name in names):
            error_msg = f'The local rule {self.label!r} sets {self.state!r} and reads {self.reads!r}. Give state names.'
            raise ValidationError(error_msg)
        if not isinstance(self.neighborhood, BaseNeighborhood):
            error_msg = f'The neighbourhood of the local rule {self.label!r} is {self.neighborhood!r}. Give one.'
            raise ValidationError(error_msg)

    def _get_laws(self, namespace: SimpleNamespace) -> list[BaseLaw]:
        """Return the local rule law of the states of the target.

        Args:
            namespace:
                The namespace of the symbols the laws can use, with `namespace.target`.

        Returns:
            laws:
                The local rule law.
        """
        targets = tuple(getattr(self._get_member(namespace, 'target'), name) for name in self._get_names())
        reads = (
            None
            if self.reads is None
            else tuple(getattr(self._get_member(namespace, 'target'), name) for name in self.reads)
        )
        return [
            LocalRuleLaw(
                targets[0] if isinstance(self.state, str) else targets,
                self.neighborhood,
                self.rule,
                reads=reads,
                schedule=self.schedule,
                stochastic=self.stochastic,
                label=self.label,
            ),
        ]