IoT Sensor Data Collector

Build a data collection system for IoT sensors that efficiently writes time-series data to InfluxDB with automatic batching.

Problem Description

You need to build a Python application that collects measurements from multiple IoT temperature sensors and writes them to InfluxDB. The sensors produce measurements at irregular intervals, and the system must efficiently batch these writes to optimize network usage and database performance.

The system should automatically batch sensor data and flush to InfluxDB based on:

1. A maximum batch size (number of data points)
2. A time interval (to ensure data isn't held too long)

Requirements

Sensor Data Collection

Your application should collect sensor readings with the following structure:

• Sensor ID (e.g., "sensor-001", "sensor-002")
• Location (e.g., "warehouse-a", "office-b")
• Temperature value (float)
• Timestamp

Batching Configuration

Implement a data writer that:

• Collects data points into batches
• Automatically flushes when the batch reaches 500 data points
• Automatically flushes every 5 seconds (5000 milliseconds) if data is waiting
• Properly cleans up resources when done

Data Writing

Write sensor data to InfluxDB where:

• Measurement name is "temperature"
• Tags include sensor_id and location
• Field is "value" containing the temperature
• Each reading has a timestamp

Test Cases

• When 500 sensor readings are written, a batch flush occurs before the time interval @test
• When fewer than 500 readings are written but 5 seconds elapse, a batch flush occurs @test
• All data is successfully flushed when the writer is properly closed @test
• Multiple sensor readings with different locations are correctly tagged @test

Implementation

@generates

API

def create_sensor_writer(url: str, token: str, org: str, bucket: str):
    """
    Creates and returns a configured writer for sensor data with batching enabled.

    Args:
        url: InfluxDB server URL
        token: Authentication token
        org: Organization name
        bucket: Bucket name for storing data

    Returns:
        A tuple of (client, write_api) that should be used with context managers
    """
    pass

def write_sensor_reading(write_api, sensor_id: str, location: str, temperature: float, timestamp):
    """
    Writes a single sensor reading to the batch.

    Args:
        write_api: The write API instance from create_sensor_writer
        sensor_id: Unique identifier for the sensor
        location: Physical location of the sensor
        temperature: Temperature reading in Celsius
        timestamp: Python datetime object for the reading
    """
    pass

Dependencies { .dependencies }

influxdb-client { .dependency }

Provides Python client for InfluxDB 2.x with batching capabilities.

@satisfied-by