python-tdd/src/apps/epic/consumers.py

from channels.db import database_sync_to_async
from channels.generic.websocket import AsyncJsonWebsocketConsumer


LEVITY_ROLES = {"PC", "NC", "SC"}
GRAVITY_ROLES = {"BC", "EC", "AC"}


class RoomConsumer(AsyncJsonWebsocketConsumer):
    async def connect(self):
        self.room_id = self.scope["url_route"]["kwargs"]["room_id"]
        self.group_name = f"room_{self.room_id}"
        await self.channel_layer.group_add(self.group_name, self.channel_name)

        self.cursor_group = None
        user = self.scope.get("user")
        if user and user.is_authenticated:
            seat = await self._get_seat(user)
            if seat:
                if seat.role in LEVITY_ROLES:
                    self.cursor_group = f"cursors_{self.room_id}_levity"
                elif seat.role in GRAVITY_ROLES:
                    self.cursor_group = f"cursors_{self.room_id}_gravity"
                if self.cursor_group:
                    await self.channel_layer.group_add(self.cursor_group, self.channel_name)

        await self.accept()

    async def disconnect(self, close_code):
        await self.channel_layer.group_discard(self.group_name, self.channel_name)
        if self.cursor_group:
            await self.channel_layer.group_discard(self.cursor_group, self.channel_name)

    async def receive_json(self, content):
        if content.get("type") == "cursor_move" and self.cursor_group:
            await self.channel_layer.group_send(
                self.cursor_group,
                {"type": "cursor_move", "x": content.get("x"), "y": content.get("y")},
            )

    @database_sync_to_async
    def _get_seat(self, user):
        from apps.epic.models import TableSeat
        return TableSeat.objects.filter(room_id=self.room_id, gamer=user).first()

    async def gate_update(self, event):
        await self.send_json(event)

    async def role_select_start(self, event):
        await self.send_json(event)

    async def turn_changed(self, event):
        await self.send_json(event)

    async def all_roles_filled(self, event):
        await self.send_json(event)

    async def sig_select_started(self, event):
        await self.send_json(event)

    async def sig_selected(self, event):
        await self.send_json(event)

    async def cursor_move(self, event):
        await self.send_json(event)