Coverage for server/parts/contest.py : 91%
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 typing import List, Optional, Sequence, TYPE_CHECKING, Tuple
6import logging
8# -- third party --
9from mypy_extensions import TypedDict
11# -- own --
12from server.base import Game, HumanPlayer
13from server.endpoint import Client
14from server.utils import command
15from utils.events import EventHub
16import wire
18# -- typing --
19if TYPE_CHECKING:
20 from server.core import Core # noqa: F401
23# -- code --
24log = logging.getLogger('Contest')
27class ContestAssocOnGame(TypedDict):
28 uids: List[int]
31def A(self: Contest, g: Game) -> ContestAssocOnGame:
32 return g._[self]
35class Contest(object):
36 def __init__(self, core: Core):
37 self.core = core
39 core.events.user_state_transition += self.handle_user_state_transition
40 core.events.game_started += self.handle_game_started
41 core.events.game_aborted += self.handle_game_aborted
42 core.events.game_ended += self.handle_game_ended
43 core.events.game_successive_create += self.handle_game_successive_create
45 D = core.events.client_command
46 D[wire.SetupContest] += self._contest
47 D[wire.JoinRoom].subscribe(self._room_join_contest_limit, -3)
49 def __repr__(self) -> str:
50 return self.__class__.__name__
52 def handle_user_state_transition(self, ev: Tuple[Client, str, str]) -> Tuple[Client, str, str]:
53 c, f, t = ev
54 return ev
56 def handle_game_started(self, g: Game) -> Game:
57 core = self.core
59 name = core.room.name_of(g)
60 flags = core.room.flags_of(g)
61 users = core.room.users_of(g)
63 if flags.get('contest'):
64 core.connect.speaker(
65 '文文', '“%s”开始了！参与玩家：%s' % (
66 name, '，'.join([core.auth.name_of(u) for u in users])
67 )
68 )
70 return g
72 def handle_game_aborted(self, g: Game) -> Game:
73 core = self.core
74 if core.room.is_started(g) and core.room.flags_of(g).get('contest'): 74 ↛ 75line 74 didn't jump to line 75, because the condition on line 74 was never true
75 core.connect.speaker(
76 '文文', '“%s”意外终止了，比赛结果作废！' % core.room.name_of(g)
77 )
79 return g
81 def handle_game_ended(self, g: Game) -> Game:
82 core = self.core
84 if not core.room.flags_of(g).get('contest'):
85 return g
87 if core.game.is_aborted(g): 87 ↛ 88line 87 didn't jump to line 88, because the condition on line 87 was never true
88 return g
90 winners: List[Client] = [
91 u.client for u in core.game.winners_of(g)
92 if isinstance(u, HumanPlayer)
93 ]
95 core.connect.speaker(
96 '文文',
97 '“%s”结束了！获胜玩家：%s' % (
98 core.room.name_of(g),
99 '，'.join([core.auth.name_of(u) for u in winners])
100 )
101 )
103 return g
105 def handle_game_successive_create(self, ev: Tuple[Game, Game]) -> Tuple[Game, Game]:
106 core = self.core
107 old, g = ev
108 flags = core.room.flags_of(g)
109 if flags.get('contest'):
110 fields = old._[self]
111 g._[self] = fields
112 self._start_poll(g, fields['uids'])
113 return ev
115 # ----- Client Commands -----
116 @command('*')
117 def _contest(self, c: Client, ev: wire.SetupContest) -> None:
118 core = self.core
119 from thb import modes
120 gamecls = modes[ev.mode]
121 if len(ev.uids) != gamecls.n_persons:
122 c.write(wire.Error(msg='wrong_players_count'))
123 return
125 g = core.room.create_game(gamecls, ev.name, {'contest': True})
127 assoc: ContestAssocOnGame = {
128 'uids': ev.uids,
129 }
130 g._[self] = assoc
131 self._start_poll(g, ev.uids)
133 @command('*')
134 def _room_join_contest_limit(self, u: Client, ev: wire.JoinRoom) -> Optional[EventHub.StopPropagation]:
135 core = self.core
137 g = core.room.get(ev.gid)
138 if not g: 138 ↛ 139line 138 didn't jump to line 139, because the condition on line 138 was never true
139 return None
141 flags = core.room.flags_of(g)
142 uid = core.auth.uid_of(u)
144 if flags.get('contest'):
145 uid = core.auth.uid_of(u)
146 if uid not in A(self, g)['uids']: 146 ↛ 150line 146 didn't jump to line 150, because the condition on line 146 was never false
147 u.write(wire.Error('not_competitor'))
148 return EventHub.STOP_PROPAGATION
150 return None
152 # ----- Methods -----
153 def _start_poll(self, g: Game, uids: Sequence[int]) -> None:
154 core = self.core
155 gid = core.room.gid_of(g)
156 name = core.room.name_of(g)
158 core.runner.spawn(core.connect.speaker, '文文', f'“{name}”房间已经建立，请相关玩家就位！')
160 @core.runner.spawn
161 def pull() -> None:
162 while core.room.get(gid) is g: 162 ↛ 198line 162 didn't jump to line 198, because the condition on line 162 was never false
163 users = core.room.online_users_of(g)
164 uids = {core.auth.uid_of(u) for u in users}
165 contest_uids = set(A(self, g)['uids'])
166 pending = contest_uids - uids
167 if not pending:
168 break
170 log.debug("Contest: Ready to pull %s", pending)
172 for uid in pending:
173 u = core.lobby.get(uid)
174 if not u:
175 log.debug("Contest: %s not found", uid)
176 continue
178 if core.lobby.state_of(u) == 'lobby':
179 log.debug("Contest: Pulling %s", uid)
180 core.room.join_game(g, u)
181 elif core.lobby.state_of(u) in ('ready', 'room'): 181 ↛ 182line 181 didn't jump to line 182, because the condition on line 181 was never true
182 log.debug("Contest: Pulling %s from state [%s]", uid, core.lobby.state_of(u).state)
183 core.room.exit_game(u)
184 core.room.join_game(g, u)
185 elif core.lobby.state_of(u) == 'ob':
186 log.debug("Contest: Pulling %s from state [%s]", uid, core.lobby.state_of(u).state)
187 core.observe.observe_detach(u)
188 core.room.join_game(g, u)
189 elif core.lobby.state_of(u) == 'game': 189 ↛ 172line 189 didn't jump to line 172, because the condition on line 189 was never false
190 log.debug("Contest: %s in game, sending warning", uid)
191 core.runner.spawn(u.write, wire.SystemMsg('你有比赛房间，请尽快结束游戏参与比赛'))
193 if core.options.testing: 193 ↛ 196line 193 didn't jump to line 196, because the condition on line 193 was never false
194 core.runner.sleep(0.01)
195 else:
196 core.runner.sleep(30)
198 log.debug("Contest: Finished pulling for game %s", gid)