Skip to content

sio_events

SioEvents #

Bases: AsyncClientNamespace

Handle all SocketIO events received by Planner.

Source code in cogip/tools/planner/sio_events.py 
 22
 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
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_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_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 
135
136
137
138
139
140
141
142
143
144
145
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 
111
112
113
114
115
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 
117
118
119
120
121
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 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
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 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
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 
76
77
78
79
80
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 
82
83
84
85
86
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 
52
53
54
55
56
57
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 
129
130
131
132
133
async def on_game_end(self):
    """
    Callback on game end message.
    """
    await self.planner.game_end()

on_pose_reached() async #

Callback on pose reached message.

Source code in cogip/tools/planner/sio_events.py 
105
106
107
108
109
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 
 99
100
101
102
103
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 
88
89
90
91
92
93
94
95
96
97
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 
123
124
125
126
127
async def on_wizard(self, message: dict[str, Any]):
    """
    Callback on wizard message.
    """
    await self.planner.wizard_response(message)