1# -*- coding: utf-8 -*- 

2from __future__ import annotations 

3 

4# -- stdlib -- 

5from typing import Dict, List, Set, TYPE_CHECKING, Tuple 

6import logging 

7 

8# -- third party -- 

9from typing_extensions import TypedDict 

10 

11# -- own -- 

12from server.endpoint import Client 

13from server.utils import command 

14from utils.misc import LoopBreaker 

15import wire 

16 

17# -- typing -- 

18if TYPE_CHECKING: 

19 from server.core import Core # noqa: F401 

20 

21 

22# -- code -- 

23log = logging.getLogger('Matching') 

24 

25 

26class MatchingAssocOnClient(TypedDict): 

27 modes: List[str] 

28 

29 

30def Au(self: Matching, u: Client) -> MatchingAssocOnClient: 

31 return u._[self] 

32 

33 

34class Matching(object): 

35 def __init__(self, core: Core): 

36 self.core = core 

37 

38 core.events.user_state_transition += self.handle_user_state_transition 

39 core.events.core_initialized += self.handle_core_initialized 

40 

41 D = core.events.client_command 

42 D[wire.StartMatching] += self._start_matching 

43 

44 self.outstanding: Dict[str, Set[Client]] = {} 

45 

46 def __repr__(self) -> str: 

47 return self.__class__.__name__ 

48 

49 def handle_core_initialized(self, ev: Core) -> Core: 

50 from thb import modes 

51 order = list(modes) 

52 order.sort(key=lambda v: modes[v].n_persons, reverse=True) 

53 

54 for m in order: 

55 self.outstanding[m] = set() 

56 

57 return ev 

58 

59 def handle_user_state_transition(self, ev: Tuple[Client, str, str]) -> Tuple[Client, str, str]: 

60 c, f, t = ev 

61 

62 if t in ('room', 'ready', 'game'): 

63 self._clear(c) 

64 

65 elif t == 'lobby': 

66 c.write(wire.StartMatching(modes=Au(self, c)['modes'])) 

67 

68 elif t == 'connected': 

69 assoc: MatchingAssocOnClient = { 

70 'modes': [], 

71 } 

72 c._[self] = assoc 

73 

74 return ev 

75 

76 # ----- Public Methods ----- 

77 def do_match(self) -> None: 

78 core = self.core 

79 from thb import modes 

80 

81 for loop in LoopBreaker(False): 

82 for m in self.outstanding: 

83 candidates = [] 

84 avail = self.outstanding[m] 

85 cls = modes[m] 

86 if len(avail) < cls.n_persons: 

87 continue 

88 

89 for _ in range(cls.n_persons): 

90 u = avail.pop() 

91 assert core.lobby.state_of(u) == 'lobby' 

92 candidates.append(u) 

93 

94 for u in candidates: 

95 self._clear(u) 

96 

97 g = core.room.create_game(cls, "匹配的游戏", {}) 

98 for u in candidates: 

99 core.room.join_game(g, u) 

100 

101 loop.cont() 

102 break 

103 

104 # ----- Methods ----- 

105 def _clear(self, u: Client) -> None: 

106 Au(self, u)['modes'] = [] 

107 for m, s in self.outstanding.items(): 

108 s.discard(u) 

109 

110 # ----- Client Commands ----- 

111 @command('lobby') 

112 def _start_matching(self, u: Client, ev: wire.StartMatching) -> None: 

113 from thb import modes 

114 

115 self._clear(u) 

116 filtered = list(set(modes) & set(ev.modes)) 

117 Au(self, u)['modes'] = filtered 

118 u.write(wire.StartMatching(modes=filtered)) 

119 if not filtered: 119 ↛ 120line 119 didn't jump to line 120, because the condition on line 119 was never true

120 return 

121 

122 for m in filtered: 

123 self.outstanding.setdefault(m, set()).add(u) 

124 

125 self.do_match()