LobotomyLabs/classquiz-ai
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

🚧 Added code for new ClassQuizController

Browse Source
This commit is contained in:
Mawoka
2023-04-04 12:29:27 +02:00
parent 2e38192835
commit ac18621672
7 changed files with 211 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
classquiz/db/models.py
+16
View File
@@ -318,3 +318,19 @@ class GameResults(ormar.Model):
tablename = "game_results"
metadata = metadata
database = database
class Controllers(ormar.Model):
id: uuid.UUID = ormar.UUID(primary_key=True)
user: uuid.UUID | User = ormar.ForeignKey(User)
secret_key: str = ormar.String(nullable=False, max_length=24, min_length=24)
player_name: str = ormar.Text(nullable=False)
last_seen: datetime | None = ormar.DateTime(nullable=True)
first_seen: datetime | None = ormar.DateTime(nullable=True)
name: str = ormar.Text(nullable=False)
os_version: str | None = ormar.Text(nullable=True)
class Meta:
tablename = "controllers"
metadata = metadata
database = database
classquiz/helpers/box_controller.py
+21
View File
@@ -0,0 +1,21 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
import random
def generate_code(specified_length: int) -> str:
buttons = [
"B",
"b",
"G",
"g",
"Y",
"y",
"R",
"r",
] # Capital stands for long press, lowercase letter for short press
resulting_code = ""
for i in range(specified_length):
resulting_code += random.choice(buttons)
return resulting_code
classquiz/routers/box_controller/embedded/__init__.py
+68
View File
@@ -0,0 +1,68 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
import os
import uuid
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from classquiz.config import redis
from classquiz.db.models import PlayGame, Controllers
from classquiz.routers.box_controller.embedded.socket import router as socket_router
router = APIRouter()
router.include_router(socket_router, prefix="/socket")
class JoinGameInput(BaseModel):
id: uuid.UUID
secret_key: str
code: str
class JoinGameResponse(BaseModel):
id: str
@router.post("/join")
async def join_game(data: JoinGameInput):
controller = await Controllers.objects.get_or_none(id=data.id, secret_key=data.secret_key)
game_pin = await redis.get(f"game:cqc:code:{data.code}")
if game_pin is None:
raise HTTPException(status_code=404, detail="Game not found")
game = await redis.get(f"game:{game_pin}")
game = PlayGame.parse_raw(game)
# Check if game is already running
if game.started:
raise HTTPException(status_code=400, detail="Game started already")
# check if username already exists
if await redis.get(f"game_session:{game_pin}:players:{controller.player_name}") is not None:
raise HTTPException(status_code=409, detail="Username already exists")
player_id = os.urandom(5).hex()
await redis.set(f"game:cqc:player:{player_id}", controller.player_name)
return JoinGameResponse(id=f"{player_id}:{game_pin}")
class RegisterWithCodeInput(BaseModel):
code: str
class RegisterWithCodeResponse(BaseModel):
id: uuid.UUID
secret_key: str
@router.post("/register")
async def register_with_code(data: RegisterWithCodeInput) -> RegisterWithCodeResponse:
c_id = await redis.get(f"controller_setup:{data.code}")
if c_id is None:
raise HTTPException(status_code=404, detail="Code not found")
await redis.delete(f"controller_setup:{data.code}")
c_id = uuid.UUID(c_id)
controller = await Controllers.objects.get(id=c_id)
return RegisterWithCodeResponse(id=controller.id, secret_key=controller.secret_key)
classquiz/routers/box_controller/embedded.py → classquiz/routers/box_controller/embedded/socket.py
+4 -32
View File
@@ -2,7 +2,6 @@
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
import enum
import os
import typing
from datetime import datetime
@@ -15,34 +14,6 @@ from classquiz.socket_server import sio, calculate_score, set_answer
router = APIRouter()
class JoinGameInput(BaseModel):
code: str
name: str
class JoinGameResponse(BaseModel):
id: str
@router.post("/join")
async def join_game(data: JoinGameInput):
game_pin = await redis.get(f"game:cqc:code:{data.code}")
if game_pin is None:
raise HTTPException(status_code=404, detail="Game not found")
game = await redis.get(f"game:{game_pin}")
game = PlayGame.parse_raw(game)
# Check if game is already running
if game.started:
raise HTTPException(status_code=400, detail="Game started already")
# check if username already exists
if await redis.get(f"game_session:{game_pin}:players:{data.name}") is not None:
raise HTTPException(status_code=409, detail="Username already exists")
player_id = os.urandom(5).hex()
await redis.set(f"game:cqc:player:{player_id}", data.name)
return JoinGameResponse(id=f"{player_id}:{game_pin}")
class SubmitAnswerInput(BaseModel):
answer: int
@@ -91,6 +62,9 @@ async def submit_answer_fn(data_answer: int, game_pin: str, player_id: str, now:
await sio.emit("everyone_answered", {})
button_to_index_map = {"b": 0, "g": 2, "y": 1, "r": 3}
class WebSocketTypes(enum.Enum):
ButtonPress = "bp"
Error = "e"
@@ -103,10 +77,8 @@ class WebSocketRequest(BaseModel):
wss_clients = {}
button_to_index_map = {"b": 0, "g": 1, "y": 2, "r": 3}
@router.websocket("/socket/{id}")
@router.websocket("/{id}")
async def websocket_endpoint(ws: WebSocket, game_id: str, id: str):
try:
if id in wss_clients.keys():
classquiz/routers/box_controller/web.py
+59 -34
View File
@@ -1,52 +1,77 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
import random
import os
import uuid
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from classquiz.auth import get_current_user
from classquiz.db.models import User, PlayGame
from classquiz.db.models import User, Controllers
from classquiz.config import redis
from classquiz.helpers.box_controller import generate_code
router = APIRouter()
def generate_code() -> str:
specified_length = 6
buttons = [
"B",
"b",
"G",
"g",
"Y",
"y",
"R",
"r",
] # Capital stands for long press, lowercase letter for short press
resulting_code = ""
for i in range(specified_length):
resulting_code += random.choice(buttons)
return resulting_code
class SetControllerUpInput(BaseModel):
player_name: str | None
name: str
class ActivateCqbForCurrentQuizInput(BaseModel):
game_pin: int
class ActivateCqbForCurrentQuizResponse(BaseModel):
class SetControllerUpResponse(BaseModel):
code: str
id: uuid.UUID
@router.post("/activate-for-quiz", response_model=ActivateCqbForCurrentQuizResponse)
async def activate_cqc_for_current_quiz(data: ActivateCqbForCurrentQuizInput, user: User = Depends(get_current_user)):
redis_res = await redis.get(f"game:{data.game_pin}")
if redis_res is None:
raise HTTPException(status_code=404, detail="Game not found")
game_data = PlayGame.parse_raw(redis_res)
if game_data.user_id != user.id:
raise HTTPException(status_code=401, detail="The quiz wasn't started by you")
code = generate_code()
await redis.set(f"game:cqc:code:{code}", game_data.game_pin, ex=3600)
return ActivateCqbForCurrentQuizResponse(code=code)
@router.post("/setup")
async def set_controller_up(
input_data: SetControllerUpInput, user: User = Depends(get_current_user)
) -> SetControllerUpResponse:
code = generate_code(10)
if input_data.player_name is None:
input_data.player_name = user.username
data = Controllers(
id=uuid.uuid4(),
user=user,
secret_key=os.urandom(12).hex(),
player_name=input_data.player_name,
last_seen=None,
first_seen=None,
name=input_data.name,
os_version=None,
)
await data.save()
await redis.set(f"controller_setup:{code}", data.id.hex, ex=900)
return SetControllerUpResponse(code=code, id=data.id)
GetControllerResponse = Controllers.get_pydantic(exclude={"secret_key", "user"})
@router.get("/controller")
async def get_controller(id: uuid.UUID, user: User = Depends(get_current_user)) -> GetControllerResponse:
controller = await Controllers.objects.get_or_none(id=id, user=user.id)
if controller is None:
raise HTTPException(status_code=404, detail="Controller not found")
return GetControllerResponse(**controller.dict())
class ModifyControllerInput(BaseModel):
id: uuid.UUID
player_name: str
name: str
@router.post("/modify")
async def modify_controller(
data: ModifyControllerInput, user: User = Depends(get_current_user)
) -> GetControllerResponse:
controller = await Controllers.objects.get_or_none(id=data.id, user=user.id)
if controller is None:
raise HTTPException(status_code=404, detail="Controller not found")
controller.player_name = data.player_name
controller.name = data.name
await controller.update()
return GetControllerResponse(**controller.dict())
classquiz/routers/quiz.py
+2 -20
View File
@@ -3,7 +3,6 @@
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
import json
import random
import re
import uuid
from datetime import datetime
@@ -20,6 +19,7 @@ import bleach
from classquiz.auth import get_current_user
from classquiz.config import redis, settings, storage, meilisearch
from classquiz.db.models import Quiz, QuizInput, User, PlayGame, GameInLobby, QuizQuestion
from classquiz.helpers.box_controller import generate_code
from classquiz.kahoot_importer.import_quiz import import_quiz
import html
import urllib.parse
@@ -99,24 +99,6 @@ async def get_public_quiz(quiz_id: str):
return PublicQuizResponse(**quiz.dict())
def generate_code() -> str:
specified_length = 6
buttons = [
"B",
"b",
"G",
"g",
"Y",
"y",
"R",
"r",
] # Capital stands for long press, lowercase letter for short press
resulting_code = ""
for i in range(specified_length):
resulting_code += random.choice(buttons)
return resulting_code
@router.post("/start/{quiz_id}")
async def start_quiz(
quiz_id: str,
@@ -160,7 +142,7 @@ async def start_quiz(
)
code = None
if cqcs_enabled:
code = generate_code()
code = generate_code(6)
await redis.set(f"game:cqc:code:{code}", game_pin, ex=3600)
await redis.set(f"game:{str(game.game_pin)}", game.json(), ex=18000)
await redis.set(f"game_pin:{user.id}:{quiz_id}", game_pin, ex=18000)
migrations/versions/b9ca06dfa179_added_controllers_table.py
+41
View File
@@ -0,0 +1,41 @@
"""Added controllers table
Revision ID: b9ca06dfa179
Revises: 7afe98d04169
Create Date: 2023-04-03 13:16:49.585163
"""
from alembic import op
import sqlalchemy as sa
import ormar
# revision identifiers, used by Alembic.
revision = "b9ca06dfa179"
down_revision = "7afe98d04169"
branch_labels = None
depends_on = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table(
"controllers",
sa.Column("id", ormar.fields.sqlalchemy_uuid.CHAR(32), nullable=False),
sa.Column("user", ormar.fields.sqlalchemy_uuid.CHAR(32), nullable=True),
sa.Column("secret_key", sa.String(length=24), nullable=False),
sa.Column("player_name", sa.Text(), nullable=False),
sa.Column("last_seen", sa.DateTime(), nullable=True),
sa.Column("first_seen", sa.DateTime(), nullable=True),
sa.Column("name", sa.Text(), nullable=False),
sa.Column("os_version", sa.Text(), nullable=True),
sa.ForeignKeyConstraint(["user"], ["users.id"], name="fk_controllers_users_id_user"),
sa.PrimaryKeyConstraint("id"),
)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table("controllers")
# ### end Alembic commands ###