phyelds.experimental¶

`phyelds.experimental` ¶

Studies in progress: ideas we are still trying out, to understand how they fit phyelds.

What lives here works and is tested, but its API is not settled: it will probably change, and it may change in any release, minor ones included, without a deprecation period. Pin the version of phyelds if you depend on it. When a study settles, it moves to phyelds.libraries.

`phyelds.experimental.state_machines` ¶

Collective state machines: a finite state machine run by the whole network, whose states are collective behaviours and whose transitions are collective agreements.

The model of Macroscopic Design of Swarms with Collective State Machines (Aguzzi et al., ICCCI 2026, https://doi.org/10.1007/978-3-032-36868-3_28), whose definitions the comments cite. Each device runs the program of its state and extends the history of the machine with the prioritised state the program produces; the devices take the maximum of the histories of their neighbors, so that a transition proposed by one device spreads to all of them. Two devices that propose different transitions at once are in conflict: the one with the higher priority wins, then the greater state.

`Next(state, priority=float('-inf'))` `dataclass` ¶

A transition to state, with the priority that settles conflicts.

A transition function that returns a bare state proposes it with the lowest priority.

`StateMachine(*states, window=None)` ¶

A collective state machine described state by state: each state is a function that does what the device does in that state and, from what happens, proposes where to go.

Example

mission = StateMachine("wait", "explore", "defend")

@mission.state("wait", to=["explore", "defend"])
def wait():
    if sense("attack"):
        mission.go("defend", priority=math.inf)
    elif sense("alarm"):
        mission.go("explore")
    return distance_to(sense("base"), neighbors_distances())

state, distance = mission.run()

Parameters:

Name	Type	Description	Default
`*states`	`Any`	The states, such as strings or the members of an enum. The machine starts in the first one, and a conflict of equal priority goes to the one declared later. Devices send the position of a state, not the state, so long names cost nothing in the messages.	`()`
`window`	`float \| None`	Forget the history older than `window` time units, as in `collective_state_machine`; `None` keeps it all.	`None`

Raises:

Type	Description
`ValueError`	If there are no states, or the same state is declared twice.

`state(name, *, to=None)` ¶

Decorate the function of the state name.

The function takes no arguments, and returns what the device does in the state (run returns it). It runs on the state the device agrees on with its neighbors, aligned on it, and proposes a transition with go; one that does not call go stays. A state without a function does nothing, and stays.

Parameters:

Name	Type	Description	Default
`name`	`Any`	The state.	required
`to`	`Sequence[Any] \| None`	The states the function may go to; `go` rejects any other. `None` allows any, but leaves the transitions out of `to_mermaid`.	`None`

Returns:

Type	Description
`Callable`	The decorator, which leaves the function unchanged.

Raises:

Type	Description
`ValueError`	If `name`, or a state in `to`, is not a state of the machine.

`go(target, *, priority=float('-inf'))` ¶

Propose a transition to target, from the function of the current state.

The state produces the greatest of staying, with the lowest priority, and the transitions it proposes (Remark 1 of the paper): a transition to an earlier state needs a higher priority, such as the negated time, to happen.

Parameters:

Name	Type	Description	Default
`target`	`Any`	The next state.	required
`priority`	`float`	The priority of the transition, the lowest by default.	`float('-inf')`

Raises:

Type	Description
`RuntimeError`	If no function of a state is running.
`ValueError`	If `target` is not a state of the machine.

`run(*, clock=None)` ¶

Run a round of the machine on this device.

Parameters:

Name	Type	Description	Default
`clock`	`float \| None`	The time used with the window; `local_time()` if `None`.	`None`

Returns:

Type	Description
`tuple[Any, Any]`	The state the device agreed on, and what its function returned.

Raises:

Type	Description
`ValueError`	If the function of the state goes to a state it does not declare.

`to_mermaid()` ¶

The diagram of the transitions declared with to, in Mermaid syntax; the members of an enum are drawn by name.

`collective_state_machine(initial, transition, *, window=None, clock=None)` ¶

Run a collective state machine (Definition 6): every device agrees on the history of the machine with its neighbors, runs the program of its last state, and extends the history with what the program produced.

Parameters:

Name	Type	Description	Default
`initial`	`S`	The state the machine starts in, with the lowest priority. States are compared with `==` and, to settle conflicts of equal priority, with `<`: use ints, strings or an `IntEnum`. A state received from a device over a codec arrives as its plain value (an `IntEnum` as an `int`).	required
`transition`	`Callable[[S], S \| Next]`	The program of the states: given the current state, the prioritised state it produces, as a `Next` or as a bare state with the lowest priority; the current state to stay. It runs aligned on the current state, so the aggregate operations it calls in a state do not interact with those of the devices in other states. Staying is recorded too, with its priority: a transition with the same priority to a smaller state loses to it, and does not spread.	required
`window`	`float \| None`	Compress the history (Section 3.1): forget the entries older than `window` time units, keeping at least two. If `window` is longer than the time a message takes to cross the network, the machine behaves as with the whole history, which `None` keeps.	`None`
`clock`	`float \| None`	The time on an approximately shared clock, used with `window`; `local_time()` if `None`.	`None`

Returns:

Type	Description
`S`	The state the device agreed on, and ran `transition` on, in this round.

Raises:

Type	Description
`ValueError`	If `window` is not positive, or there is a window but no clock.

`phyelds.experimental.swarm.vector` ¶

Vectors in space: velocities, headings and the offsets of the neighbors.

A Vector is three-dimensional; a planar swarm leaves z at zero, so that the same blocks move robots on the ground and drones in the air.

`Vector(x=0.0, y=0.0, z=0.0)` `dataclass` ¶

A vector in space, Vector() being zero; it travels in messages.

Vectors add, subtract and scale with the usual operators. Vector(*position) is the vector of a position of two or three coordinates, and a vector unpacks into its three components, so that tuple(velocity) is what a simulator or a robot moves by.

`position(dimensions)` ¶

The first dimensions components, as a position of that many coordinates.

`dot(other)` ¶

The dot product.

`cross(other)` ¶

The cross product.

`length()` ¶

The Euclidean length.

`normalize()` ¶

The vector of length one with the same direction; zero stays zero.

`limit(maximum)` ¶

The same vector, shortened to maximum if it is longer.

`rotate(angle)` ¶

The vector rotated counterclockwise by angle radians around the vertical axis.

`phyelds.experimental.swarm.movement` ¶

Movement blocks: what a robot does on its own, or knowing where it is (MacroSwarm, §4.3).

Most need the position sensor: go_to, is_close and explore read it, and neighbors_vectors shares it. The others combine with any block, for instance::

wander = maintain_trajectory(brownian, period=5.0)
velocity = (wander + avoid_obstacles(obstacles, 2.0)).normalize()

`neighbors_vectors(position=None)` ¶

The vector from the device to each neighbor, zero to itself.

Parameters:

Name	Type	Description	Default
`position`	`Sequence[float] \| None`	The position of the device; `local_position()` if `None`. Only relative positions matter, so any frame shared by the neighbors works.	`None`

Returns:

Type	Description
`NeighborhoodField[Vector]`	The field of the vectors from the device to its neighbors.

`stand_still()` ¶

Stay where you are: zero.

`brownian(scale=1.0, *, dimensions=2)` ¶

A random direction, uniform in the plane or in space, of length scale.

Parameters:

Name	Type	Description	Default
`scale`	`float`	The length of the vector.	`1.0`
`dimensions`	`int`	2 for a direction in the horizontal plane, 3 for one in space.	`2`

Returns:

Type	Description
`Vector`	A new random vector at every call.

`go_to(target, speed=1.0)` ¶

Head to target at speed; zero once on it.

Parameters:

Name	Type	Description	Default
`target`	`Sequence[float]`	The position to reach.	required
`speed`	`float`	The length of the velocity.	`1.0`

Returns:

Type	Description
`Vector`	The velocity towards the target.

`is_close(target, distance)` ¶

Whether the device is within distance of target.

`explore(minimum, maximum, speed=1.0, *, tolerance=None)` ¶

Wander inside a box: head to a random goal in it, and draw a new one on reaching it.

Parameters:

Name	Type	Description	Default
`minimum`	`Sequence[float]`	The corner of the box with the lowest coordinates.	required
`maximum`	`Sequence[float]`	The opposite corner.	required
`speed`	`float`	The length of the velocity.	`1.0`
`tolerance`	`float \| None`	How close a goal counts as reached; `speed` if `None`.	`None`

Returns:

Type	Description
`Vector`	The velocity towards the current goal.

`maintain_trajectory(velocity, period)` ¶

Keep a velocity for period time units, then draw the next one.

Parameters:

Name	Type	Description	Default
`velocity`	`Callable[[], Vector]`	What draws a velocity, such as `brownian`. It runs in every round, so that the aggregate operations it calls stay aligned.	required
`period`	`float`	How long each velocity lasts.	required

Returns:

Type	Description
`Vector`	The velocity drawn last.

`maintain_until(velocity, condition)` ¶

Keep velocity until condition holds, then stop.

`avoid_obstacles(obstacles, min_distance, norm=100.0)` ¶

Move away from the obstacles closer than min_distance, the more the closer.

Parameters:

Name	Type	Description	Default
`obstacles`	`Iterable[Vector]`	The vector from the device to each obstacle it senses.	required
`min_distance`	`float`	The distance below which an obstacle repels.	required
`norm`	`float`	What the repulsion is divided by: each obstacle at distance `d` adds `(min_distance - d) / norm` away from it.	`100.0`

Returns:

Type	Description
`Vector`	The sum of the repulsions.

`phyelds.experimental.swarm.flocking` ¶

Flocking blocks: moving together, as Reynolds, Vicsek and Cucker-Smale (MacroSwarm, §4.4).

Each block takes the velocity of the device in the previous round, since flocking iterates on velocities; remember it across rounds::

velocity = remember_and_evolve(Vector(), lambda previous: vicsek(previous.value)).value

The neighbors a block looks at can be restricted, as the queries of MacroSwarm: to those within a distance, or to the nearest few. The device itself never counts as one, but in vicsek, which averages its velocity with theirs.

`separation(velocity, *, within=None, nearest=None)` ¶

Steer away from the neighbors: the heading away from their mean, minus velocity.

Parameters:

Name	Type	Description	Default
`velocity`	`Vector`	The velocity of the device in the previous round.	required
`within`	`float \| None`	Only the neighbors closer than this.	`None`
`nearest`	`int \| None`	Only this many neighbors, the closest.	`None`

Returns:

Type	Description
`Vector`	The steering; zero without neighbors.

`alignment(velocity, *, within=None, nearest=None)` ¶

Steer as the neighbors move: the heading of their mean velocity, minus velocity.

Parameters:

Name	Type	Description	Default
`velocity`	`Vector`	The velocity of the device in the previous round, shared with them.	required
`within`	`float \| None`	Only the neighbors closer than this.	`None`
`nearest`	`int \| None`	Only this many neighbors, the closest.	`None`

Returns:

Type	Description
`Vector`	The steering; zero without neighbors.

`cohesion(velocity, *, within=None, nearest=None)` ¶

Steer towards the neighbors: the heading to their mean position, minus velocity.

Parameters:

Name	Type	Description	Default
`velocity`	`Vector`	The velocity of the device in the previous round.	required
`within`	`float \| None`	Only the neighbors closer than this.	`None`
`nearest`	`int \| None`	Only this many neighbors, the closest.	`None`

Returns:

Type	Description
`Vector`	The steering; zero without neighbors.

`reynolds(velocity, *, vision=None, separation_within=None, separation_weight=1.0, alignment_weight=1.0, cohesion_weight=1.0)` ¶

The boids of Reynolds: velocity plus the weighted headings of separation, alignment and cohesion.

Parameters:

Name	Type	Description	Default
`velocity`	`Vector`	The velocity of the device in the previous round.	required
`vision`	`float \| None`	The range of alignment and cohesion; all the neighbors if `None`.	`None`
`separation_within`	`float \| None`	The range of separation, usually shorter; all the neighbors if `None`.	`None`
`separation_weight`	`float`	The weight of separation.	`1.0`
`alignment_weight`	`float`	The weight of alignment.	`1.0`
`cohesion_weight`	`float`	The weight of cohesion.	`1.0`

Returns:

Type	Description
`Vector`	The new velocity, not normalized.

`vicsek(velocity, *, alpha=0.1, noise=0.1, within=None, nearest=None)` ¶

The model of Vicsek: move towards the mean velocity of the neighbors and the device, with some noise.

Parameters:

Name	Type	Description	Default
`velocity`	`Vector`	The velocity of the device in the previous round, shared with them.	required
`alpha`	`float`	How much of the mean replaces `velocity`, from 0 to 1.	`0.1`
`noise`	`float`	The length of the random vector added.	`0.1`
`within`	`float \| None`	Only the neighbors closer than this.	`None`
`nearest`	`int \| None`	Only this many neighbors, the closest.	`None`

Returns:

Type	Description
`Vector`	The new velocity.

`cucker_smale(velocity, strength, *, noise=0.1, within=None, nearest=None)` ¶

The model of Cucker and Smale: move towards the velocity of each neighbor, the more the closer it is, with some noise.

Parameters:

Name	Type	Description	Default
`velocity`	`Vector`	The velocity of the device in the previous round, shared with them.	required
`strength`	`float`	How much a neighbor at distance `d` pulls: `strength / d`. The pulls of all the neighbors should add up to less than one, or the velocities overshoot each other and oscillate.	required
`noise`	`float`	The length of the random vector added.	`0.1`
`within`	`float \| None`	Only the neighbors closer than this.	`None`
`nearest`	`int \| None`	Only this many neighbors, the closest.	`None`

Returns:

Type	Description
`Vector`	The new velocity.

`phyelds.experimental.swarm.leader` ¶

Leader-based blocks: following a leader, and splitting the swarm in teams (MacroSwarm, §4.5 and §4.6).

A leader is a Boolean field, true on the leaders: chosen in advance, sensed, or elected. Each device follows the nearest leader, along the network.

`Team(leader, formed, velocity)` `dataclass` ¶

The team of a device: its leader, whether it is formed, and the velocity that gathers it.

`inside(behaviour)` ¶

Run behaviour once the team is formed, apart from the other teams.

Parameters:

Name	Type	Description	Default
`behaviour`	`Callable[[int], Vector]`	Given the leader of the team, the velocity of the device; it runs aligned on the leader, so that its aggregate operations stay in the team.	required

Returns:

Type	Description
`Vector`	What `behaviour` returns; the velocity gathering the team until it is formed.

`cast_along(source, value, step, offsets=None)` ¶

Spread value from the nearest source along the shortest paths, changing it on the way by where each neighbor is, such as summing the vectors to the source.

Parameters:

Name	Type	Description	Default
`source`	`bool`	Whether the device is a source.	required
`value`	`object`	The value of a source, and of a device away from any.	required
`step`	`Callable[[object, Vector], object]`	Given the value of the neighbor on the shortest path to the source and the vector to that neighbor, the value of the device.	required
`offsets`	`NeighborhoodField[Vector] \| None`	The vectors from the device to its neighbors, which also measure the paths; `neighbors_vectors()` if `None`.	`None`

Returns:

Type	Description
`tuple[float, object]`	The distance to the nearest source, infinite away from any, and the value.

`align_with_leader(leader, velocity)` ¶

Move as the leader: its velocity, spread to its area of influence.

Parameters:

Name	Type	Description	Default
`leader`	`bool`	Whether the device is a leader.	required
`velocity`	`Vector`	The velocity of the device; that of the leader is the one spread.	required

Returns:

Type	Description
`Vector`	The velocity of the nearest leader; `velocity` away from any.

`sink_at(leader)` ¶

Head to the nearest leader: the direction of the shortest path to it, of length one.

Parameters:

Name	Type	Description	Default
`leader`	`bool`	Whether the device is a leader.	required

Returns:

Type	Description
`Vector`	The heading to the leader; zero on the leader and away from any.

`spin_around(center)` ¶

Turn clockwise around the nearest center, seen from above.

Parameters:

Name	Type	Description	Default
`center`	`bool`	Whether the device is a center.	required

Returns:

Type	Description
`Vector`	The heading perpendicular to that of `sink_at`, of length one.

`is_team_formed(leader, target_distance, necessary=1)` ¶

Whether every member of the team of the nearest leader has its closest necessary neighbors within target_distance on average.

Parameters:

Name	Type	Description	Default
`leader`	`bool`	Whether the device is a leader.	required
`target_distance`	`float`	The mean distance to the closest neighbors that is close enough.	required
`necessary`	`int`	How many neighbors count.	`1`

Returns:

Type	Description
`bool`	What the leader gathered from its team; false until the leader is known.

`team_formation(intra_distance, extra_distance, *, separation_weight=1.0, condition=_no_condition, leader=None)` ¶

Split the swarm in teams: elect leaders, and gather each team around its leader until condition holds; from then on, the device stays in its team.

Parameters:

Name	Type	Description	Default
`intra_distance`	`float`	The distance the members keep from each other.	required
`extra_distance`	`float`	The area of influence of a leader, where no other is elected.	required
`separation_weight`	`float`	The weight of keeping apart, against gathering.	`1.0`
`condition`	`Callable[[bool], bool]`	Given whether the device is a leader, whether its team is formed, such as `lambda leader: is_team_formed(leader, 2.0)`.	`_no_condition`
`leader`	`bool \| None`	Whether the device is a leader, to choose the leaders instead of electing them.	`None`

Returns:

Type	Description
`Team`	The team of the device.

phyelds.experimental¶

phyelds.experimental ¶

phyelds.experimental.state_machines ¶

Next(state, priority=float('-inf')) dataclass ¶

StateMachine(*states, window=None) ¶

state(name, *, to=None) ¶

go(target, *, priority=float('-inf')) ¶

run(*, clock=None) ¶

to_mermaid() ¶

collective_state_machine(initial, transition, *, window=None, clock=None) ¶

phyelds.experimental.swarm.vector ¶

Vector(x=0.0, y=0.0, z=0.0) dataclass ¶

position(dimensions) ¶

dot(other) ¶

cross(other) ¶

length() ¶

normalize() ¶

limit(maximum) ¶

rotate(angle) ¶

phyelds.experimental.swarm.movement ¶

neighbors_vectors(position=None) ¶

stand_still() ¶

brownian(scale=1.0, *, dimensions=2) ¶

go_to(target, speed=1.0) ¶

is_close(target, distance) ¶

explore(minimum, maximum, speed=1.0, *, tolerance=None) ¶

maintain_trajectory(velocity, period) ¶

maintain_until(velocity, condition) ¶

avoid_obstacles(obstacles, min_distance, norm=100.0) ¶

phyelds.experimental.swarm.flocking ¶

separation(velocity, *, within=None, nearest=None) ¶

alignment(velocity, *, within=None, nearest=None) ¶

cohesion(velocity, *, within=None, nearest=None) ¶

reynolds(velocity, *, vision=None, separation_within=None, separation_weight=1.0, alignment_weight=1.0, cohesion_weight=1.0) ¶

vicsek(velocity, *, alpha=0.1, noise=0.1, within=None, nearest=None) ¶

cucker_smale(velocity, strength, *, noise=0.1, within=None, nearest=None) ¶

phyelds.experimental.swarm.leader ¶

Team(leader, formed, velocity) dataclass ¶

inside(behaviour) ¶

cast_along(source, value, step, offsets=None) ¶

align_with_leader(leader, velocity) ¶

sink_at(leader) ¶

spin_around(center) ¶

is_team_formed(leader, target_distance, necessary=1) ¶

team_formation(intra_distance, extra_distance, *, separation_weight=1.0, condition=_no_condition, leader=None) ¶

`phyelds.experimental` ¶

`phyelds.experimental.state_machines` ¶

`Next(state, priority=float('-inf'))` `dataclass` ¶

`StateMachine(*states, window=None)` ¶

`state(name, *, to=None)` ¶

`go(target, *, priority=float('-inf'))` ¶

`run(*, clock=None)` ¶

`to_mermaid()` ¶

`collective_state_machine(initial, transition, *, window=None, clock=None)` ¶

`phyelds.experimental.swarm.vector` ¶

`Vector(x=0.0, y=0.0, z=0.0)` `dataclass` ¶

`position(dimensions)` ¶

`dot(other)` ¶

`cross(other)` ¶

`length()` ¶

`normalize()` ¶

`limit(maximum)` ¶

`rotate(angle)` ¶

`phyelds.experimental.swarm.movement` ¶

`neighbors_vectors(position=None)` ¶

`stand_still()` ¶

`brownian(scale=1.0, *, dimensions=2)` ¶

`go_to(target, speed=1.0)` ¶

`is_close(target, distance)` ¶

`explore(minimum, maximum, speed=1.0, *, tolerance=None)` ¶

`maintain_trajectory(velocity, period)` ¶

`maintain_until(velocity, condition)` ¶

`avoid_obstacles(obstacles, min_distance, norm=100.0)` ¶

`phyelds.experimental.swarm.flocking` ¶

`separation(velocity, *, within=None, nearest=None)` ¶

`alignment(velocity, *, within=None, nearest=None)` ¶

`cohesion(velocity, *, within=None, nearest=None)` ¶

`reynolds(velocity, *, vision=None, separation_within=None, separation_weight=1.0, alignment_weight=1.0, cohesion_weight=1.0)` ¶

`vicsek(velocity, *, alpha=0.1, noise=0.1, within=None, nearest=None)` ¶

`cucker_smale(velocity, strength, *, noise=0.1, within=None, nearest=None)` ¶

`phyelds.experimental.swarm.leader` ¶

`Team(leader, formed, velocity)` `dataclass` ¶

`inside(behaviour)` ¶

`cast_along(source, value, step, offsets=None)` ¶

`align_with_leader(leader, velocity)` ¶

`sink_at(leader)` ¶

`spin_around(center)` ¶

`is_team_formed(leader, target_distance, necessary=1)` ¶

`team_formation(intra_distance, extra_distance, *, separation_weight=1.0, condition=_no_condition, leader=None)` ¶