Analyze Edge Sensor Data with DuckDB and Polars

                      
                     
from typing import List, Dict
import duckdb
import polars as pl
import os

# Configuration
DB_PATH = os.getenv('DUCKDB_PATH', 'sensor_data.duckdb')

# Initialize DuckDB connection
def init_db() -> None:
    try:
        duckdb.connect(DB_PATH)
        print(f'Connected to DuckDB database at {DB_PATH}')
    except Exception as e:
        print(f'Error connecting to database: {e}')

# Function to analyze sensor data
def analyze_sensor_data() -> pl.DataFrame:
    try:
        # Read data from DuckDB into Polars DataFrame
        query = "SELECT * FROM sensor_data"
        df = duckdb.query(query).to_df()
        polars_df = pl.from_pandas(df)
        # Perform analysis (e.g., computing averages)
        result = polars_df.groupby('sensor_id').agg(pl.avg('value').alias('avg_value'))
        return result
    except Exception as e:
        print(f'Error analyzing data: {e}')

if __name__ == '__main__':
    init_db()
    analysis_result = analyze_sensor_data()
    print(analysis_result)