Coverage for integrations/social/game

1"""

2HevolveSocial - Game Type Strategies

3Each game type implements initialize, validate_move, apply_move,

4check_round_end, check_game_end, calculate_results.

5"""

6import logging

7import random

8from datetime import datetime

9from typing import Dict, List, Tuple, Optional

11logger = logging.getLogger('hevolve_social')

14class BaseGameType:

15 """Base class for game types."""

17 def initialize(self, config: Dict, total_rounds: int,

18 player_ids: List[str]) -> Dict:

19 """Create initial game state."""

20 raise NotImplementedError

22 def validate_move(self, game_state: Dict, user_id: str,

23 move_data: Dict) -> Tuple[bool, str]:

24 """Check if a move is valid. Returns (valid, reason)."""

25 raise NotImplementedError

27 def apply_move(self, game_state: Dict, user_id: str,

28 move_data: Dict) -> Tuple[Dict, int]:

29 """Apply move. Returns (new_state, score_delta)."""

30 raise NotImplementedError

32 def check_round_end(self, game_state: Dict) -> bool:

33 """Check if current round is over."""

34 raise NotImplementedError

36 def check_game_end(self, game_state: Dict, current_round: int,

37 total_rounds: int) -> bool:

38 """Check if game should end."""

39 return current_round > total_rounds

41 def calculate_results(self, game_state: Dict,

42 participants) -> Dict[str, Dict]:

43 """Calculate final results. Returns {user_id: {result, ...}}."""

44 raise NotImplementedError

47class TriviaGame(BaseGameType):

48 """Timed trivia rounds. First correct answer scores."""

50 # Pre-seeded question categories

51 CATEGORIES = {

52 'general': [

53 {'q': 'What is the largest planet in our solar system?', 'a': 'jupiter', 'options': ['mars', 'jupiter', 'saturn', 'neptune']},

54 {'q': 'What language has the most native speakers?', 'a': 'mandarin', 'options': ['english', 'mandarin', 'spanish', 'hindi']},

55 {'q': 'What is the chemical symbol for gold?', 'a': 'au', 'options': ['ag', 'au', 'fe', 'cu']},

56 {'q': 'Which ocean is the deepest?', 'a': 'pacific', 'options': ['atlantic', 'pacific', 'indian', 'arctic']},

57 {'q': 'How many bits in a byte?', 'a': '8', 'options': ['4', '8', '16', '32']},

58 ],

59 'tech': [

60 {'q': 'Who created Linux?', 'a': 'linus torvalds', 'options': ['bill gates', 'linus torvalds', 'steve jobs', 'dennis ritchie']},

61 {'q': 'What does HTTP stand for?', 'a': 'hypertext transfer protocol', 'options': ['hypertext transfer protocol', 'high tech transfer protocol', 'hypertext transport protocol', 'high text transfer program']},

62 {'q': 'Which company created Python?', 'a': 'none', 'options': ['google', 'microsoft', 'none', 'sun microsystems']},

63 {'q': 'What year was the first iPhone released?', 'a': '2007', 'options': ['2005', '2007', '2008', '2010']},

64 {'q': 'What does GPU stand for?', 'a': 'graphics processing unit', 'options': ['graphics processing unit', 'general processing unit', 'graphical power unit', 'graphics protocol unit']},

65 ],

66 'science': [

67 {'q': 'What is the speed of light in km/s (approx)?', 'a': '300000', 'options': ['150000', '300000', '500000', '1000000']},

68 {'q': 'What is the powerhouse of the cell?', 'a': 'mitochondria', 'options': ['nucleus', 'mitochondria', 'ribosome', 'golgi']},

69 {'q': 'What gas do plants absorb?', 'a': 'carbon dioxide', 'options': ['oxygen', 'nitrogen', 'carbon dioxide', 'hydrogen']},

70 {'q': 'How many chromosomes do humans have?', 'a': '46', 'options': ['23', '46', '48', '44']},

71 {'q': 'What planet is known as the Red Planet?', 'a': 'mars', 'options': ['venus', 'mars', 'jupiter', 'mercury']},

72 ],

73 }

75 def initialize(self, config, total_rounds, player_ids):

76 category = config.get('category', 'general')

77 questions = list(self.CATEGORIES.get(category, self.CATEGORIES['general']))

78 random.shuffle(questions)

79 return {

80 'questions': questions[:total_rounds],

81 'current_question_idx': 0,

82 'answers': {}, # {round_idx: {user_id: answer}}

83 'round_scores': {}, # {user_id: [round_score, ...]}

84 'round_answered': False,

85 'players': player_ids,

86 }

88 def validate_move(self, game_state, user_id, move_data):

89 if 'answer' not in move_data:

90 return False, "Missing 'answer' in move_data"

91 idx = game_state.get('current_question_idx', 0)

92 answers = game_state.get('answers', {})

93 round_answers = answers.get(str(idx), {})

94 if user_id in round_answers:

95 return False, "Already answered this round"

96 return True, ""

98 def apply_move(self, game_state, user_id, move_data):

99 idx = game_state.get('current_question_idx', 0)

100 questions = game_state.get('questions', [])

101 if idx >= len(questions):

102 return game_state, 0

103

104 answer = str(move_data['answer']).lower().strip()

105 correct_answer = questions[idx]['a'].lower().strip()

106

107 answers = game_state.get('answers', {})

108 round_answers = answers.get(str(idx), {})

109 round_answers[user_id] = answer

110 answers[str(idx)] = round_answers

111 game_state['answers'] = answers

112

113 score_delta = 0

114 if answer == correct_answer and not game_state.get('round_answered'):

115 score_delta = 10 # First correct answer gets points

116 game_state['round_answered'] = True

117

118 return game_state, score_delta

119

120 def check_round_end(self, game_state):

121 idx = game_state.get('current_question_idx', 0)

122 answers = game_state.get('answers', {})

123 round_answers = answers.get(str(idx), {})

124 players = game_state.get('players', [])

125 if len(round_answers) >= len(players):

126 game_state['current_question_idx'] = idx + 1

127 game_state['round_answered'] = False

128 return True

129 return game_state.get('round_answered', False)

130

131 def check_game_end(self, game_state, current_round, total_rounds):

132 idx = game_state.get('current_question_idx', 0)

133 questions = game_state.get('questions', [])

134 return idx >= len(questions) or current_round > total_rounds

135

136 def calculate_results(self, game_state, participants):

137 scores = {p.user_id: p.score for p in participants}

138 max_score = max(scores.values()) if scores else 0

139 results = {}

140 for uid, score in scores.items():

141 if score == max_score and max_score > 0:

142 results[uid] = {'result': 'win', 'score': score}

143 elif score == max_score:

144 results[uid] = {'result': 'draw', 'score': score}

145 else:

146 results[uid] = {'result': 'loss', 'score': score}

147 return results

148

149

150class WordChainGame(BaseGameType):

151 """Turn-based word chain. Each word starts with last letter of previous."""

152

153 def initialize(self, config, total_rounds, player_ids):

154 topic = config.get('topic', '')

155 return {

156 'words': [],

157 'turn_order': player_ids,

158 'current_turn_idx': 0,

159 'topic': topic,

160 'used_words': set(),

161 'skips': {uid: 0 for uid in player_ids},

162 'max_skips': 2,

163 'players': player_ids,

164 }

165

166 def validate_move(self, game_state, user_id, move_data):

167 turn_order = game_state.get('turn_order', [])

168 current_idx = game_state.get('current_turn_idx', 0)

169 if not turn_order:

170 return False, "No players"

171 expected_player = turn_order[current_idx % len(turn_order)]

172 if user_id != expected_player:

173 return False, "Not your turn"

174 if 'word' not in move_data and not move_data.get('skip'):

175 return False, "Missing 'word' in move_data"

176 return True, ""

177

178 def apply_move(self, game_state, user_id, move_data):

179 words = game_state.get('words', [])

180 # Handle serialized set

181 used_words = game_state.get('used_words', [])

182 if isinstance(used_words, list):

183 used_words = set(used_words)

184

185 if move_data.get('skip'):

186 skips = game_state.get('skips', {})

187 skips[user_id] = skips.get(user_id, 0) + 1

188 game_state['skips'] = skips

189 game_state['current_turn_idx'] = game_state.get('current_turn_idx', 0) + 1

190 return game_state, 0

191

192 word = move_data['word'].lower().strip()

193

194 # Check duplicate

195 if word in used_words:

196 return game_state, -2 # penalty

197

198 # Check chain rule

199 if words:

200 last_letter = words[-1][-1]

201 if word[0] != last_letter:

202 return game_state, -1 # wrong starting letter

203

204 words.append(word)

205 used_words.add(word)

206 game_state['words'] = words

207 game_state['used_words'] = list(used_words) # JSON-serializable

208 game_state['current_turn_idx'] = game_state.get('current_turn_idx', 0) + 1

209

210 score = len(word) # longer words = more points

211 return game_state, score

212

213 def check_round_end(self, game_state):

214 turn_idx = game_state.get('current_turn_idx', 0)

215 players = game_state.get('players', [])

216 # Round ends after each player has had a turn

217 return turn_idx > 0 and turn_idx % len(players) == 0

218

219 def check_game_end(self, game_state, current_round, total_rounds):

220 # Also end if all players exceeded max skips

221 skips = game_state.get('skips', {})

222 max_skips = game_state.get('max_skips', 2)

223 all_exhausted = all(v >= max_skips for v in skips.values()) if skips else False

224 return current_round > total_rounds or all_exhausted

225

226 def calculate_results(self, game_state, participants):

227 scores = {p.user_id: p.score for p in participants}

228 max_score = max(scores.values()) if scores else 0

229 results = {}

230 for uid, score in scores.items():

231 if score == max_score and max_score > 0:

232 results[uid] = {'result': 'win', 'score': score}

233 elif score == max_score:

234 results[uid] = {'result': 'draw', 'score': score}

235 else:

236 results[uid] = {'result': 'loss', 'score': score}

237 return results

238

239

240class CollabPuzzleGame(BaseGameType):

241 """Cooperative: players collectively arrange ideas into a thought experiment.

242 Everyone wins together — no competition."""

243

244 # Pre-seeded puzzle fragments

245 PUZZLES = [

246 {

247 'theme': 'community',

248 'fragments': [

249 'If every neighborhood had a shared AI assistant',

250 'that learned from local traditions and needs',

251 'it could help coordinate resources during emergencies',

252 'while preserving cultural identity and privacy',

253 'leading to stronger bonds between neighbors',

254 ],

255 },

256 {

257 'theme': 'environment',

258 'fragments': [

259 'If idle computing power from millions of devices',

260 'was donated to climate modeling simulations',

261 'scientists could predict weather patterns more accurately',

262 'helping farmers optimize crop yields',

263 'reducing food waste by 30% globally',

264 ],

265 },

266 {

267 'theme': 'education',

268 'fragments': [

269 'If students could create AI tutors trained on their own learning style',

270 'these personal agents would adapt to individual pace',

271 'teachers would be freed to focus on emotional support',

272 'every child would have access to world-class instruction',

273 'closing the education gap between urban and rural areas',

274 ],

275 },

276 ]

277

278 def initialize(self, config, total_rounds, player_ids):

279 puzzle = random.choice(self.PUZZLES)

280 fragments = list(puzzle['fragments'])

281 random.shuffle(fragments)

282 return {

283 'theme': puzzle['theme'],

284 'fragments': fragments,

285 'correct_order': puzzle['fragments'],

286 'submitted_order': [],

287 'contributions': {}, # {user_id: count}

288 'players': player_ids,

289 'moves_made': 0,

290 }

291

292 def validate_move(self, game_state, user_id, move_data):

293 if 'fragment_index' not in move_data or 'position' not in move_data:

294 return False, "Need 'fragment_index' and 'position'"

295 idx = move_data['fragment_index']

296 fragments = game_state.get('fragments', [])

297 if idx < 0 or idx >= len(fragments):

298 return False, "Invalid fragment index"

299 return True, ""

300

301 def apply_move(self, game_state, user_id, move_data):

302 fragments = game_state.get('fragments', [])

303 submitted = game_state.get('submitted_order', [])

304 idx = move_data['fragment_index']

305 position = move_data['position']

306

307 fragment = fragments[idx]

308 # Insert at position (or append)

309 if position >= len(submitted):

310 submitted.append(fragment)

311 else:

312 submitted.insert(position, fragment)

313

314 game_state['submitted_order'] = submitted

315 game_state['moves_made'] = game_state.get('moves_made', 0) + 1

316

317 # Track who contributed

318 contributions = game_state.get('contributions', {})

319 contributions[user_id] = contributions.get(user_id, 0) + 1

320 game_state['contributions'] = contributions

321

322 # Score: +5 for each fragment in correct position

323 score_delta = 0

324 correct = game_state.get('correct_order', [])

325 if position < len(correct) and fragment == correct[position]:

326 score_delta = 5

327

328 return game_state, score_delta

329

330 def check_round_end(self, game_state):

331 submitted = game_state.get('submitted_order', [])

332 correct = game_state.get('correct_order', [])

333 return len(submitted) >= len(correct)

334

335 def check_game_end(self, game_state, current_round, total_rounds):

336 return self.check_round_end(game_state)

337

338 def calculate_results(self, game_state, participants):

339 # Cooperative: everyone wins if puzzle is assembled

340 submitted = game_state.get('submitted_order', [])

341 correct = game_state.get('correct_order', [])

342 # Calculate similarity

343 matches = sum(1 for s, c in zip(submitted, correct) if s == c)

344 total = len(correct)

345 accuracy = matches / total if total > 0 else 0

346

347 result = 'win' if accuracy >= 0.6 else 'draw'

348 return {p.user_id: {'result': result, 'accuracy': accuracy}

349 for p in participants}

350

351

352class ComputeChallengeGame(BaseGameType):

353 """Race to complete micro-compute tasks from the distributed coordinator.

354 Bridge to compute lending — shows users what distributed compute does."""

355

356 def initialize(self, config, total_rounds, player_ids):

357 return {

358 'task_completions': {uid: 0 for uid in player_ids},

359 'tasks_verified': {uid: 0 for uid in player_ids},

360 'target_tasks': config.get('target_tasks', 5),

361 'players': player_ids,

362 'moves_made': 0,

363 }

364

365 def validate_move(self, game_state, user_id, move_data):

366 if 'task_result' not in move_data:

367 return False, "Missing 'task_result'"

368 return True, ""

369

370 def apply_move(self, game_state, user_id, move_data):

371 completions = game_state.get('task_completions', {})

372 completions[user_id] = completions.get(user_id, 0) + 1

373 game_state['task_completions'] = completions

374 game_state['moves_made'] = game_state.get('moves_made', 0) + 1

375

376 # If task was verified, extra points

377 verified = game_state.get('tasks_verified', {})

378 if move_data.get('verified'):

379 verified[user_id] = verified.get(user_id, 0) + 1

380 game_state['tasks_verified'] = verified

381 return game_state, 15 # verified task = more points

382 return game_state, 10

383

384 def check_round_end(self, game_state):

385 # No distinct rounds — continuous play

386 return False

387

388 def check_game_end(self, game_state, current_round, total_rounds):

389 target = game_state.get('target_tasks', 5)

390 completions = game_state.get('task_completions', {})

391 # End when any player hits the target

392 return any(v >= target for v in completions.values())

393

394 def calculate_results(self, game_state, participants):

395 completions = game_state.get('task_completions', {})

396 max_completions = max(completions.values()) if completions else 0

397 results = {}

398 for p in participants:

399 count = completions.get(p.user_id, 0)

400 if count == max_completions and max_completions > 0:

401 results[p.user_id] = {'result': 'win', 'tasks_completed': count}

402 else:

403 results[p.user_id] = {'result': 'loss', 'tasks_completed': count}

404 return results

405

406

407# ─── Registry ───

408

409_GAME_TYPES = {

410 'trivia': TriviaGame(),

411 'word_chain': WordChainGame(),

412 'collab_puzzle': CollabPuzzleGame(),

413 'compute_challenge': ComputeChallengeGame(),

414 'quick_match': TriviaGame(), # quick_match defaults to trivia

415}

416

417

418def _ensure_extended_types():

419 """Lazy-register extended game type engines."""

420 if 'opentdb_trivia' in _GAME_TYPES:

421 return # already registered

422 try:

423 from .game_types_extended import (

424 OpenTDBTriviaGame, BoardGameType, PhaserGameType,

425 WordScrambleGame, WordSearchGame, SudokuGame,

426 )

427 _GAME_TYPES.update({

428 'opentdb_trivia': OpenTDBTriviaGame(),

429 'boardgame': BoardGameType(),

430 'phaser': PhaserGameType(),

431 'word_scramble': WordScrambleGame(),

432 'word_search': WordSearchGame(),

433 'sudoku': SudokuGame(),

434 })

435 logger.info("Extended game types registered: %d total engines",

436 len(_GAME_TYPES))

437 except ImportError as e:

438 logger.warning("Could not load extended game types: %s", e)

439

440

441def get_game_type(game_type: str) -> BaseGameType:

442 """Get handler for a game type (engine name or catalog ID)."""

443 _ensure_extended_types()

444

445 # Direct engine match

446 handler = _GAME_TYPES.get(game_type)

447 if handler:

448 return handler

449

450 # Try resolving via catalog (catalog ID → engine name)

451 try:

452 from .game_catalog import get_engine_for_catalog_entry

453 engine = get_engine_for_catalog_entry(game_type)

454 if engine:

455 handler = _GAME_TYPES.get(engine)

456 if handler:

457 return handler

458 except ImportError:

459 pass

460

461 raise ValueError(f"Unknown game type: {game_type}")

462

463

464def is_valid_game_type(game_type: str) -> bool:

465 """Check if a game type or catalog ID is valid."""

466 _ensure_extended_types()

467 if game_type in _GAME_TYPES:

468 return True

469 try:

470 from .game_catalog import get_catalog_entry

471 return get_catalog_entry(game_type) is not None

472 except ImportError:

473 return False

Coverage for integrations / social / game_types.py: 0.0%

241 statements