Data Processing

Common patterns for processing sensor data in Arc

Arc programs typically process sensor data: converting units, smoothing noise, combining readings, detecting changes. This guide covers the patterns you’ll use for these tasks.

Basic Flows

A flow connects a source to a destination through a function:

temperature -> to_celsius{} -> temperature_celsius

When temperature receives new data, Arc runs to_celsius and writes the result to temperature_celsius. You don’t write loops or polling code.

The function:

func to_celsius(f f64) f64 {
    return(f - 32.0) * 5.0 / 9.0
}

The {} after to_celsius instantiates the function as a node in the dataflow graph. The input comes from the flow (the value of temperature), not from parentheses.

Inputs

Input parameters make functions reusable. They’re set when you instantiate the function, not when data flows through:

func scale{factor f64} (value f64) f64 {
    return value * factor
}

pressure -> scale{factor=2.0} -> pressure_doubled
temperature -> scale{factor=0.5} -> temperature_halved

Both flows use the same function with different scaling factors.

Use input parameters for:

• Calibration constants (gain, offset)
• Thresholds
• Conversion factors
• Anything that varies per sensor but stays constant during execution

func linear_scale{gain f64, offset f64} (raw f64) f64 {
    return raw * gain + offset
}

pressure_raw -> linear_scale{gain=100.0, offset=5.0} -> pressure_psi

Expression Flows

Simple transformations don’t need a separate function. Write the expression directly in the flow:

temperature * 9.0 / 5.0 + 32.0 -> temperature_fahrenheit

This creates an implicit function that runs whenever temperature updates.

Expressions can include comparisons:

pressure > 500 -> pressure_high

Now pressure_high outputs true when pressure exceeds 500, false otherwise.

You can combine channels:

(pressure_1 + pressure_2) / 2.0 -> pressure_avg

Stateful Variables

Some calculations need to remember values across executions. Use $= to declare a stateful variable.

The Variables reference page explains persistence in more depth.

Tracking Previous Values

A common pattern is comparing the current value to the previous one. Initialize the stateful variable to the input parameter:

func delta(value f64) f64 {
    prev $= value
    d := value - prev
    prev = value
    return d
}

pressure -> delta{} -> pressure_change

On the first execution, prev is set to value, so d is 0. On subsequent executions, prev holds the previous value and d is the actual change.

This works because $= evaluates its right-hand side only on the first execution. After that, prev retains whatever was assigned to it.

Running Calculations

Track running statistics with math.avg and math.max, which emit the updated statistic on each input:

temperature -> math.avg{} -> temperature_avg
temperature -> math.max{} -> temperature_max

Both accumulate over every sample by default. Set duration or count to control the window:

temperature -> math.avg{duration=5s} -> temperature_avg
temperature -> math.avg{count=20} -> temperature_avg

Exponential Moving Average

An exponential moving average (EMA) smooths noisy data while responding to changes. It weights recent values more heavily:

func ema{alpha f64} (value f64) f64 {
    avg $= value
    avg = alpha * value + (1.0 - alpha) * avg
    return avg
}

pressure -> ema{alpha=0.2} -> pressure_smooth

The alpha parameter controls responsiveness:

• Lower alpha (0.1): Heavy smoothing, slow response
• Higher alpha (0.5): Light smoothing, fast response

Multi-Input Functions

When a function needs multiple sensor values, pass the channels as input parameters:

func pressure_diff{inlet chan f64, outlet chan f64} () f64 {
    p1 := inlet
    p2 := outlet
    return p1 - p2
}

time.interval{period=5s} -> pressure_diff{
    inlet = inlet_pressure,
    outlet = outlet_pressure
} -> delta_p

Inside the function, inlet and outlet read the latest values from those channels. The function has no flow input, so time.interval triggers it on a schedule.

Flow-Driven vs Interval-Driven

Two ways to trigger execution:

Flow-driven: The source channel triggers execution when it updates.

temperature -> to_celsius{} -> temperature_celsius

Good when you want to process every sample from a sensor.

Interval-driven: A timer triggers execution at a fixed rate.

time.interval{period=100ms} -> read_sensors{} -> output

Good when:

• The function reads multiple channels (no single source to trigger it)
• You want a consistent sample rate regardless of when sensors update
• You’re implementing a control loop that should run at a fixed frequency

Reading Channels in Functions

Inside a function body, reading a channel returns its latest value immediately:

func check_limits{sensor chan f64, limit f64} () bool {
    value := sensor // reads latest value, doesn't block
    return value > limit
}

This is different from flow-driven execution. In a flow like sensor -> process{} -> output, the function receives data pushed through the flow. Inside a function body, reading a channel pulls the current value.

If nothing has been written to the channel yet, reading returns zero.

Control Flow

Use if/else for conditional logic:

func clamp{min f64, max f64} (value f64) f64 {
    if value < min {
        return min
    }
    if value > max {
        return max
    }
    return value
}

For more complex decisions, chain conditions:

func categorize(value f64) i64 {
    if value < 100 {
        return 0
    } else if value < 500 {
        return 1
    } else if value < 900 {
        return 2
    }
    return 3
}

Sensor Voting

When you have redundant sensors, use voting to reject outliers. A median of three readings ignores a single faulty sensor:

func median3{sensor_a chan f64, sensor_b chan f64, sensor_c chan f64} () f64 {
    a := sensor_a
    b := sensor_b
    c := sensor_c

    // Find the middle value
    if (a >= b and a <= c) or (a >= c and a <= b) {
        return a
    }
    if (b >= a and b <= c) or (b >= c and b <= a) {
        return b
    }
    return c
}

time.interval{period=50ms} -> median3{
    sensor_a = pressure_1,
    sensor_b = pressure_2,
    sensor_c = pressure_3
} -> pressure_voted

If one sensor fails and reads 0 while the others read 500, the median is 500. The faulty reading is ignored.

Rate of Change

math.derivative computes how fast a value is changing from the samples’ actual timestamps:

pressure -> math.derivative{} -> pressure_rate

The output is in units per second (e.g., psi/second if pressure is in psi).

Formatting Status Strings

Format strings build human-readable status messages from live channel values. Placeholders inside {...} accept any expression, and a :spec after the expression controls formatting:

func fmt_pressure(value f64) str {
    return f"Pressure: {value:.2f} psi"
}

pressure -> fmt_pressure{} -> pressure_status

Each time pressure updates, pressure_status receives a string like "Pressure: 523.70 psi". Format specs are validated against the placeholder type at compile time, so a .2f against an integer is caught before the program runs. A bool placeholder renders as true or false.

Format strings are useful for writing diagnostic outputs:

func alert(value f64) str {
    if value > 800.0 {
        return f"HIGH PRESSURE: {value:.1f} psi"
    }
    return f"pressure nominal: {value:.1f} psi"
}

pressure -> alert{} -> log

Putting It Together

Here’s a complete pipeline that reads a pressure sensor, converts units, smooths the signal, computes rate of change, and outputs everything:

import math

// Unit conversion: voltage to psi
func volts_to_psi{v_min f64, v_max f64, p_max f64} (v f64) f64 {
    ratio := (v - v_min) / (v_max - v_min)
    return ratio * p_max
}
// Exponential moving average
func ema{alpha f64} (value f64) f64 {
    avg $= value
    avg = alpha * value + (1.0 - alpha) * avg
    return avg
}

// Pipeline: raw voltage -> psi -> smoothed -> rate
pressure_raw -> volts_to_psi{v_min=0.5, v_max=4.5, p_max=1000.0} -> pressure_psi
pressure_psi -> ema{alpha=0.3} -> pressure_smooth
pressure_smooth -> math.derivative{} -> pressure_rate

// Also output a high-pressure flag
pressure_smooth > 800 -> pressure_high

Each stage feeds the next. When pressure_raw updates:

1. volts_to_psi converts to psi
2. ema smooths the result
3. math.derivative computes how fast the smoothed value is changing
4. The comparison outputs true if pressure exceeds 800

The raw value flows through the entire pipeline automatically.