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 = 0 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.get('tolerance_radius', 1000) # Default 1km if not set # Parse coordinates if they are strings def parse_coords(coord): if isinstance(coord, str): try: # Handle both comma and space separated coordinates if ',' in coord: lat, lng = map(float, [x.strip() for x in coord.split(',')]) else: # Try splitting by space if no comma parts = coord.strip().split() if len(parts) >= 2: lat, lng = map(float, parts[:2]) else: raise ValueError("Invalid coordinate format") return (lat, lng) except (ValueError, AttributeError) as e: print(f"Error parsing coordinates {coord}: {e}") return None elif isinstance(coord, (list, tuple)) and len(coord) >= 2: return (float(coord[0]), float(coord[1])) return None # Parse the coordinates target_location = parse_coords(target_location) answer_location = parse_coords(answer_location) if not target_location or not answer_location: print("Invalid coordinates, cannot calculate score") is_correct = False score = 0 else: # Configuration - Make these easily adjustable MAX_DISTANCE_KM = 200 # Maximum distance for scoring (in km) DIRECT_HIT_BONUS = 200 # Extra points for direct hit within tolerance try: # Calculate distance distance_km = geodesic(answer_location, target_location).kilometers tolerance_km = float(tolerance_radius) / 1000 # Convert meters to kilometers print(f"Distance: {distance_km} km, Tolerance: {tolerance_km} km") # Calculate base score with non-linear decrease if distance_km <= tolerance_km: # Direct hit - full points + bonus base_score = 1000 + DIRECT_HIT_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 progress = (distance_km - tolerance_km) / (MAX_DISTANCE_KM - tolerance_km) falloff = progress ** 1.5 # 1.5 makes the falloff steeper at the beginning base_score = 1000 * (1 - falloff) is_correct = True # 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 # Apply max score limit after all calculations (including bonus) final_score = min(final_score, 1200) # Cap at 1200 (1000 + 200 bonus) score = int(round(final_score)) except Exception as e: print(f"Error calculating score: {e}") is_correct = False score = 0 # 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}) } )