Sensor Data Ingestion System

A system that processes time-series sensor data from CSV files and writes them to InfluxDB using efficient batch operations.

Problem

You are building a data pipeline that needs to ingest large CSV files containing sensor measurements into InfluxDB. The CSV files contain temperature, humidity, and pressure readings from multiple sensors across different locations. You need to implement a function that reads the CSV data, transforms it appropriately, and writes it efficiently to InfluxDB.

Requirements

Data Transformation

The input CSV file has the following columns:

• timestamp - ISO 8601 formatted timestamp
• sensor_id - Unique sensor identifier
• location - Geographic location of the sensor
• temperature - Temperature in Celsius
• humidity - Relative humidity percentage
• pressure - Atmospheric pressure in hPa

The function should:

• Read the CSV file into a suitable data structure
• Convert the timestamp column to proper datetime objects
• Write all measurements to InfluxDB in a single batch operation
• Use sensor_id and location as tags
• Use temperature, humidity, and pressure as fields
• Store all data points with the measurement name "environmental_data"

Implementation

• It reads a CSV file and returns the number of data points written @test
• It correctly handles timestamp conversion from ISO 8601 format @test
• It writes data with proper tag and field mapping @test
• It efficiently handles large datasets (1000+ rows) @test

@generates

API

from influxdb_client import InfluxDBClient
import pandas as pd

def ingest_sensor_data(
    csv_file_path: str,
    influx_client: InfluxDBClient,
    bucket: str,
    org: str
) -> int:
    """
    Ingest sensor data from CSV file into InfluxDB.

    Args:
        csv_file_path: Path to the CSV file containing sensor data
        influx_client: Initialized InfluxDB client instance
        bucket: Target bucket name for the data
        org: Organization name

    Returns:
        Number of data points successfully written

    Raises:
        FileNotFoundError: If the CSV file does not exist
        ValueError: If the CSV format is invalid
    """
    pass

Dependencies { .dependencies }

influxdb-client { .dependency }

Provides InfluxDB 2.x client functionality for time-series data storage.

pandas { .dependency }

Provides data analysis and CSV reading support.