Solution Timeline

All versions (edits) of solutions to Python Script to Run Command N Times a Second appear below in the order they were created. Comments that appear under revisions were those created when that particular revision was current.

To see the revision history of a single solution (with diffs), click on the solution number (ie. "#1") in the upper right corner of a solution revision below.

← Bounty Expand all edits

Super simple, but works flawlessly.

import os, time
def do_loop(hz, dofor, cmd):
    times_to_run = hz * dofor
    for i in range(times_to_run):
        time.sleep(1/hz)
        os.system(cmd)

# do_loop(5, 1, "echo hi")

Super simple, works flawlessly.
I wrote this on Windows, but I'm sure it works the same with bash as it does for batch.

import os, time
def do_loop(hz, dofor, cmd):
    outputs = []
    times_to_run = hz * dofor
    for i in range(times_to_run):
        time.sleep(1/hz)
        code = os.system(cmd)
        outputs += [code]
    print(f"done! return codes: {outputs}")

# do_loop(5, 1, "echo hi")

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__":
    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

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)
    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 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.

Winning solution

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 = USB_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.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 argparse

    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, period)
    total = sum(return_codes.values())
    print("Return codes:")
    for rc, n in return_codes.items():
        if rc is not None:
            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.