Skip to content

firmware_calibration

FirmwareCalibrationNamespace #

Bases: AsyncNamespace

Handle all SocketIO events related to firmware calibration clients. Only one client connection is allowed at a time.

Source code in cogip/tools/server/namespaces/firmware_calibration.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
class FirmwareCalibrationNamespace(socketio.AsyncNamespace):
    """
    Handle all SocketIO events related to firmware calibration clients.
    Only one client connection is allowed at a time.
    """

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

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

    async def on_connected(self, sid):
        logger.info("Calibration client connected.")
        self.context.calibration_sid = sid

    async def on_disconnect(self, sid):
        self.context.calibration_sid = None
        logger.info("Calibration client disconnected.")

    async def on_pose_start(self, sid, pose: dict[str, Any]):
        """
        Callback on pose start (reset robot position).
        Forward to copilot.
        """
        logger.info(f"[calibration => copilot] Pose start: {pose}")
        await self.emit("pose_start", pose, namespace="/copilot")

    async def on_pose_order(self, sid, pose: dict[str, Any]):
        """
        Callback on pose order.
        Forward to pose to copilot and dashboards.
        """
        logger.info(f"[calibration => copilot] Pose order: {pose}")
        await self.emit("pose_order", pose, namespace="/copilot")
        await self.emit("pose_order", (self.context.robot_id, pose), namespace="/dashboard")

    async def on_speed_order(self, sid, data: dict[str, Any]):
        """
        Callback on speed order.
        Forward to copilot.
        """
        logger.info(f"[calibration => copilot] Speed order: {data}")
        await self.emit("speed_order", data, namespace="/copilot")

    async def on_game_end(self, sid):
        """
        Callback on game_end message.
        Forward to copilot.
        """
        logger.info("[calibration => copilot] Game reset.")
        await self.emit("game_end", namespace="/copilot")

    async def on_set_controller(self, sid, controller: int):
        """
        Callback on set_controller message.
        Forward to copilot.
        """
        logger.info(f"[calibration => copilot] controller: {controller}")
        await self.emit("set_controller", controller, namespace="/copilot")

on_game_end(sid) async #

Callback on game_end message. Forward to copilot.

Source code in cogip/tools/server/namespaces/firmware_calibration.py 
59
60
61
62
63
64
65
async def on_game_end(self, sid):
    """
    Callback on game_end message.
    Forward to copilot.
    """
    logger.info("[calibration => copilot] Game reset.")
    await self.emit("game_end", namespace="/copilot")

on_pose_order(sid, pose) async #

Callback on pose order. Forward to pose to copilot and dashboards.

Source code in cogip/tools/server/namespaces/firmware_calibration.py 
42
43
44
45
46
47
48
49
async def on_pose_order(self, sid, pose: dict[str, Any]):
    """
    Callback on pose order.
    Forward to pose to copilot and dashboards.
    """
    logger.info(f"[calibration => copilot] Pose order: {pose}")
    await self.emit("pose_order", pose, namespace="/copilot")
    await self.emit("pose_order", (self.context.robot_id, pose), namespace="/dashboard")

on_pose_start(sid, pose) async #

Callback on pose start (reset robot position). Forward to copilot.

Source code in cogip/tools/server/namespaces/firmware_calibration.py 
34
35
36
37
38
39
40
async def on_pose_start(self, sid, pose: dict[str, Any]):
    """
    Callback on pose start (reset robot position).
    Forward to copilot.
    """
    logger.info(f"[calibration => copilot] Pose start: {pose}")
    await self.emit("pose_start", pose, namespace="/copilot")

on_set_controller(sid, controller) async #

Callback on set_controller message. Forward to copilot.

Source code in cogip/tools/server/namespaces/firmware_calibration.py 
67
68
69
70
71
72
73
async def on_set_controller(self, sid, controller: int):
    """
    Callback on set_controller message.
    Forward to copilot.
    """
    logger.info(f"[calibration => copilot] controller: {controller}")
    await self.emit("set_controller", controller, namespace="/copilot")

on_speed_order(sid, data) async #

Callback on speed order. Forward to copilot.

Source code in cogip/tools/server/namespaces/firmware_calibration.py 
51
52
53
54
55
56
57
async def on_speed_order(self, sid, data: dict[str, Any]):
    """
    Callback on speed order.
    Forward to copilot.
    """
    logger.info(f"[calibration => copilot] Speed order: {data}")
    await self.emit("speed_order", data, namespace="/copilot")