Coverage for pyquickhelper/loghelper/time_helper.py: 91%

44 statements  

« prev     ^ index     » next       coverage.py v7.2.7, created at 2023-06-03 02:21 +0200

1""" 

2@file 

3@brief Helpers around time. 

4""" 

5import os 

6import time 

7from .run_cmd import run_script 

8 

9 

10def repeat_execution(fct, every_second=1, stop_after_second=5, 

11 verbose=0, fLOG=None, exc=True): 

12 """ 

13 Runs a function on a regular basis. The function 

14 is not multithreaded, it returns when all execution 

15 are done. 

16 

17 @param fct function to run 

18 @param every_second every second 

19 @param stop_after_second stop after a given time or never if None 

20 @param verbose prints out every execution 

21 @param fLOG logging function 

22 @param exc if False, catch exception, 

23 else does not catch them 

24 @return results of the function if 

25 *stop_after_second* is not None 

26 """ 

27 iter = 0 

28 start = time.monotonic() 

29 end = None if stop_after_second is None else start + stop_after_second 

30 current = start 

31 res = [] 

32 while end is None or current < end: 

33 iter += 1 

34 if exc: 

35 r = fct() 

36 if verbose > 0 and fLOG is not None: 

37 fLOG( 

38 f"[repeat_execution] iter={iter} time={current} end={end}") 

39 if stop_after_second is not None: 

40 res.append(r) 

41 else: 

42 try: 

43 r = fct() 

44 if verbose > 0 and fLOG is not None: 

45 fLOG( 

46 f"[repeat_execution] iter={iter} time={current} end={end}") 

47 if stop_after_second is not None: 

48 res.append(r) 

49 except Exception as e: 

50 if verbose > 0 and fLOG is not None: 

51 fLOG("[repeat_execution] iter={} time={} end={} error={}".format( 

52 iter, current, end, str(e))) 

53 while current <= time.monotonic(): 

54 current += every_second 

55 while time.monotonic() < current: 

56 time.sleep(every_second / 2) 

57 

58 return res if res else None 

59 

60 

61def repeat_script_execution(script, every_second=1, stop_after_second=5, 

62 outfile=None, errfile=None, 

63 verbose=0, fLOG=None, exc=True): 

64 """ 

65 Runs a python script on a regular basis. The function 

66 is not multithreaded, it returns when all execution 

67 are done. 

68 

69 @param script script to run 

70 @param every_second every second 

71 @param stop_after_second stop after a given time or never if None 

72 @param outfile file which receives the standard output 

73 @param errfile file which receives the standard error 

74 @param verbose prints out every execution 

75 @param fLOG logging function 

76 @param exc if False, catch exception, 

77 else does not catch them 

78 @return all outputs if 

79 *stop_after_second* is not None 

80 """ 

81 if not os.path.exists(script): 

82 raise FileNotFoundError( # pragma: no cover 

83 f"Unable to find '{script}'.") 

84 

85 iter = [0] 

86 

87 def fct_(): 

88 out, err = run_script(script, wait=True) 

89 if out and outfile: 

90 with open(outfile, "a", encoding="utf-8") as f: 

91 f.write(f'[repeat_script_execution] iter={iter[0]}\n') 

92 f.write(out) 

93 if err and errfile: 

94 with open(errfile, "a", encoding="utf-8") as f: # pragma: no cover 

95 f.write(f'[repeat_script_execution] iter={iter[0]}\n') 

96 f.write(err) 

97 iter[0] += 1 

98 return out 

99 

100 return repeat_execution(fct_, every_second=every_second, 

101 stop_after_second=stop_after_second, 

102 verbose=verbose, fLOG=fLOG, exc=exc)