Async Data Aggregator

Build a data aggregator that processes async streams of sensor readings and computes summary statistics.

Problem

You need to implement a function that processes async streams of numeric sensor readings and returns aggregated statistics. The function should efficiently handle async iterables and compute the sum, maximum, and minimum values from the stream.

Requirements

Implement aggregate_sensor_data(readings) that:

• Accepts an async iterable of numeric sensor readings
• Returns a dictionary with three keys: "sum", "max", and "min"
• The "sum" key contains the total sum of all readings
• The "max" key contains the largest reading value
• The "min" key contains the smallest reading value
• Handles empty streams by returning {"sum": 0, "max": None, "min": None}

Test Cases

• Given async readings [10, 20, 30], returns {"sum": 60, "max": 30, "min": 10} @test
• Given async readings [5.5, 2.3, 8.1, 3.0], returns {"sum": 18.9, "max": 8.1, "min": 2.3} (with floating point tolerance) @test
• Given async readings [-10, -5, -20], returns {"sum": -35, "max": -5, "min": -20} @test
• Given empty async readings, returns {"sum": 0, "max": None, "min": None} @test
• Given async readings with single value [42], returns {"sum": 42, "max": 42, "min": 42} @test

Implementation

@generates

API

async def aggregate_sensor_data(readings):
    """
    Aggregate sensor readings from an async iterable.

    Args:
        readings: An async iterable of numeric values

    Returns:
        dict: A dictionary with keys "sum", "max", and "min"
    """
    pass

Dependencies { .dependencies }

asyncstdlib { .dependency }

Provides async-compatible versions of standard library functions for working with async iterables.

@satisfied-by