IoT Data Storage with Time-Series Databases — InfluxDB & TimescaleDB Guide

-- Create a measurement for temperature sensors
-- Tags: device_id, location (indexed)
-- Fields: temperature, humidity (values)

from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
import random, time, json

INFLUX_URL = "http://localhost:8086"
INFLUX_TOKEN = "your-token"
INFLUX_ORG = "dodatech"
INFLUX_BUCKET = "iot_sensors"

client = InfluxDBClient(url=INFLUX_URL, token=INFLUX_TOKEN, org=INFLUX_ORG)
write_api = client.write_api(write_type=SYNCHRONOUS)

for i in range(10):
    point = (
        Point("environment")
        .tag("device_id", f"sensor-{i:03d}")
        .tag("location", random.choice(["warehouse", "office", "outdoor"]))
        .field("temperature", round(random.uniform(15.0, 35.0), 1))
        .field("humidity", round(random.uniform(30.0, 80.0), 1))
        .field("battery", round(random.uniform(2.5, 3.7), 2))
    )
    write_api.write(bucket=INFLUX_BUCKET, record=point)
    time.sleep(0.1)

print("10 data points written to InfluxDB")

# Query recent data
query_api = client.query_api()
result = query_api.query(
    f'from(bucket:"{INFLUX_BUCKET}")
     |> range(start: -1h)
     |> filter(fn: (r) => r._measurement == "environment")
     |> filter(fn: (r) => r._field == "temperature")'
)
for table in result:
    for record in table.records:
        print(f"{record.values.get('device_id')}: {record.get_value()}°C at {record.get_time()}")

-- Install extension
CREATE EXTENSION IF NOT EXISTS timescaledb;

-- Create sensor table
CREATE TABLE sensor_data (
    time        TIMESTAMPTZ NOT NULL,
    device_id   TEXT NOT NULL,
    location    TEXT NOT NULL,
    temperature DOUBLE PRECISION,
    humidity    DOUBLE PRECISION,
    battery     DOUBLE PRECISION
);

-- Convert to hypertable partitioned by time
SELECT create_hypertable('sensor_data', 'time', chunk_time_interval => INTERVAL '1 day');

import psycopg2
import random, time
from datetime import datetime, timezone

conn = psycopg2.connect(
    host="localhost",
    port=5432,
    dbname="iot",
    user="dodatech",
    password="your-password"
)
cur = conn.cursor()

# Insert batch of sensor readings
readings = []
for i in range(100):
    readings.append((
        datetime.now(timezone.utc),
        f"sensor-{i:03d}",
        random.choice(["warehouse", "office", "outdoor"]),
        round(random.uniform(15.0, 35.0), 1),
        round(random.uniform(30.0, 80.0), 1),
        round(random.uniform(2.5, 3.7), 2)
    ))

cur.executemany(
    """INSERT INTO sensor_data (time, device_id, location, temperature, humidity, battery)
       VALUES (%s, %s, %s, %s, %s, %s)""",
    readings
)
conn.commit()

# Query hourly average temperature by location
cur.execute("""
    SELECT
        location,
        time_bucket('1 hour', time) AS hour,
        AVG(temperature) AS avg_temp,
        COUNT(*) AS readings
    FROM sensor_data
    WHERE time > NOW() - INTERVAL '1 day'
    GROUP BY location, hour
    ORDER BY hour, location
""")
for row in cur.fetchall():
    print(f"{row[0]} | {row[1]} | Avg: {row[2]:.1f}°C | Count: {row[3]}")

cur.close()
conn.close()

-- Create a downsampled table
CREATE TABLE sensor_data_hourly (
    time        TIMESTAMPTZ NOT NULL,
    device_id   TEXT NOT NULL,
    location    TEXT NOT NULL,
    avg_temp    DOUBLE PRECISION,
    min_temp    DOUBLE PRECISION,
    max_temp    DOUBLE PRECISION,
    reading_count INTEGER
);

SELECT create_hypertable('sensor_data_hourly', 'time');

-- Periodic aggregation job
INSERT INTO sensor_data_hourly
SELECT
    time_bucket('1 hour', time) AS time,
    device_id,
    location,
    AVG(temperature) AS avg_temp,
    MIN(temperature) AS min_temp,
    MAX(temperature) AS max_temp,
    COUNT(*) AS reading_count
FROM sensor_data
WHERE time < NOW() - INTERVAL '30 days'
GROUP BY time_bucket('1 hour', time), device_id, location;

-- Drop raw data older than 90 days
DELETE FROM sensor_data WHERE time < NOW() - INTERVAL '90 days';

SELECT
    time_bucket('5 minutes', time) AS time,
    AVG(temperature) AS avg_temp,
    MAX(temperature) AS max_temp,
    MIN(temperature) AS min_temp
FROM sensor_data
WHERE $__timeFilter(time)
GROUP BY time_bucket('5 minutes', time)
ORDER BY time

import random, json, sqlite3
from datetime import datetime, timedelta

class IoTTimeSeries:
    def __init__(self, db_path: str = ":memory:"):
        self.conn = sqlite3.connect(db_path)
        self.cur = self.conn.cursor()
        self.cur.execute("""
            CREATE TABLE IF NOT EXISTS sensor_data (
                timestamp TEXT,
                device_id TEXT,
                location TEXT,
                temperature REAL,
                humidity REAL
            )
        """)
        self.cur.execute("CREATE INDEX IF NOT EXISTS idx_time ON sensor_data(timestamp)")

    def generate_data(self, num_devices: int = 5, hours: int = 24):
        start_time = datetime.utcnow() - timedelta(hours=hours)
        for hour_offset in range(hours):
            for device in range(num_devices):
                ts = start_time + timedelta(hours=hour_offset)
                self.cur.execute(
                    """INSERT INTO sensor_data VALUES (?, ?, ?, ?, ?)""",
                    (
                        ts.isoformat(),
                        f"sensor-{device:03d}",
                        random.choice(["warehouse", "office", "outdoor"]),
                        round(random.uniform(15.0, 35.0), 1),
                        round(random.uniform(30.0, 80.0), 1)
                    )
                )
        self.conn.commit()
        print(f"Generated {num_devices * hours} data points")

    def query_hourly_average(self, device_id: str) -> list:
        self.cur.execute("""
            SELECT SUBSTR(timestamp, 0, 14) || ':00:00' AS hour,
                   AVG(temperature) AS avg_temp,
                   AVG(humidity) AS avg_hum
            FROM sensor_data
            WHERE device_id = ?
            GROUP BY hour
            ORDER BY hour
        """, (device_id,))
        return self.cur.fetchall()

ts = IoTTimeSeries()
ts.generate_data(num_devices=3, hours=48)
results = ts.query_hourly_average("sensor-000")
for row in results[:5]:
    print(f"{row[0]} | Temp: {row[1]:.1f}°C | Hum: {row[2]:.1f}%")