Writing Custom Programs

Writing Custom Programs

This guide explains how to write your own aggregate programs using the computational fields framework.

Basic Structure

An aggregate program is a Python function that takes a Context and returns a value:

from computational_fields.blocks.gradient import gradient

def my_program(ctx):
    is_source = ctx.sense("is_source")
    distance = gradient(ctx, is_source)
    return distance

Setting Up a Network

from computational_fields.simulation.network import Network
from computational_fields.simulation.engine import SimulationEngine

# Create an 8x8 grid
net = Network.grid(
    rows=8, cols=8, spacing=1.0, comm_range=1.5,
    sensors_fn=lambda did, row, col: {"is_source": did == 0}
)

# Run the simulation
engine = SimulationEngine(net, my_program)
results = engine.run(num_rounds=20)

Using Sensors

Sensors are the input mechanism. Set them when creating the network:

def sensors(device_id, row, col):
    return {
        "is_source": device_id == 0,
        "temperature": 20.0 + row * 0.5,
        "is_exit": row == 0 or col == 0,
    }

net = Network.grid(8, 8, sensors_fn=sensors)

Read sensors in your program:

def program(ctx):
    temp = ctx.sense("temperature")
    return temp

Composing Blocks

The power of aggregate computing comes from composition:

from computational_fields.blocks.gradient import gradient, broadcast
from computational_fields.blocks.collection import collect
from computational_fields.blocks.sparse import sparse

def monitoring_program(ctx):
    # Step 1: Elect leaders
    is_leader = sparse(ctx, grain=3.0)

    # Step 2: Compute distance to nearest leader
    dist_to_leader = gradient(ctx, is_leader)

    # Step 3: Collect temperature readings at leaders
    temp = ctx.sense("temperature")
    avg_temp = collect(ctx, dist_to_leader, lambda a, b: a + b, temp, 0.0)

    # Step 4: Broadcast alerts from leaders
    alert = broadcast(ctx, is_leader, avg_temp > 30.0)

    return {"distance": dist_to_leader, "alert": alert}

Returning Multiple Fields

Return a dictionary to produce multiple named fields:

def multi_field_program(ctx):
    return {
        "gradient": gradient(ctx, ctx.sense("is_source")),
        "leader": sparse(ctx, 3.0),
    }

Access individual fields from the engine:

engine.step()
gradient_field = engine.get_field("gradient")
leader_field = engine.get_field("leader")

Tips

1. Always use ctx — pass it as the first argument to every primitive and block call
2. Avoid side effects — programs should be pure functions of sensor inputs and neighbour state
3. Think in fields — every expression produces a value on every device simultaneously
4. Compose blocks — build complex behaviour by chaining G, C, S, T blocks
5. Test convergence — run enough rounds for the field to stabilise (typically O(diameter))