Example: Setting Up the Data Ingestion Agent

from datetime import datetime
import psycopg2
from psycopg2.extras import execute_values
import json

class TimescaleDBAgent:
    def __init__(self, db_config):
        self.conn = psycopg2.connect(**db_config)
        self.cursor = self.conn.cursor()
    
    def normalize_sensor_data(self, raw_data):
        # Example normalization logic
        return {
            'timestamp': datetime.utcnow(),
            'sensor_id': raw_data.get('sensor_id'),
            'value': float(raw_data.get('value', 0)),
            'unit': raw_data.get('unit', 'unknown'),
            'metadata': json.dumps(raw_data.get('metadata', {}))
        }
    
    def store_data(self, data):
        normalized = self.normalize_sensor_data(data)
        query = """
            INSERT INTO sensor_readings 
            (time, sensor_id, value, unit, metadata)
            VALUES (%s, %s, %s, %s, %s)
        """
        self.cursor.execute(query, (
            normalized['timestamp'],
            normalized['sensor_id'],
            normalized['value'],
            normalized['unit'],
            normalized['metadata']
        ))
        self.conn.commit()
    
    def close(self):
        self.cursor.close()
        self.conn.close()

Benefits