Skip to content

copilot

CopilotNamespace #

Bases: AsyncNamespace

Handle all SocketIO events related to copilot.

Source code in cogip/tools/server/namespaces/copilot.py 
 10
 11
 12
 13
 14
 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
 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
class CopilotNamespace(socketio.AsyncNamespace):
    """
    Handle all SocketIO events related to copilot.
    """

    def __init__(self, cogip_server: "server.Server"):
        super().__init__("/copilot")
        self.cogip_server = cogip_server
        self.context = Context()
        self.context.copilot_sid = None

    async def on_connect(self, sid, environ):
        if self.context.copilot_sid:
            message = "A copilot is already connected"
            logger.error(f"Copilot connection refused: {message}")
            raise ConnectionRefusedError(message)
        self.context.copilot_sid = sid

    async def on_connected(self, sid):
        logger.info("Copilot connected.")
        await self.emit("copilot_connected", namespace="/planner")

    async def on_disconnect(self, sid):
        self.context.copilot_sid = None
        self.context.shell_menu = None
        await self.emit("copilot_disconnected", namespace="/planner")
        logger.info("Copilot disconnected.")

    async def on_reset(self, sid) -> None:
        """
        Callback on reset event.
        """
        logger.info("[copilot => planner] reset.")
        await self.emit("reset", namespace="/planner")

    async def on_register_menu(self, sid, data: dict[str, Any]):
        """
        Callback on register_menu.
        """
        await self.cogip_server.register_menu("copilot", data)

    async def on_pose_reached(self, sid) -> None:
        """
        Callback on pose reached message.
        """
        logger.info("[copilot => planner] Pose reached.")
        await self.emit("pose_reached", namespace="/planner")

    async def on_intermediate_pose_reached(self, sid) -> None:
        """
        Callback on intermediate pose reached message.
        """
        logger.info("[copilot => planner] Intermediate pose reached.")
        await self.emit("intermediate_pose_reached", namespace="/planner")

    async def on_blocked(self, sid) -> None:
        """
        Callback on blocked message.
        """
        logger.info("[copilot => planner] Blocked.")
        await self.emit("blocked", namespace="/planner")

    async def on_menu(self, sid, menu):
        """
        Callback on menu event.
        """
        self.context.shell_menu = models.ShellMenu.model_validate(menu)
        await self.emit("shell_menu", (self.context.robot_id, menu), namespace="/dashboard")

    async def on_state(self, sid, state):
        """
        Callback on state event.
        """
        await self.emit("state", (self.context.robot_id, state), namespace="/dashboard")

    async def on_actuator_state(self, sid, actuator_state: dict[str, Any]):
        """
        Callback on actuator_state message.
        """
        await self.emit("actuator_state", actuator_state, namespace="/planner")
        await self.emit("actuator_state", actuator_state, namespace="/dashboard")

    async def on_pid(self, sid, pid: dict[str, Any]):
        """
        Callback on pid message.
        """
        await self.emit("pid", pid, namespace="/dashboard")

    async def on_config(self, sid, config: dict[str, Any]):
        """
        Callback on config message.
        """
        await self.emit("config", config, namespace="/dashboard")

    async def on_game_end(self, sid) -> None:
        """
        Callback on game end message.
        """
        logger.info("[copilot => planner] Game end.")
        await self.emit("game_end", namespace="/planner")

on_actuator_state(sid, actuator_state) async #

Callback on actuator_state message.

Source code in cogip/tools/server/namespaces/copilot.py 
85
86
87
88
89
90
async def on_actuator_state(self, sid, actuator_state: dict[str, Any]):
    """
    Callback on actuator_state message.
    """
    await self.emit("actuator_state", actuator_state, namespace="/planner")
    await self.emit("actuator_state", actuator_state, namespace="/dashboard")

on_blocked(sid) async #

Callback on blocked message.

Source code in cogip/tools/server/namespaces/copilot.py 
65
66
67
68
69
70
async def on_blocked(self, sid) -> None:
    """
    Callback on blocked message.
    """
    logger.info("[copilot => planner] Blocked.")
    await self.emit("blocked", namespace="/planner")

on_config(sid, config) async #

Callback on config message.

Source code in cogip/tools/server/namespaces/copilot.py 
 98
 99
100
101
102
async def on_config(self, sid, config: dict[str, Any]):
    """
    Callback on config message.
    """
    await self.emit("config", config, namespace="/dashboard")

on_game_end(sid) async #

Callback on game end message.

Source code in cogip/tools/server/namespaces/copilot.py 
104
105
106
107
108
109
async def on_game_end(self, sid) -> None:
    """
    Callback on game end message.
    """
    logger.info("[copilot => planner] Game end.")
    await self.emit("game_end", namespace="/planner")

on_intermediate_pose_reached(sid) async #

Callback on intermediate pose reached message.

Source code in cogip/tools/server/namespaces/copilot.py 
58
59
60
61
62
63
async def on_intermediate_pose_reached(self, sid) -> None:
    """
    Callback on intermediate pose reached message.
    """
    logger.info("[copilot => planner] Intermediate pose reached.")
    await self.emit("intermediate_pose_reached", namespace="/planner")

on_menu(sid, menu) async #

Callback on menu event.

Source code in cogip/tools/server/namespaces/copilot.py 
72
73
74
75
76
77
async def on_menu(self, sid, menu):
    """
    Callback on menu event.
    """
    self.context.shell_menu = models.ShellMenu.model_validate(menu)
    await self.emit("shell_menu", (self.context.robot_id, menu), namespace="/dashboard")

on_pid(sid, pid) async #

Callback on pid message.

Source code in cogip/tools/server/namespaces/copilot.py 
92
93
94
95
96
async def on_pid(self, sid, pid: dict[str, Any]):
    """
    Callback on pid message.
    """
    await self.emit("pid", pid, namespace="/dashboard")

on_pose_reached(sid) async #

Callback on pose reached message.

Source code in cogip/tools/server/namespaces/copilot.py 
51
52
53
54
55
56
async def on_pose_reached(self, sid) -> None:
    """
    Callback on pose reached message.
    """
    logger.info("[copilot => planner] Pose reached.")
    await self.emit("pose_reached", namespace="/planner")

on_register_menu(sid, data) async #

Callback on register_menu.

Source code in cogip/tools/server/namespaces/copilot.py 
45
46
47
48
49
async def on_register_menu(self, sid, data: dict[str, Any]):
    """
    Callback on register_menu.
    """
    await self.cogip_server.register_menu("copilot", data)

on_reset(sid) async #

Callback on reset event.

Source code in cogip/tools/server/namespaces/copilot.py 
38
39
40
41
42
43
async def on_reset(self, sid) -> None:
    """
    Callback on reset event.
    """
    logger.info("[copilot => planner] reset.")
    await self.emit("reset", namespace="/planner")

on_state(sid, state) async #

Callback on state event.

Source code in cogip/tools/server/namespaces/copilot.py 
79
80
81
82
83
async def on_state(self, sid, state):
    """
    Callback on state event.
    """
    await self.emit("state", (self.context.robot_id, state), namespace="/dashboard")