Skip to content

beacon

BeaconNamespace #

Bases: AsyncNamespace

Handle all SocketIO events related to beacon server.

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

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

    async def on_connect(self, sid, environ):
        if self.context.beacon_sid:
            message = "A beacon server is already connected"
            logger.error(f"Beacon verser connection refused: {message}")
            raise ConnectionRefusedError(message)
        self.context.beacon_sid = sid

    async def on_connected(self, sid):
        logger.info("Beacon connected.")

    def on_disconnect(self, sid):
        self.context.beacon_sid = None
        logger.info("Beacon disconnected.")

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

    async def on_soft_reset(self, sid):
        """
        Callback on soft reset message.
        """
        logger.info("[beacon => planner] soft reset.")
        await self.emit("soft_reset", namespace="/planner")

    async def on_command(self, sid, cmd, *args):
        """
        Callback on command.
        """
        logger.info(f"[beacon => planner] Command: {cmd}")
        await self.emit("command", (cmd, *args), namespace="/planner")

    async def on_wizard(self, sid, message):
        """
        Callback on wizard message.
        """
        await self.emit("wizard", message, namespace="/planner")

    async def on_crates_from_granary_available(self, sid, value: bool):
        """
        Callback on crates_from_granary_available message.
        This message is sent by the Ninja Planner through the beacon server.
        """
        await self.emit("crates_from_granary_available", value, namespace="/planner")

on_command(sid, cmd, *args) async #

Callback on command.

Source code in cogip/tools/server/namespaces/beacon.py 
46
47
48
49
50
51
async def on_command(self, sid, cmd, *args):
    """
    Callback on command.
    """
    logger.info(f"[beacon => planner] Command: {cmd}")
    await self.emit("command", (cmd, *args), namespace="/planner")

on_crates_from_granary_available(sid, value) async #

Callback on crates_from_granary_available message. This message is sent by the Ninja Planner through the beacon server.

Source code in cogip/tools/server/namespaces/beacon.py 
59
60
61
62
63
64
async def on_crates_from_granary_available(self, sid, value: bool):
    """
    Callback on crates_from_granary_available message.
    This message is sent by the Ninja Planner through the beacon server.
    """
    await self.emit("crates_from_granary_available", value, namespace="/planner")

on_reset(sid) async #

Callback on reset message.

Source code in cogip/tools/server/namespaces/beacon.py 
32
33
34
35
36
37
async def on_reset(self, sid):
    """
    Callback on reset message.
    """
    logger.info("[beacon => planner] reset.")
    await self.emit("reset", namespace="/planner")

on_soft_reset(sid) async #

Callback on soft reset message.

Source code in cogip/tools/server/namespaces/beacon.py 
39
40
41
42
43
44
async def on_soft_reset(self, sid):
    """
    Callback on soft reset message.
    """
    logger.info("[beacon => planner] soft reset.")
    await self.emit("soft_reset", namespace="/planner")

on_wizard(sid, message) async #

Callback on wizard message.

Source code in cogip/tools/server/namespaces/beacon.py 
53
54
55
56
57
async def on_wizard(self, sid, message):
    """
    Callback on wizard message.
    """
    await self.emit("wizard", message, namespace="/planner")