Coverage for thb/common.py : 96%
Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# -*- coding: utf-8 -*-
2from __future__ import annotations
4# -- stdlib --
5from enum import Enum
6from itertools import cycle
7from typing import Any, Dict, Generic, Iterable, List, Optional, Sequence, TYPE_CHECKING, Tuple
8from typing import Type, TypeVar
9import logging
10import random
12# -- third party --
13from mypy_extensions import TypedDict
15# -- own --
16from game.base import GameViralContext, Player, get_seed_for, sync_primitive
17from thb.item import GameItem
18from thb.mode import THBattle
19from utils.misc import BatchList, partition
20import settings
22# -- typing --
23if TYPE_CHECKING:
24 from thb.characters.base import Character # noqa: F401
27# -- code --
28log = logging.getLogger('thb.common')
31class CharChoice(GameViralContext):
32 chosen: Any = None
33 char_cls: Optional[Type[Character]]
34 akari: bool = False
36 def __init__(self, char_cls=None, akari=False) -> None:
37 self.set(char_cls, akari)
39 def dump(self):
40 return self.char_cls.__name__ if not self.akari else 'Akari'
42 def sync(self, data) -> None:
43 from thb.characters.base import Character
44 self.set(Character.classes[data], False)
46 def conceal(self) -> None:
47 self.char_cls = None
48 self.chosen = None
49 self.akari = False
51 def set(self, char_cls, akari=False) -> None:
52 self.char_cls = char_cls
53 g = self.game
55 if akari:
56 self.akari = True
57 if g.is_client_side():
58 from thb import characters
59 self.char_cls = characters.akari.Akari
61 def __repr__(self):
62 return '<Choice: {}{}>'.format(
63 'None' if not self.char_cls else self.char_cls.__name__,
64 '[Akari]' if self.akari else '',
65 )
68T = TypeVar('T', bound=Enum)
71class PlayerRole(Generic[T], GameViralContext):
72 _role: T
74 def __init__(self, typ: Type[T]):
75 self._typ = typ
76 self._role = typ(0)
78 def __str__(self) -> str:
79 return self._role.name
81 def __eq__(self, other: object) -> bool:
82 if not isinstance(other, self._typ): 82 ↛ 83line 82 didn't jump to line 83, because the condition on line 82 was never true
83 return False
85 return self._role == other
87 def dump(self) -> Any:
88 return self._role.value
90 def sync(self, data) -> None:
91 self._role = self._typ(data)
93 '''
94 def is_type(self, t: Enum) -> bool:
95 g = self.game
96 pl = g.players
97 return sync_primitive(self.identity == t, pl)
98 '''
100 def set(self, t: T) -> None:
101 assert isinstance(t, self._typ)
102 if self.game.is_server_side():
103 self._role = self._typ(t)
105 def get(self) -> T:
106 return self._role
109def roll(g: THBattle, pl: BatchList[Player], items: Dict[Player, List[GameItem]]) -> BatchList[Player]:
110 from thb.item import European
111 roll = list(range(len(pl)))
112 g.random.shuffle(roll)
113 eu = European.get_european(g, items)
114 if eu:
115 i = pl.index(eu)
116 roll.remove(i)
117 roll.insert(0, i)
119 roll = sync_primitive(roll, pl)
120 roll = BatchList(pl[i] for i in roll)
121 g.emit_event('game_roll', roll)
122 return roll
125class BuildChoicesSpec(TypedDict):
126 num: int
127 akaris: int
130class VirtualPlayer(Player):
132 def __init__(self, players: Sequence[Player]):
133 self.players = players
134 super().__init__()
136 def reveal(self, obj: Any):
137 for p in self.players:
138 p.reveal(obj)
141def build_choices_shared(g: THBattle,
142 players: BatchList[Player],
143 items: Dict[Player, List[GameItem]],
144 candidates: List[Type[Character]],
145 spec: BuildChoicesSpec,
146 ) -> Tuple[List[CharChoice], Dict[Player, CharChoice]]:
147 p = VirtualPlayer(players)
148 choices, imperial = build_choices(g, players, items, candidates, {p: spec})
149 return choices[p], imperial
152def build_choices(g: THBattle,
153 players: BatchList[Player],
154 items: Dict[Player, List[GameItem]],
155 candidates: List[Type[Character]],
156 spec: Dict[Player, BuildChoicesSpec],
157 ) -> Tuple[Dict[Player, List[CharChoice]], Dict[Player, CharChoice]]:
159 from thb.item import ImperialChoice
161 # ----- testing -----
162 from thb.characters.base import Character
164 testing_lst: Iterable[str] = settings.TESTING_CHARACTERS
165 testing = list(Character.classes[i] for i in testing_lst)
166 candidates, _ = partition(lambda c: c not in testing, candidates)
168 if g.is_server_side():
169 candidates = list(candidates)
170 g.random.shuffle(candidates)
171 else:
172 candidates = [None] * len(candidates)
174 assert sum(s['num'] for p, s in spec.items()) <= len(candidates) + len(testing), 'Insufficient choices'
176 result: Dict[Player, List[CharChoice]] = {p: [] for p in spec}
178 players_for_testing = list(spec)
180 candidates = list(candidates)
181 seed = get_seed_for(g, players)
182 shuffler = random.Random(seed)
183 shuffler.shuffle(players_for_testing)
185 for e, cls in zip(cycle(players_for_testing), testing): 185 ↛ 186line 185 didn't jump to line 186, because the loop on line 185 never started
186 result[e].append(CharChoice(cls))
188 # ----- imperial (force chosen by ImperialChoice) -----
189 imperial = ImperialChoice.get_chosen(items, players)
190 imperial = {p: CharChoice(cls) for p, cls in imperial.items()}
192 for p, c in imperial.items(): 192 ↛ 193line 192 didn't jump to line 193, because the loop on line 192 never started
193 result[p].append(c)
195 # ----- normal -----
196 for p, s in spec.items():
197 for _ in range(len(result[p]), s['num']):
198 result[p].append(CharChoice(candidates.pop()))
200 # ----- akaris -----
201 if g.is_server_side():
202 rest = candidates
203 else:
204 rest = [None] * len(candidates)
206 g.random.shuffle(rest)
208 for p, s in spec.items():
209 for i in range(-s['akaris'], 0):
210 result[p][i].set(rest.pop(), True)
212 # ----- compose final result, reveal, and return -----
213 for p, l in result.items():
214 p.reveal(l)
216 return result, imperial