655 lines
24 KiB
Python
655 lines
24 KiB
Python
import json
|
|
import asyncio
|
|
from channels.generic.websocket import AsyncWebsocketConsumer
|
|
from channels.db import database_sync_to_async
|
|
from django.utils import timezone
|
|
from django.urls import reverse
|
|
from play.models import QuizGame, QuizGameParticipant, QuizAnswer
|
|
from django.conf import settings
|
|
from geopy.distance import geodesic
|
|
|
|
class GameConsumer(AsyncWebsocketConsumer):
|
|
|
|
async def connect(self):
|
|
self.join_code = self.scope['url_route']['kwargs']['join_code']
|
|
self.game_group_name = f'game_{self.join_code}'
|
|
|
|
# Join game group
|
|
await self.channel_layer.group_add(
|
|
self.game_group_name,
|
|
self.channel_name
|
|
)
|
|
|
|
await self.accept()
|
|
|
|
async def disconnect(self, close_code):
|
|
# Leave game group
|
|
await self.channel_layer.group_discard(
|
|
self.game_group_name,
|
|
self.channel_name
|
|
)
|
|
|
|
# Check if game should be deleted
|
|
await self.cleanup_game()
|
|
|
|
|
|
@database_sync_to_async
|
|
def get_question_data(self):
|
|
quiz_game = QuizGame.objects.get(join_code=self.join_code)
|
|
current_question = quiz_game.quiz_id.questions.all()[quiz_game.current_question_index]
|
|
return json.loads(current_question.data)
|
|
|
|
async def receive(self, text_data):
|
|
text_data_json = json.loads(text_data)
|
|
message_type = text_data_json['type']
|
|
|
|
if message_type == 'ping':
|
|
# Respond with pong to keep connection alive
|
|
await self.send(text_data=json.dumps({
|
|
'type': 'pong'
|
|
}))
|
|
return
|
|
|
|
if message_type == 'leave_game':
|
|
participant_id = text_data_json.get('participant_id')
|
|
if participant_id:
|
|
try:
|
|
# Get participant info before deletion
|
|
participant = await database_sync_to_async(QuizGameParticipant.objects.get)(participant_id=participant_id)
|
|
display_name = participant.display_name
|
|
join_code = self.join_code
|
|
|
|
# Delete participant
|
|
await database_sync_to_async(participant.delete)()
|
|
|
|
# Update participants list
|
|
participants = await self.get_participants()
|
|
await self.channel_layer.group_send(
|
|
self.game_group_name,
|
|
{
|
|
'type': 'participant_list_update',
|
|
'participants': participants
|
|
}
|
|
)
|
|
|
|
# Send redirect to home
|
|
await self.send(text_data=json.dumps({
|
|
'type': 'redirect',
|
|
'url': '/'
|
|
}))
|
|
except QuizGameParticipant.DoesNotExist:
|
|
pass
|
|
return
|
|
|
|
if message_type == 'submit_rating':
|
|
participant_id = text_data_json['participant_id']
|
|
rating = text_data_json['rating']
|
|
success = await self.save_rating(participant_id, rating)
|
|
if success:
|
|
await self.send(text_data=json.dumps({
|
|
'type': 'rating_confirmed'
|
|
}))
|
|
|
|
elif message_type == 'submit_answer':
|
|
participant_id = text_data_json['participant_id']
|
|
answer = text_data_json['answer']
|
|
time_remaining = text_data_json.get('time_remaining', 0)
|
|
|
|
|
|
question_data = await self.get_question_data()
|
|
if question_data['type'] == "map":
|
|
await self.save_answer_map(participant_id, answer, time_remaining)
|
|
else:
|
|
await self.save_answer(participant_id, answer, time_remaining)
|
|
|
|
|
|
# Notify host about the new answer
|
|
await self.channel_layer.group_send(
|
|
self.game_group_name,
|
|
{
|
|
'type': 'participant_answer',
|
|
'answer': answer
|
|
}
|
|
)
|
|
|
|
elif message_type == 'start_game':
|
|
host_id = text_data_json['host_id']
|
|
if await self.verify_host(host_id):
|
|
await self.advance_to_next_question()
|
|
|
|
elif message_type == 'get_answer_stats':
|
|
stats = await self.get_answer_stats()
|
|
await self.send(text_data=json.dumps({
|
|
'type': 'answer_stats',
|
|
'stats': stats,
|
|
}))
|
|
|
|
elif message_type == 'update_participants':
|
|
participants = await self.get_participants()
|
|
await self.channel_layer.group_send(
|
|
self.game_group_name,
|
|
{
|
|
'type': 'participant_list_update',
|
|
'participants': participants
|
|
}
|
|
)
|
|
|
|
elif message_type == 'next_question':
|
|
host_id = text_data_json['host_id']
|
|
if await self.verify_host(host_id):
|
|
await self.advance_to_next_question()
|
|
elif message_type == 'winner_podest':
|
|
host_id = text_data_json['host_id']
|
|
if await self.verify_host(host_id):
|
|
await self.show_winner_podest()
|
|
|
|
elif message_type == 'scoreboard':
|
|
host_id = text_data_json['host_id']
|
|
if await self.verify_host(host_id):
|
|
await self.show_scoreboard()
|
|
|
|
elif message_type == 'finish_game':
|
|
host_id = text_data_json['host_id']
|
|
if await self.verify_host(host_id):
|
|
await self.finish_game()
|
|
|
|
elif message_type == 'advance_to_scores':
|
|
host_id = text_data_json['host_id']
|
|
if await self.verify_host(host_id):
|
|
await self.show_scores()
|
|
|
|
async def participant_answer(self, event):
|
|
await self.send(text_data=json.dumps({
|
|
'type': 'participant_answer',
|
|
'answer': event['answer']
|
|
}))
|
|
|
|
async def participant_list_update(self, event):
|
|
await self.send(text_data=json.dumps({
|
|
'type': 'participant_list_update',
|
|
'participants': event['participants']
|
|
}))
|
|
|
|
async def player_left(self, event):
|
|
await self.send(text_data=json.dumps({
|
|
'type': 'player_left',
|
|
'player_name': event['player_name'],
|
|
'was_kicked': event.get('was_kicked', False)
|
|
}))
|
|
|
|
async def game_state_update(self, event):
|
|
await self.send(text_data=json.dumps({
|
|
'type': 'game_state_update',
|
|
'action': event['action'],
|
|
'redirect_url': event.get('redirect_url')
|
|
}))
|
|
|
|
@database_sync_to_async
|
|
def verify_host(self, host_id):
|
|
try:
|
|
quiz_game = QuizGame.objects.get(join_code=self.join_code)
|
|
return quiz_game.host_id == host_id
|
|
except QuizGame.DoesNotExist:
|
|
return False
|
|
|
|
@database_sync_to_async
|
|
def save_answer_map(self, participant_id, answer_location, time_remaining):
|
|
answer_index = -1
|
|
try:
|
|
# Get all necessary data first
|
|
participant = QuizGameParticipant.objects.get(
|
|
quiz_game__join_code=self.join_code,
|
|
participant_id=participant_id
|
|
)
|
|
quiz_game = participant.quiz_game
|
|
current_question = quiz_game.quiz_id.questions.all()[quiz_game.current_question_index]
|
|
question_data = json.loads(current_question.data)
|
|
target_location = question_data['target_location']
|
|
tolerance_radius = question_data['tolerance_radius']
|
|
print(f"target_location: {target_location}")
|
|
print(f"answer_location: {answer_location}")
|
|
print(f"tolerance_radius: {tolerance_radius}")
|
|
|
|
# Configuration - Make these easily adjustable
|
|
MAX_DISTANCE_KM = 1000 # Maximum distance for scoring (in km)
|
|
DIRECT_HIT_BONUS = 200 # Extra points for direct hit within tolerance
|
|
|
|
# Calculate base score
|
|
distance_km = geodesic(answer_location, target_location).kilometers
|
|
tolerance_km = int(tolerance_radius) / 1000 # Convert meters to kilometers
|
|
|
|
# Calculate base score with non-linear decrease
|
|
if distance_km <= tolerance_km:
|
|
# Direct hit - full points + bonus
|
|
base_score = 1000 + DIRECT_HIT_BONUS # 1000 + 200 bonus
|
|
is_correct = True
|
|
else:
|
|
if distance_km > MAX_DISTANCE_KM:
|
|
# Beyond max distance - 0 points
|
|
base_score = 0
|
|
is_correct = False
|
|
else:
|
|
# Non-linear decrease using exponential falloff
|
|
# This creates a steeper drop-off at the beginning
|
|
progress = (distance_km - tolerance_km) / (MAX_DISTANCE_KM - tolerance_km)
|
|
# Using a power function to create non-linear falloff
|
|
falloff = progress ** 1.5 # 1.5 makes the falloff steeper at the beginning
|
|
base_score = 1000 * (1 - falloff)
|
|
is_correct = True
|
|
base_score = max(0, min(1000, int(base_score)))
|
|
|
|
# Apply time penalty (reduces score based on time taken)
|
|
if is_correct and base_score > 0:
|
|
# Calculate time taken as a fraction of total time (0-1)
|
|
time_taken_sec = (int(current_question.time_per_question) * 1000 - time_remaining) / 1000
|
|
max_time_sec = int(current_question.time_per_question)
|
|
time_fraction = min(1.0, time_taken_sec / max_time_sec) if max_time_sec > 0 else 1.0
|
|
|
|
# Reduce points based on time taken (up to 50% reduction for answering at the last second)
|
|
time_penalty = base_score * 0.5 * time_fraction # Up to 50% penalty
|
|
final_score = max(1, base_score - time_penalty) # At least 1 point if correct
|
|
else:
|
|
final_score = base_score # 0 points for incorrect answers
|
|
|
|
score = int(round(final_score))
|
|
|
|
# Save the answer
|
|
answer_obj, created = QuizAnswer.objects.get_or_create(
|
|
participant=participant,
|
|
question_index=quiz_game.current_question_index,
|
|
defaults={
|
|
'answer_index': answer_index,
|
|
'is_correct': is_correct,
|
|
'score': score,
|
|
'time_remaining': time_remaining
|
|
}
|
|
)
|
|
|
|
if not created:
|
|
answer_obj.answer_index = answer_index
|
|
answer_obj.is_correct = is_correct
|
|
answer_obj.score = score
|
|
answer_obj.time_remaining = time_remaining
|
|
answer_obj.save()
|
|
|
|
participant.last_score = score
|
|
participant.score += score
|
|
participant.last_answer_correct = is_correct
|
|
participant.save()
|
|
|
|
return True
|
|
|
|
except (QuizGameParticipant.DoesNotExist, IndexError):
|
|
return False
|
|
|
|
|
|
@database_sync_to_async
|
|
def save_answer(self, participant_id, answer, time_remaining):
|
|
try:
|
|
participant = QuizGameParticipant.objects.get(
|
|
quiz_game__join_code=self.join_code,
|
|
participant_id=participant_id
|
|
)
|
|
quiz_game = participant.quiz_game
|
|
current_question = quiz_game.quiz_id.questions.all()[quiz_game.current_question_index]
|
|
|
|
question_data = json.loads(current_question.data)
|
|
question_type = question_data.get("type")
|
|
|
|
is_correct = False
|
|
answer_index = None
|
|
answers_order = None
|
|
answer_text = None
|
|
|
|
if question_type in ["multiple_choice", "true_false"]:
|
|
# answer ist hier eine Integer-Antwort (Index)
|
|
answer_index = int(answer)
|
|
if answer_index >= 0:
|
|
is_correct = question_data['options'][answer_index].get('is_correct', False)
|
|
|
|
elif question_type == "input":
|
|
user_input = str(answer).strip().lower().replace(" ", "")
|
|
correct_value = str(question_data['options'][0].get('value', '')).strip().lower().replace(" ", "")
|
|
is_correct = (user_input == correct_value)
|
|
answer_text = user_input
|
|
answer_index = 0 if is_correct else 1
|
|
|
|
elif question_type == "order":
|
|
# answer ist hier eine Liste von Strings mit der Reihenfolge
|
|
answers_order = answer
|
|
correct_sequence = sorted(question_data["options"], key=lambda x: x["order"])
|
|
correct_values = [opt["value"] for opt in correct_sequence]
|
|
is_correct = (answers_order == correct_values)
|
|
|
|
elif question_type == "guess":
|
|
try:
|
|
user_val = float(answer)
|
|
correct_val = float(question_data['options'][0].get('value', 0))
|
|
tolerance = float(question_data.get('tolerance', 0))
|
|
percentage = min(user_val, correct_val) / (max(user_val, correct_val) + 1e-5)
|
|
|
|
|
|
score_percentage= min(round(percentage, 3), 1.0)
|
|
if score_percentage>= abs(tolerance/100):
|
|
is_correct=True
|
|
else:
|
|
is_correct=False
|
|
|
|
print(score_percentage)
|
|
|
|
|
|
except ValueError:
|
|
is_correct = False # falls keine Zahl eingegeben wurde
|
|
answer_index = 0 if is_correct else 1
|
|
|
|
|
|
# Score berechnen
|
|
try:
|
|
if score_percentage==None:
|
|
score_percentage=1
|
|
except:
|
|
score_percentage=1
|
|
|
|
|
|
score = 0
|
|
if is_correct:
|
|
base_score = 1000
|
|
time_factor = time_remaining / int(current_question.time_per_question) / 1000
|
|
score = int(base_score * (0.75 + 0.25 * time_factor))*score_percentage
|
|
print(score)
|
|
|
|
# Antwort speichern oder aktualisieren
|
|
answer_obj, created = QuizAnswer.objects.get_or_create(
|
|
participant=participant,
|
|
question_index=quiz_game.current_question_index,
|
|
defaults={
|
|
'answer_index': answer_index,
|
|
'answers_order': answers_order,
|
|
'is_correct': is_correct,
|
|
'score': score,
|
|
'time_remaining': time_remaining
|
|
}
|
|
)
|
|
if not created:
|
|
answer_obj.answer_index = answer_index
|
|
answer_obj.answers_order = answers_order
|
|
answer_obj.is_correct = is_correct
|
|
answer_obj.score = score
|
|
answer_obj.time_remaining = time_remaining
|
|
answer_obj.save()
|
|
|
|
# Teilnehmer aktualisieren
|
|
participant.last_score = score
|
|
participant.score += score
|
|
participant.last_answer=answer_index
|
|
participant.last_answer_correct = is_correct
|
|
participant.save()
|
|
|
|
return True
|
|
except (QuizGameParticipant.DoesNotExist, IndexError):
|
|
return False
|
|
|
|
|
|
@database_sync_to_async
|
|
def get_answer_stats(self):
|
|
try:
|
|
quiz_game = QuizGame.objects.get(join_code=self.join_code)
|
|
answers = QuizAnswer.objects.filter(
|
|
participant__quiz_game=quiz_game,
|
|
question_index=quiz_game.current_question_index
|
|
)
|
|
|
|
|
|
|
|
|
|
current_question = quiz_game.quiz_id.questions.all()[quiz_game.current_question_index]
|
|
|
|
question_data = json.loads(current_question.data)
|
|
question_type = question_data.get("type")
|
|
stats = {}
|
|
|
|
for single_answer in answers:
|
|
|
|
|
|
if question_type == "order":
|
|
correct_answer=[str(opt["value"]) for opt in sorted(question_data["options"], key=lambda x: x["order"])]
|
|
|
|
if single_answer.answers_order== correct_answer:
|
|
key = 0 # richtig
|
|
else:
|
|
key = 1 # falsch
|
|
stats[key] = stats.get(key, 0) + 1
|
|
else:
|
|
key = str(single_answer.answer_index if single_answer.answer_index is not None else "unanswered")
|
|
stats[key] = stats.get(key, 0) + 1
|
|
return stats
|
|
except:
|
|
pass
|
|
|
|
|
|
|
|
|
|
|
|
@database_sync_to_async
|
|
def save_rating(self, participant_id, rating):
|
|
from django.db import models
|
|
if not settings.QIVIP_CONFIG['ENABLE_RATING_SYSTEM']:
|
|
return False
|
|
from library.models import QuizRating
|
|
try:
|
|
game = QuizGame.objects.get(join_code=self.join_code)
|
|
participant = QuizGameParticipant.objects.get(
|
|
quiz_game=game,
|
|
participant_id=participant_id
|
|
)
|
|
|
|
# Update or create rating
|
|
"""
|
|
rating_obj, created = QuizRating.objects.get_or_create(
|
|
quiz=game.quiz_id,
|
|
participant_id=participant_id,
|
|
defaults={'rating': rating}
|
|
)
|
|
"""
|
|
rating_obj=QuizRating.objects.create(
|
|
quiz=game.quiz_id,
|
|
participant_id=participant_id,
|
|
rating=rating)
|
|
|
|
rating_obj.rating = rating
|
|
rating_obj.save()
|
|
|
|
|
|
|
|
return True
|
|
except (QuizGame.DoesNotExist, QuizGameParticipant.DoesNotExist):
|
|
return False
|
|
|
|
@database_sync_to_async
|
|
def cleanup_game(self):
|
|
try:
|
|
game = QuizGame.objects.get(join_code=self.join_code)
|
|
|
|
active_participants = QuizGameParticipant.objects.filter(
|
|
quiz_game=game
|
|
).count()
|
|
|
|
if active_participants == 0:
|
|
game.delete()
|
|
except QuizGame.DoesNotExist:
|
|
pass
|
|
|
|
|
|
@database_sync_to_async
|
|
def get_participants(self):
|
|
try:
|
|
quiz_game = QuizGame.objects.get(join_code=self.join_code)
|
|
participants = QuizGameParticipant.objects.filter(quiz_game=quiz_game)
|
|
return [
|
|
{
|
|
'id': p.participant_id,
|
|
'display_name': p.display_name,
|
|
'score': p.score
|
|
}
|
|
for p in participants
|
|
]
|
|
except QuizGame.DoesNotExist:
|
|
return []
|
|
|
|
@database_sync_to_async
|
|
def _advance_to_next_question(self):
|
|
try:
|
|
quiz_game = QuizGame.objects.get(join_code=self.join_code)
|
|
quiz = quiz_game.quiz_id
|
|
|
|
# Move to next question
|
|
quiz_game.current_question_index += 1
|
|
|
|
# Check if we've reached the end
|
|
if quiz_game.current_question_index >= quiz.questions.count():
|
|
quiz_game.current_state = 'finished'
|
|
quiz_game.save()
|
|
return 'finished'
|
|
else:
|
|
# Update game state and start time
|
|
quiz_game.current_state = 'question'
|
|
quiz_game.question_start_time = timezone.now()
|
|
quiz_game.save()
|
|
return 'question'
|
|
except QuizGame.DoesNotExist:
|
|
return None
|
|
|
|
async def advance_to_next_question(self):
|
|
result = await self._advance_to_next_question()
|
|
if result == 'finished':
|
|
# Notify clients to redirect to finished page
|
|
await self.channel_layer.group_send(
|
|
self.game_group_name,
|
|
{
|
|
'type': 'game_state_update',
|
|
'action': 'finish_game',
|
|
'redirect_url': reverse('play:finished', kwargs={'join_code': self.join_code})
|
|
}
|
|
)
|
|
elif result == 'question':
|
|
# Notify clients to redirect to next question
|
|
await self.channel_layer.group_send(
|
|
self.game_group_name,
|
|
{
|
|
'type': 'game_state_update',
|
|
'action': 'next_question',
|
|
'redirect_url': reverse('play:question', kwargs={'join_code': self.join_code})
|
|
}
|
|
)
|
|
elif result == 'winner_podest':
|
|
# Notify clients to redirect to next question
|
|
await self.channel_layer.group_send(
|
|
self.game_group_name,
|
|
{
|
|
'type': 'game_state_update',
|
|
'action': 'show_winner_podest',
|
|
'redirect_url': reverse('play:winner_podest', kwargs={'join_code': self.join_code})
|
|
}
|
|
)
|
|
elif result == 'scoreboard':
|
|
# Notify clients to redirect to next question
|
|
await self.channel_layer.group_send(
|
|
self.game_group_name,
|
|
{
|
|
'type': 'game_state_update',
|
|
'action': 'show_scoreboard',
|
|
'redirect_url': reverse('play:scoreboard', kwargs={'join_code': self.join_code})
|
|
}
|
|
)
|
|
|
|
@database_sync_to_async
|
|
def _show_scores(self):
|
|
try:
|
|
quiz_game = QuizGame.objects.get(join_code=self.join_code)
|
|
quiz_game.current_state = 'scores'
|
|
quiz_game.save()
|
|
return True
|
|
except QuizGame.DoesNotExist:
|
|
return False
|
|
|
|
async def show_scores(self):
|
|
success = await self._show_scores()
|
|
if success:
|
|
# Notify clients to redirect to scores page
|
|
await self.channel_layer.group_send(
|
|
self.game_group_name,
|
|
{
|
|
'type': 'game_state_update',
|
|
'action': 'show_scores',
|
|
'redirect_url': reverse('play:scores', kwargs={'join_code': self.join_code})
|
|
}
|
|
)
|
|
|
|
@database_sync_to_async
|
|
def _finish_game(self):
|
|
try:
|
|
quiz_game = QuizGame.objects.get(join_code=self.join_code)
|
|
quiz_game.current_state = 'finished'
|
|
quiz_game.save()
|
|
return True
|
|
except QuizGame.DoesNotExist:
|
|
return False
|
|
|
|
async def finish_game(self):
|
|
success = await self._finish_game()
|
|
if success:
|
|
# Notify clients to redirect to finished page
|
|
await self.channel_layer.group_send(
|
|
self.game_group_name,
|
|
{
|
|
'type': 'game_state_update',
|
|
'action': 'finish_game',
|
|
'redirect_url': reverse('play:finished', kwargs={'join_code': self.join_code})
|
|
}
|
|
)
|
|
@database_sync_to_async
|
|
def _show_winner_podest(self):
|
|
try:
|
|
quiz_game = QuizGame.objects.get(join_code=self.join_code)
|
|
quiz_game.current_state = 'winner_podest'
|
|
quiz_game.save()
|
|
return True
|
|
except QuizGame.DoesNotExist:
|
|
return False
|
|
|
|
async def show_winner_podest(self):
|
|
success = await self._show_winner_podest()
|
|
if success:
|
|
# Notify clients to redirect to finished page
|
|
await self.channel_layer.group_send(
|
|
self.game_group_name,
|
|
{
|
|
'type': 'game_state_update',
|
|
'action': 'show_winner_podest',
|
|
'redirect_url': reverse('play:winner_podest', kwargs={'join_code': self.join_code})
|
|
}
|
|
)
|
|
|
|
@database_sync_to_async
|
|
def _show_scoreboard(self):
|
|
try:
|
|
quiz_game = QuizGame.objects.get(join_code=self.join_code)
|
|
quiz_game.current_state = 'scoreboard'
|
|
quiz_game.save()
|
|
return True
|
|
except QuizGame.DoesNotExist:
|
|
return False
|
|
|
|
async def show_scoreboard(self):
|
|
success = await self._show_scoreboard()
|
|
if success:
|
|
# Notify clients to redirect to finished page
|
|
await self.channel_layer.group_send(
|
|
self.game_group_name,
|
|
{
|
|
'type': 'game_state_update',
|
|
'action': 'show_scoreboard',
|
|
'redirect_url': reverse('play:scoreboard', kwargs={'join_code': self.join_code})
|
|
}
|
|
)
|