chumicro-runner

Tick-based scheduling without async. Every state change is one print() away.

Register your services once, call runner.tick() in your main loop, and each one gets a turn each tick. Every networked library in ChuMicro (wifi, sockets, mqtt, requests, http_server, websockets) registers here, so your LED keeps blinking through TLS handshakes (the encrypted-connection setup), slow HTTP responses, and stalled peers because every service gets a fair share of every tick.

Works on CircuitPython, MicroPython, and CPython. Built on chumicro-timing.

Part of the ChuMicro family: small, focused Python libraries for microcontrollers and laptops. Browse all libraries.

Install

# CircuitPython (after `circup bundle-add ChuMicro/ChuMicro-Bundle`)
circup install chumicro_runner

# MicroPython
mpremote mip install github:ChuMicro/ChuMicro-Bundle/chumicro_runner

# CPython
pip install chumicro-runner

For bundle setup, pre-compiled .mpy bundles, the experimental channel, and details on PyPI naming, see the chumicro INSTALL guide.

Quick example

from chumicro_runner import Runner

runner = Runner()
runner.add_periodic(lambda now_ms: print("blink!"), period_ms=500)

while True:
    now_ms = runner.tick()
    runner.wait(now_ms)

tick() fires every due handler. wait() then idles the CPU until the next deadline, or until a registered socket is ready for networked services. Together they give every service a fair share of every tick without burning the loop.

For a bounded run, runner.run_until(predicate, timeout_ms=...) is the one-call form of that while loop. It ticks and idles until predicate() is truthy (returns True) or the timeout elapses (returns False):

handle = runner.add_generator(echo_run(host, port, radio=wifi.adapter.radio))
runner.run_until(lambda: handle.done)

That's all you need for simple tasks. For services with conditional logic (only do something when a condition is met), implement check() and handle():

from chumicro_runner import Runner

class TemperatureSensor:
    """Alert when temperature exceeds a threshold.

    Args:
        threshold: Temperature in °C that triggers an alert.
    """

    def __init__(self, threshold: float = 30.0) -> None:
        self._threshold = threshold
        self._last_reading = 0.0

    def read_temperature(self) -> float:
        """Read from hardware: fast I2C or ADC operation."""
        # On a real board: return self._i2c_device.temperature
        return self._last_reading

    def check(self, now_ms: int) -> bool:
        """Return True when the reading exceeds the threshold.

        Args:
            now_ms: Current tick timestamp (unused here).

        Returns:
            True if the last reading exceeds the threshold.
        """
        self._last_reading = self.read_temperature()
        return self._last_reading > self._threshold

    def handle(self, now_ms: int) -> None:
        """Print an alert with the current reading.

        Args:
            now_ms: Current tick timestamp.
        """
        print(f"ALERT: {self._last_reading}°C exceeds {self._threshold}°C")

runner = Runner()
sensor = TemperatureSensor(threshold=30.0)
runner.add(sensor, period_ms=5000)  # check every 5 seconds


while True:
    now_ms = runner.tick()
    runner.wait(now_ms)

Both loops isolate a handler that raises: the fault is counted in handler_errors, the exception itself lands in last_handler_error, and the other services keep ticking, but neither says a word when it happens. A real app should pass on_handler_error(handle, exception) to Runner(...) (see the API table below) so a fire-and-forget service that dies surfaces a line instead of stalling in silence.

What's included

Core

Generator helpers (opt-in sub-module)

chumicro_runner.generators carries the scheduler-side sleep; the completion-wait vocabulary lives in chumicro_timing.waits and the socket-driven helpers in chumicro_sockets.generators (with the raw read/write wait markers in chumicro_sockets.waits). Import explicitly so plain-runner consumers stay free of the load:

A Signal bridges callback-land into a generator body: hand signal.set to a service as its callback, then value = yield from wait_for(signal):

link_up = Signal()
wifi.on_state_change(lambda old, new: link_up.set(new))
state = yield from wait_for(link_up)

Testing

Where this fits

Depends on chumicro-timing for the tick source. Every networked library in ChuMicro registers here (wifi, sockets, requests, http_server, mqtt, websockets), so the runner lives at the center of any multi-service app.

Platform support

All classes use only basic Python features. Works identically on CPython, MicroPython, and CircuitPython. Designed to be lightweight: uses minimal memory per task, suitable for boards with limited RAM.

Examples

The full user guide covers registration patterns, generator-driven sequential I/O, idling between ticks, runtime mutation, and testing your components in depth.

Contributing

Issues, bug reports, and pull requests are welcome, and so is "I ran it on this board and here's what happened", some of the most useful feedback a hardware project can get. Development happens in the ChuMicro repository, whose contributing guide covers setup and the test workflow.

Docs

📖 Stable docs · Experimental docs

Find this library

• PyPI: chumicro-runner
• Bundle: ChuMicro-Bundle (CircuitPython & MicroPython)
• Experimental bundle: ChuMicro-Bundle-Experimental
• Source: libraries/runner

License

MIT