Skip to content

dashboard

DashboardNamespace #

Bases: AsyncNamespace

Handle all SocketIO events related to dashboards.

Source code in cogip/tools/server/namespaces/dashboard.py 
 9
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
class DashboardNamespace(socketio.AsyncNamespace):
    """
    Handle all SocketIO events related to dashboards.
    """

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

    async def on_connect(self, sid, environ):
        pass

    async def on_connected(self, sid):
        logger.info("Dashboard connected.")
        await self.emit("tool_menu", self.context.tool_menus[self.context.current_tool_menu].model_dump(), to=sid)

        if self.context.shell_menu:
            await self.emit("shell_menu", (self.context.robot_id, self.context.shell_menu.model_dump()), to=sid)

    def on_disconnect(self, sid):
        logger.info("Dashboard disconnected.")

    async def on_tool_cmd(self, sid, cmd: str) -> None:
        """
        Callback on tool command message from dashboard.
        """
        # Find entry in current menu
        entry = None
        for entry in self.context.tool_menus[self.context.current_tool_menu].entries:
            if entry.cmd == cmd:
                break

        # Check if it corresponds to a menu or a command
        if entry and entry.cmd in self.context.tool_menus:
            # Enter a menu
            self.context.current_tool_menu = cmd
            await self.emit(
                "tool_menu",
                self.context.tool_menus[self.context.current_tool_menu].model_dump(),
                namespace="/dashboard",
            )
        else:
            # Forward command to corresponding namespace
            if cmd == "exit":
                self.context.current_tool_menu = "root"
                await self.emit(
                    "tool_menu",
                    self.context.tool_menus[self.context.current_tool_menu].model_dump(),
                    namespace="/dashboard",
                )
            else:
                split_ns = self.context.current_tool_menu.split("/")
                namespace = split_ns.pop(0)
                await self.emit("command", cmd, namespace=f"/{namespace}")

    async def on_shell_cmd(self, sid, cmd: str) -> None:
        """
        Callback on shell command message from dashboard.
        """
        await self.emit("shell_command", cmd, namespace="/copilot")

    async def on_config_updated(self, sid, config: dict[str, Any]) -> None:
        namespace = config.pop("namespace")
        await self.emit("config_updated", config, namespace=namespace)

    async def on_actuators_start(self, sid):
        """
        Callback on actuators_start message.
        """
        await self.emit("actuators_start", namespace="/copilot")

    async def on_actuators_stop(self, sid):
        """
        Callback on actuators_stop message.
        """
        await self.emit("actuators_stop", namespace="/copilot")

    async def on_actuator_command(self, sid, data):
        """
        Callback on actuator_command message.
        """
        await self.emit("actuator_command", data, namespace="/copilot")

    async def on_wizard(self, sid, data: dict[str, Any]):
        """
        Callback on wizard message.
        """
        namespace = data.pop("namespace")
        await self.emit("wizard", data, namespace=namespace)
        await self.emit("close_wizard")

on_actuator_command(sid, data) async #

Callback on actuator_command message.

Source code in cogip/tools/server/namespaces/dashboard.py 
87
88
89
90
91
async def on_actuator_command(self, sid, data):
    """
    Callback on actuator_command message.
    """
    await self.emit("actuator_command", data, namespace="/copilot")

on_actuators_start(sid) async #

Callback on actuators_start message.

Source code in cogip/tools/server/namespaces/dashboard.py 
75
76
77
78
79
async def on_actuators_start(self, sid):
    """
    Callback on actuators_start message.
    """
    await self.emit("actuators_start", namespace="/copilot")

on_actuators_stop(sid) async #

Callback on actuators_stop message.

Source code in cogip/tools/server/namespaces/dashboard.py 
81
82
83
84
85
async def on_actuators_stop(self, sid):
    """
    Callback on actuators_stop message.
    """
    await self.emit("actuators_stop", namespace="/copilot")

on_shell_cmd(sid, cmd) async #

Callback on shell command message from dashboard.

Source code in cogip/tools/server/namespaces/dashboard.py 
65
66
67
68
69
async def on_shell_cmd(self, sid, cmd: str) -> None:
    """
    Callback on shell command message from dashboard.
    """
    await self.emit("shell_command", cmd, namespace="/copilot")

on_tool_cmd(sid, cmd) async #

Callback on tool command message from dashboard.

Source code in cogip/tools/server/namespaces/dashboard.py 
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
async def on_tool_cmd(self, sid, cmd: str) -> None:
    """
    Callback on tool command message from dashboard.
    """
    # Find entry in current menu
    entry = None
    for entry in self.context.tool_menus[self.context.current_tool_menu].entries:
        if entry.cmd == cmd:
            break

    # Check if it corresponds to a menu or a command
    if entry and entry.cmd in self.context.tool_menus:
        # Enter a menu
        self.context.current_tool_menu = cmd
        await self.emit(
            "tool_menu",
            self.context.tool_menus[self.context.current_tool_menu].model_dump(),
            namespace="/dashboard",
        )
    else:
        # Forward command to corresponding namespace
        if cmd == "exit":
            self.context.current_tool_menu = "root"
            await self.emit(
                "tool_menu",
                self.context.tool_menus[self.context.current_tool_menu].model_dump(),
                namespace="/dashboard",
            )
        else:
            split_ns = self.context.current_tool_menu.split("/")
            namespace = split_ns.pop(0)
            await self.emit("command", cmd, namespace=f"/{namespace}")

on_wizard(sid, data) async #

Callback on wizard message.

Source code in cogip/tools/server/namespaces/dashboard.py 
93
94
95
96
97
98
99
async def on_wizard(self, sid, data: dict[str, Any]):
    """
    Callback on wizard message.
    """
    namespace = data.pop("namespace")
    await self.emit("wizard", data, namespace=namespace)
    await self.emit("close_wizard")