Coverage for selenible/modules/screencast.py: 12%

1import io 

2import threading 

3import time 

4from logging import getLogger 

5 

6from PIL import Image 

7 

8 

9class screencast(threading.Thread): 

10 def __init__(self, drvobj, crop, interval=1.0, thumbnail=None): 

11 super().__init__() 

12 self.drvobj = drvobj 

13 if crop is None: 

14 self.crop = crop 

15 else: 

16 self.crop = tuple(crop) 

17 self.thumbnail = thumbnail 

18 self.log = getLogger(self.name) 

19 self.stop = False 

20 self.interval = interval 

21 self.frames = [] 

22 

23 def run(self): 

24 self.start_ts = time.time() 

25 while not self.stop: 

26 t1 = time.time() 

27 with self.drvobj.lock: 

28 p = self.drvobj.saveshot() 

29 self.log.info("shot: %d bytes, %f sec", len(p), time.time() - t1) 

30 if len(p) != 0: 

31 img = Image.open(io.BytesIO(p)) 

32 if self.crop is not None: 

33 self.log.info("crop %s, %f sec", self.crop, time.time() - t1) 

34 img = img.crop(self.crop) 

35 if self.thumbnail is not None: 

36 img.thumbnail(self.thumbnail, Image.ANTIALIAS) 

37 self.frames.append(img) 

38 td = time.time() - t1 

39 if self.interval is not None and self.interval > td: 

40 self.log.info("sleep %f - %f", self.interval, td) 

41 time.sleep(self.interval - td) 

42 self.finished_ts = time.time() 

43 

44 def savefile(self, output_fn, optimize=False, loop=0, speed=1.0): 

45 if len(self.frames) == 0: 

46 raise Exception("no image found") 

47 d = 1000 * (self.finished_ts - self.start_ts) / len(self.frames) / speed 

48 self.log.info( 

49 "%d frames, %d sec. duration=%f(ms)", 

50 len(self.frames), 

51 self.finished_ts - self.start_ts, 

52 d, 

53 ) 

54 self.frames[0].save( 

55 output_fn, 

56 save_all=True, 

57 duration=d, 

58 optimize=optimize, 

59 loop=loop, 

60 append_images=self.frames[1:], 

61 ) 

62 

63 

64scr_th = None 

65 

66 

67def Base_screencast(self, param): 

68 """ 

69 - name: start screencast 

70 screencast: 

71 interval: 0.5 

72 thumbnail: [100, 100] 

73 - name: sleep 

74 sleep: 3 

75 - name: save screencast 

76 screencast: output.gif 

77 """ 

78 global scr_th 

79 if isinstance(param, dict) and "output" not in param: 

80 if scr_th is not None: 

81 raise Exception("screencast already working") 

82 scr_th = screencast( 

83 self, param.get("crop"), param.get("interval"), param.get("thumbnail") 

84 ) 

85 scr_th.start() 

86 return "started" 

87 if scr_th is None: 

88 raise Exception("screencast is not working") 

89 self.log.debug("stop cast") 

90 scr_th.stop = True 

91 scr_th.join(2) 

92 if scr_th.is_alive(): 

93 self.log.warning("cannot stop screencast thread.") 

94 else: 

95 self.log.debug("done join. saving") 

96 if isinstance(param, str): 

97 self.log.info("save to %s", param) 

98 scr_th.savefile(param) 

99 elif isinstance(param, dict): 

100 output = param.get("output") 

101 assert output is not None 

102 self.log.info("save to %s", output) 

103 scr_th.savefile( 

104 output, 

105 optimize=param.get("optimize", False), 

106 loop=param.get("loop", 0), 

107 speed=param.get("speed", 1.0), 

108 ) 

109 scr_th = None 

110 return "finished"