Skip to content

sio_events

SioEvents #

Bases: AsyncClientNamespace

Handle all SocketIO events received by Planner.

Source code in cogip/tools/planner/sio_events.py 
 23
 24
 25
 26
 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
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
class SioEvents(socketio.AsyncClientNamespace):
    """
    Handle all SocketIO events received by Planner.
    """

    def __init__(self, planner: "Planner"):
        super().__init__("/planner")
        self.planner = planner
        self.game_context = context.GameContext()

    async def on_connect(self):
        """
        On connection to cogip-server.
        """
        await asyncio.to_thread(
            polling2.poll,
            lambda: self.client.connected is True,
            step=0.2,
            poll_forever=True,
        )
        logger.info("Connected to cogip-server")
        await self.emit("connected")
        await self.emit("register_menu", {"name": "planner", "menu": menu.model_dump()})
        await self.emit("register_menu", {"name": "wizard", "menu": wizard_test_menu.model_dump()})
        if self.planner.robot_id == 1:
            await self.emit("register_menu", {"name": "actuators", "menu": robot_actuators_menu.model_dump()})
        else:
            await self.emit("register_menu", {"name": "actuators", "menu": pami_actuators_menu.model_dump()})
        await self.emit("register_menu", {"name": "cameras", "menu": cameras_menu.model_dump()})

    async def on_disconnect(self):
        """
        On disconnection from cogip-server.
        """
        await self.planner.stop()
        logger.info("Disconnected from cogip-server")

    async def on_connect_error(self, data: dict[str, Any]):
        """
        On connection error, check if a Planner is already connected and exit,
        or retry connection.
        """
        if (
            data
            and isinstance(data, dict)
            and (message := data.get("message"))
            and message == "A planner is already connected"
        ):
            logger.error(f"Connection to cogip-server failed: {message}")
            self.planner.retry_connection = False
            return
        else:
            logger.error(f"Connection to cogip-server failed: {data = }")

    async def on_copilot_connected(self):
        """
        Copilot connected, start planner.
        """
        await self.planner.start()

    async def on_copilot_disconnected(self):
        """
        Copilot disconnected, stop planner.
        """
        await self.planner.stop()

    def on_starter_changed(self, pushed: bool):
        """
        Signal received from the Monitor when the starter state changes in emulation mode.
        """
        if not self.planner.virtual:
            return
        if pushed:
            self.planner.starter.pin.drive_low()
        else:
            self.planner.starter.pin.drive_high()

    async def on_reset(self):
        """
        Callback on reset message from copilot.
        """
        await self.planner.reset()

    async def on_pose_current(self, pose: dict[str, Any]):
        """
        Callback on pose current message.
        """
        self.planner.set_pose_current(models.Pose.model_validate(pose))

    async def on_pose_reached(self):
        """
        Callback on pose reached message.
        """
        await self.planner.sio_receiver_queue.put(self.planner.set_pose_reached())

    async def on_command(self, cmd: str):
        """
        Callback on command message from dashboard.
        """
        await self.planner.command(cmd)

    async def on_config_updated(self, config: dict[str, Any]):
        """
        Callback on config update from dashboard.
        """
        self.planner.update_config(config)

    async def on_obstacles(self, obstacles: dict[str, Any]):
        """
        Callback on obstacles message.
        """
        self.planner.set_obstacles(TypeAdapter(list[models.Vertex]).validate_python(obstacles))

    async def on_wizard(self, message: dict[str, Any]):
        """
        Callback on wizard message.
        """
        await self.planner.wizard_response(message)

    async def on_game_end(self):
        """
        Callback on game end message.
        """
        await self.planner.game_end()

    async def on_actuator_state(self, actuator_state: dict[str, Any]):
        """
        Callback on actuator_state message.
        """
        try:
            state = TypeAdapter(ActuatorState).validate_python(actuator_state)
        except ValidationError as exc:
            logger.warning(f"Failed to decode ActuatorState: {exc}")
            return

        await self.planner.update_actuator_state(state)

on_actuator_state(actuator_state) async #

Callback on actuator_state message.

Source code in cogip/tools/planner/sio_events.py 
148
149
150
151
152
153
154
155
156
157
158
async def on_actuator_state(self, actuator_state: dict[str, Any]):
    """
    Callback on actuator_state message.
    """
    try:
        state = TypeAdapter(ActuatorState).validate_python(actuator_state)
    except ValidationError as exc:
        logger.warning(f"Failed to decode ActuatorState: {exc}")
        return

    await self.planner.update_actuator_state(state)

on_command(cmd) async #

Callback on command message from dashboard.

Source code in cogip/tools/planner/sio_events.py 
118
119
120
121
122
async def on_command(self, cmd: str):
    """
    Callback on command message from dashboard.
    """
    await self.planner.command(cmd)

on_config_updated(config) async #

Callback on config update from dashboard.

Source code in cogip/tools/planner/sio_events.py 
124
125
126
127
128
async def on_config_updated(self, config: dict[str, Any]):
    """
    Callback on config update from dashboard.
    """
    self.planner.update_config(config)

on_connect() async #

On connection to cogip-server.

Source code in cogip/tools/planner/sio_events.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
async def on_connect(self):
    """
    On connection to cogip-server.
    """
    await asyncio.to_thread(
        polling2.poll,
        lambda: self.client.connected is True,
        step=0.2,
        poll_forever=True,
    )
    logger.info("Connected to cogip-server")
    await self.emit("connected")
    await self.emit("register_menu", {"name": "planner", "menu": menu.model_dump()})
    await self.emit("register_menu", {"name": "wizard", "menu": wizard_test_menu.model_dump()})
    if self.planner.robot_id == 1:
        await self.emit("register_menu", {"name": "actuators", "menu": robot_actuators_menu.model_dump()})
    else:
        await self.emit("register_menu", {"name": "actuators", "menu": pami_actuators_menu.model_dump()})
    await self.emit("register_menu", {"name": "cameras", "menu": cameras_menu.model_dump()})

on_connect_error(data) async #

On connection error, check if a Planner is already connected and exit, or retry connection.

Source code in cogip/tools/planner/sio_events.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
async def on_connect_error(self, data: dict[str, Any]):
    """
    On connection error, check if a Planner is already connected and exit,
    or retry connection.
    """
    if (
        data
        and isinstance(data, dict)
        and (message := data.get("message"))
        and message == "A planner is already connected"
    ):
        logger.error(f"Connection to cogip-server failed: {message}")
        self.planner.retry_connection = False
        return
    else:
        logger.error(f"Connection to cogip-server failed: {data = }")

on_copilot_connected() async #

Copilot connected, start planner.

Source code in cogip/tools/planner/sio_events.py 
77
78
79
80
81
async def on_copilot_connected(self):
    """
    Copilot connected, start planner.
    """
    await self.planner.start()

on_copilot_disconnected() async #

Copilot disconnected, stop planner.

Source code in cogip/tools/planner/sio_events.py 
83
84
85
86
87
async def on_copilot_disconnected(self):
    """
    Copilot disconnected, stop planner.
    """
    await self.planner.stop()

on_disconnect() async #

On disconnection from cogip-server.

Source code in cogip/tools/planner/sio_events.py 
53
54
55
56
57
58
async def on_disconnect(self):
    """
    On disconnection from cogip-server.
    """
    await self.planner.stop()
    logger.info("Disconnected from cogip-server")

on_game_end() async #

Callback on game end message.

Source code in cogip/tools/planner/sio_events.py 
142
143
144
145
146
async def on_game_end(self):
    """
    Callback on game end message.
    """
    await self.planner.game_end()

on_obstacles(obstacles) async #

Callback on obstacles message.

Source code in cogip/tools/planner/sio_events.py 
130
131
132
133
134
async def on_obstacles(self, obstacles: dict[str, Any]):
    """
    Callback on obstacles message.
    """
    self.planner.set_obstacles(TypeAdapter(list[models.Vertex]).validate_python(obstacles))

on_pose_current(pose) async #

Callback on pose current message.

Source code in cogip/tools/planner/sio_events.py 
106
107
108
109
110
async def on_pose_current(self, pose: dict[str, Any]):
    """
    Callback on pose current message.
    """
    self.planner.set_pose_current(models.Pose.model_validate(pose))

on_pose_reached() async #

Callback on pose reached message.

Source code in cogip/tools/planner/sio_events.py 
112
113
114
115
116
async def on_pose_reached(self):
    """
    Callback on pose reached message.
    """
    await self.planner.sio_receiver_queue.put(self.planner.set_pose_reached())

on_reset() async #

Callback on reset message from copilot.

Source code in cogip/tools/planner/sio_events.py 
100
101
102
103
104
async def on_reset(self):
    """
    Callback on reset message from copilot.
    """
    await self.planner.reset()

on_starter_changed(pushed) #

Signal received from the Monitor when the starter state changes in emulation mode.

Source code in cogip/tools/planner/sio_events.py 
89
90
91
92
93
94
95
96
97
98
def on_starter_changed(self, pushed: bool):
    """
    Signal received from the Monitor when the starter state changes in emulation mode.
    """
    if not self.planner.virtual:
        return
    if pushed:
        self.planner.starter.pin.drive_low()
    else:
        self.planner.starter.pin.drive_high()

on_wizard(message) async #

Callback on wizard message.

Source code in cogip/tools/planner/sio_events.py 
136
137
138
139
140
async def on_wizard(self, message: dict[str, Any]):
    """
    Callback on wizard message.
    """
    await self.planner.wizard_response(message)