# Coverage for client/parts/warpgate.py : 38%

## 107 statements 46 run 61 missing 2 excluded 0 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 Any, Callable, List, TYPE_CHECKING, Tuple

6 import random

7 import sys

8 import typing

9

10# -- third party --

11 from gevent.event import Event

12 from mypy_extensions import TypedDict

13 from typing_extensions import Literal

14 import gevent

15 import gevent.hub

16

17# -- own --

18 from client.base import Game

19 from game.base import EventHandler

20

21

22# -- typing --

23 if TYPE_CHECKING:

24 from client.core import Core # noqa: F401

25

26

27# -- code --

28 class GameEvent(TypedDict):

29 t: Literal['g']

30 g: Game

31 evt: str

32 arg: Any

33

34

35 class InputEvent(TypedDict):

36 t: Literal['i']

37 g: Game

38 arg: Any

39 done: Callable

40

41

42 class SystemEvent(TypedDict):

43 t: Literal['s']

44 evt: str

45 arg: Tuple

46

47

48 class UnityUIEventHook(EventHandler):

49 game: Game

50

51 def __init__(self, g: Game):

52 EventHandler.__init__(self, g)

53 self.game = g

54 self.live = False

55

56 def evt_user_input(self, arg: Any) -> None:

57 evt = Event()

58 g = self.game

59 core = g.core

60 core.warpgate.feed_ev({'t': 'i', 'g': g, 'arg': arg, 'done': evt.set})

61 evt.wait()

62

63 def handle(self, evt: str, arg: Any) -> Any:

64 if not self.live and evt not in ('game_begin', 'switch_character', 'reseat'):

65 return arg

66

67 g = self.game

68 core = g.core

69

70 if evt == 'user_input':

71 self.evt_user_input(arg)

72 elif evt == '__game_live':

73 self.live = True

74 return None

75 else:

76 core.warpgate.feed_ev({'t': 'g', 'g': g, 'evt': evt, 'arg': arg})

77

78 if random.random() < 0.01:

79 core.runner.idle()

80

81 return arg

82

83 def set_live(self) -> None:

84 self.live = True

85 core = self.game.core

86 core.warpgate.feed_ev({'t': 'g', 'g': self.game, 'evt': 'game_live', 'arg': None})

87

88

89 sys.argv = []

90

91

92

93 class ExecutiveWrapper(object):

94 def connect_server(self, host, port):

95 from UnityEngine import Debug

96 Debug.Log(repr((host, port)))

97

98 @gevent.spawn

99 def do():

100 Q = self.warpgate.queue_system_event

101 Q('connect', self.executive.connect_server((host, port), Q))

102

103 def start_replay(self, rep):

104 self.executive.start_replay(rep, self.warpgate.queue_system_event)

105

106 def ignite(self, g):

107 g.event_observer = UnityUIEventHook(self.warpgate, g)

108

109 @gevent.spawn

110 def start():

111 gevent.sleep(0.3)

112 svr = g.me.server

113 if svr.gamedata_piled():

114 g.start()

115 svr.wait_till_live()

116 gevent.sleep(0.1)

117 svr.wait_till_live()

118 g.event_observer.set_live()

119 else:

120 g.event_observer.set_live()

121 g.start()

122

123

124 class Warpgate(object):

125 def __init__(self, core: Core):

126 self.core = core

127 self.events: List[Any] = []

128

129 core.events.core_initialized += self.init_warpgate

130

131 def init_warpgate(self, core: Core) -> Core:

132 for name, hub in core.events.__dict__.items():

133 if name in ('core_initialized', 'server_command'):

134 continue

135

136 hub += self.forward_event(name)

137

138 return core

139

140 def forward_event(self, evt: str) -> Callable:

141 def forwarder(arg: Any) -> Any:

142 self.feed_ev({'t': 's', 'evt': evt, 'arg': arg})

143 return arg

144 return forwarder

145

146 @typing.overload

147 def feed_ev(self, ev: GameEvent) -> None: ...

148

149 @typing.overload

150 def feed_ev(self, ev: InputEvent) -> None: ...

151

152 @typing.overload

153 def feed_ev(self, ev: SystemEvent) -> None: ...

154

155 def feed_ev(self, ev: Any) -> None:

156 self.events.append(ev)

157

158 def get_events(self) -> List[Any]:

159 l = self.events

160 self.events = []

161 return l