# Coverage for server/parts/matching.py : 98%

## 74 statements 73 run 1 missing 4 excluded 1 partial

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 -*-

2 from __future__ import annotations

3

4# -- stdlib --

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

6 import logging

7

8# -- third party --

9 from typing_extensions import TypedDict

10

11# -- own --

12 from server.endpoint import Client

13 from server.utils import command

14 from utils.misc import LoopBreaker

15 import wire

16

17# -- typing --

18 if TYPE_CHECKING:

19 from server.core import Core # noqa: F401

20

21

22# -- code --

23 log = logging.getLogger('Matching')

24

25

26 class MatchingAssocOnClient(TypedDict):

27 modes: List[str]

28

29

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

31 return u._[self]

32

33

34 class 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()