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

2from __future__ import annotations 

3 

4# -- stdlib -- 

5from typing import Any, Dict, Sequence, Tuple, Type, TypeVar 

6 

7# -- third party -- 

8# -- own -- 

9from client import parts 

10from client.base import Game 

11from game.base import Player 

12from utils.events import EventHub 

13import core 

14import wire 

15 

16 

17# -- code -- 

18class Options(object): 

19 def __init__(self, options: Dict[str, Any]): 

20 self.disables = options.get('disables', []) 

21 self.testing = options.get('testing', False) # In tests? 

22 

23 

24T = TypeVar('T') 

25 

26 

27class _ServerCommandMapping(dict): 

28 __slots__ = ('core',) 

29 

30 def __init__(self, core: Core): 

31 self.core = core 

32 super().__init__() 

33 

34 def __getitem__(self, k: Type[T]) -> EventHub[T]: 

35 if k in self: 

36 return dict.__getitem__(self, k) 

37 

38 hub = EventHub[T]() 

39 hub.name = f'{self.core}::server_command' 

40 self[k] = hub 

41 return hub 

42 

43 

44class Events(object): 

45 def __init__(self, core: Core) -> None: 

46 self.core = core 

47 

48 self.core_initialized = EventHub[Core]() 

49 

50 # Fires when server send some command 

51 self.server_command = _ServerCommandMapping(core) 

52 

53 # Server connected 

54 self.server_connected = EventHub[bool]() 

55 

56 # Server timed-out or actively rejects 

57 self.server_refused = EventHub[bool]() 

58 

59 # Server dropped 

60 self.server_dropped = EventHub[bool]() 

61 

62 # Server & client version mismatch 

63 self.version_mismatch = EventHub[bool]() 

64 

65 # Server error 

66 self.server_error = EventHub[str]() 

67 

68 # Server info 

69 self.server_info = EventHub[str]() 

70 

71 # Lobby status 

72 self.lobby_updated = EventHub[Tuple[Sequence[wire.model.User], Sequence[wire.model.Game]]]() 

73 

74 # Observer request 

75 self.observe_request = EventHub[int]() 

76 

77 # Observer state 

78 self.observer_enter = EventHub[Tuple[int, int]]() 

79 self.observer_leave = EventHub[Tuple[int, int]]() 

80 

81 # Joined a game 

82 self.game_joined = EventHub[Game]() 

83 

84 # Player presence changed 

85 self.player_presence = EventHub[Tuple[Game, Dict[Player, bool]]]() 

86 

87 # Left a game 

88 self.game_left = EventHub[Game]() 

89 

90 # Left a game 

91 # ev = (g: Game, users: [server.core.view.User(u), ...]) 

92 self.room_users = EventHub[Tuple[Game, Sequence[wire.model.User]]]() 

93 

94 # Game is up and running 

95 # ev = (g: Game) 

96 self.game_started = EventHub[Game]() 

97 

98 # ev = (g: Game) 

99 self.game_crashed = EventHub[Game]() 

100 

101 # Client game finished, 

102 # Server will send `game_end` soon if everything goes right 

103 self.client_game_finished = EventHub[Game]() 

104 

105 # ev = (g: Game) 

106 self.game_ended = EventHub[Game]() 

107 

108 # ev = uid: int 

109 self.auth_success = EventHub[int]() 

110 

111 # ev = reason: str 

112 self.auth_error = EventHub[str]() 

113 

114 def __setattr__(self, name: str, v: Any) -> None: 

115 if hasattr(v, 'name'): 

116 v.name = f'{repr(self.core)}::{name}' 

117 object.__setattr__(self, name, v) 

118 

119 

120class Core(core.Core): 

121 core_type = 'C' 

122 

123 def __init__(self, **options: Dict[str, Any]) -> None: 

124 super().__init__() 

125 

126 self.options = Options(options) 

127 self.events = Events(self) 

128 

129 disables = self.options.disables 

130 

131 if 'server' not in disables: 131 ↛ 134line 131 didn't jump to line 134, because the condition on line 131 was never false

132 self.server = parts.server.Server(self) 

133 

134 if 'auth' not in disables: 134 ↛ 137line 134 didn't jump to line 137, because the condition on line 134 was never false

135 self.auth = parts.auth.Auth(self) 

136 

137 if 'lobby' not in disables: 137 ↛ 140line 137 didn't jump to line 140, because the condition on line 137 was never false

138 self.lobby = parts.lobby.Lobby(self) 

139 

140 if 'room' not in disables: 140 ↛ 143line 140 didn't jump to line 143, because the condition on line 140 was never false

141 self.room = parts.room.Room(self) 

142 

143 if 'observe' not in disables: 143 ↛ 146line 143 didn't jump to line 146, because the condition on line 143 was never false

144 self.observe = parts.observe.Observe(self) 

145 

146 if 'game' not in disables: 146 ↛ 149line 146 didn't jump to line 149, because the condition on line 146 was never false

147 self.game = parts.game.GamePart(self) 

148 

149 if 'contest' not in disables: 149 ↛ 152line 149 didn't jump to line 152, because the condition on line 149 was never false

150 self.contest = parts.contest.Contest(self) 

151 

152 if 'replay' not in disables: 152 ↛ 155line 152 didn't jump to line 155, because the condition on line 152 was never false

153 self.replay = parts.replay.Replay(self) 

154 

155 if 'admin' not in disables: 155 ↛ 158line 155 didn't jump to line 158, because the condition on line 155 was never false

156 self.admin = parts.admin.Admin(self) 

157 

158 if 'warpgate' not in disables: 158 ↛ 159line 158 didn't jump to line 159, because the condition on line 158 was never true

159 self.warpgate = parts.warpgate.Warpgate(self) 

160 

161 self.events.core_initialized.emit(self)