Crowds

Crowds: pedestrians driven by social forces, evacuations, and criminality against security forces.

CriminalContagionInteraction(label, *, candidate, field, classes=5, eta=1.0, attraction=0.3, redemption=0.1)

Bases: EncounterInteraction

Conservative encounters between citizens, in which crime attracts and honesty redeems.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
candidate	Annotated[str, RoleSpec()]	The path of the distribution of citizens whose members move.	required
field	Annotated[str, RoleSpec()]	The path of the distribution of citizens they meet, usually the same distribution.	required
classes	Annotated[int, ParameterSpec(domain=open(0, oo))]	The number of criminality classes.	5
eta	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The encounter rate.	1.0
attraction	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The probability that a citizen who meets a more criminal one moves one class up.	0.3
redemption	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The probability that a citizen who meets a more honest one moves one class down.	0.1

Examples:

>>> from skcomplex.systems.social.crowds import CriminalContagionInteraction
>>> CriminalContagionInteraction('contagion', candidate='citizens', field='citizens').attraction
0.3

Source code in src/skcomplex/systems/social/crowds/_criminality.py

def __init__(
    self,
    label: str,
    *,
    candidate: Annotated[str, RoleSpec()],
    field: Annotated[str, RoleSpec()],
    classes: Annotated[int, ParameterSpec(domain=RealSet.open(0, oo))] = 5,
    eta: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    attraction: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.3,
    redemption: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.1,
) -> None:
    super().__init__(label, candidate=candidate, field=field, rate=self._get_rate, kernel=self._get_kernel)
    self.classes = classes
    self.eta = eta
    self.attraction = attraction
    self.redemption = redemption

CrowdingInteraction(label, *, candidate, field, rooms=7, speeds=3, crowding=2.0)

Bases: EncounterInteraction

Encounters that slow pedestrians down in crowded rooms.

It is an EncounterInteraction of the crowd with itself. Two pedestrians in the same room meet at the rate crowding, and the candidate slows down one speed class. Pedestrians outside keep their speed.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
candidate	Annotated[str, RoleSpec()]	The path of the crowd, a distribution over rooms and speed classes.	required
field	Annotated[str, RoleSpec()]	The path of the crowd that the candidates meet, the crowd again.	required
rooms	Annotated[int, ParameterSpec(domain=open(0, oo))]	The number of rooms.	7
speeds	Annotated[int, ParameterSpec(domain=open(0, oo))]	The number of speed classes.	3
crowding	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The rate of the encounters within a room that slow pedestrians down.	2.0

Examples:

>>> from skcomplex.systems.social.crowds import CrowdingInteraction
>>> CrowdingInteraction('crowding', candidate='crowd', field='crowd').crowding
2.0

Source code in src/skcomplex/systems/social/crowds/_evacuation.py

def __init__(
    self,
    label: str,
    *,
    candidate: Annotated[str, RoleSpec()],
    field: Annotated[str, RoleSpec()],
    rooms: Annotated[int, ParameterSpec(domain=RealSet.open(0, oo))] = 7,
    speeds: Annotated[int, ParameterSpec(domain=RealSet.open(0, oo))] = 3,
    crowding: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 2.0,
) -> None:
    super().__init__(label, candidate=candidate, field=field, rate=self._get_rate, kernel=self._get_kernel)
    self.rooms = rooms
    self.speeds = speeds
    self.crowding = crowding

DeterrenceInteraction(label, *, candidate, field, classes=5, forces=3, deterrence=2.0)

Bases: EncounterInteraction

Encounters of citizens with security forces, which move criminal citizens toward honesty.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
candidate	Annotated[str, RoleSpec()]	The path of the distribution of citizens.	required
field	Annotated[str, RoleSpec()]	The path of the distribution of the security forces over efficacy.	required
classes	Annotated[int, ParameterSpec(domain=open(0, oo))]	The number of criminality classes.	5
forces	Annotated[int, ParameterSpec(domain=open(0, oo))]	The number of efficacy classes of the security forces.	3
deterrence	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The rate at which the most effective agents deter a criminal citizen.	2.0

Examples:

>>> from skcomplex.systems.social.crowds import DeterrenceInteraction
>>> DeterrenceInteraction('deterrence', candidate='citizens', field='security').deterrence
2.0

Source code in src/skcomplex/systems/social/crowds/_criminality.py

def __init__(
    self,
    label: str,
    *,
    candidate: Annotated[str, RoleSpec()],
    field: Annotated[str, RoleSpec()],
    classes: Annotated[int, ParameterSpec(domain=RealSet.open(0, oo))] = 5,
    forces: Annotated[int, ParameterSpec(domain=RealSet.open(0, oo))] = 3,
    deterrence: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 2.0,
) -> None:
    super().__init__(label, candidate=candidate, field=field, rate=self._get_rate, kernel=self._get_kernel)
    self.classes = classes
    self.forces = forces
    self.deterrence = deterrence

GuidanceInteraction(label, *, candidate, field, rooms=7, speeds=3, recovery=0.5)

Bases: EncounterInteraction

Encounters with the exit signs that speed pedestrians up again.

It is an EncounterInteraction of the crowd with the exit signs. A pedestrian meets the exit signs at the rate recovery and speeds up one class. Pedestrians outside keep their speed.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
candidate	Annotated[str, RoleSpec()]	The path of the crowd, a distribution over rooms and speed classes.	required
field	Annotated[str, RoleSpec()]	The path of the group of one member whose fixed density is the guidance of the exit signs.	required
rooms	Annotated[int, ParameterSpec(domain=open(0, oo))]	The number of rooms.	7
speeds	Annotated[int, ParameterSpec(domain=open(0, oo))]	The number of speed classes.	3
recovery	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The rate at which pedestrians speed up again.	0.5

Examples:

>>> from skcomplex.systems.social.crowds import GuidanceInteraction
>>> GuidanceInteraction('guidance', candidate='crowd', field='signs').recovery
0.5

Source code in src/skcomplex/systems/social/crowds/_evacuation.py

def __init__(
    self,
    label: str,
    *,
    candidate: Annotated[str, RoleSpec()],
    field: Annotated[str, RoleSpec()],
    rooms: Annotated[int, ParameterSpec(domain=RealSet.open(0, oo))] = 7,
    speeds: Annotated[int, ParameterSpec(domain=RealSet.open(0, oo))] = 3,
    recovery: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
) -> None:
    super().__init__(label, candidate=candidate, field=field, rate=self._get_rate, kernel=self._get_kernel)
    self.rooms = rooms
    self.speeds = speeds
    self.recovery = recovery

KineticCrowdInteraction(label, *, target, length=20.0, width=5.0, cell=2.5, packing=4.0, quality=1.0, stress=0.5, deceleration=0.95, rate=1.0, wall_distance=0.5, dt=0.1, panic_time=None, panic_stress=0.9, band_width=1.0)

Bases: Interaction

Decision process of Bellomo's kinetic crowd model, run by a direct simulation Monte Carlo of the walkers.

The density rho and the mean velocity xi are counted on square cells of side cell of a street of length length, periodic along it, and width width, between walls, with rho = 1 at packing walkers per unit area. Every walker interacts with the crowd with the probability eta_A dt in a step of length dt. It first turns to the preferred direction nu_p ~ rho nu_v + (1 - rho) (beta nu_s + (1 - beta) nu_t) / |.|, where nu_v points down the gradient of the density, nu_s along the mean velocity, and nu_t to its target, and then adjusts its speed to the perceived density rho_theta = rho + d / sqrt(1 + d^2) ((1 - rho) H(d) + rho H(-d)) of the derivative d of the density along its new direction: a walker no faster than the mean speed accelerates to xi + alpha beta (1 - rho_theta) (alpha beta - xi) with the probability alpha beta (1 - rho_theta), and a faster one decelerates to xi - rho_theta xi with the probability (1 - c alpha beta) rho_theta. A walker at a distance d < d_w from a wall and heading to it then reduces the normal component of its velocity to d / d_w of it at constant speed, and every walker moves along its velocity. From the time panic_time on, the stress parameter is panic_stress in place of stress. The interaction reports the band index, the mean of |rho_1 - rho_2| / (rho_1 + rho_2) over the cells of the street where walkers are, of the densities smoothed by a Gaussian of width band_width, and the mean speed.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
target	Annotated[str, RoleSpec()]	The path of the group of walkers.	required
length	Annotated[float, ParameterSpec(domain=open(0, oo))]	The length of the street, periodic along it.	20.0
width	Annotated[float, ParameterSpec(domain=open(0, oo))]	The width of the street, between the walls.	5.0
cell	Annotated[float, ParameterSpec(domain=open(0, oo))]	The side of the square cells of the density and the mean velocity.	2.5
packing	Annotated[float, ParameterSpec(domain=open(0, oo))]	The number of walkers per unit area at the largest density 1.	4.0
quality	Annotated[float, ParameterSpec(domain=closed(0, 1))]	The quality alpha of the venue.	1.0
stress	Annotated[float, ParameterSpec(domain=closed(0, 1))]	The stress beta, the weight of the stream against the target.	0.5
deceleration	Annotated[float, ParameterSpec(domain=closed_open(0, 1))]	The constant c < 1 of the probability of deceleration.	0.95
rate	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The interaction rate eta_A between walkers.	1.0
wall_distance	Annotated[float, ParameterSpec(domain=open(0, oo))]	The distance d_w from a wall below which walkers turn away from it.	0.5
dt	Annotated[float, ParameterSpec(domain=open(0, oo))]	The length of a step.	0.1
panic_time	float | None	The time t_c from which the stress is panic_stress, or None for no panic.	None
panic_stress	Annotated[float, ParameterSpec(domain=closed(0, 1))]	The stress beta_2 after the panic time.	0.9
band_width	Annotated[float, ParameterSpec(domain=open(0, oo))]	The width of the Gaussian smoothing of the densities of the band index.	1.0

Examples:

>>> from skcomplex.systems.social.crowds import KineticCrowdInteraction
>>> KineticCrowdInteraction('decisions', target='walkers').stress
0.5

Source code in src/skcomplex/systems/social/crowds/_kinetic_crowd.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    length: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 20.0,
    width: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 5.0,
    cell: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 2.5,
    packing: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 4.0,
    quality: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 1.0,
    stress: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 0.5,
    deceleration: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, 1))] = 0.95,
    rate: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    wall_distance: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 0.5,
    dt: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 0.1,
    panic_time: float | None = None,
    panic_stress: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 0.9,
    band_width: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
) -> None:
    super().__init__(label)
    self.target = target
    self.length = length
    self.width = width
    self.cell = cell
    self.packing = packing
    self.quality = quality
    self.stress = stress
    self.deceleration = deceleration
    self.rate = rate
    self.wall_distance = wall_distance
    self.dt = dt
    self.panic_time = panic_time
    self.panic_stress = panic_stress
    self.band_width = band_width

walk(system)

Run the decisions, the turns at the walls, and the motion of one step.

Parameters:

Name	Type	Description	Default
system	SystemView	The system.	required

Source code in src/skcomplex/systems/social/crowds/_kinetic_crowd.py

def walk(self, system: SystemView) -> None:
    """Run the decisions, the turns at the walls, and the motion of one step.

    Args:
        system:
            The system.
    """
    views = system.agents(self.target, KineticWalkerView)
    if not views:
        return
    walkers = {
        name: np.array([getattr(view, name) for view in views])
        for name in ('x', 'y', 'speed', 'direction', 'target')
    }
    stress = self.stress
    if self.panic_time is not None and system.time >= self.panic_time:
        stress = self.panic_stress
    velocity = np.stack([np.cos(walkers['direction']), np.sin(walkers['direction'])], axis=1)
    fields = self._get_fields(walkers['x'], walkers['y'], walkers['speed'][:, None] * velocity, walkers['target'])
    self._interact(walkers, fields, stress, system.rng)
    self._avoid_walls(walkers)
    walkers['x'] = np.mod(walkers['x'] + self.dt * walkers['speed'] * np.cos(walkers['direction']), self.length)
    walkers['y'] = np.clip(
        walkers['y'] + self.dt * walkers['speed'] * np.sin(walkers['direction']),
        0.0,
        self.width,
    )
    for position, view in enumerate(views):
        for name in ('x', 'y', 'speed', 'direction'):
            setattr(view, name, float(walkers[name][position]))
    system.report('band_index', self._get_band_index(fields['first'], fields['second']))
    system.report('mean_speed', float(walkers['speed'].mean()))

KineticWalkerElement(label, visual=VisualSpec('agents', {'color_by': 'target', 'color_range': (0.0, 1.0)}), *, x=0.0, y=0.0, speed=0.5, direction=0.0, target=0.0)

Bases: Element

Walker of the kinetic crowd system, with its position, its speed, its direction, and the side it walks to.

Parameters:

Name	Type	Description	Default
label	str	The label of the walker or the group.	required
visual	VisualSpec | None	How the agent is drawn, or None for default_visual_.	VisualSpec('agents', {'color_by': 'target', 'color_range': (0.0, 1.0)})
x	Annotated[float, StateSpec()]	The position along the street.	0.0
y	Annotated[float, StateSpec()]	The position across the street, between the walls at 0 and the width.	0.0
speed	Annotated[float, StateSpec(domain=closed(0, 1))]	The dimensionless speed v, between 0 and the largest speed 1.	0.5
direction	Annotated[float, StateSpec()]	The direction theta of the velocity.	0.0
target	Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))]	The direction of the target, 0 for the exit at the right end and 1 for the exit at the left end.	0.0

Examples:

>>> from skcomplex.systems.social.crowds import KineticWalkerElement
>>> KineticWalkerElement('walker', target=1.0).target
1.0

Source code in src/skcomplex/systems/social/crowds/_kinetic_crowd.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'target', 'color_range': (0.0, 1.0)}),
    *,
    x: Annotated[float, StateSpec()] = 0.0,
    y: Annotated[float, StateSpec()] = 0.0,
    speed: Annotated[float, StateSpec(domain=RealSet.closed(0, 1))] = 0.5,
    direction: Annotated[float, StateSpec()] = 0.0,
    target: Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))] = 0.0,
) -> None:
    super().__init__(label, visual=visual)
    self.x = x
    self.y = y
    self.speed = speed
    self.direction = direction
    self.target = target

KineticWalkerView(agent, stepper)

Bases: AgentView

View of a walker of the kinetic crowd system, with its states as attributes.

Source code in src/skcomplex/core/engine/solvers/_mesa.py

def __init__(self, agent: mesa.Agent, stepper: 'MesaSolver') -> None:
    self._agent = agent
    self._stepper = stepper

MotivationInteraction(label, *, target, speed=1.34, direction=0.0, tau=0.5)

Bases: RelaxationInteraction

Driving force of the social force system, which relaxes a walker's velocity to its desired velocity.

The force per unit mass is (speed * direction - x_d1) / tau, where the velocity x_d1 is the first derivative of the position x.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
target	Annotated[str, RoleSpec()]	The path of the walker or group of walkers.	required
speed	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The desired speed.	1.34
direction	Annotated[ArrayLike, ParameterSpec(vector=True)]	The unit vector of the desired direction.	0.0
tau	Annotated[float, ParameterSpec(domain=open(0, oo))]	The relaxation time.	0.5

Examples:

>>> from skcomplex.systems.social.crowds import MotivationInteraction
>>> MotivationInteraction('drive', target='walkers', direction=[1.0, 0.0]).speed
1.34

Source code in src/skcomplex/systems/social/crowds/_pedestrians.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    speed: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.34,
    direction: Annotated[npt.ArrayLike, ParameterSpec(vector=True)] = 0.0,
    tau: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 0.5,
) -> None:
    super().__init__(label, target=target, state='x_d1')
    self.speed = speed
    self.direction = direction
    self.tau = tau

RecruitmentInteraction(label, *, target, citizens, classes=5, recruitment=0.5, retirement=0.1)

Bases: RateInteraction

Recruitment of security agents in proportion to the mean criminality, a macro-scale external action.

It is a RateInteraction of the density f of the security forces that reads the citizens. The security forces grow at the rate recruitment m share_k in efficacy class k, where m is the mean criminality of the citizens, a moment of their distribution, and retire at the rate retirement.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
target	Annotated[str, RoleSpec()]	The path of the distribution of the security forces over efficacy.	required
citizens	Annotated[str, RoleSpec()]	The path of the distribution of citizens.	required
classes	Annotated[int, ParameterSpec(domain=open(0, oo))]	The number of criminality classes.	5
recruitment	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The recruitment rate per unit of mean criminality.	0.5
retirement	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The retirement rate.	0.1

Examples:

>>> from skcomplex.systems.social.crowds import RecruitmentInteraction
>>> RecruitmentInteraction('recruitment', target='security', citizens='citizens').retirement
0.1

Source code in src/skcomplex/systems/social/crowds/_criminality.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    citizens: Annotated[str, RoleSpec()],
    classes: Annotated[int, ParameterSpec(domain=RealSet.open(0, oo))] = 5,
    recruitment: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
    retirement: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.1,
) -> None:
    super().__init__(label, target=target, state='f', rate=self._get_rate)
    self.citizens = citizens
    self.classes = classes
    self.recruitment = recruitment
    self.retirement = retirement

SecurityAgentsElement(label, visual=None, *, f=0.0, share=1.0)

Bases: Element

Density of security agents at one efficacy class, with the share of new recruits that join the class.

Parameters:

Name	Type	Description	Default
label	str	The label of the distribution.	required
visual	VisualSpec | None	How the agent is drawn, or None for default_visual_.	None
f	Annotated[float, StateSpec(domain=closed_open(0, oo))]	The initial density.	0.0
share	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The share of new recruits that join the efficacy class.	1.0

Examples:

>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.social.crowds import SecurityAgentsElement
>>> GroupElement('security', element=SecurityAgentsElement, size=3, values={'share': [0.5, 0.3, 0.2]}).size
3

Source code in src/skcomplex/systems/social/crowds/_criminality.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    f: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    share: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
) -> None:
    super().__init__(label, visual=visual)
    self.f = f
    self.share = share

SocialRepulsionInteraction(label, first, second, cutoff=1.0, position='x', *, period=None, strength=2.0, falloff=0.3, radius=0.3)

Bases: PairCouplingInteraction

Repulsion between two walkers, the force strength * exp((2 * radius - r) / falloff) of the social force system.

It derives from the pair energy strength * falloff * exp((2 * radius - r) / falloff).

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
first	str	The path of the group or walker of the first members.	required
second	str	The path of the group or walker of the second members.	required
cutoff	float	The distance beyond which walkers do not interact.	1.0
position	str	The name of the vector state that holds the positions of the walkers.	'x'
period	float | None	The side of a periodic box, for minimum image distances, or None for an open space.	None
strength	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The strength of the repulsion.	2.0
falloff	Annotated[float, ParameterSpec(domain=open(0, oo))]	The distance over which the repulsion falls off.	0.3
radius	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The radius of a walker.	0.3

Examples:

>>> from skcomplex.systems.social.crowds import SocialRepulsionInteraction
>>> SocialRepulsionInteraction('repulsion', first='walkers', second='walkers').strength
2.0

Source code in src/skcomplex/systems/social/crowds/_pedestrians.py

def __init__(
    self,
    label: str,
    first: str,
    second: str,
    cutoff: float = 1.0,
    position: str = 'x',
    *,
    period: float | None = None,
    strength: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 2.0,
    falloff: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 0.3,
    radius: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.3,
) -> None:
    super().__init__(
        label,
        first,
        second,
        cutoff=cutoff,
        position=position,
        potential=self._get_pair_energy,
        period=period,
    )
    self.strength = strength
    self.falloff = falloff
    self.radius = radius

WalkingInteraction(label, *, candidate, field, branching=2, depth=2, speeds=3, mobility=0.5)

Bases: EncounterInteraction

Walking of pedestrians to the next room on the way out, an encounter with the exit signs.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
candidate	Annotated[str, RoleSpec()]	The path of the crowd, a distribution over rooms and speed classes.	required
field	Annotated[str, RoleSpec()]	The path of the group of one member whose fixed density is the guidance of the exit signs.	required
branching	Annotated[int, ParameterSpec(domain=open(0, oo))]	The number of rooms behind every door of the building, a balanced tree of rooms with the exit in room 0.	2
depth	Annotated[int, ParameterSpec(domain=open(0, oo))]	The number of doors between the exit room and the rooms furthest from it.	2
speeds	Annotated[int, ParameterSpec(domain=open(0, oo))]	The number of speed classes.	3
mobility	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The rate at which pedestrians at the free speed leave a room.	0.5

Examples:

>>> from skcomplex.systems.social.crowds import WalkingInteraction
>>> WalkingInteraction('walking', candidate='crowd', field='signs').mobility
0.5

Source code in src/skcomplex/systems/social/crowds/_evacuation.py

def __init__(
    self,
    label: str,
    *,
    candidate: Annotated[str, RoleSpec()],
    field: Annotated[str, RoleSpec()],
    branching: Annotated[int, ParameterSpec(domain=RealSet.open(0, oo))] = 2,
    depth: Annotated[int, ParameterSpec(domain=RealSet.open(0, oo))] = 2,
    speeds: Annotated[int, ParameterSpec(domain=RealSet.open(0, oo))] = 3,
    mobility: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
) -> None:
    super().__init__(label, candidate=candidate, field=field, rate=self._get_rate, kernel=self._get_kernel)
    self.branching = branching
    self.depth = depth
    self.speeds = speeds
    self.mobility = mobility

WallsInteraction(label, *, target, width=10.0, strength=10.0, falloff=0.2, radius=0.3)

Bases: GradientInteraction

Repulsion of walkers by two walls of a corridor at y = 0 and y = width.

It is a GradientInteraction of the position x down the potential strength falloff (exp((radius - y) / falloff) + exp((radius - width + y) / falloff)), so the force per unit mass from a wall at distance d is strength * exp((radius - d) / falloff), pointing away from the wall.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
target	Annotated[str, RoleSpec()]	The path of the walker or group of walkers.	required
width	Annotated[float, ParameterSpec(domain=open(0, oo))]	The distance between the walls.	10.0
strength	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The strength of the repulsion.	10.0
falloff	Annotated[float, ParameterSpec(domain=open(0, oo))]	The distance over which the repulsion falls off.	0.2
radius	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The radius of a walker.	0.3

Examples:

>>> from skcomplex.systems.social.crowds import WallsInteraction
>>> WallsInteraction('walls', target='walkers', width=5.0).width
5.0

Source code in src/skcomplex/systems/social/crowds/_pedestrians.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    width: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 10.0,
    strength: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 10.0,
    falloff: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 0.2,
    radius: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.3,
) -> None:
    super().__init__(label, target=target, potential=self._get_potential)
    self.width = width
    self.strength = strength
    self.falloff = falloff
    self.radius = radius

building_of(branching, depth)

Return a building whose rooms form a balanced tree, with the exit door in room 0, the root.

Parameters:

Name	Type	Description	Default
branching	int	The number of rooms behind every door of a room further from the exit.	required
depth	int	The number of doors between the exit room and the rooms furthest from it.	required

Returns:

Name	Type	Description
building	Graph	The network of rooms, labelled 0 to rooms - 1, whose edges are the doors.

Examples:

>>> from skcomplex.systems.social.crowds import building_of
>>> building_of(2, 2).number_of_nodes()
7

Source code in src/skcomplex/systems/social/crowds/_evacuation.py

def building_of(branching: int, depth: int) -> nx.Graph:
    """Return a building whose rooms form a balanced tree, with the exit door in room 0, the root.

    Args:
        branching:
            The number of rooms behind every door of a room further from the exit.

        depth:
            The number of doors between the exit room and the rooms furthest from it.

    Returns:
        building:
            The network of rooms, labelled `0` to `rooms - 1`, whose edges are the doors.

    Examples:
        >>> from skcomplex.systems.social.crowds import building_of
        >>> building_of(2, 2).number_of_nodes()
        7
    """
    return nx.balanced_tree(branching, depth)

contagion_table(classes, attraction, redemption)

Return the transition table of encounters between citizens, A[h, k, i].

A citizen of class h who meets a more criminal citizen moves one class up with probability attraction, and one who meets a more honest citizen moves one class down with probability redemption.

Parameters:

Name	Type	Description	Default
classes	int	The number of criminality classes.	required
attraction	object	The probability that crime attracts, a number or a parameter.	required
redemption	object	The probability that honesty redeems, a number or a parameter.	required

Returns:

Name	Type	Description
table	NDArray[object_]	The transition probabilities, which sum to one over the outcomes.

Examples:

>>> from skcomplex.systems.social.crowds import contagion_table
>>> contagion_table(3, 0.3, 0.1)[0, 2].astype(float).tolist()
[0.7, 0.3, 0.0]

Source code in src/skcomplex/systems/social/crowds/_criminality.py

def contagion_table(classes: int, attraction: object, redemption: object) -> npt.NDArray[np.object_]:
    """Return the transition table of encounters between citizens, `A[h, k, i]`.

    A citizen of class `h` who meets a more criminal citizen moves one class up with probability `attraction`, and one
    who meets a more honest citizen moves one class down with probability `redemption`.

    Args:
        classes:
            The number of criminality classes.

        attraction:
            The probability that crime attracts, a number or a parameter.

        redemption:
            The probability that honesty redeems, a number or a parameter.

    Returns:
        table:
            The transition probabilities, which sum to one over the outcomes.

    Examples:
        >>> from skcomplex.systems.social.crowds import contagion_table
        >>> contagion_table(3, 0.3, 0.1)[0, 2].astype(float).tolist()
        [0.7, 0.3, 0.0]
    """
    table = np.zeros((classes, classes, classes), dtype=object)
    for candidate in range(classes):
        for field in range(classes):
            if field > candidate:
                table[candidate, field, candidate + 1] += attraction
            elif field < candidate:
                table[candidate, field, candidate - 1] += redemption
            table[candidate, field, candidate] += 1 - table[candidate, field].sum()
    return table

criminal_fraction(s, criminality)

Return the fraction of citizens whose criminality is at least one half.

Parameters:

Name	Type	Description	Default
s	SimpleNamespace	The namespace of one member of the distribution.	required
criminality	NDArray[float64]	The criminality of every class.	required

Returns:

Name	Type	Description
fraction	Expression	The moment of the indicator of the criminal classes, divided by the density.

Source code in src/skcomplex/systems/social/crowds/_criminality.py

def criminal_fraction(s: SimpleNamespace, criminality: npt.NDArray[np.float64]) -> Expression:
    """Return the fraction of citizens whose criminality is at least one half.

    Args:
        s:
            The namespace of one member of the distribution.

        criminality:
            The criminality of every class.

    Returns:
        fraction:
            The moment of the indicator of the criminal classes, divided by the density.
    """
    return moment(s.f, 1, (criminality >= CRIMINAL_THRESHOLD).astype(float), normalized=True)

criminality(classes=5, forces=3, *, criminals=0.2, security=0.05, attraction=0.3, redemption=0.1, deterrence=2.0, recruitment=0.5, retirement=0.1)

Return the kinetic system of criminality in a society and the security forces that act against it.

Citizens are a distribution over criminality classes and security agents a distribution over efficacy classes, held by site groups on lattices of the classes. Encounters between citizens spread crime when it attracts more than honesty redeems, encounters with security agents move criminal citizens toward honesty, a micro-scale action, and the security forces grow in proportion to the mean criminality, a macro-scale action that reads a moment of the citizens. The security forces grow while crime is high, crime recedes, and the forces retire, so crime and security oscillate toward a balance where some crime persists.

Parameters:

Name	Type	Description	Default
classes	int	The number of criminality classes.	5
forces	int	The number of efficacy classes of the security forces.	3
criminals	float	The initial fraction of citizens in the most criminal class, the others being honest.	0.2
security	float	The initial density of the security forces, spread evenly over their efficacy classes.	0.05
attraction	float	The probability that a citizen who meets a more criminal one moves one class up.	0.3
redemption	float	The probability that a citizen who meets a more honest one moves one class down.	0.1
deterrence	float	The rate at which the most effective agents deter a criminal citizen.	2.0
recruitment	float	The recruitment rate of security agents per unit of mean criminality.	0.5
retirement	float	The retirement rate of security agents.	0.1

Returns:

Name	Type	Description
system	System	The system, with the distributions citizens and security and the interactions contagion, deterrence, and recruitment. Its standard observables are the mean criminality, the criminal fraction, and the density of the security forces.

Examples:

>>> from skcomplex.systems.social.crowds import criminality
>>> [item.label for item in criminality().interactions]
['contagion', 'deterrence', 'recruitment']

Source code in src/skcomplex/systems/social/crowds/_criminality.py

@register_system(
    field='social/crowds',
    tags=('criminality', 'security', 'kinetic', 'distribution'),
    references=CRIMINALITY_REFERENCES,
    defaults={'until': 60, 'dt': 0.5},
    observables=lambda system: criminality_observables(*_sizes_of(system)),
)
def criminality(
    classes: int = 5,
    forces: int = 3,
    *,
    criminals: float = 0.2,
    security: float = 0.05,
    attraction: float = 0.3,
    redemption: float = 0.1,
    deterrence: float = 2.0,
    recruitment: float = 0.5,
    retirement: float = 0.1,
) -> System:
    """Return the kinetic system of criminality in a society and the security forces that act against it.

    Citizens are a distribution over criminality classes and security agents a distribution over efficacy classes,
    held by site groups on lattices of the classes. Encounters between citizens spread crime when it attracts more than
    honesty redeems, encounters with security agents move criminal citizens toward honesty, a micro-scale action, and
    the security forces grow in proportion to the mean criminality, a macro-scale action that reads a moment of the
    citizens. The security forces grow while crime is high, crime recedes, and the forces retire, so crime and
    security oscillate toward a balance where some crime persists.

    Args:
        classes:
            The number of criminality classes.

        forces:
            The number of efficacy classes of the security forces.

        criminals:
            The initial fraction of citizens in the most criminal class, the others being honest.

        security:
            The initial density of the security forces, spread evenly over their efficacy classes.

        attraction:
            The probability that a citizen who meets a more criminal one moves one class up.

        redemption:
            The probability that a citizen who meets a more honest one moves one class down.

        deterrence:
            The rate at which the most effective agents deter a criminal citizen.

        recruitment:
            The recruitment rate of security agents per unit of mean criminality.

        retirement:
            The retirement rate of security agents.

    Returns:
        system:
            The system, with the distributions `citizens` and `security` and the interactions `contagion`,
            `deterrence`, and `recruitment`. Its standard observables are the mean criminality, the criminal fraction,
            and the density of the security forces.

    Examples:
        >>> from skcomplex.systems.social.crowds import criminality
        >>> [item.label for item in criminality().interactions]
        ['contagion', 'deterrence', 'recruitment']
    """
    citizens = np.zeros(classes)
    citizens[0], citizens[-1] = 1.0 - criminals, criminals
    return System(
        elements=[
            GroupElement('citizens', element=DensityElement, size=classes, on='sites', values={'f': citizens}),
            GroupElement(
                'security',
                element=SecurityAgentsElement,
                size=forces,
                on='sites',
                space=LatticeSpace(forces, boundary='fixed'),
                values={'f': np.full(forces, security / forces), 'share': np.full(forces, 1.0 / forces)},
            ),
        ],
        interactions=[
            CriminalContagionInteraction(
                'contagion',
                candidate='citizens',
                field='citizens',
                classes=classes,
                attraction=attraction,
                redemption=redemption,
            ),
            DeterrenceInteraction(
                'deterrence',
                candidate='citizens',
                field='security',
                classes=classes,
                forces=forces,
                deterrence=deterrence,
            ),
            RecruitmentInteraction(
                'recruitment',
                target='security',
                citizens='citizens',
                classes=classes,
                recruitment=recruitment,
                retirement=retirement,
            ),
        ],
        space=LatticeSpace(classes, boundary='fixed'),
        time=ContinuousTime(),
    )

criminality_grid(classes)

Return the criminality of every class of citizens, from 0 for honest to 1 for criminal.

Parameters:

Name	Type	Description	Default
classes	int	The number of criminality classes.	required

Returns:

Name	Type	Description
criminality	NDArray[float64]	The criminality of every class.

Examples:

>>> from skcomplex.systems.social.crowds import criminality_grid
>>> criminality_grid(3).tolist()
[0.0, 0.5, 1.0]

Source code in src/skcomplex/systems/social/crowds/_criminality.py

def criminality_grid(classes: int) -> npt.NDArray[np.float64]:
    """Return the criminality of every class of citizens, from 0 for honest to 1 for criminal.

    Args:
        classes:
            The number of criminality classes.

    Returns:
        criminality:
            The criminality of every class.

    Examples:
        >>> from skcomplex.systems.social.crowds import criminality_grid
        >>> criminality_grid(3).tolist()
        [0.0, 0.5, 1.0]
    """
    return np.linspace(0.0, 1.0, classes)

criminality_observables(classes, forces)

Return the mean criminality, the criminal fraction, and the density of the security forces of the system.

Parameters:

Name	Type	Description	Default
classes	int	The number of criminality classes.	required
forces	int	The number of efficacy classes of the security forces.	required

Returns:

Name	Type	Description
observables	list[ObservableSpec]	The observables, each reading the moments of a distribution.

Examples:

>>> from skcomplex.systems.social.crowds import criminality_observables
>>> [item.label for item in criminality_observables(5, 3)]
['mean_criminality', 'criminal_fraction', 'security_forces']

Source code in src/skcomplex/systems/social/crowds/_criminality.py

def criminality_observables(classes: int, forces: int) -> list[ObservableSpec]:
    """Return the mean criminality, the criminal fraction, and the density of the security forces of the system.

    Args:
        classes:
            The number of criminality classes.

        forces:
            The number of efficacy classes of the security forces.

    Returns:
        observables:
            The observables, each reading the moments of a distribution.

    Examples:
        >>> from skcomplex.systems.social.crowds import criminality_observables
        >>> [item.label for item in criminality_observables(5, 3)]
        ['mean_criminality', 'criminal_fraction', 'security_forces']
    """
    criminality = criminality_grid(classes)
    return [
        ObservableSpec('mean_criminality', partial(mean_criminality, criminality=criminality), group='citizens'),
        ObservableSpec('criminal_fraction', partial(criminal_fraction, criminality=criminality), group='citizens'),
        ObservableSpec('security_forces', partial(security_density, forces=forces), group='security'),
    ]

crowding_table(rooms, speeds, step)

Return the rate pattern and the transition table of encounters that change the speed class by one step.

With step = -1, two pedestrians in the same room meet and the candidate slows down one class, so crowded rooms walk slower. With step = 1, a pedestrian meets the exit signs and speeds up one class.

Parameters:

Name	Type	Description	Default
rooms	int	The number of rooms.	required
speeds	int	The number of speed classes.	required
step	int	-1 to slow down in encounters within a room, 1 to speed up in encounters with the exit signs.	required

Returns:

Name	Type	Description
pattern	NDArray[float64]	The rate pattern, 1 for the encounters that act.
table	...	The transition table.

Examples:

>>> from skcomplex.systems.social.crowds import crowding_table
>>> crowding_table(1, 2, 1)[0].ravel().tolist()
[1.0, 0.0, 0.0, 0.0]

Source code in src/skcomplex/systems/social/crowds/_evacuation.py

def crowding_table(rooms: int, speeds: int, step: int) -> tuple[npt.NDArray[np.float64], ...]:
    """Return the rate pattern and the transition table of encounters that change the speed class by one step.

    With `step = -1`, two pedestrians in the same room meet and the candidate slows down one class, so crowded rooms
    walk slower. With `step = 1`, a pedestrian meets the exit signs and speeds up one class.

    Args:
        rooms:
            The number of rooms.

        speeds:
            The number of speed classes.

        step:
            `-1` to slow down in encounters within a room, `1` to speed up in encounters with the exit signs.

    Returns:
        pattern:
            The rate pattern, 1 for the encounters that act.

        table:
            The transition table.

    Examples:
        >>> from skcomplex.systems.social.crowds import crowding_table
        >>> crowding_table(1, 2, 1)[0].ravel().tolist()
        [1.0, 0.0, 0.0, 0.0]
    """
    sites = (rooms + 1) * speeds
    fields = sites if step < 0 else 1
    pattern = np.zeros((sites, fields))
    table = np.zeros((sites, fields, sites))
    for site in range(sites):
        room, speed = divmod(site, speeds)
        outcome = min(max(speed + step, 0), speeds - 1)
        table[site, :, room * speeds + outcome] = 1.0
        if room == rooms or outcome == speed:
            continue
        if step > 0:
            pattern[site, 0] = 1.0
        else:
            pattern[site, room * speeds : (room + 1) * speeds] = 1.0
    return pattern, table

deterrence_table(classes, forces)

Return the rate pattern and the transition table of the encounters of citizens with security forces.

A citizen of a criminal class h > 0 who meets a security agent of efficacy v_k moves one class down at a rate proportional to v_k. Honest citizens are left alone.

Parameters:

Name	Type	Description	Default
classes	int	The number of criminality classes.	required
forces	int	The number of efficacy classes of the security forces.	required

Returns:

Name	Type	Description
pattern	NDArray[float64]	The rate pattern eta[h, k], the efficacy v_k for criminal classes and 0 for honest ones.
table	NDArray[float64]	The transition table A[h, k, i].

Examples:

>>> from skcomplex.systems.social.crowds import deterrence_table
>>> deterrence_table(2, 2)[0].tolist()
[[0.0, 0.0], [0.5, 1.0]]

Source code in src/skcomplex/systems/social/crowds/_criminality.py

def deterrence_table(classes: int, forces: int) -> tuple[npt.NDArray[np.float64], npt.NDArray[np.float64]]:
    """Return the rate pattern and the transition table of the encounters of citizens with security forces.

    A citizen of a criminal class `h > 0` who meets a security agent of efficacy `v_k` moves one class down at a rate
    proportional to `v_k`. Honest citizens are left alone.

    Args:
        classes:
            The number of criminality classes.

        forces:
            The number of efficacy classes of the security forces.

    Returns:
        pattern:
            The rate pattern `eta[h, k]`, the efficacy `v_k` for criminal classes and 0 for honest ones.

        table:
            The transition table `A[h, k, i]`.

    Examples:
        >>> from skcomplex.systems.social.crowds import deterrence_table
        >>> deterrence_table(2, 2)[0].tolist()
        [[0.0, 0.0], [0.5, 1.0]]
    """
    efficacy = np.linspace(1.0 / forces, 1.0, forces)
    pattern = np.zeros((classes, forces))
    pattern[1:] = efficacy
    table = np.zeros((classes, forces, classes))
    for candidate in range(classes):
        table[candidate, :, max(candidate - 1, 0)] = 1.0
    return pattern, table

evacuated_fraction(s, rooms, speeds)

Return the fraction of the crowd that has left the building.

Parameters:

Name	Type	Description	Default
s	SimpleNamespace	The namespace of one site of the crowd.	required
rooms	int	The number of rooms.	required
speeds	int	The number of speed classes.	required

Returns:

Name	Type	Description
fraction	Expression	The moment of the indicator of the outside sites, divided by the density.

Source code in src/skcomplex/systems/social/crowds/_evacuation.py

def evacuated_fraction(s: SimpleNamespace, rooms: int, speeds: int) -> Expression:
    """Return the fraction of the crowd that has left the building.

    Args:
        s:
            The namespace of one site of the crowd.

        rooms:
            The number of rooms.

        speeds:
            The number of speed classes.

    Returns:
        fraction:
            The moment of the indicator of the outside sites, divided by the density.
    """
    outside = (np.arange((rooms + 1) * speeds) >= rooms * speeds).astype(float)
    return moment(s.f, 1, outside, normalized=True)

evacuation(branching=2, depth=2, speeds=3, *, mobility=0.5, crowding=2.0, recovery=0.5)

Return the kinetic system of the evacuation of a building whose rooms form a network.

The crowd is a distribution over pairs of a room and a speed class, held by a site group on a two-dimensional lattice indexed by room and speed, with one last row of sites for the outside. The rooms form a balanced tree with the exit in the root room, and the walking encounters read its doors, so pedestrians walk to the next room on the way out at a rate proportional to their speed. Encounters within a room slow pedestrians down, and the exit signs speed them up again, so the crowd queues in the rooms near the exit and the evacuated fraction grows along an S-shaped curve.

The rooms are the nodes and the speed classes the discrete activity of the kinetic structure for a network of Bellomo et al. (2017, sections 3.4 and 3.4.1), and the slowing and guiding encounters follow the behavioural crowd of Bellomo and Gibelli (2015). The crowd of chapter 6 of Bellomo et al. (2017), which moves in continuous space around walls and obstacles, is a different model.

Parameters:

Name	Type	Description	Default
branching	int	The number of rooms behind every door of the building.	2
depth	int	The number of doors between the exit room and the rooms furthest from it.	2
speeds	int	The number of speed classes.	3
mobility	float	The rate at which pedestrians at the free speed leave a room.	0.5
crowding	float	The rate of the encounters within a room that slow pedestrians down.	2.0
recovery	float	The rate at which pedestrians speed up again.	0.5

Returns:

Name	Type	Description
system	System	The system, with the crowd crowd, the exit signs signs, and the interactions walking, crowding, and guidance. Its standard observables are the evacuated fraction and the mean walking speed.

Examples:

>>> from skcomplex.systems.social.crowds import evacuation
>>> evacuation().elements[0].prepare().size_
24

Source code in src/skcomplex/systems/social/crowds/_evacuation.py

@register_system(
    field='social/crowds',
    tags=('crowds', 'evacuation', 'kinetic', 'network', 'distribution'),
    references=EVACUATION_REFERENCES,
    defaults={'until': 40, 'dt': 0.5},
    observables=lambda system: evacuation_observables(*_sizes_of(system)),
)
def evacuation(
    branching: int = 2,
    depth: int = 2,
    speeds: int = 3,
    *,
    mobility: float = 0.5,
    crowding: float = 2.0,
    recovery: float = 0.5,
) -> System:
    """Return the kinetic system of the evacuation of a building whose rooms form a network.

    The crowd is a distribution over pairs of a room and a speed class, held by a site group on a two-dimensional
    lattice indexed by room and speed, with one last row of sites for the outside. The rooms form a balanced tree with
    the exit in the root room, and the walking encounters read its doors, so pedestrians walk to the next room on the
    way out at a rate proportional to their speed. Encounters within a room slow pedestrians down, and the exit signs
    speed them up again, so the crowd queues in the rooms near the exit and the evacuated fraction grows along an
    S-shaped curve.

    The rooms are the nodes and the speed classes the discrete activity of the kinetic structure for a network of
    Bellomo et al. (2017, sections 3.4 and 3.4.1), and the slowing and guiding encounters follow the behavioural crowd
    of Bellomo and Gibelli (2015). The crowd of chapter 6 of Bellomo et al. (2017), which moves in continuous space
    around walls and obstacles, is a different model.

    Args:
        branching:
            The number of rooms behind every door of the building.

        depth:
            The number of doors between the exit room and the rooms furthest from it.

        speeds:
            The number of speed classes.

        mobility:
            The rate at which pedestrians at the free speed leave a room.

        crowding:
            The rate of the encounters within a room that slow pedestrians down.

        recovery:
            The rate at which pedestrians speed up again.

    Returns:
        system:
            The system, with the crowd `crowd`, the exit signs `signs`, and the interactions `walking`, `crowding`, and
            `guidance`.
            Its standard observables are the evacuated fraction and the mean walking speed.

    Examples:
        >>> from skcomplex.systems.social.crowds import evacuation
        >>> evacuation().elements[0].prepare().size_
        24
    """
    rooms = building_of(branching, depth).number_of_nodes()
    initial = np.zeros((rooms + 1, speeds))
    initial[:rooms, speeds // 2] = 1.0 / rooms
    crowd_space = LatticeSpace((rooms + 1, speeds), boundary='fixed')
    return System(
        elements=[
            GroupElement(
                'crowd',
                element=DensityElement,
                on='sites',
                space=crowd_space,
                values={'f': initial.ravel()},
            ),
            GroupElement('signs', element=DensityElement, size=1, values={'f': 1.0}),
        ],
        interactions=[
            WalkingInteraction(
                'walking',
                candidate='crowd',
                field='signs',
                branching=branching,
                depth=depth,
                speeds=speeds,
                mobility=mobility,
            ),
            CrowdingInteraction(
                'crowding',
                candidate='crowd',
                field='crowd',
                rooms=rooms,
                speeds=speeds,
                crowding=crowding,
            ),
            GuidanceInteraction(
                'guidance',
                candidate='crowd',
                field='signs',
                rooms=rooms,
                speeds=speeds,
                recovery=recovery,
            ),
        ],
        space=crowd_space,
        time=ContinuousTime(),
    )

evacuation_observables(rooms, speeds)

Return the evacuated fraction and the mean walking speed of a crowd crowd.

Parameters:

Name	Type	Description	Default
rooms	int	The number of rooms.	required
speeds	int	The number of speed classes.	required

Returns:

Name	Type	Description
observables	list[ObservableSpec]	The observables, each reading the moments of the crowd.

Examples:

>>> from skcomplex.systems.social.crowds import evacuation_observables
>>> [item.label for item in evacuation_observables(7, 3)]
['evacuated', 'mean_speed']

Source code in src/skcomplex/systems/social/crowds/_evacuation.py

def evacuation_observables(rooms: int, speeds: int) -> list[ObservableSpec]:
    """Return the evacuated fraction and the mean walking speed of a crowd `crowd`.

    Args:
        rooms:
            The number of rooms.

        speeds:
            The number of speed classes.

    Returns:
        observables:
            The observables, each reading the moments of the crowd.

    Examples:
        >>> from skcomplex.systems.social.crowds import evacuation_observables
        >>> [item.label for item in evacuation_observables(7, 3)]
        ['evacuated', 'mean_speed']
    """
    return [
        ObservableSpec('evacuated', partial(evacuated_fraction, rooms=rooms, speeds=speeds), group='crowd'),
        ObservableSpec('mean_speed', partial(mean_speed, rooms=rooms, speeds=speeds), group='crowd'),
    ]

kinetic_crowd(n_walkers=50, *, length=20.0, width=5.0, cell=2.5, packing=4.0, quality=1.0, stress=0.5, deceleration=0.95, rate=1.0, wall_distance=0.5, panic_time=None, panic_stress=0.9, random_state=0)

Return Bellomo's kinetic crowd of two groups walking in opposite directions along a street with walls.

Walkers in a continuous venue choose first a direction and then a speed (Bellomo et al. 2017 sections 6.4 and 6.5). The preferred direction weighs the search of less crowded areas, the attraction of the mean stream, and the target, by the density rho and the stress beta (Eq. 6.19). The speed then adjusts to the density perceived along that direction, with the venue quality alpha and the stress (Eqs. 6.22 to 6.24), and walkers near a wall turn parallel to it at constant speed (Eq. 6.25). Panic raises the stress from beta_1 to beta_2 at a critical time (Eqs. 6.27 and 6.28). The kinetic equation is solved, as in the book, by a particle method close to direct simulation Monte Carlo, here with one particle per walker. In space-homogeneous conditions the mean speed follows the fundamental diagram alpha^3 beta^3 (1 - rho)^2 / (alpha^2 beta^2 (1 - rho)^2 + (1 - c alpha beta) rho^2) (Eq. 6.30), which a single cell spanning the street reproduces. The default is the counterflow of section 6.5.2, 50 walkers uniformly distributed in a street periodic along its length, for which the book reports segregation into lanes, measured by the band index of Eq. 6.31, most pronounced at beta = 0.5. The densities are counted on cells of side cell, an assumption of this particle method. The venue is a distinct system from evacuation, a network of rooms. One step is a tenth of a time unit.

Parameters:

Name	Type	Description	Default
n_walkers	int	The number of walkers, half walking to each end.	50
length	float	The length of the street, periodic along it.	20.0
width	float	The width of the street, between the walls.	5.0
cell	float	The side of the square cells of the density and the mean velocity.	2.5
packing	float	The number of walkers per unit area at the largest density 1.	4.0
quality	float	The quality alpha of the venue.	1.0
stress	float	The stress beta, the weight of the stream against the target.	0.5
deceleration	float	The constant c < 1 of the probability of deceleration.	0.95
rate	float	The interaction rate eta_A between walkers.	1.0
wall_distance	float	The distance d_w from a wall below which walkers turn away from it.	0.5
panic_time	float | None	The time t_c, in steps, from which the stress is panic_stress, or None for no panic.	None
panic_stress	float	The stress beta_2 after the panic time.	0.9
random_state	int | Generator | None	The seed or generator of the initial positions and speeds.	0

Returns:

Name	Type	Description
system	System	The system, with the group walkers and the reporters decisions.band_index and decisions.mean_speed.

Examples:

>>> from skcomplex.systems.social.crowds import kinetic_crowd
>>> kinetic_crowd(10).elements[0].prepare().size_
10

Source code in src/skcomplex/systems/social/crowds/_kinetic_crowd.py

@register_system(
    field='social/crowds',
    tags=('pedestrians', 'kinetic theory', 'agents', 'lanes', 'panic', 'monte carlo'),
    references=('Bellomo, Knopoff, and Soler 2017', 'Bellomo, Bellouquid, and Knopoff 2013'),
    defaults={'until': 300},
)
def kinetic_crowd(
    n_walkers: int = 50,
    *,
    length: float = 20.0,
    width: float = 5.0,
    cell: float = 2.5,
    packing: float = 4.0,
    quality: float = 1.0,
    stress: float = 0.5,
    deceleration: float = 0.95,
    rate: float = 1.0,
    wall_distance: float = 0.5,
    panic_time: float | None = None,
    panic_stress: float = 0.9,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return Bellomo's kinetic crowd of two groups walking in opposite directions along a street with walls.

    Walkers in a continuous venue choose first a direction and then a speed (Bellomo et al. 2017 sections 6.4 and
    6.5). The preferred direction weighs the search of less crowded areas, the attraction of the mean stream, and the
    target, by the density `rho` and the stress `beta` (Eq. 6.19). The speed then adjusts to the density perceived
    along that direction, with the venue quality `alpha` and the stress (Eqs. 6.22 to 6.24), and walkers near a wall
    turn parallel to it at constant speed (Eq. 6.25). Panic raises the stress from `beta_1` to `beta_2` at a critical
    time (Eqs. 6.27 and 6.28). The kinetic equation is solved, as in the book, by a particle method close to direct
    simulation Monte Carlo, here with one particle per walker. In space-homogeneous conditions the mean speed follows
    the fundamental diagram `alpha^3 beta^3 (1 - rho)^2 / (alpha^2 beta^2 (1 - rho)^2 + (1 - c alpha beta) rho^2)`
    (Eq. 6.30), which a single cell spanning the street reproduces. The default is the counterflow of section 6.5.2, 50
    walkers uniformly distributed in a street periodic along its length, for which the book reports segregation into
    lanes, measured by the band index of Eq. 6.31, most pronounced at `beta = 0.5`. The densities are counted on cells
    of side `cell`, an assumption of this particle method. The venue is a distinct system from `evacuation`, a network
    of rooms. One step is a tenth of a time unit.

    Args:
        n_walkers:
            The number of walkers, half walking to each end.

        length:
            The length of the street, periodic along it.

        width:
            The width of the street, between the walls.

        cell:
            The side of the square cells of the density and the mean velocity.

        packing:
            The number of walkers per unit area at the largest density 1.

        quality:
            The quality `alpha` of the venue.

        stress:
            The stress `beta`, the weight of the stream against the target.

        deceleration:
            The constant `c < 1` of the probability of deceleration.

        rate:
            The interaction rate `eta_A` between walkers.

        wall_distance:
            The distance `d_w` from a wall below which walkers turn away from it.

        panic_time:
            The time `t_c`, in steps, from which the stress is `panic_stress`, or `None` for no panic.

        panic_stress:
            The stress `beta_2` after the panic time.

        random_state:
            The seed or generator of the initial positions and speeds.

    Returns:
        system:
            The system, with the group `walkers` and the reporters `decisions.band_index` and `decisions.mean_speed`.

    Examples:
        >>> from skcomplex.systems.social.crowds import kinetic_crowd
        >>> kinetic_crowd(10).elements[0].prepare().size_
        10
    """
    rng = np.random.default_rng(random_state)
    target = (np.arange(n_walkers) % 2).astype(float)
    walkers = GroupElement(
        'walkers',
        element=KineticWalkerElement,
        size=n_walkers,
        values={
            'x': rng.random(n_walkers) * length,
            'y': rng.random(n_walkers) * width,
            'speed': 0.5 * rng.random(n_walkers),
            'direction': np.pi * target,
            'target': target,
        },
    )
    decisions = KineticCrowdInteraction(
        'decisions',
        target='walkers',
        length=length,
        width=width,
        cell=cell,
        packing=packing,
        quality=quality,
        stress=stress,
        deceleration=deceleration,
        rate=rate,
        wall_distance=wall_distance,
        panic_time=panic_time,
        panic_stress=panic_stress,
    )
    return System(elements=[walkers], interactions=[decisions])

mean_criminality(s, criminality)

Return the mean criminality of a distribution of citizens.

Parameters:

Name	Type	Description	Default
s	SimpleNamespace	The namespace of one member of the distribution.	required
criminality	NDArray[float64]	The criminality of every class.	required

Returns:

Name	Type	Description
mean	Expression	The normalized first moment.

Source code in src/skcomplex/systems/social/crowds/_criminality.py

def mean_criminality(s: SimpleNamespace, criminality: npt.NDArray[np.float64]) -> Expression:
    """Return the mean criminality of a distribution of citizens.

    Args:
        s:
            The namespace of one member of the distribution.

        criminality:
            The criminality of every class.

    Returns:
        mean:
            The normalized first moment.
    """
    return moment(s.f, 1, criminality, normalized=True)

mean_speed(s, rooms, speeds)

Return the mean walking speed of the crowd, as a fraction of the free speed.

Parameters:

Name	Type	Description	Default
s	SimpleNamespace	The namespace of one site of the crowd.	required
rooms	int	The number of rooms.	required
speeds	int	The number of speed classes.	required

Returns:

Name	Type	Description
speed	Expression	The normalized first moment of the speeds of the sites.

Source code in src/skcomplex/systems/social/crowds/_evacuation.py

def mean_speed(s: SimpleNamespace, rooms: int, speeds: int) -> Expression:
    """Return the mean walking speed of the crowd, as a fraction of the free speed.

    Args:
        s:
            The namespace of one site of the crowd.

        rooms:
            The number of rooms.

        speeds:
            The number of speed classes.

    Returns:
        speed:
            The normalized first moment of the speeds of the sites.
    """
    return moment(s.f, 1, np.tile(walking_speeds(speeds), rooms + 1), normalized=True)

next_rooms(building, exit_room)

Return the next room on the shortest path to the exit room from every room, and the outside for the exit room.

Parameters:

Name	Type	Description	Default
building	Graph	The network of rooms, labelled 0 to rooms - 1, whose edges are the doors.	required
exit_room	int	The room whose door leads outside.	required

Returns:

Name	Type	Description
next_rooms	list[int]	The next room of every room, with rooms standing for the outside.

Examples:

>>> import networkx as nx
>>> from skcomplex.systems.social.crowds import next_rooms
>>> next_rooms(nx.path_graph(3), 0)
[3, 0, 1]

Source code in src/skcomplex/systems/social/crowds/_evacuation.py

def next_rooms(building: nx.Graph, exit_room: int) -> list[int]:
    """Return the next room on the shortest path to the exit room from every room, and the outside for the exit room.

    Args:
        building:
            The network of rooms, labelled `0` to `rooms - 1`, whose edges are the doors.

        exit_room:
            The room whose door leads outside.

    Returns:
        next_rooms:
            The next room of every room, with `rooms` standing for the outside.

    Examples:
        >>> import networkx as nx
        >>> from skcomplex.systems.social.crowds import next_rooms
        >>> next_rooms(nx.path_graph(3), 0)
        [3, 0, 1]
    """
    paths = nx.shortest_path(building, target=exit_room)
    rooms = building.number_of_nodes()
    return [rooms if room == exit_room else int(paths[room][1]) for room in range(rooms)]

security_density(s, forces)

Return the density of the security forces.

Parameters:

Name	Type	Description	Default
s	SimpleNamespace	The namespace of one member of the distribution of the security forces.	required
forces	int	The number of efficacy classes.	required

Returns:

Name	Type	Description
density	Expression	The order-0 moment.

Source code in src/skcomplex/systems/social/crowds/_criminality.py

def security_density(s: SimpleNamespace, forces: int) -> Expression:
    """Return the density of the security forces.

    Args:
        s:
            The namespace of one member of the distribution of the security forces.

        forces:
            The number of efficacy classes.

    Returns:
        density:
            The order-0 moment.
    """
    return moment(s.f, 0, np.arange(forces, dtype=float))

social_force(n_walkers=20, *, length=10.0, speed=1.34, random_state=0)

Return two crowds walking in opposite directions along a periodic corridor, which form lanes.

Parameters:

Name	Type	Description	Default
n_walkers	int	The number of walkers in each direction.	20
length	float	The length and the width of the corridor, periodic along its length.	10.0
speed	float	The desired speed.	1.34
random_state	int | Generator | None	The seed or generator of the initial positions.	0

Returns:

Name	Type	Description
system	System	The system, with the groups rightward and leftward.

Examples:

>>> from skcomplex.systems.social.crowds import social_force
>>> [group.size for group in social_force(5).elements]
[5, 5]

Source code in src/skcomplex/systems/social/crowds/_pedestrians.py

@register_system(
    field='social/crowds',
    tags=('pedestrians', 'particles', 'lanes', 'self-organization'),
    references=('Helbing and Molnár 1995', 'Helbing 2001'),
    defaults={'until': 20.0, 'dt': 0.2, 'solver': RK4Solver(dt=0.05)},
)
def social_force(
    n_walkers: int = 20,
    *,
    length: float = 10.0,
    speed: float = 1.34,
    random_state: int | np.random.Generator | None = 0,
) -> System:
    """Return two crowds walking in opposite directions along a periodic corridor, which form lanes.

    Args:
        n_walkers:
            The number of walkers in each direction.

        length:
            The length and the width of the corridor, periodic along its length.

        speed:
            The desired speed.

        random_state:
            The seed or generator of the initial positions.

    Returns:
        system:
            The system, with the groups `rightward` and `leftward`.

    Examples:
        >>> from skcomplex.systems.social.crowds import social_force
        >>> [group.size for group in social_force(5).elements]
        [5, 5]
    """
    rng = np.random.default_rng(random_state)
    cutoff = 2.0
    walkers = [
        GroupElement(
            label,
            element=PointElement,
            size=n_walkers,
            values={'x': rng.uniform([0.0, 0.5], [length, length - 0.5], (n_walkers, 2))},
        )
        for label in ('rightward', 'leftward')
    ]
    interactions: list[BaseInteraction] = []
    for label, sign in (('rightward', 1.0), ('leftward', -1.0)):
        interactions += [
            MotivationInteraction(f'{label}_drive', target=label, speed=speed, direction=[sign, 0.0]),
            WallsInteraction(f'{label}_walls', target=label, width=length),
        ]
    pairs = (('rightward', 'rightward'), ('leftward', 'leftward'), ('rightward', 'leftward'))
    interactions += [
        SocialRepulsionInteraction(f'{first}_{second}', first=first, second=second, cutoff=cutoff, period=length)
        for first, second in pairs
    ]
    return System(
        elements=walkers,
        interactions=interactions,
        space=ContinuousSpace(extent=(length, length), boundary='periodic'),
    )

walking_speeds(speeds)

Return the walking speed of every speed class, evenly spaced from a quarter of the free speed to the free speed.

Parameters:

Name	Type	Description	Default
speeds	int	The number of speed classes.	required

Returns:

Name	Type	Description
speeds	NDArray[float64]	The walking speed of every class, as a fraction of the free speed.

Examples:

>>> from skcomplex.systems.social.crowds import walking_speeds
>>> walking_speeds(4).tolist()
[0.25, 0.5, 0.75, 1.0]

Source code in src/skcomplex/systems/social/crowds/_evacuation.py

def walking_speeds(speeds: int) -> npt.NDArray[np.float64]:
    """Return the walking speed of every speed class, evenly spaced from a quarter of the free speed to the free speed.

    Args:
        speeds:
            The number of speed classes.

    Returns:
        speeds:
            The walking speed of every class, as a fraction of the free speed.

    Examples:
        >>> from skcomplex.systems.social.crowds import walking_speeds
        >>> walking_speeds(4).tolist()
        [0.25, 0.5, 0.75, 1.0]
    """
    return np.linspace(0.25, 1.0, speeds)

walking_table(building, exit_room, speeds)

Return the rate pattern and the transition table of walking to the next room, an encounter with exit signs.

The sites of the crowd are indexed by room and speed class in row-major order, with one last row of sites for the outside. A pedestrian of speed v in a room walks to the next room on the way to the exit at the rate v.

Parameters:

Name	Type	Description	Default
building	Graph	The network of rooms, labelled 0 to rooms - 1.	required
exit_room	int	The room whose door leads outside.	required
speeds	int	The number of speed classes.	required

Returns:

Name	Type	Description
pattern	NDArray[float64]	The rate pattern eta[h, 0], the walking speed of the site, 0 outside.
table	...	The transition table A[h, 0, i].

Examples:

>>> import networkx as nx
>>> from skcomplex.systems.social.crowds import walking_table
>>> pattern, table = walking_table(nx.path_graph(2), 0, 2)
>>> pattern.ravel().tolist()
[0.25, 1.0, 0.25, 1.0, 0.0, 0.0]

Source code in src/skcomplex/systems/social/crowds/_evacuation.py

def walking_table(building: nx.Graph, exit_room: int, speeds: int) -> tuple[npt.NDArray[np.float64], ...]:
    """Return the rate pattern and the transition table of walking to the next room, an encounter with exit signs.

    The sites of the crowd are indexed by room and speed class in row-major order, with one last row of sites for the
    outside. A pedestrian of speed `v` in a room walks to the next room on the way to the exit at the rate `v`.

    Args:
        building:
            The network of rooms, labelled `0` to `rooms - 1`.

        exit_room:
            The room whose door leads outside.

        speeds:
            The number of speed classes.

    Returns:
        pattern:
            The rate pattern `eta[h, 0]`, the walking speed of the site, 0 outside.

        table:
            The transition table `A[h, 0, i]`.

    Examples:
        >>> import networkx as nx
        >>> from skcomplex.systems.social.crowds import walking_table
        >>> pattern, table = walking_table(nx.path_graph(2), 0, 2)
        >>> pattern.ravel().tolist()
        [0.25, 1.0, 0.25, 1.0, 0.0, 0.0]
    """
    following = next_rooms(building, exit_room)
    velocity = walking_speeds(speeds)
    sites = (len(following) + 1) * speeds
    pattern = np.zeros((sites, 1))
    table = np.zeros((sites, 1, sites))
    for site in range(sites):
        room, speed = divmod(site, speeds)
        if room == len(following):
            table[site, 0, site] = 1.0
            continue
        pattern[site, 0] = velocity[speed]
        table[site, 0, following[room] * speeds + speed] = 1.0
    return pattern, table