Coverage for src/lib/executor.py: 100.00%

1#!/usr/bin/env python3 

2 

3# Standard libraries 

4from errno import ENOENT 

5from getpass import getuser 

6from os import chdir, environ 

7from time import sleep 

8from typing import List, Optional 

9 

10# Components 

11from ..engines.engine import Engine 

12from ..package.bundle import Bundle 

13from ..prints.colors import Colors 

14from ..system.platform import Platform 

15 

16# Executor 

17class Executor: 

18 

19 # Key constants (arrows) 

20 KEY_UP: str = '\033[A' 

21 KEY_DOWN: str = '\033[B' 

22 KEY_LEFT: str = '\033[D' 

23 KEY_RIGHT: str = '\033[C' 

24 

25 # Key constants (actions) 

26 KEY_BACKSPACE: str = '\b' 

27 KEY_DELETE: str = '\033[3~' 

28 KEY_END: str = '\033[F' 

29 KEY_END_2: str = '\033[4~' 

30 KEY_ENTER: str = '\r' 

31 KEY_ESC: str = '\033' 

32 KEY_HOME: str = '\033[H' 

33 KEY_HOME_2: str = '\033[1~' 

34 KEY_INSERT: str = '\033[2~' 

35 KEY_PAGEUP: str = '\033[5~' 

36 KEY_PAGEDOWN: str = '\033[6~' 

37 KEY_MACRO: str = '\033[M' 

38 KEY_PAUSE: str = '\033[P' 

39 KEY_SPACE: str = ' ' 

40 KEY_TAB: str = '\t' 

41 

42 # Key constants (Ctrl+?) 

43 KEY_CTRL_A: str = '\x01' 

44 KEY_CTRL_B: str = '\x02' 

45 KEY_CTRL_C: str = '\x03' 

46 KEY_CTRL_D: str = '\x04' 

47 KEY_CTRL_E: str = '\x05' 

48 KEY_CTRL_F: str = '\x06' 

49 KEY_CTRL_G: str = '\x07' 

50 KEY_CTRL_H: str = '\x08' 

51 KEY_CTRL_I: str = '\x09' 

52 KEY_CTRL_J: str = '\x0A' 

53 KEY_CTRL_K: str = '\x0B' 

54 KEY_CTRL_L: str = '\x0C' 

55 KEY_CTRL_M: str = '\x0D' 

56 KEY_CTRL_N: str = '\x0E' 

57 KEY_CTRL_O: str = '\x0F' 

58 KEY_CTRL_P: str = '\x10' 

59 KEY_CTRL_Q: str = '\x11' 

60 KEY_CTRL_R: str = '\x12' 

61 KEY_CTRL_S: str = '\x13' 

62 KEY_CTRL_T: str = '\x14' 

63 KEY_CTRL_U: str = '\x15' 

64 KEY_CTRL_V: str = '\x16' 

65 KEY_CTRL_W: str = '\x17' 

66 KEY_CTRL_X: str = '\x18' 

67 KEY_CTRL_Y: str = '\x19' 

68 KEY_CTRL_Z: str = '\x1A' 

69 

70 # Key constants (F?) 

71 KEY_F1: str = '\033[[A' 

72 KEY_F2: str = '\033[[B' 

73 KEY_F3: str = '\033[[C' 

74 KEY_F4: str = '\033[[D' 

75 KEY_F5: str = '\033[[E' 

76 KEY_F6: str = '\033[17~' 

77 KEY_F7: str = '\033[18~' 

78 KEY_F8: str = '\033[19~' 

79 KEY_F9: str = '\033[20~' 

80 KEY_F10: str = '\033[21~' 

81 KEY_F11: str = '\033[23~' 

82 KEY_F12: str = '\033[24~' 

83 

84 # Delays 

85 DELAY_INIT: float = 1.0 

86 DELAY_PRESS: float = 0.5 

87 DELAY_PROMPT: float = 1.0 

88 

89 # Labels 

90 LABEL_HOST: str = 'preview' 

91 LABEL_TOOL: str = 'executor' 

92 

93 # Members 

94 __delay_init: float 

95 __delay_press: float 

96 __delay_prompt: float 

97 __engine: Optional[Engine] 

98 __host: str = environ[Bundle.ENV_HOST] if Bundle.ENV_HOST in environ else LABEL_HOST 

99 __masks: List[str] = [] 

100 __tool: str = environ[Bundle.ENV_TOOL] if Bundle.ENV_TOOL in environ else LABEL_TOOL 

101 

102 # Constructor, pylint: disable=too-many-arguments,too-many-positional-arguments 

103 def __init__( 

104 self, 

105 command: str = '', 

106 delay_init: float = DELAY_INIT, 

107 delay_press: float = DELAY_PRESS, 

108 delay_prompt: float = DELAY_PROMPT, 

109 hold_prompt: bool = False, 

110 workdir: str = '', 

111 ) -> None: 

112 

113 # Prepare delays 

114 self.__delay_init = float(delay_init) 

115 self.__delay_press = float(delay_press) 

116 self.__delay_prompt = float(delay_prompt) 

117 

118 # Prepare workdir 

119 if workdir: 

120 self.__prompt(f'cd {workdir}', hold_prompt=hold_prompt) 

121 chdir(workdir) 

122 

123 # Prepare members 

124 self.__engine = None 

125 

126 # Prepare command 

127 self.__prompt(command, hold_prompt=hold_prompt) 

128 if command: 

129 

130 # Spawn command 

131 self.__engine = Engine(command) 

132 

133 # Delay executor initialization 

134 if self.__delay_init > 0.0: 

135 Executor.sleep(self.__delay_init) 

136 self.read() 

137 

138 # Configure 

139 @staticmethod 

140 def configure( 

141 host: str = LABEL_HOST, 

142 tool: str = LABEL_TOOL, 

143 masks: Optional[List[str]] = None, 

144 strips: Optional[List[str]] = None, 

145 ) -> None: 

146 

147 # Prepare host 

148 Executor.__host = host 

149 

150 # Deprecate strips 

151 if strips: # pragma: no cover 

152 raise SystemError('Parameter "strips" is deprecated, use "masks" instead') 

153 

154 # Prepare masks 

155 if masks: 

156 Executor.__masks = masks 

157 else: 

158 Executor.__masks = [] 

159 

160 # Prepare tool 

161 Executor.__tool = tool 

162 

163 # Prepare colors 

164 Colors.prepare() 

165 

166 # Control key, pylint: disable=no-self-use 

167 def __control_key(self, key: str) -> bytes: 

168 

169 # Acquire key value 

170 key = key.lower() 

171 try: 

172 value = ord(key) 

173 except TypeError: 

174 value = 0 

175 

176 # Handle alphabetical key 

177 if 97 <= value <= 122: 

178 value = value - ord('a') + 1 

179 return bytes([value]) 

180 

181 # List specific keys 

182 mappings = { 

183 '@': 0, 

184 '`': 0, 

185 '[': 27, 

186 '{': 27, 

187 '\\': 28, 

188 '|': 28, 

189 ']': 29, 

190 '}': 29, 

191 '^': 30, 

192 '~': 30, 

193 '_': 31, 

194 '?': 127 

195 } 

196 

197 # Handle specific keys 

198 if key in mappings: 

199 return bytes([mappings[key]]) 

200 

201 # Unknown fallback 

202 return bytes() 

203 

204 # Prompt 

205 def __prompt(self, command: str, hold_prompt: bool = False) -> None: 

206 

207 # Display prompt 

208 print( 

209 f'{Colors.GREEN_THIN}{getuser()}{Colors.RESET}' 

210 f'@{Colors.RED_THIN}{self.__host}{Colors.RESET}' 

211 f':{Colors.YELLOW_THIN}~/{self.__tool}{Colors.RESET}$ ', end='') 

212 Platform.flush() 

213 

214 # Delay prompt 

215 Executor.sleep(self.__delay_prompt) 

216 

217 # Display command 

218 if command: 

219 print(f'{command} ', end='') 

220 Platform.flush() 

221 Executor.sleep(self.__delay_prompt) 

222 print(' ') 

223 Platform.flush() 

224 

225 # Return prompt 

226 elif not hold_prompt: 

227 print(' ') 

228 Platform.flush() 

229 

230 # Press 

231 def press(self, key: str, control: bool = False) -> 'Executor': 

232 

233 # Execution check 

234 if not self.__engine: 

235 return self 

236 

237 # Delay press 

238 Executor.sleep(self.__delay_press) 

239 

240 # Press Ctrl+key 

241 if control: 

242 self.__engine.send(self.__control_key(key)) 

243 

244 # Press key 

245 else: 

246 self.__engine.send(key) 

247 

248 # Result 

249 return self 

250 

251 # Read 

252 def read(self) -> 'Executor': 

253 

254 # Execution check 

255 if not self.__engine: 

256 return self 

257 

258 # Read stream 

259 self.__engine.read(Executor.__masks) 

260 

261 # Result 

262 return self 

263 

264 # Wait 

265 def wait(self, delay: float) -> 'Executor': 

266 

267 # Delay execution 

268 Executor.sleep(delay) 

269 

270 # Result 

271 return self 

272 

273 # Finish 

274 def finish(self, force: bool = False) -> int: 

275 

276 # Execution check 

277 if not self.__engine: 

278 return ENOENT 

279 

280 # Read and wait execution 

281 if not force: 

282 try: 

283 while self.__engine.isalive(): 

284 self.read() 

285 except KeyboardInterrupt: 

286 pass 

287 

288 # Terminate process 

289 self.__engine.terminate(force=force) 

290 

291 # Result 

292 return self.__engine.status() 

293 

294 # Sleep 

295 @staticmethod 

296 def sleep(delay: float) -> None: 

297 

298 # Delay execution 

299 sleep(delay)