chumicro-runner 0.1.0

Tick-based task runner with shared timestamps for Chumicro applications.

chumicro-runner

A tick-based service pattern for Chumicro libraries.

Components implement a check(now_ms) -> bool check that gates when a handler fires. A Runner captures time once per tick, checks each service, and batch-fires all due handlers — replacing ad-hoc polling loops with a single standard contract.

Installation

# CPython (pip)
pip install chumicro-runner

# CircuitPython (circup)
circup bundle-add ChuMicro/chumicro-bundle
circup install chumicro-runner

# MicroPython (mip)
mpremote mip install github:ChuMicro/chumicro-bundle/chumicro_runner

For experimental (pre-release) versions from the develop branch:

pip install chumicro-runner-experimental
circup install chumicro-runner-experimental
mpremote mip install github:ChuMicro/chumicro-bundle/chumicro_runner_experimental

Quick example

from chumicro_runner import Runner


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

    def __init__(self, threshold=30.0):
        self._threshold = threshold
        self._last_reading = 0.0

    def read_temperature(self):
        """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):
        self._last_reading = self.read_temperature()
        return self._last_reading > self._threshold

    def handle(self, now_ms):
        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:
    runner.tick()

For simple periodic tasks, no service class is needed:

from chumicro_runner import Runner

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

while True:
    runner.tick()

What's included

Core

Symbol	Description
Runner(ticks=None)	Tick-based service loop with shared timestamps
Runner.add(task, handler=None, period_ms=None, start_after_ms=None, run_count=None)	Register a task; returns a TaskHandle
Runner.add_periodic(handler, period_ms, start_after_ms=None, run_count=None)	Register a periodic handler; returns a TaskHandle
Runner.tick()	Capture time, check services, batch-fire handlers; returns now_ms
TaskHandle	Opaque handle for runtime mutation of a registered service
TaskHandle.set_period(period_ms)	Add, change, or remove the period (None to remove)
TaskHandle.remove()	Remove this service from the runner
TaskHandle.period_ms	Read-only: the service period, or None
TaskHandle.run_count	Read-only: remaining run count, or None if unlimited
TaskHandle.active	Read-only: whether the service is still registered

Testing

Symbol	Description
CallRecorder()	Callable that records handler invocations for test assertions
CallRecorder.calls	Direct access to the list of recorded now_ms values

Registration patterns

Object-based (with .check() and .handle())

Pass an object that has check(now_ms) -> bool and handle(now_ms) methods. The runner calls .check(); if it returns True, .handle() is queued:

class MotionDetector:
    def __init__(self):
        # On a real board: self._pin = digitalio.DigitalInOut(board.D5)
        pass

    def detect_motion(self):
        """Read PIR sensor pin — fast digital read."""
        # On a real board: return self._pin.value
        return False

    def check(self, now_ms):
        return self.detect_motion()

    def handle(self, now_ms):
        print("Motion!")

runner.add(MotionDetector())

Callable-based (check function + handler)

Pass a callable check function and a handler. Both can be lambdas, bound methods, or regular functions:

runner.add(
    lambda now_ms: sensor.ready(),
    handler=lambda now_ms: process(sensor.read()),
)

Handler-only (no check, fires every tick)

Pass just a handler with no service check:

runner.add(handler=lambda now_ms: poll_buttons(now_ms))

Periodic (fires every N milliseconds)

No service check needed — the handler fires on schedule:

handle = runner.add_periodic(toggle_led, period_ms=500)
handle.set_period(1000)  # change rate at runtime

Runtime mutation

add() and add_periodic() return a TaskHandle for runtime changes:

handle = runner.add(sensor, period_ms=5000)

# Speed up.
handle.set_period(1000)

# Remove the period — service runs every tick.
handle.set_period(None)

# Remove entirely.
handle.remove()

Testing your components

The chumicro_runner.testing module provides CallRecorder for verifying that handlers fire at the right times:

from chumicro_runner.testing import CallRecorder
from chumicro_timing.testing import FakeTicks

fake = FakeTicks()
recorder = CallRecorder()
runner = Runner(ticks=fake)
runner.add_periodic(recorder, period_ms=100)

runner.tick()
assert len(recorder) == 0  # not due yet

fake.advance(100)
runner.tick()
assert recorder.calls == [100]

Platform support

All classes use only basic Python features. Works identically on CPython, MicroPython, and CircuitPython.

Memory notes

• _TaskEntry and TaskHandle use __slots__ to minimise per-instance memory.
• Handlers are collected into a pre-allocated list and batch-fired, avoiding per-tick allocation.

Docs

• User guide — the pattern, getting started, writing components
• API reference — full API documentation
• Testing helpers — using CallRecorder in your tests