CyteBode's solution to "Python Script to Run Command N Times a Second"

Here's my solution. It doesn't block because it uses Popen<\/code> instead of system<\/code> and it can use a busy loop in case the granularity is too high (that should only be a problem with Windows). The cmd<\/code> can be passed as a string or a function that takes the iteration number and returns a string.<\/p>\n

#!/usr/bin/env python3<\/span>\n\nimport<\/span> subprocess<\/span>\nimport<\/span> time<\/span>\n\n\ndef<\/span> hifreq_cmd<\/span>(<\/span>cmd<\/span>,<\/span> freq<\/span>,<\/span> period<\/span>,<\/span> use_busy_loop<\/span> =<\/span> False<\/span>):<\/span>\n    if<\/span> isinstance<\/span>(<\/span>cmd<\/span>,<\/span> str<\/span>):<\/span>\n        cmd_fn<\/span> =<\/span> lambda<\/span> i<\/span>:<\/span> cmd<\/span>\n    else<\/span>:<\/span>\n        assert<\/span> callable<\/span>(<\/span>cmd<\/span>)<\/span>\n        cmd_fn<\/span> =<\/span> cmd<\/span>\n\n    end_time<\/span> =<\/span> time<\/span>.<\/span>time<\/span>()<\/span> +<\/span> period<\/span>\n    deadline<\/span> =<\/span> time<\/span>.<\/span>time<\/span>()<\/span>\n\n    i<\/span> =<\/span> 0<\/span>\n    while<\/span> time<\/span>.<\/span>time<\/span>()<\/span> <<\/span> end_time<\/span>:<\/span>\n        deadline<\/span> +=<\/span> 1.0<\/span> /<\/span> freq<\/span>\n        if<\/span> use_busy_loop<\/span>:<\/span>\n            while<\/span> (<\/span>time<\/span>.<\/span>time<\/span>()<\/span> <<\/span> deadline<\/span>):<\/span>\n                pass<\/span>\n        else<\/span>:<\/span>\n            if<\/span> time<\/span>.<\/span>time<\/span>()<\/span> <<\/span> deadline<\/span>:<\/span>\n                time<\/span>.<\/span>sleep<\/span>(<\/span>deadline<\/span> -<\/span> time<\/span>.<\/span>time<\/span>())<\/span>\n        subprocess<\/span>.<\/span>Popen<\/span>(<\/span>cmd_fn<\/span>(<\/span>i<\/span>)<\/span>.<\/span>split<\/span>(<\/span>&<\/span>quot<\/span>;<\/span> &<\/span>quot<\/span>;))<\/span>\n        i<\/span> +=<\/span> 1<\/span>\n\n\nif<\/span> __name__<\/span> ==<\/span> &<\/span>quot<\/span>;<\/span>__main__<\/span>&<\/span>quot<\/span>;:<\/span>\n    hifreq_cmd<\/span>(<\/span>&<\/span>quot<\/span>;<\/span>echo<\/span> 3<\/span> times<\/span> once<\/span> a<\/span> second<\/span>&<\/span>quot<\/span>;,<\/span> 1<\/span>,<\/span> 3.0<\/span>)<\/span>\n\n    start<\/span> =<\/span> time<\/span>.<\/span>time<\/span>()<\/span>\n    hifreq_cmd<\/span>(<\/span>lambda<\/span> i<\/span>:<\/span> &<\/span>quot<\/span>;<\/span>echo<\/span> %<\/span>d<\/span>&<\/span>quot<\/span>;<\/span> %<\/span> i<\/span>,<\/span> 1000<\/span>,<\/span> 1.0<\/span>,<\/span> True<\/span>)<\/span>\n    end<\/span> =<\/span> time<\/span>.<\/span>time<\/span>()<\/span>\n\n    print<\/span>(<\/span>end<\/span> -<\/span> start<\/span>)<\/span> # Should be ~1.0<\/span>\n<\/pre><\/div>
Here's my solution. It doesn't block because it uses `Popen` instead of `system` and it can use a busy loop in case the granularity is too high (that should only be a problem with Windows). The `cmd` can be passed as a string or a function that takes the iteration number and returns a string. #!/usr/bin/env python3 import subprocess import time def hifreq_cmd(cmd, freq, period, use_busy_loop = False): if isinstance(cmd, str): cmd_fn = lambda i: cmd else: assert callable(cmd) cmd_fn = cmd end_time = time.time() + period deadline = time.time() i = 0 while time.time() < end_time: deadline += 1.0 / freq if use_busy_loop: while (time.time() < deadline): pass else: if time.time() < deadline: time.sleep(deadline - time.time()) subprocess.Popen(cmd_fn(i).split(" ")) i += 1 if __name__ == "__main__": ": """ Usage examples of hifreq_cmd hifreq_cmd("echo 3 times once a second", 1, 3.0) start = time.time() hifreq_cmd(lambda i: "echo %d" % i, 1000, 1.0, True) end = time.time() print(end - start) # Should be ~1.0 """ import sys freq = float(sys.argv[1]) period = float(sys.argv[2]) cmd = " ".join(sys.argv[3:]) hifreq_cmd(cmd, freq, period) **Edit**: Added script argument handling.
Here's my solution. It doesn't block because it uses `Popen` instead of `system` and it can use a busy loop in case the granularity is too high (that should only be a problem with Windows). The `cmd` can be passed as a string or a function that takes the iteration number and returns a string. #!/usr/bin/env python3 import subprocess import time def hifreq_cmd(cmd, freq, period, use_busy_loop = False): if isinstance(cmd, str): cmd_fn = lambda i: cmd else: assert callable(cmd) cmd_fn = cmd end_time = time.time() + period deadline = time.time() children = [] i = 0 while time.time() < end_time: deadline += 1.0 / freq if use_busy_loop: while (time.time() < deadline): pass else: if time.time() < deadline: time.sleep(deadline - time.time()) child = subprocess.Popen(cmd_fn(i).split(" ")) children.append(child) i += 1 return_codes = {} for child in children: child.poll() rc = child.returncode while rc is None: time.sleep(0.001) child.poll() child.returncode return_codes[rc] = return_codes.setdefault(rc, 0) + 1 return return_codes if __name__ == "__main__": """ Usage examples of hifreq_cmd hifreq_cmd("echo 3 times once a second", 1, 3.0) start = time.time() hifreq_cmd(lambda i: "echo %d" % i, 1000, 1.0, True) end = time.time() print(end - start) # Should be ~1.0 """ import sys freq = float(sys.argv[1]) period = float(sys.argv[2]) cmd = " ".join(sys.argv[3:]) hifreq_cmd(lambda i: "echo %d" % i, 1000, 1.0, True) """ import sys freq = float(sys.argv[1]) period = float(sys.argv[2]) cmd = " ".join(sys.argv[3:]) return_codes = hifreq_cmd(cmd, freq, period) **) total = sum(return_codes.values()) print("Return codes:") for rc, n in return_codes.items(): print(" %d: %d/%d" % (rc, n, total)) **Edit 1**: Added script argument handling. **Edit 2**: Added summary of return codes.
Here's my solution. It doesn't block because it uses `Popen` instead of `system` and it can use a busy loop in case the granularity is too high (that should only be a problem with Windows). The `cmd` can be passed as a string or a function that takes the iteration number and returns a string. #!/usr/bin/env python3 import subprocess import time TIMEOUT_TIME = 1.0 MAX_TIMEOUTS = 10 def hifreq_cmd(cmd, freq, period, use_busy_loop = False): if isinstance(cmd, str): cmd_fn = lambda i: cmd else: assert callable(cmd) cmd_fn = cmd end_time = time.time() + period deadline = time.time() children = [] i = 0 while time.time() < end_time: deadline += 1.0 / freq if use_busy_loop: while (time.time() < deadline): pass else: if time.time() < deadline: time.sleep(deadline - time.time()) child = subprocess.Popen(cmd_fn(i).split(" ")) children.append(child) i += 1 return_codes = {} timing_out = {} while children: child = children.pop(0) if timing_out.get(child, 0) > MAX_TIMEOUTS: return_codes[None] = return_codes.setdefault(None, 0) + 1 continue try: rc = child.wait(TIMEOUT_TIME) return_codes[rc] = return_codes.setdefault(rc, 0) + 1 except subprocess.TimeoutExpired: children.append(child) timing_out[child] = timing_out.get(child, 0) + 1 return return_codes if __name__ == "__main__": """ Usage examples of hifreq_cmd hifreq_cmd("echo 3 times once a second", 1, 3.0) hifreq_cmd(lambda i: "echo %d" % i, 1000, 1.0, True) """ import sys freq = float(sys.argv[1]) period = float(sys.argv[2]) cmd = " ".join(sys.argv[3:]) return_codes = hifreq_cmd(cmd, freq, period) total = sum(return_codes.values()) print("Return codes:") for child in children: child.poll() rc = child.returncode while rc is None: time.sleep(0.001) child.poll() child.returncode return_codes[rc] = return_codes.setdefault(rc, 0) + 1 return return_codes if __name__ == "__main__": """ Usage examples of hifreq_cmd hifreq_cmd("echo 3 times once a second", 1, 3.0) hifreq_cmd(lambda i: "echo %d" % i, 1000, 1.0, True) """ import sys freq = float(sys.argv[1]) period = float(sys.argv[2]) cmd = " ".join(sys.argv[3:]) return_codes = hifreq_cmd(cmd, freq, period) total = sum(return_codes.values()) print("Return codes:") for rc, n in return_codes.items(): if rc is None: print("Timed out: %d/%d" % (n, total)) else: print(" %d: %d/%d" % (rc, n, total)) **)) **Edit 1**: Added script argument handling. **Edit 2**: Added summary of return codes. **Edit 3**: Changed `poll` to `wait` and added timeout handling.
Here's my solution. It doesn't block because it uses `Popen` instead of `system` and it can use a busy loop in case the granularity is too high (that should only be a problem with Windows). The `cmd` can be passed as a string or a function that takes the iteration number and returns a string. #!/usr/bin/env python3 import subprocess import time TIMEOUT_TIME = 1.0 MAX_TIMEOUTS = 10 USB_BUSY_LOOP = False def hifreq_cmd(cmd, freq, period, use_busy_loop = FalseUSB_BUSY_LOOP): if isinstance(cmd, str): cmd_fn = lambda i: cmd else: assert callable(cmd) cmd_fn = cmd end_time = time.time() + period deadline = time.time() children = [] i = 0 while time.time() < end_time: deadline += 1.0 / freq if use_busy_loop: while (time.time() < deadline): pass else: if time.time() < deadline: time.sleep(deadline - time.time()) child = subprocess.Popen(cmd_fn(i).split(" ")) children.append(child) i += 1 return_codes = {} timing_out = {} while children: child = children.pop(0) if timing_out.get(child, 0) > MAX_TIMEOUTS: return_codes[None] = return_codes.setdefault(None, 0) + 1 continue try: rc = child.wait(TIMEOUT_TIME) return_codes[rc] = return_codes.setdefault(rc, 0) + 1 except subprocess.TimeoutExpired: children.append(child) timing_out[child] = timing_out.get(None, 0) + 1 continue try: rc = child.wait(TIMEOUT_TIME) return_codes[rc] = return_codes.get(rc, 0) + 1 except subprocess.TimeoutExpired: children.append(child) timing_out[child] = timing_out.get(child, 0) + 1 return return_codes if __name__ == "__main__": """ Usage examples of hifreq_cmd hifreq_cmd("echo 3 times once a second", 1, 3.0) hifreq_cmd(lambda i: "echo %d" % i, 1000, 1.0, True) """ import sysargparse parser = argparse.ArgumentParser( description="Runs a command repeatedly at a given rate, for a given" + " amount of time.") parser.add_argument("frequency", type=float, help="Rate at which to run the command, in times/sec") parser.add_argument("period", type=float, help="Amount of time to run the command, in seconds") parser.add_argument("command", type=str, nargs="+", help="The command to run, between quotes or not") args = parser.parse_args() freq = args.frequency period = args.period cmd = " ".join(args.command) return_codes = hifreq_cmd(cmd, freq = float(sys.argv[1]) period = float(sys.argv[2]) cmd = " ".join(sys.argv[3:]) return_codes = hifreq_cmd(cmd, freq, period) total = sum(return_codes.values()) print("Return codes:") for rc, n in return_codes.items(): if rc is not None: print("Timed out: %d/%d" % (n, total)) else: print(" %d: %d/%d" % (rc, n, total)) if None in return_codes: print(" Timed out: %d/%d" % (return_codes[None], total)) **Edit 1**: Added script argument handling. **Edit 2**: Added summary of return codes. **Edit 3**: Changed `poll` to `wait` and added timeout handling.. **Edit 4**: Cleaned up the script, added argument handling with `argparse`.

User: CyteBode

Question: Python Script to Run Command N Times a Second

Back to question