Coverage for src/engines/base.py: 100.00%

1#!/usr/bin/env python3 

2 

3# Standard libraries 

4from typing import Any, List, Optional 

5 

6# Components 

7from ..system.platform import Platform 

8 

9# Base class 

10class Base: 

11 

12 # Members 

13 _child: Optional[Any] = None 

14 

15 # Constructor 

16 def __init__(self, command: str) -> None: 

17 

18 # Virtual method 

19 raise NotImplementedError() # pragma: no cover 

20 

21 # Read 

22 def _read(self) -> bytes | List[bytes | str] | str: 

23 

24 # Virtual method 

25 raise NotImplementedError() # pragma: no cover 

26 

27 # Is alive 

28 def isalive(self) -> bool: 

29 

30 # Result 

31 assert self._child is not None 

32 return bool(self._child.isalive()) 

33 

34 # Read 

35 def read(self, masks: Optional[List[str]] = None) -> None: 

36 

37 # Read stream 

38 while True: 

39 

40 # Acquire output 

41 output = self._read() 

42 

43 # Interrupted stream 

44 if not output: 

45 break 

46 

47 # Print stream 

48 if isinstance(output, (bytes, str)): 

49 if isinstance(output, bytes): 

50 output = output.decode('utf-8', errors='ignore') 

51 output = output.replace('\x1b[6n', '') 

52 if masks: 

53 for mask in masks: 

54 output = output.replace(mask, '') 

55 print(output, end='') 

56 Platform.flush() 

57 

58 # Lines lines 

59 elif isinstance(output, list): # pragma: no cover 

60 for line in output: 

61 if isinstance(line, bytes): 

62 line = line.decode('utf-8', errors='ignore') 

63 line = line.replace('\x1b[6n', '') 

64 if masks: 

65 for mask in masks: 

66 line = line.replace(mask, '') 

67 print(line) 

68 Platform.flush() 

69 

70 # Send 

71 def send(self, key: bytes | str) -> None: 

72 

73 # Send key 

74 assert self._child is not None 

75 self._child.send(key) 

76 

77 # Status 

78 def status(self) -> int: 

79 

80 # Fallback status 

81 assert self._child is not None 

82 if self._child.exitstatus is None: 

83 return 1 

84 

85 # Process status 

86 return int(self._child.exitstatus) 

87 

88 # Terminate 

89 def terminate(self, force: bool = False) -> None: 

90 

91 # Terminate process 

92 assert self._child is not None 

93 self._child.terminate(force=force)