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
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
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", self.planner.virtual)
        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()})
            await self.emit(
                "register_menu", {"name": "actuators_multi", "menu": robot_actuators_multi_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.
        """
        logger.info("[SIO] Copilot connected.")
        await self.planner.start()

    async def on_copilot_disconnected(self):
        """
        Copilot disconnected, stop planner.
        """
        logger.info("[SIO] Copilot disconnected.")
        await self.planner.stop()

    def on_starter_changed(self, pushed: bool):
        """
        Signal received from the Monitor when the starter state changes in emulation mode.
        """
        logger.info(f"[SIO] Starter changed: {pushed}")
        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.
        """
        logger.info("[SIO] Reset.")
        await self.planner.reset()

    async def on_pose_reached(self):
        """
        Callback on pose reached message.
        """
        logger.info("[SIO] Pose reached.")
        await self.planner.sio_receiver_queue.put(self.planner.set_pose_reached())

    async def on_intermediate_pose_reached(self):
        """
        Callback on intermediate pose reached message.
        """
        logger.info("[SIO] Intermediate pose reached.")
        await self.planner.sio_receiver_queue.put(self.planner.set_intermediate_pose_reached())

    async def on_blocked(self):
        """
        Callback on blocked message.
        """
        logger.info("[SIO] Blocked.")
        await self.planner.sio_receiver_queue.put(self.planner.blocked())

    async def on_command(self, cmd: str, *args):
        """
        Callback on command message from dashboard.
        """
        logger.info(f"[SIO] Command: {cmd}")
        await self.planner.command(cmd, *args)

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

    async def on_scservo_updated(self, scservo: dict[str, Any]):
        """
        Callback on scservo update from dashboard.
        """
        await self.planner.update_scservo(scservo)

    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.
        """
        logger.info("[SIO] Game ended.")
        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 
166
167
168
169
170
171
172
173
174
175
176
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_blocked() async #

Callback on blocked message.

Source code in cogip/tools/planner/sio_events.py 
127
128
129
130
131
132
async def on_blocked(self):
    """
    Callback on blocked message.
    """
    logger.info("[SIO] Blocked.")
    await self.planner.sio_receiver_queue.put(self.planner.blocked())

on_command(cmd, *args) async #

Callback on command message from dashboard.

Source code in cogip/tools/planner/sio_events.py 
134
135
136
137
138
139
async def on_command(self, cmd: str, *args):
    """
    Callback on command message from dashboard.
    """
    logger.info(f"[SIO] Command: {cmd}")
    await self.planner.command(cmd, *args)

on_config_updated(config) async #

Callback on config update from dashboard.

Source code in cogip/tools/planner/sio_events.py 
141
142
143
144
145
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
52
53
54
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", self.planner.virtual)
    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()})
        await self.emit(
            "register_menu", {"name": "actuators_multi", "menu": robot_actuators_multi_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 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
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 
80
81
82
83
84
85
async def on_copilot_connected(self):
    """
    Copilot connected, start planner.
    """
    logger.info("[SIO] Copilot connected.")
    await self.planner.start()

on_copilot_disconnected() async #

Copilot disconnected, stop planner.

Source code in cogip/tools/planner/sio_events.py 
87
88
89
90
91
92
async def on_copilot_disconnected(self):
    """
    Copilot disconnected, stop planner.
    """
    logger.info("[SIO] Copilot disconnected.")
    await self.planner.stop()

on_disconnect() async #

On disconnection from cogip-server.

Source code in cogip/tools/planner/sio_events.py 
56
57
58
59
60
61
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 
159
160
161
162
163
164
async def on_game_end(self):
    """
    Callback on game end message.
    """
    logger.info("[SIO] Game ended.")
    await self.planner.game_end()

on_intermediate_pose_reached() async #

Callback on intermediate pose reached message.

Source code in cogip/tools/planner/sio_events.py 
120
121
122
123
124
125
async def on_intermediate_pose_reached(self):
    """
    Callback on intermediate pose reached message.
    """
    logger.info("[SIO] Intermediate pose reached.")
    await self.planner.sio_receiver_queue.put(self.planner.set_intermediate_pose_reached())

on_pose_reached() async #

Callback on pose reached message.

Source code in cogip/tools/planner/sio_events.py 
113
114
115
116
117
118
async def on_pose_reached(self):
    """
    Callback on pose reached message.
    """
    logger.info("[SIO] Pose reached.")
    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 
106
107
108
109
110
111
async def on_reset(self):
    """
    Callback on reset message from copilot.
    """
    logger.info("[SIO] Reset.")
    await self.planner.reset()

on_scservo_updated(scservo) async #

Callback on scservo update from dashboard.

Source code in cogip/tools/planner/sio_events.py 
147
148
149
150
151
async def on_scservo_updated(self, scservo: dict[str, Any]):
    """
    Callback on scservo update from dashboard.
    """
    await self.planner.update_scservo(scservo)

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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def on_starter_changed(self, pushed: bool):
    """
    Signal received from the Monitor when the starter state changes in emulation mode.
    """
    logger.info(f"[SIO] Starter changed: {pushed}")
    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 
153
154
155
156
157
async def on_wizard(self, message: dict[str, Any]):
    """
    Callback on wizard message.
    """
    await self.planner.wizard_response(message)