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

2from __future__ import annotations 

3 

4# -- stdlib -- 

5from typing import Any, Callable, List, TYPE_CHECKING, Tuple 

6import random 

7import sys 

8import typing 

9 

10# -- third party -- 

11from gevent.event import Event 

12from mypy_extensions import TypedDict 

13from typing_extensions import Literal 

14import gevent 

15import gevent.hub 

16 

17# -- own -- 

18from client.base import Game 

19from game.base import EventHandler 

20 

21 

22# -- typing -- 

23if TYPE_CHECKING: 

24 from client.core import Core # noqa: F401 

25 

26 

27# -- code -- 

28class GameEvent(TypedDict): 

29 t: Literal['g'] 

30 g: Game 

31 evt: str 

32 arg: Any 

33 

34 

35class InputEvent(TypedDict): 

36 t: Literal['i'] 

37 g: Game 

38 arg: Any 

39 done: Callable 

40 

41 

42class SystemEvent(TypedDict): 

43 t: Literal['s'] 

44 evt: str 

45 arg: Tuple 

46 

47 

48class 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 

89sys.argv = [] 

90 

91 

92 

93class 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 

124class 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