Visualizzazione post con etichetta RaspberryPI. Mostra tutti i post
Visualizzazione post con etichetta RaspberryPI. Mostra tutti i post

mercoledì 30 settembre 2026

Architettura IoT ed Edge AI: Creare una Stazione Meteo Predittiva con SwitchBot e Raspberry Pi 5

L'Edge Computing sta rivoluzionando il modo in cui gestiamo e analizziamo i dati in locale, riducendo la dipendenza dal cloud e garantendo latenze minime. Se disponi di un Raspberry Pi 5 e di sensori ambientali di qualità, come i celebri SwitchBot Indoor/Outdoor Thermo-Hygrometer, hai già l'hardware necessario per costruire un sistema di Machine Learning autonomo.

In questo articolo esploreremo come implementare una pipeline completa di Data Ingestion e Predictive Modeling. Utilizzeremo due sensori SwitchBot, posizionati su due balconi differenti (con esposizioni climatiche diverse, ad esempio Nord e Sud), per addestrare un'Intelligenza Artificiale in grado non solo di storicizzare temperatura e umidità, ma di imparare l'inerzia termica del microclima locale per prevedere accuratamente il meteo futuro.

L'architettura del progetto si divide in quattro layer fondamentali:

  • Data Ingestion Layer: Uno script per l'estrazione asincrona dei dati tramite le API REST di SwitchBot.
  • Short-Term Inference Engine (1 Ora): Un modello predittivo basato su Random Forest Regressor per le variazioni rapide, dotato di un loop di feedback per l'auto-correzione del bias.
  • Long-Term Inference Engine (24 Ore): Un modello di aggregazione giornaliera per il calcolo della stagionalità e la stima degli estremi termici (Max/Min) del giorno successivo.
  • Interface & Alerting Layer: Un demone Telegram per l'invio di bollettini schedulati e l'esecuzione asincrona (On-Demand) delle previsioni tramite comandi interattivi.

1. Environment Setup: Preparazione del Raspberry Pi 5

Il Raspberry Pi 5 offre capacità di elaborazione multi-core ideali per l'addestramento locale di modelli di machine learning su dataset tabulari. Il primo passo consiste nel configurare l'ambiente di runtime.

Aggiorniamo i repository del sistema operativo basato su Debian Bookworm:

sudo apt update
sudo apt upgrade -y

Installiamo i pacchetti core per la manipolazione vettoriale dei dati e il modeling statistico. Utilizzeremo pandas per il time-series processing e scikit-learn per gli algoritmi predittivi:

sudo apt install python3-pandas python3-sklearn -y

Per l'integrazione con Telegram, installeremo la libreria pyTelegramBotAPI. Considerando le stringenti policy PEP 668 introdotte in Debian 12 per la gestione dei pacchetti Python globali, bypasseremo il blocco con il flag --break-system-packages (per ambienti di produzione complessi si raccomanda l'uso di un venv):

pip3 install pyTelegramBotAPI --break-system-packages

Creiamo l'alberatura delle directory per separare log, database e script:

mkdir -p ~/weather_project/data
mkdir -p ~/weather_project/log
cd ~/weather_project

2. Creazione del Bot Telegram e Sicurezza dell'Endpoint

Per interagire con i nostri modelli e ricevere l'output dell'inferenza, implementeremo un bot Telegram.

Generazione del Token:

  1. Interagisci con @BotFather su Telegram tramite il comando /newbot.
  2. Segui il wizard per definire nome e username (es. EdgeWeather_bot).
  3. Conserva il Token HTTP API (es. 123456789:ABCDefGhiJkL-MnoPqRstUvwXyz) generato al termine del processo.

Mappatura del Chat ID:

Per ragioni di sicurezza, il nostro listener rifiuterà i comandi provenienti da utenti non autorizzati. È necessario recuperare il proprio identificativo numerico:

  1. Invia un messaggio al tuo nuovo bot.
  2. Interroga l'API di Telegram via browser: https://api.telegram.org/bot<IL_TUO_TOKEN>/getUpdates
  3. Estrai il valore numerico associato alla chiave "chat":{"id": (es. 1193645047).

3. Data Ingestion Layer (weather_extractor.py)

Questo script agisce come un job ETL (Extract, Transform, Load). Si connette all'API v1.1 di SwitchBot, generando dinamicamente gli header di autenticazione HMAC-SHA256 richiesti (sfruttando un nonce e un timestamp epoch).

Come accennato, l'infrastruttura prevede due sensori fisici mappati su due balconi (denominati courtyard e balcony). I dati vengono estratti e inseriti in un database SQLite, scelto per la sua natura ACID-compliant, serverless e ideale per applicazioni edge.

import json
import time
import hashlib
import hmac
import base64
import uuid
import requests
import sqlite3
import urllib3
import os 
from pathlib import Path
from datetime import datetime, timezone

urllib3.disable_warnings()

# Database initialization
db_path = str(Path(__file__).parent.absolute()) + "/data"
if not os.path.exists(db_path):
   os.makedirs(db_path)

conn = sqlite3.connect(db_path + "/weather.db")
cursor = conn.cursor()

# Schema definition
cursor.execute('''
    CREATE TABLE IF NOT EXISTS weather (
        id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
        ts TEXT NOT NULL,
        temperature NUMERIC,
        humidity INTEGER,
        device_id TEXT,
        name TEXT
    )
''')

# SwitchBot API Credentials (REPLACE WITH YOUR OWN)
token = 'YOUR_SWITCHBOT_TOKEN_HERE'
secret = 'YOUR_SWITCHBOT_SECRET_HERE'

# HMAC Authentication Generation for SwitchBot API v1.1
nonce = uuid.uuid4()
t = int(round(time.time() * 1000))
string_to_sign = bytes(f'{token}{t}{nonce}', 'utf-8')
secret_bytes = bytes(secret, 'utf-8')
sign = base64.b64encode(hmac.new(secret_bytes, msg=string_to_sign, digestmod=hashlib.sha256).digest())

api_header = {
    'Authorization': token,
    'Content-Type': 'application/json',
    'charset': 'utf8',
    't': str(t),
    'sign': str(sign, 'utf-8'),
    'nonce': str(nonce)
}

# Topology: Two sensors mapping the two balconies
sensors = {
    'courtyard': 'YOUR_DEVICE_ID_1',
    'balcony': 'YOUR_DEVICE_ID_2'
}

for name, device_id in sensors.items():
    try:
        response = requests.get(f"https://api.switch-bot.com/v1.1/devices/{device_id}/status", headers=api_header, verify=False)
        out_data = response.json()["body"]
        
        current_ts = datetime.now(timezone.utc).isoformat()
        
        # Load operation
        cursor.execute("""
            INSERT INTO weather (ts, temperature, humidity, device_id, name) 
            VALUES (?, ?, ?, ?, ?)
        """, (current_ts, out_data["temperature"], out_data["humidity"], device_id, name))
        print(f"[{current_ts}] Data ingested for {name}: {out_data['temperature']}C, {out_data['humidity']}%")
        
    except Exception as e:
        print(f"API Extraction Error for {name}: {e}")

conn.commit()
conn.close()

4. Short-Term Inference Engine (hourly_predictor.py)

L'analisi predittiva a breve termine sfrutta un algoritmo Random Forest Regressor, un metodo di ensemble learning eccezionale per catturare correlazioni non lineari senza richiedere pesanti normalizzazioni dei dati.

Le peculiarità di questo script sono due:

  1. Feature Engineering Dinamico: Il dataframe viene processato aggiungendo variabili di lag (ritardo). Il modello non guarda solo la temperatura attuale, ma anche quella di 10, 20 e 30 minuti fa. Questo permette all'algoritmo di apprendere matematicamente l'inerzia termica del singolo balcone.
  2. Feedback Loop (Auto-correzione): L'intelligenza artificiale verifica continuamente i propri errori passati. Se nelle ore precedenti ha sistematicamente sovrastimato la temperatura reale di 0.5°C, calcolerà un bias correttivo e lo applicherà dinamicamente alla nuova predizione.
import sqlite3
import pandas as pd
import os
import logging
from logging.handlers import TimedRotatingFileHandler
from pathlib import Path
from sklearn.ensemble import RandomForestRegressor
from datetime import datetime, timedelta, timezone
import warnings
import requests
import time

warnings.filterwarnings('ignore')

# --- CONFIGURATION ---
DB_PATH = str(Path(__file__).parent.absolute()) + "/data/weather.db"
LOG_DIR = str(Path(__file__).parent.absolute()) + "/log"
STEPS_AHEAD = 6                  # Prediction horizon: 6 steps of 10 min = 1 hour
SENSORS = ['courtyard', 'balcony'] 
HISTORY_DAYS = 30              

# --- TELEGRAM CONFIGURATION ---
TELEGRAM_TOKEN = "YOUR_TELEGRAM_BOT_TOKEN"
TELEGRAM_CHAT_ID = "YOUR_CHAT_ID"

# --- LOGGING SETUP ---
if not os.path.exists(LOG_DIR):
    os.makedirs(LOG_DIR)

logger = logging.getLogger("HourlyPredictor")
logger.setLevel(logging.INFO)
log_file = os.path.join(LOG_DIR, "hourly_predictor.log")
handler = TimedRotatingFileHandler(log_file, when="midnight", interval=1, backupCount=2)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
handler.setFormatter(formatter)
logger.addHandler(handler)

def send_telegram_message(message):
    """Network-resilient Telegram dispatcher."""
    url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage"
    payload = {"chat_id": TELEGRAM_CHAT_ID, "text": message, "parse_mode": "HTML"}
    for attempt in range(3):
        try:
            res = requests.post(url, json=payload, timeout=10)
            res.raise_for_status() 
            break 
        except Exception as e:
            if attempt < 2:
                time.sleep(3) 
            else:
                logger.error(f"Webhook dispatch failed: {e}")

def setup_database(conn):
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS hourly_predictions (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT,
            calc_time TEXT,
            target_time TEXT,
            pred_temp NUMERIC, real_temp NUMERIC, temp_err NUMERIC,
            pred_hum NUMERIC, real_hum NUMERIC, hum_err NUMERIC
        )
    ''')
    conn.commit()

def calculate_recent_bias(conn, df_real, sensor_name):
    """Dynamic feedback loop for bias correction based on historical error."""
    cursor = conn.cursor()
    query = f"SELECT * FROM hourly_predictions WHERE name = '{sensor_name}' AND real_temp IS NULL"
    df_pending = pd.read_sql_query(query, conn)
    
    bias = {'temperature': 0.0, 'humidity': 0.0}
    counts = {'temperature': 0, 'humidity': 0}
    
    for _, row in df_pending.iterrows():
        target_ts = pd.to_datetime(row['target_time'])
        if target_ts <= df_real.index[-1] + pd.Timedelta(minutes=5):
            try:
                idx = df_real.index.get_indexer([target_ts], method='nearest', tolerance=pd.Timedelta('10min'))[0]
                if idx != -1:
                    real_t = float(df_real.iloc[idx]['temperature'])
                    real_h = float(df_real.iloc[idx]['humidity'])
                    
                    err_t = real_t - row['pred_temp']
                    err_h = real_h - row['pred_hum']
                    
                    cursor.execute("""
                        UPDATE hourly_predictions 
                        SET real_temp = ?, temp_err = ?, real_hum = ?, hum_err = ? 
                        WHERE id = ?
                    """, (real_t, err_t, real_h, err_h, row['id']))
                    
                    bias['temperature'] += err_t
                    counts['temperature'] += 1
                    bias['humidity'] += err_h
                    counts['humidity'] += 1
            except Exception:
                pass
    conn.commit()
    
    return {
        'temperature': bias['temperature'] / counts['temperature'] if counts['temperature'] > 0 else 0.0,
        'humidity': bias['humidity'] / counts['humidity'] if counts['humidity'] > 0 else 0.0
    }

def process_sensor(conn, sensor_name):
    logger.info(f"Initiating inference pipeline for: {sensor_name.upper()}")
    
    date_limit = (datetime.now(timezone.utc) - timedelta(days=HISTORY_DAYS)).isoformat()
    query = f"SELECT ts, temperature, humidity FROM weather WHERE name = '{sensor_name}' AND ts > '{date_limit}' ORDER BY ts ASC"
    df = pd.read_sql_query(query, conn)
    
    if len(df) < 50:
        logger.warning(f"Insufficient telemetry data for {sensor_name}.")
        return

    # Data transformation and interpolation
    df['ts'] = pd.to_datetime(df['ts'], utc=True)
    df.set_index('ts', inplace=True)
    df = df.resample('10min').mean().interpolate()

    recent_bias = calculate_recent_bias(conn, df, sensor_name)

    # Time-series feature engineering (Lags)
    for m in ['temperature', 'humidity']:
        df[f'{m}_lag_1'] = df[m].shift(1)
        df[f'{m}_lag_2'] = df[m].shift(2)
        df[f'{m}_lag_3'] = df[m].shift(3)
        df[f'target_{m}'] = df[m].shift(-STEPS_AHEAD)

    df_train = df.dropna()
    if len(df_train) < 10:
        return

    features = [
        'temperature', 'temperature_lag_1', 'temperature_lag_2', 'temperature_lag_3',
        'humidity', 'humidity_lag_1', 'humidity_lag_2', 'humidity_lag_3'
    ]
    targets = ['target_temperature', 'target_humidity']

    X = df_train[features]
    y = df_train[targets]

    # Model training (Ensemble Learning)
    ai_agent = RandomForestRegressor(n_estimators=100, random_state=42, n_jobs=-1)
    ai_agent.fit(X, y)

    last_reading = df.iloc[-1]
    X_today = last_reading[features].values.reshape(1, -1)
    raw_predictions = ai_agent.predict(X_today)[0]
    
    # Heuristic adjustment leveraging moving bias
    pred_temp = raw_predictions[0] + (recent_bias['temperature'] * 0.5)
    pred_hum = raw_predictions[1] + (recent_bias['humidity'] * 0.5)
    
    calc_time = datetime.now(timezone.utc)
    target_time = calc_time + timedelta(minutes=STEPS_AHEAD * 10)
    
    logger.info(f"[{sensor_name.upper()}] Temp: {last_reading['temperature']:.1f}C -> Pred: {pred_temp:.1f}C")
    
    str_last = df.index[-1].strftime('%d/%m %H:%M')
    str_target = target_time.astimezone(timezone.utc).astimezone(tz=None).strftime('%d/%m %H:%M')

    msg = (
        f"📊 <b>1-Hour Forecast: {sensor_name.upper()}</b>\n"
        f"🕒 <i>Last reading: {str_last}</i>\n"
        f"🎯 <i>Target time: {str_target}</i>\n\n"
        f"🌡 <b>Temperature:</b>\n"
        f"   Current: {last_reading['temperature']:.1f}°C\n"
        f"   Expected: {pred_temp:.1f}°C\n"
        f"   Bias: {recent_bias['temperature']:+.2f}°C\n\n"
        f"💧 <b>Humidity:</b>\n"
        f"   Current: {last_reading['humidity']:.1f}%\n"
        f"   Expected: {pred_hum:.1f}%\n"
        f"   Bias: {recent_bias['humidity']:+.2f}%"
    )
    send_telegram_message(msg)
        
    cursor = conn.cursor()
    cursor.execute("""
        INSERT INTO hourly_predictions (name, calc_time, target_time, pred_temp, pred_hum) 
        VALUES (?, ?, ?, ?, ?)
    """, (sensor_name, calc_time.isoformat(), target_time.isoformat(), round(pred_temp, 2), round(pred_hum, 2)))
    conn.commit()

if __name__ == "__main__":
    conn = sqlite3.connect(DB_PATH)
    setup_database(conn)
    for s in SENSORS:
        process_sensor(conn, s)
    conn.close()

5. Long-Term Inference Engine (daily_predictor.py)

A differenza del modello orario, la stima a 24 ore richiede la comprensione dei cicli macroscopici. Lo script utilizza la libreria pandas per eseguire un downsampling dei dati: raggruppa le oltre 140 misurazioni giornaliere calcolando le funzioni di aggregazione max, min e mean. Integrando come feature il "giorno dell'anno" (dayofyear), l'algoritmo impara a mappare la stagionalità climatica dei due balconi per stimare gli estremi termici del giorno successivo.

import sqlite3
import pandas as pd
import os
import logging
from logging.handlers import TimedRotatingFileHandler
from pathlib import Path
from sklearn.ensemble import RandomForestRegressor
from datetime import datetime, timedelta, timezone
import warnings
import requests
import time

warnings.filterwarnings('ignore')

# --- CONFIGURATION ---
DB_PATH = str(Path(__file__).parent.absolute()) + "/data/weather.db"
LOG_DIR = str(Path(__file__).parent.absolute()) + "/log"
SENSORS = ['courtyard', 'balcony'] 

# --- TELEGRAM CONFIGURATION ---
TELEGRAM_TOKEN = "YOUR_TELEGRAM_BOT_TOKEN"
TELEGRAM_CHAT_ID = "YOUR_CHAT_ID"

# --- LOGGING SETUP ---
if not os.path.exists(LOG_DIR):
    os.makedirs(LOG_DIR)

logger = logging.getLogger("DailyPredictor")
logger.setLevel(logging.INFO)
log_file = os.path.join(LOG_DIR, "daily_predictor.log")
handler = TimedRotatingFileHandler(log_file, when="midnight", interval=1, backupCount=7)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
handler.setFormatter(formatter)
logger.addHandler(handler)

def send_telegram_message(message):
    url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage"
    payload = {"chat_id": TELEGRAM_CHAT_ID, "text": message, "parse_mode": "HTML"}
    for attempt in range(3):
        try:
            res = requests.post(url, json=payload, timeout=10)
            res.raise_for_status() 
            break 
        except Exception as e:
            if attempt < 2:
                time.sleep(3) 
            else:
                logger.error(f"Telegram dispatch failed: {e}")

def process_daily_sensor(conn, sensor_name):
    logger.info(f"Initiating daily aggregation for: {sensor_name.upper()}")
    
    # Load complete dataset for seasonality extraction
    query = f"SELECT ts, temperature, humidity FROM weather WHERE name = '{sensor_name}' ORDER BY ts ASC"
    df = pd.read_sql_query(query, conn)
    
    if len(df) < 144 * 7: 
        logger.warning(f"Insufficient history for {sensor_name}.")
        return

    # TZ Conversion and normalization for precise 1D resampling
    df['ts'] = pd.to_datetime(df['ts'], utc=True).dt.tz_convert('Europe/Rome')
    df['ts'] = df['ts'].dt.normalize() 
    df.set_index('ts', inplace=True)
    
    # Downsampling logic
    df_daily = df.resample('1D').agg({
        'temperature': ['max', 'min'],
        'humidity': ['mean']
    })
    df_daily.columns = ['t_max', 't_min', 'h_mean']
    df_daily = df_daily.dropna()

    if len(df_daily) < 7:
        return

    # Macro-Feature Engineering
    df_daily['day_of_year'] = df_daily.index.dayofyear
    df_daily['t_max_yesterday'] = df_daily['t_max'].shift(1)
    df_daily['t_min_yesterday'] = df_daily['t_min'].shift(1)
    
    df_daily['target_t_max'] = df_daily['t_max'].shift(-1)
    df_daily['target_t_min'] = df_daily['t_min'].shift(-1)

    df_train = df_daily.dropna()
    if len(df_train) < 5:
        return

    features = ['day_of_year', 't_max', 't_min', 't_max_yesterday', 't_min_yesterday']
    targets = ['target_t_max', 'target_t_min']

    X = df_train[features]
    y = df_train[targets]

    ai_agent = RandomForestRegressor(n_estimators=100, random_state=42, n_jobs=-1)
    ai_agent.fit(X, y)

    today_data = df_daily.iloc[-1]
    if pd.isna(today_data[features]).any():
        return

    X_today = today_data[features].values.reshape(1, -1)
    tomorrow_pred = ai_agent.predict(X_today)[0]
    
    t_max_tomorrow = tomorrow_pred[0]
    t_min_tomorrow = tomorrow_pred[1]
    
    target_date_str = (datetime.now() + timedelta(days=1)).strftime('%d/%m/%Y')

    msg = (
        f"📅 <b>Tomorrow's Forecast: {sensor_name.upper()} ({target_date_str})</b>\n\n"
        f"🔴 <b>Max Temp:</b> {t_max_tomorrow:.1f}°C\n"
        f"🔵 <b>Min Temp:</b> {t_min_tomorrow:.1f}°C\n\n"
        f"<i>(Today's recorded extremes: Max {today_data['t_max']:.1f}°C, Min {today_data['t_min']:.1f}°C)</i>"
    )
    send_telegram_message(msg)

if __name__ == "__main__":
    conn = sqlite3.connect(DB_PATH)
    for s in SENSORS:
        process_daily_sensor(conn, s)
    conn.close()

6. Interface Layer: Il Listener Interattivo (telegram_bot.py)

Avere dei task in esecuzione silente è utile, ma l'aggiunta di un'interfaccia interattiva eleva drasticamente la flessibilità del sistema. Creeremo un listener Telegram basato su Long-Polling che resterà in ascolto 24/7. Per evitare di bloccare il loop asincrono del bot durante l'elaborazione dei modelli (che può richiedere alcuni secondi per caricare il database in memoria e ricalcolare i gradienti dell'albero decisionale), utilizzeremo subprocess.Popen per eseguire gli script Python in processi background paralleli.

import telebot
import subprocess
import os
from pathlib import Path

# --- CONFIGURATION ---
TOKEN = "YOUR_TELEGRAM_BOT_TOKEN"
MY_CHAT_ID = int("YOUR_CHAT_ID_NUMERIC_HERE")  # Authorization Whitelist
BASE_DIR = str(Path(__file__).parent.absolute())

bot = telebot.TeleBot(TOKEN)

@bot.message_handler(commands=['hourly'])
def force_hourly_prediction(message):
    if message.chat.id == MY_CHAT_ID:
        bot.reply_to(message, "⏳ Processing short-term forecast in background...")
        script_path = os.path.join(BASE_DIR, "hourly_predictor.py")
        # Asynchronous execution to prevent I/O blocking
        subprocess.Popen(["/usr/bin/python3", script_path])
    else:
        bot.reply_to(message, "⛔ HTTP 403: Unauthorized access attempt.")

@bot.message_handler(commands=['daily'])
def force_daily_prediction(message):
    if message.chat.id == MY_CHAT_ID:
        bot.reply_to(message, "⏳ Processing daily forecast in background...")
        script_path = os.path.join(BASE_DIR, "daily_predictor.py")
        subprocess.Popen(["/usr/bin/python3", script_path])
    else:
        bot.reply_to(message, "⛔ HTTP 403: Unauthorized access attempt.")

@bot.message_handler(commands=['start', 'help'])
def welcome_message(message):
    if message.chat.id == MY_CHAT_ID:
        text = (
            "🤖 <b>Edge AI Weather Bot Operational</b>\n\n"
            "Supported Endpoints:\n"
            "🔹 /hourly - Trigger 1-hour Inference\n"
            "🔹 /daily - Trigger Daily Extremes Inference"
        )
        bot.reply_to(message, text, parse_mode="HTML")

if __name__ == "__main__":
    print("Telegram Bot Daemon Initialized...")
    bot.infinity_polling(timeout=10, long_polling_timeout=5)

7. Orchestrazione: Cron e Systemd

Per garantire che l'architettura sia solida e autonoma in caso di riavvii del Raspberry Pi, divideremo i compiti utilizzando gli scheduler nativi di Linux.

Scheduled Tasks (Crontab)

I processi batch (l'estrazione dei dati e le predizioni schedulate) verranno gestiti tramite Cron. Apri la tabella di scheduling eseguendo crontab -e e aggiungi le policy di esecuzione:

# Data Ingestion: Trigger API extraction every 10 minutes
*/10 * * * * /usr/bin/python3 /home/pi/weather_project/weather_extractor.py  

# Short-term Inference: Trigger every 10 minutes (offset by 2 minutes for DB sync)
2-59/10 * * * * /usr/bin/python3 /home/pi/weather_project/hourly_predictor.py

# Long-term Inference: Trigger daily batch aggregation at 23:50
50 23 * * * /usr/bin/python3 /home/pi/weather_project/daily_predictor.py

Daemon Management (Systemd)

Il listener di Telegram, essendo un processo continuo, verrà avvolto in un servizio systemd per gestirne in modo robusto il ciclo di vita (startup e riavvii automatici post-crash). Crea il file di servizio:

sudo nano /etc/systemd/system/weather-bot.service

Inserisci il descrittore del demone:

[Unit]
Description=Telegram Edge AI Weather Bot Daemon
After=network.target

[Service]
Type=simple
User=pi
ExecStart=/usr/bin/python3 /home/pi/weather_project/telegram_bot.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

Inizializza l'ambiente e avvia il servizio:

sudo systemctl daemon-reload
sudo systemctl enable weather-bot.service
sudo systemctl start weather-bot.service

8. Il Risultato Finale

Il deploy è completato. Ora disponi di una micro-infrastruttura edge avanzata in grado di monitorare contemporaneamente i due sensori sui tuoi balconi. Tramite l'integrazione Telegram, hai una dashboard di comando direttamente nello smartphone: digita /hourly in chat per innescare un task predittivo on-demand.

Con il passare delle settimane e la crescita del volume dei dati storicizzati nel database SQLite, osserverai la Random Forest convergere sempre di più, calcolando con estrema precisione i gradienti termici e i delta di umidità specifici del tuo ambiente.

Ecco un esempio del payload inviato dal bot, dove si evince la precisione della stima e il tuning dinamico del Bias Engine:



sabato 14 febbraio 2015

Raspberry Pi configurare due wifi dongle



Il primo dongle è un Tenda il secondo un piccolo Netgear, entrambi collegati alla stessa rete.


sudo apt-get purge ifplugd


file: /etc/network/interfaces
auto lo
iface lo inet loopback
iface eth0 inet dhcp


allow-hotplug wlan0
iface wlan0 inet manual
wpa-roam /etc/wpa_supplicant/wpa_supplicant.conf
iface default inet dhcp

allow-hotplug wlan1
iface wlan1 inet manual
wpa-roam /etc/wpa_supplicant/wpa_supplicant.conf
ifcae default inet dhcp





file: /etc/wpa_supplicant/wpa_supplicant.conf
ctrl_interface=DIR=/var/run/wpa_supplicant GROUP=netdev
update_config=1

network={
ssid="Digicom_5B36"
psk="xxxxxxxxxxxxxx"
}







sabato 8 marzo 2014

Raspberry Rover + Accelerometro


Proseguono gli sviluppi sul progetto RaspiRoverDuinio. 

Integrato accelerometro MPU6050



Accelerometro collegato al GPIO del Raspberry PI MPU6050


Il prossimo passo è l'integrazione del sensore di prossimità HC-SR04.

tutto i sorgenti sempre disponibili su github


giovedì 16 gennaio 2014

Rover + Nodejs + Socket.io

Prosegue la parte di sviluppo del piccolo rover con Raspberry Pi + Arduino e Dagu 5.
Nella prossima fase prevede l'integrazione del giroscopio MPU5060

La parte di gestione da browser adesso utilizza Express.io appunto con le chiamate in socket. Molto più veloce rispetto alle classiche chiamate ajax.

Sulla sinistra i comandi principali per muovere il rover sulle 4 ruote motrici. In basso a sinistra il controllo del Tilt pannel della RaspiCam

Sulla destra lo stream della camera.





tutti i sorgente e le foto sul gitHub

lunedì 23 dicembre 2013

Arduino + Raspberry PI + Dagu 5 Rover + Node.js



Dopo un lungo periodo di esperimenti e di fatica finalmente sono arrivato alla prima fase del piccolo progetto.

Dai questo video il rover che muove i suoi primi passi in casa... con qualche problema..
L'hw è un mix di arduino per il controllo dei motori edella tilt pan per la Cam, Raspberry che gestisce lo streaming della cam sulla rete, e la parte web con node.js. Mancano alcuni dettagli che sistemerò a giorni.




Tutti i sorgenti, le specifiche e l'hw potete trovarlo sul GitHub



 

  










link on GitHub Site

domenica 1 dicembre 2013

Raspberry Pi - Dagu 5 - Primi Passi

UPDATE: la gestione hardware è stata demandata ad Arduino.Il Raspberry comanderà arduino. Nel prossimo post altri dettagli.

Proseguono gli esperimenti con il Raspberry PI, preso da una frenesia di acquisti notturni ho comprato questo spettacolare rover con la scheda di controllo per i motori.




Hardware


Per il Raspberry Utilizzo la distribuzione standard presente sul sito, ho tolto X ed aggiunto la java-openjdk per utilizzare appunto java più alcune utility (joe, htop, etc), tutto comodamente utilizzando apt-get install.

La comunicazione avviene tramite wifi applicando una chiavetta al raspberry in modo da mantenerlo autonomo da cavo di rete.

Prima problema.. mi sono accorto che l'uscita digitale GPIO del Rapsberry è a 3.3V invece la scheda di controllo dei motori del  dagu funziona su un segnale a 5 volt like Arduino. Quindi devo cercare un convertitore di segnale.. eccolo!!! Tindie converter gpio da 3 a 5v


Primi test.

Dopo vari esperimenti e ricerche trovo questa schedina che converte tutti i segnali del gpio a 5v. Proprio quello che serve per il controller del rover.

I primi esperimenti veloci dell'hardware sono con Nodejs e una piccola lib node-gpio semplice e veloce.


Materiale

Cavetti
Raspberry pi





ecco il tutto montato e pronto:







Primi esperimenti con python e nodejs... subito salta fuori il problema della programmazione parallela del gpio non molto semplice da risolvere. Se non risolvo la cosa sarà facile che passerò ad Arduino.... ufff....




martedì 19 marzo 2013

Raspberry Pi - configurare - NetGear 150N - RTL8188CUS


Preso il Rapsberry PI direttamente da RS. Chiavetta USB WI-FI  - NetGear150N.

Configurato in dhcp sulla mia rete interna.

Inserire la chiavetta nel Raspberry,

lsusb e vedete se la chiavetta viene vista dal sistema

NetGear, Inc. WNA1000M 802.11bgn [Realtek RTL8188CUS]

le configurazioni di

/etc/network/interfaces


auto lo

iface lo inet loopback
iface eth0 inet dhcp

allow-hotplug wlan0
iface wlan0 inet dhcp
wpa-conf /etc/wpa_supplicant/wpa_supplicant.conf


il file wpa-conf /etc/wpa_supplicant/wpa_supplicant.conf


ctrl_interface=DIR=/var/run/wpa_supplicant GROUP=netdev
update_config=1
network={
ssid="xxxxx"
proto=RSN
key_mgmt=WPA-PSK
pairwise=CCMP TKIP
group=CCMP TKIP
psk="xxxxxxxxx"
}



riavviare e togliere il cavo di rete.



[via omer.me ]