# 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:

```arc
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:

```arc
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:

```arc
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](https://docs.synnaxlabs.com/reference/control/arc/reference/functions#inputs) 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

```arc
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:

```arc
temperature * 9.0 / 5.0 + 32.0 -> temperature_fahrenheit
```

This creates an implicit function that runs whenever `temperature` updates.

Expressions can include comparisons:

```arc
pressure > 500 -> pressure_high
```

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

You can combine channels:

```arc
(pressure_1 + pressure_2) / 2.0 -> pressure_avg
```

> Expression flows execute when any referenced channel updates. If `pressure_1` and `pressure_2` update at different times, the average recalculates on each update using the latest values from both.

## Stateful variables

Some calculations need to remember values across executions. Use `$=` to declare a [stateful variable](https://docs.synnaxlabs.com/reference/control/arc/reference/variables#stateful).

**Flow**

```arc
sequence monitor {
    stage s1 {
        count $= 0
        count + 1 -> count
        time.wait{5s} => s2
    }
    stage s2 {
        restart => s1
    }
}
```

**Func**

```arc
func counter() i64 {
    count $= 0
    count = count + 1
    return count
}

trigger -> counter{} -> count_ch
```

## Tracking previous values

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

```arc
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`](https://docs.synnaxlabs.com/reference/control/arc/reference/standard-library/math#avg) and [`math.max`](https://docs.synnaxlabs.com/reference/control/arc/reference/standard-library/math#max), which emit the updated statistic on each input:

```arc
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:

```arc
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:

```arc
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:

```arc
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.

```arc
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.

```arc
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:

```arc
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:

```arc
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:

```arc
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:

```arc
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`](https://docs.synnaxlabs.com/reference/control/arc/reference/standard-library/math#derivative) computes how fast a value is changing from the samples’ actual timestamps:

```arc
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](https://docs.synnaxlabs.com/reference/control/arc/reference/syntax#format-strings) build human-readable status messages from live channel values. Placeholders inside `{...}` accept any expression, and a `:spec` after the expression controls formatting:

```arc
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:

```arc
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:

```arc
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.