Skip to content

Follow Mobility

The Follow Mobility plugin implements a leader/follower pattern where one or more leader nodes broadcast their position and follower nodes autonomously track and move towards them.

How It Works

The plugin is split into two independent components, a leader and a follower, that communicate through the simulation's broadcast messaging system via the Dispatcher plugin.

Leader (MobilityLeaderPlugin)

The leader periodically broadcasts its current position and orientation to all nodes in range. It does not control the node's movement, your protocol is free to move the leader however you like (e.g., via a mission mobility plugin or manual commands). The leader also tracks which followers are currently connected by listening for acknowledgment messages.

Key behaviors:

  • Broadcasts position + orientation at a configurable interval (broadcast_interval).
  • Automatically tracks its own position via telemetry updates.
  • Culls followers that haven't responded within follower_timeout seconds.
  • Exposes a followers property to query currently connected followers.

Follower (MobilityFollowerPlugin)

The follower listens for leader broadcasts and moves to maintain a configurable relative position offset from the followed leader. It periodically scans for available leaders and handles leader disconnections.

Key behaviors:

  • Maintains a list of available leaders based on received broadcasts.
  • By default, auto-follows the first available leader (auto_follow=True). This can be disabled to require explicit calls to follow_leader().
  • Moves to leader_position + relative_position on each leader broadcast.
  • Sends an acknowledgment message back to the leader on each update.
  • Detects leader disconnection after leader_timeout seconds of silence.

Orientation Support

The leader has a configurable orientation (in degrees, counter-clockwise from the +X axis) that is included in every broadcast. When orientation-aware following is enabled (follow_orientation=True, the default), the follower's relative position offset is rotated by the leader's orientation before being applied.

This means the follower formation rotates together with the leader's heading. For example, a follower at relative position (10, 0, 0) (10 units ahead) will always stay 10 units ahead of the leader in the direction the leader is facing, rather than always staying at a fixed world-coordinate offset.

The leader's orientation can be changed at any time via set_orientation().

Mobility Control

The follower plugin controls your protocol's mobility. You should not use any other mobility plugin alongside it or implement custom mobility behavior in a follower protocol. The leader plugin does not affect movement and is safe to combine with other mobility plugins.

Quick Usage Example

from gradysim.protocol.interface import IProtocol
from gradysim.protocol.plugin.follow_mobility import (
    MobilityLeaderPlugin,
    MobilityLeaderConfiguration,
    MobilityFollowerPlugin,
    MobilityFollowerConfiguration,
)


class LeaderProtocol(IProtocol):
    def initialize(self):
        self.leader = MobilityLeaderPlugin(
            self,
            MobilityLeaderConfiguration(
                broadcast_interval=0.1,
                initial_orientation=90.0,  # facing +Y
            ),
        )

    def handle_timer(self, timer):
        pass

    def handle_packet(self, message):
        pass

    def handle_telemetry(self, telemetry):
        pass

    def finish(self):
        pass


class FollowerProtocol(IProtocol):
    def initialize(self):
        self.follower = MobilityFollowerPlugin(
            self,
            MobilityFollowerConfiguration(
                follow_orientation=True,
            ),
        )
        # Stay 10 units behind the leader (relative to leader's heading)
        self.follower.set_relative_position((-10, 0, 0))

    def handle_timer(self, timer):
        pass

    def handle_packet(self, message):
        pass

    def handle_telemetry(self, telemetry):
        pass

    def finish(self):
        pass

API Reference

gradysim.protocol.plugin.follow_mobility

This module declares two plugin for the protocol: a leader and a follower. The leader broadcasts its position and the follower follows it.

Beware that this plugin controls your protocol's mobility to implement its behaviour, so you should not use any other mobility plugin with it or implement any mobility behaviour in your protocol. The MobilityLeaderPlugin does not affect the node's movement and thus should be fine to use with other mobility plugin or mobility behaviour.

MobilityFollowerConfiguration dataclass

Source code in gradysim/protocol/plugin/follow_mobility/follower.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@dataclass
class MobilityFollowerConfiguration:
    scanning_interval: float = 0.5
    """
    Interval between leader scans, in simulation seconds. The follower will update the list of leaders and the current 
    leader.
    """

    leader_timeout: float = 2
    """
    After this amount of simulation seconds without receiving a broadcast from the leader, we consider it
    disconnected
    """

    auto_follow: bool = True
    """
    Automatically follows the first leader available if set to True. If set to False, the user must call 
    follow_leader manually. If True the user can still call follow_leader to follow a specific leader, but if connection
    to that leader is lost the follower will automatically follow the first leader available.
    """

    follow_orientation: bool = True
    """
    Whether the follower's relative position should shift with the leader's orientation.
    If True, relative (X, Y) is rotated by the leader's orientation.
    If False, relative position remains fixed in world coordinates.
    """

auto_follow: bool = True class-attribute instance-attribute

Automatically follows the first leader available if set to True. If set to False, the user must call follow_leader manually. If True the user can still call follow_leader to follow a specific leader, but if connection to that leader is lost the follower will automatically follow the first leader available.

follow_orientation: bool = True class-attribute instance-attribute

Whether the follower's relative position should shift with the leader's orientation. If True, relative (X, Y) is rotated by the leader's orientation. If False, relative position remains fixed in world coordinates.

leader_timeout: float = 2 class-attribute instance-attribute

After this amount of simulation seconds without receiving a broadcast from the leader, we consider it disconnected

scanning_interval: float = 0.5 class-attribute instance-attribute

Interval between leader scans, in simulation seconds. The follower will update the list of leaders and the current leader.

MobilityFollowerPlugin

Source code in gradysim/protocol/plugin/follow_mobility/follower.py 
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
class MobilityFollowerPlugin:
    _leader: Optional[int] = None
    _leader_position: Optional[Position] = None
    _leader_orientation: Optional[float] = None

    _relative_position: Position = (0, 0, 0)

    _last_leader_broadcast: Dict[int, float]

    def __init__(self, protocol: IProtocol,
                 configuration: MobilityFollowerConfiguration = MobilityFollowerConfiguration()):
        self._config = configuration

        self._protocol = protocol
        self._dispatcher = create_dispatcher(protocol)

        self._last_leader_broadcast = {}

        self._initialize_following()
        self._initialize_scanning()

    def _initialize_following(self):
        """Initializes leader following behaviour"""

        def follow_handler(_instance: IProtocol, message: str):
            if not message.startswith(LEADER_TAG):
                return DispatchReturn.CONTINUE

            leader_payload = json.loads(message[len(f"{LEADER_TAG}:"):])
            leader_id = leader_payload["id"]
            self._last_leader_broadcast[leader_id] = self._protocol.provider.current_time()

            if leader_id == self._leader:
                self._leader_position = leader_payload["position"]
                self._leader_orientation = leader_payload.get("orientation", 0.0)

                if self._config.follow_orientation and self._leader_orientation is not None:
                    relative_position = rotate_position_2d(self._relative_position, self._leader_orientation)
                else:
                    relative_position = self._relative_position

                # Going to the leader's position at relative coordinates
                destination = (coord + relative_coord
                               for coord, relative_coord in zip(self._leader_position, relative_position))
                mobility_command = GotoCoordsMobilityCommand(*destination)
                self._protocol.provider.send_mobility_command(mobility_command)

                # Informing the leader that we are following him
                command = CommunicationCommand(
                    CommunicationCommandType.SEND,
                    f"{FOLLOWER_TAG}:{self._protocol.provider.get_id()}",
                    leader_id
                )
                self._protocol.provider.send_communication_command(command)

            return DispatchReturn.INTERRUPT

        self._dispatcher.register_handle_packet(follow_handler)

    def _initialize_scanning(self):
        """Periodically updates the list of current leaders"""

        def scan_handler(_instance: IProtocol, timer: str):
            if timer != FOLLOWER_TIMER_TAG:
                return DispatchReturn.CONTINUE

            self._last_leader_broadcast = {
                leader_id: last_broadcast
                for leader_id, last_broadcast in self._last_leader_broadcast.items()
                if self._protocol.provider.current_time() - last_broadcast < self._config.leader_timeout
            }

            if self._leader is not None and self._leader not in self._last_leader_broadcast:
                self._leader = None
                self._leader_position = None
                self._leader_orientation = None

            if self._leader is None and len(self.available_leaders) > 0:
                self.follow_leader(list(self.available_leaders)[0])

            self._protocol.provider.schedule_timer(
                FOLLOWER_TIMER_TAG,
                self._protocol.provider.current_time() + self._config.scanning_interval
            )
            return DispatchReturn.INTERRUPT

        self._dispatcher.register_handle_timer(scan_handler)

        self._protocol.provider.schedule_timer(FOLLOWER_TIMER_TAG, self._config.scanning_interval)

    @property
    def available_leaders(self) -> Set[int]:
        return set(self._last_leader_broadcast.keys())

    @property
    def current_leader(self) -> Optional[int]:
        return self._leader

    @property
    def relative_position(self) -> Position:
        return self._relative_position

    @property
    def current_leader_position(self) -> Optional[Position]:
        return self._leader_position

    @property
    def current_leader_orientation(self) -> Optional[float]:
        return self._leader_orientation

    def follow_leader(self, leader_id: int) -> None:
        if leader_id not in self.available_leaders:
            raise FollowMobilityException(f"Leader {leader_id} is not available")
        self._leader = leader_id

    def set_relative_position(self, position: Position) -> None:
        self._relative_position = position

MobilityLeaderConfiguration dataclass

Source code in gradysim/protocol/plugin/follow_mobility/leader.py 
14
15
16
17
18
19
20
21
22
23
24
25
@dataclass
class MobilityLeaderConfiguration:
    broadcast_interval: float = 0.02
    """The interval at which the leader broadcasts its position"""

    follower_timeout: float = 5
    """
    If we don't receive a message from a follower for this amount of simulation seconds we consider it disconnected
    """

    initial_orientation: float = 0.0
    """The initial orientation of the leader in degrees (0° along +X, 90° along +Y)"""

broadcast_interval: float = 0.02 class-attribute instance-attribute

The interval at which the leader broadcasts its position

follower_timeout: float = 5 class-attribute instance-attribute

If we don't receive a message from a follower for this amount of simulation seconds we consider it disconnected

initial_orientation: float = 0.0 class-attribute instance-attribute

The initial orientation of the leader in degrees (0° along +X, 90° along +Y)

MobilityLeaderPlugin

Source code in gradysim/protocol/plugin/follow_mobility/leader.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class MobilityLeaderPlugin:
    _position: Position
    _orientation: float

    _last_connection_from_follower: Dict[int, float]
    """Last broadcast round in which a follower was connected"""

    def __init__(self, protocol: IProtocol, configuration: MobilityLeaderConfiguration = MobilityLeaderConfiguration()):
        self._config = configuration
        self._protocol = protocol
        self._dispatcher = create_dispatcher(protocol)
        self._last_connection_from_follower = {}
        self._position = (0, 0, 0)
        self._orientation = configuration.initial_orientation
        self.is_broadcasting = False

        self._initialize_position_watching()
        self._initialize_broadcast()
        self._initialize_listening()

    @property
    def orientation(self) -> float:
        return self._orientation

    def set_orientation(self, orientation: float) -> None:
        self._orientation = orientation

    @property
    def followers(self) -> Set[int]:
        return set(self._last_connection_from_follower.keys())

    def _cull_disconnected_followers(self) -> None:
        """Culls disconnected followers"""
        self._last_connection_from_follower = {
            follower_id: last_broadcast_round
            for follower_id, last_broadcast_round in self._last_connection_from_follower.items()
            if self._broadcast_round - last_broadcast_round < self._config.follower_timeout
        }

    def _initialize_position_watching(self) -> None:
        """Listens for position updates from the position module"""

        def position_handler(_instance: IProtocol, telemetry: Telemetry) -> DispatchReturn:
            self._position = telemetry.current_position
            return DispatchReturn.CONTINUE

        self._dispatcher.register_handle_telemetry(position_handler)

    def _initialize_broadcast(self) -> None:
        """Initializes position broadcast"""

        def broadcast_handler(_instance: IProtocol, timer: str):
            if timer != BROADCAST_TIMER_TAG:
                return DispatchReturn.CONTINUE

            leader_payload = {
                "id": self._protocol.provider.get_id(),
                "position": self._position,
                "orientation": self._orientation
            }

            command = CommunicationCommand(
                CommunicationCommandType.BROADCAST,
                f"{LEADER_TAG}:{json.dumps(leader_payload)}"
            )
            self._protocol.provider.send_communication_command(command)

            self._cull_disconnected_followers()

            self._broadcast_round += 1

            self._protocol.provider.schedule_timer(
                BROADCAST_TIMER_TAG,
                self._protocol.provider.current_time() + self._config.broadcast_interval
            )
            return DispatchReturn.INTERRUPT

        self._dispatcher.register_handle_timer(broadcast_handler)

        self._protocol.provider.schedule_timer(BROADCAST_TIMER_TAG, self._config.broadcast_interval)
        self.is_broadcasting = True
        self._broadcast_round = 0

    def _initialize_listening(self) -> None:
        """Listens for messages from followers"""

        def listen_handler(_instance: IProtocol, message: str):
            if not message.startswith(FOLLOWER_TAG):
                return DispatchReturn.CONTINUE

            follower_id = int(message.split(":")[1])
            self._last_connection_from_follower[follower_id] = self._broadcast_round

            return DispatchReturn.INTERRUPT

        self._dispatcher.register_handle_packet(listen_handler)