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
« 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
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.
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)
58 return res if res else None
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.
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}'.")
85 iter = [0]
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
100 return repeat_execution(fct_, every_second=every_second,
101 stop_after_second=stop_after_second,
102 verbose=verbose, fLOG=fLOG, exc=exc)