Sensor Data Ingestion System

Build a system that ingests sensor data from multiple sources and writes it to InfluxDB using appropriate data structures for each source type.

Requirements

You need to implement a data ingestion module that handles three different sensor data sources:

Temperature Sensor Data

Temperature sensors provide readings as Python dictionaries with the following structure:

{
    "sensor_id": "temp_001",
    "temperature": 23.5,
    "humidity": 65,
    "pressure": 1013,
    "timestamp": 1609459200000000000  # nanoseconds
}

The pressure field should be stored as an unsigned integer, while temperature and humidity should be stored as floats.

Location Tracker Data

Location trackers provide data as NamedTuples:

LocationData = namedtuple('LocationData', ['device_id', 'latitude', 'longitude', 'altitude', 'timestamp'])

Environmental Monitor Data

Environmental monitors use Python dataclasses:

@dataclass
class EnvironmentalData:
    monitor_id: str
    co2_level: int
    noise_level: float
    light_intensity: int
    timestamp: int

Implementation Tasks

1. Create a function ingest_temperature_data(client, bucket, data_dict) that writes temperature sensor dictionaries to InfluxDB with:

   • Measurement name: "temperature_reading"
   • Tag: sensor_id
   • Fields: temperature, humidity, pressure (with correct types)
   • Timestamp from the data

2. Create a function ingest_location_data(client, bucket, location_tuple) that writes location tracker NamedTuples to InfluxDB with:

   • Measurement name: "location_tracking"
   • Tag: device_id
   • Fields: latitude, longitude, altitude
   • Timestamp from the data

3. Create a function ingest_environmental_data(client, bucket, env_data) that writes environmental monitor dataclasses to InfluxDB with:

   • Measurement name: "environmental_monitoring"
   • Tag: monitor_id
   • Fields: co2_level, noise_level, light_intensity
   • Timestamp from the data

@generates

API

from influxdb_client import InfluxDBClient
from typing import Dict, NamedTuple
from dataclasses import dataclass

def ingest_temperature_data(client: InfluxDBClient, bucket: str, data_dict: Dict) -> None:
    """
    Ingest temperature sensor data from dictionary format.

    Args:
        client: InfluxDBClient instance
        bucket: Target bucket name
        data_dict: Dictionary containing sensor data
    """
    pass

def ingest_location_data(client: InfluxDBClient, bucket: str, location_tuple: NamedTuple) -> None:
    """
    Ingest location tracker data from NamedTuple format.

    Args:
        client: InfluxDBClient instance
        bucket: Target bucket name
        location_tuple: NamedTuple containing location data
    """
    pass

def ingest_environmental_data(client: InfluxDBClient, bucket: str, env_data: dataclass) -> None:
    """
    Ingest environmental monitor data from dataclass format.

    Args:
        client: InfluxDBClient instance
        bucket: Target bucket name
        env_data: Dataclass containing environmental data
    """
    pass

Test Cases

• Given a temperature dictionary with sensor_id="temp_001", temperature=23.5, humidity=65, pressure=1013, and timestamp, when ingested, the data is written to InfluxDB with pressure as unsigned int @test

• Given a LocationData NamedTuple with device_id="tracker_01", latitude=37.7749, longitude=-122.4194, altitude=10, and timestamp, when ingested, the data is written to InfluxDB correctly @test

• Given an EnvironmentalData dataclass with monitor_id="env_monitor_001", co2_level=400, noise_level=45.5, light_intensity=750, and timestamp, when ingested, the data is written to InfluxDB correctly @test

Dependencies { .dependencies }

influxdb-client { .dependency }

Provides InfluxDB client functionality for writing time-series data with support for custom data type serialization.

@satisfied-by