Sateliți și API-uri meteo pentru AgriTech: date predictive cu Sentinel-2 și Python
În 2024, Agenția Spațială Europeană a achiziționat peste 1,5 petabytes de date în fiecare zi de la sateliții Sentinel ai programului Copernicus. Dintre acestea, portiunea dedicata monitorizarii agricole în Europa reprezintă cel mai mare și cel mai relevant segment strategic. Cu toate acestea, marea majoritate a companiilor agricole italiene, inclusiv a celor mai avansate din punct de vedere tehnologic, nu exploatează încă această infrastructură publică, gratuită și validată științific.
Motivul nu este lipsa datelor: este complexitatea accesării, procesării și transformării acestora decizii agronomice concrete. Un satelit Sentinel-2 trece peste podgoria ta la fiecare 5 zile cu rezoluție la 10 metri pe pixel. Fiecare trecere produce 13 benzi spectrale care codifică sănătatea de vegetație, stresul hidric, conținutul de clorofilă, prezența paraziților. Combinat cu date vreme, senzori de sol IoT și modele predictive, aceste date pot fi radical transformatoare calitatea deciziilor agronomice, reducerea costurilor de intrare cu 15-25% și creșterea recoltelor până la 10-20%.
Piața globală de observare a Pământului merită 10,07 miliarde de dolari în 2025 potrivit Straits Research, cu o creștere estimată la 17,20 miliarde până în 2033 (CAGR 6,92%). Segmentul agriculturii reprezintă 21% din aplicații și crește la cel mai mare ritm, determinat de cererea pentru agricultura de precizie, monitorizarea randamentului și optimizarea irigațiilor. În acest articol construim, pas cu pas, o conductă Python completă pentru accesarea Sentinel-2 prin CDSE (Copernicus Data Space Ecosystem), calculați NDVI și alți indici de vegetație, integrați date meteorologice de la Open-Meteo și furnizează un model predictiv pentru stresul hidric.
Ce veți învăța în acest articol
- Arhitectura programului Copernicus și cum să accesați gratuit Sentinel-2 prin CDSE
- Indici de vegetație: NDVI, EVI, SAVI, LAI - formule matematice, interpretare și praguri pentru culturile italiene
- Implementare completă Python: autentificare CDSE, descărcare bandă, calcul NDVI cu rasterio și numpy
- API-uri meteo comparate: Open-Meteo, OpenWeatherMap, Tomorrow.io și integrarea datelor satelitare
- Conductă geospațială cu GeoPandas, rasterio și PostGIS pentru date vectoriale și raster
- Model predictiv de stres hidric: combinarea Sentinel-2 + vremea + IoT cu scikit-learn
- Studiu de caz practic: podgoria din Puglia, monitorizarea sezonieră NDVI, optimizarea irigațiilor
- Legislația italiană: AGEA, SIAN, open data CAP și digital CAP
Seria FoodTech - Toate articolele
| # | Articol | Nivel | Stat |
|---|---|---|---|
| 1 | Conductă IoT pentru agricultura de precizie cu Python și MQTT | Avansat | Disponibil |
| 2 | ML Edge pentru monitorizarea culturilor: computer Vision in the Fields | Avansat | Disponibil |
| 3 | Sateliți și API-uri meteo pentru AgriTech: date predictive (sunteți aici) | Avansat | Actual |
| 4 | Trasabilitatea blockchain în alimente: de la câmp la supermarket | Intermediar | În curând |
| 5 | Viziunea computerizată pentru controlul calității în industria alimentară | Avansat | În curând |
| 6 | FSMA și Digital Compliance: Automatizarea proceselor de reglementare | Intermediar | În curând |
| 7 | Agricultura verticală: Controlul mediului cu IoT și ML | Avansat | În curând |
| 8 | Prognoza cererii pentru comerțul cu amănuntul alimentar cu Prophet și LightGBM | Intermediar | În curând |
| 9 | Tabloul de bord Farm Intelligence: analiză în timp real cu Grafana | Intermediar | În curând |
| 10 | Optimizarea lanțului de aprovizionare alimentară: ML pentru reducerea deșeurilor | Intermediar | În curând |
Programul Copernicus: Infrastructură europeană pentru observarea Pământului
Programul Copernicus al Uniunii Europene este cea mai mare infrastructură de observare din Teren niciodată construit. Gestionat de ESA (Agenția Spațială Europeană) și Comisia Europeană cu Suport operațional al EUMETSAT, Copernicus operează o constelație de sateliți numiți „Sentinel”, fiecare conceput pentru misiuni specifice de monitorizare. Pentru agricultură, satelit de referință absolută e Santinela-2.
Constelația Sentinel-2 este compusă din doi sateliți gemeni (Sentinel-2A și Sentinel-2B) în orbita sincronă cu soarele la 786 km altitudine. Configurația celor doi sateliți garantează a timp Revizuire de 5 zile la ecuator și 2-3 zile la latitudini europene (Italia incluse). Fiecare satelit poartă la bord senzorul MSI (MultiSpectral Instrument), o mătură scaner care achiziționează date cu o lățime de glisare de 290 km și 13 benzi spectrale distribuit pe trei rezoluții spațiale: 10 metri, 20 metri și 60 de metri.
Cea mai strategică caracteristică a Sentinel-2 este ea complet gratuit: din 2014 toate datele Copernicus sunt disponibile în acces deschis pentru orice utilizare, comercială sau necomercială comercial. Noul portal de acces din 2023 și Ecosistemul Copernicus Data Space (CDSE), care a înlocuit anteriorul Copernicus Open Access Hub (SciHub) dezafectat în 2023. CDSE oferă acces imediat la arhiva completă Sentinel-2, cu latență de achiziție aproximativ 3 ore disponibile pentru cele mai recente imagini.
Satellite Sentinel-2: Specificații tehnice ale senzorului MSI
| Bandă | lungime de unda (nm) | Rezoluţie | Aplicația principală |
|---|---|---|---|
| B1 - Aerosoli de coastă | 443 | 60 m | calitatea aerului, corecția atmosferică |
| B2 - Albastru | 490 | 10 m | Discriminare vegetație/sol, cartografiere |
| B3 - Verde | 560 | 10 m | Vigurozitatea vegetației, culoarea frunzelor |
| B4 -Roşu | 665 | 10 m | Clorofilă, NDVI (bandă roșie) |
| B5 - Marginea roșie a vegetației | 705 | 20 m | Stresul vegetației, Red Edge NDVI |
| B6 - Marginea roșie a vegetației | 740 | 20 m | Conținutul de clorofilă, clasificarea speciilor |
| B7 - Marginea roșie a vegetației | 783 | 20 m | Biomasă, LAI |
| B8 - NIR | 842 | 10 m | NDVI (banda NIR), biomasă, LAI |
| B8a - Marginea roșie a vegetației | 865 | 20 m | Structura baldachin, NDVI îngust |
| B9 - Vaporii de apă | 940 | 60 m | Corecția vaporilor de apă atmosferici |
| B10 - SWIR - Cirrus | 1375 | 60 m | Detectarea norilor de ciruri |
| B11 - SWIR | 1610 | 20 m | Stresul hidric, conținutul de apă din frunze |
| B12 - SWIR | 2190 | 20 m | Stresul de apă avansat, senescență |
Există în principal două niveluri de procesare a datelor Sentinel-2 relevante pentru agricultură: Nivel-1C (L1C) - strălucirea atmosferei superioare cu corecție geometrică, de ex Nivel-2A (L2A) - reflectanța în partea de jos a atmosferei (Bottom of Atmosphere, BOA) cu corecția atmosferică aplicată prin algoritmul Sen2Cor. Pentru aplicații agricole directe, Nivelul-2A este cel recomandat deoarece este deja corectat pentru influența atmosferei, făcând valori comparabile ale reflectanței între diferite date și diferite locații geografice.
Platforme prin satelit pentru AgriTech: comparație completă 2025
Sentinel-2 nu este singura opțiune disponibilă. Piața de date satelitare pentru agricultură include furnizori comerciali, cum ar fi Planet Labs, operatori precum Landsat (NASA/USGS) și platforme cloud integrate, cum ar fi Google Earth Engine. Alegerea depinde de un compromis complex între rezoluția spațială, frecvența de achiziție, acoperirea cloud gestionată și bugetul disponibil.
Comparația platformelor prin satelit pentru agricultura de precizie - 2025
| Platformă | Rezoluție spațială | Revizuiți Frecvența | Benzi spectrale | Cost | Latența datelor | API-uri |
|---|---|---|---|---|---|---|
| Sentinel-2 (ESA/Copernicus) | 10 m (vis/NIR), 20 m (SWIR) | 5 zile (2-3 în UE) | 13 benzi MSI | Gratuit (date deschise) | ~3 ore de la achiziție | CDSE OData, STAC, SentinelHub, OpenEO |
| Planeta PlanetScope | 3,7 m | Zilnic (cvasi-zilnic) | 8 benzi (PS2.SD) | Abonament Comercial; gratuit pentru cercetare | 24 de ore | Planet API v1, Subscriptions API |
| Landsat 8/9 (NASA/USGS) | 30 m (multispectral), 15 m (pan) | 16 zile | 11 benzi OLI/TIRS | Gratuit (date deschise) | 24-48 ore | USGS EarthExplorer, Google Earth Engine |
| MODIS (NASA Earth/Aqua) | 250m / 500m / 1km | 1-2 zile | 36 de benzi | Gratuit (date deschise) | 24-48 ore | NASA EarthData STAC, Google Earth Engine |
| Google Earth Engine | Depinde de setul de date (10m-1km) | Depinde de setul de date | Multi-seturi de date integrate | Gratuit necomercial; reclamă plătită | Procesare instantanee în cloud | Python ee, API JavaScript |
| Maxar WorldView-3 | 0,3 m (pan), 1,24 m (multi) | 1-4 zile | 29 de benzi (CAVIS) | ~25-40 USD/km2 | 4-8 ore | Maxar Streaming API |
| Airbus Pleiades Neo | 0,3 m | 1-2 zile | 6 benzi multispectrale | ~15-30 EUR/km2 | 24-48 ore | OneAtlas API |
Pentru marea majoritate a aplicațiilor agricole din Italia, Sentinel-2 și alegerea optim: rezoluția de 10 metri este suficientă pentru parcele mai mari de 0,5 hectare (pragul sub care statisticile pixelilor devin nesigure), frecvența de revizuire de 2-3 zile în Italia și adecvate pentru monitorizare sezonieră, iar complet gratuit elimină orice barieră economică în calea adopției. Planet Labs devine relevant doar pentru aplicații care necesită monitorizare aproape zilnică (de exemplu, detectarea debutului precoce al stresul hidric în culturile de mare valoare, cum ar fi fructele și legumele intensive) sau rezoluția sub 10 m. MODIS și Landsat rămân utile pentru analize de suprafețe mari și serii de timp multidecenale.
Indici de vegetație: NDVI, EVI, SAVI și LAI pentru culturile italiene
Indicii de vegetație (VI) sunt metrici derivate matematic din combinație de benzi spectrale care cuantifică proprietăți specifice vegetației. Ei exploatează faptul că clorofila absoarbe puternic în banda roșie (665 nm) și reflectă puternic în infraroșu apropiat (842 nm): un raport între aceste benzi codifică densitatea și vitalitatea vegetaţie cu eleganţă matematică.
NDVI - Indicele de vegetație a diferențelor normalizate
NDVI este cel mai utilizat indice de vegetație din lume, introdus de Rouse și colab. în 1973. Formula matematică este:
formula NDVI
NDVI = (NIR - Red) / (NIR + Red)
Su Sentinel-2:
NDVI = (B8 - B4) / (B8 + B4)
Range: da -1.0 a +1.0
Valorile NDVI sunt interpretate conform unei scale standardizate, dar pragurile optime variază după cultură și faza fenologică. Următorul tabel prezintă pragurile de referință pentru principalele culturi italiene, derivate din literatura științifică și validarea empirică pe zone tipice (Pianura Padana, Puglia, Sicilia, Toscana):
Interpretarea NDVI pentru culturile italiene
| Gama NDVI | Interpretare generală | grâu (Triticum) | Viță de vie (Vitis) | Roșie (Solanum) | Porumb (Zea mays) | măslin (Olea) |
|---|---|---|---|---|---|---|
| < 0,1 | Pământ gol, stâncă, zăpadă | Pământ nesemănat | Iarna, înainte de înmugurire | Post-recoltare | Pre-semănat | Pământ gol între rânduri |
| 0,1 - 0,2 | Vegetație foarte rară sau uscată | Urgență inițială | Dormință/stres sever | Transplant recent | Urgență (VE) | Stres sever hidric/termic |
| 0,2 - 0,4 | Vegetație rară, stres moderat | Târnire inițială | Pre-înflorire, stres ușor | Creșterea vegetativă inițială | Etapele V1-V3 | Vegetație săracă, stres |
| 0,4 - 0,6 | Vegetație moderată, sănătate bună | Răsărire completă | Pre-înflorire / set de fructe | Dezvoltare vegetativă activă | Etapele V4-V8 | Frunziș normal, bine udat |
| 0,6 - 0,8 | Vegetație densă, vigoare excelentă | Titlu (vârf sezonier) | Foliare completă, veraiare | Acoperire maximă (înflorire) | Etapele V10-VT (înflorire) | Frunziș dens, stare excelentă |
| > 0,8 | Vegetație foarte densă, pădure | Rare (păduri de la marginea câmpului) | Garduri vii și copaci de graniță | Sere cu acoperire totală | Condiții optime rare | Livezi de măslini cu densitate mare |
EVI - Enhanced Vegetation Index
EVI, dezvoltat de Huete et al. pentru senzorul MODIS, corectează limitele NDVI din zone cu densitate mare a vegetației (saturație NDVI peste 0,7) și în zonele cu soluri expuse (Erori NDVI cauzate de reflectanța solului). Formula este:
EVI = G * (NIR - Red) / (NIR + C1*Red - C2*Blue + L)
Su Sentinel-2:
EVI = 2.5 * (B8 - B4) / (B8 + 6*B4 - 7.5*B2 + 1)
Costanti standard:
G = 2.5 (guadagno)
C1 = 6.0 (coefficiente resistenza aerosol)
C2 = 7.5 (coefficiente resistenza aerosol)
L = 1.0 (fattore aggiustamento canopy)
Range: -1.0 a +1.0 (tipicamente 0 a 0.9 per vegetazione)SAVI - Soil Adjusted Vegetation Index
SAVI, propus de Huete în 1988, este deosebit de util în medii semi-aride precum Puglia sau Sicilia unde solul expus influențează semnificativ răspunsul spectral. Factorul de corecție L reduce influența solului:
SAVI = (NIR - Red) * (1 + L) / (NIR + Red + L)
Su Sentinel-2:
SAVI = (B8 - B4) * (1 + 0.5) / (B8 + B4 + 0.5)
L = 0.5 (valore standard per vegetazione media)
L = 1.0 per vegetazione molto rada
L = 0.25 per vegetazione densa
Vantaggioso rispetto a NDVI quando:
- Copertura vegetale < 40%
- Suoli chiari e brillanti (calcari, sabbie)
- Areali semi-aridi del Sud ItaliaLAI - Indexul zonei frunzelor
LAI (indicele suprafeței frunzelor) cuantifică suprafața totală a frunzei pe unitatea de suprafață solului. Este un parametru agronomic fundamental legat de productivitatea potențială și transpiratie. Sentinel-2 vă permite să estimați LAI folosind algoritmi empiric sau fizic (Procesor biofizic SNAP):
# Stima LAI empirica da NDVI (relazione di Boegh et al., corretta per Sentinel-2)
# Valida per cereali e colture erbacee europee
LAI_stima = -0.3 + 10.2 * NDVI # valida per NDVI in [0.2, 0.8]
# Per la vite (relazione specifica da letteratura italiana):
LAI_vite = 0.57 * EXP(3.28 * NDVI) # Dalla et al., 2019
# Soglie LAI indicative per colture italiane:
# Grano: LAI ottimale 3-5 m2/m2 alla spigatura
# Mais: LAI ottimale 3-5 m2/m2 alla fioritura
# Vite: LAI ottimale 1.5-2.5 m2/m2 durante invaiatura
# Pomodoro: LAI ottimale 3-4 m2/m2 durante copertura massimaLimitările indicilor de vegetație prin satelit
- Înnorarea: Sentinel-2 și optice - norii fac imaginea inutilizabilă. În Italia, acoperirea medie a norilor este de 48% iarna și 15% vara. Banda QA60 permite mascarea automată a norului.
- Saturație NDVI: Peste NDVI ~0,7-0,8 sensibilitatea este redusă drastic. Pentru culturile cu densitate mare folosiți EVI.
- Mix de pixeli: La o rezoluție de 10 m, un pixel poate conține atât vegetație, cât și sol. Pentru parcelele mici (< 0,5 ha) NDVI mediu este mai puțin reprezentativ.
- Sezonalitate fenologică: Compararea NDVI între diferite date necesită luarea în considerare a fazei fenologice, nu doar a valorii absolute.
- Variabilitatea interanuală: NDVI variază și în funcție de condițiile meteorologice sezoniere (temperatură, precipitații), indiferent de practicile agronomice.
Implementare Python: Acces CDSE și descărcare Sentinel-2
Ecosistemul Copernicus Data Space (CDSE) a devenit portalul unic pentru accesarea datelor Sentinel din 2023. Oferă patru API-uri principale: API-ul OData (căutați și descărcați produse), API-ul STAC (Catalogul SpatioTemporal Asset), OpenEO API (prelucrare standardizat) e API-ul Sentinel Hub (prelucrare ca serviciu). Pentru a începe, creați un cont gratuit pe dataspace.copernicus.eu și generați acreditări OAuth2.
Configurarea mediului Python și dependențe
# Installa le librerie necessarie
pip install sentinelhub rasterio numpy pandas geopandas shapely \
requests python-dotenv matplotlib scipy scikit-learn \
openmeteo-requests retry-requests xarray
# requirements.txt con versioni validate per Python 3.11+
# sentinelhub==3.11.4 - interfaccia CDSE e SentinelHub
# rasterio==1.4.0 - lettura/scrittura raster geospaziali
# numpy==2.1.0 - calcolo matriciale per bande
# geopandas==1.0.1 - dati vettoriali e spatial operations
# shapely==2.0.6 - geometrie per AOI (Area of Interest)
# openmeteo-requests==0.3.3 - client Open-Meteo API
# scikit-learn==1.5.2 - modello predittivo stress idricoAutentificare CDSE cu OAuth2
import os
import requests
from datetime import datetime, timedelta
from dotenv import load_dotenv
load_dotenv()
# Configurazione credenziali CDSE (da variabili d'ambiente)
CDSE_USERNAME = os.getenv("CDSE_USERNAME")
CDSE_PASSWORD = os.getenv("CDSE_PASSWORD")
CDSE_CLIENT_ID = "cdse-public"
CDSE_TOKEN_URL = "https://identity.dataspace.copernicus.eu/auth/realms/CDSE/protocol/openid-connect/token"
def get_cdse_access_token(username: str, password: str) -> str:
"""
Ottieni access token OAuth2 dal CDSE Identity Server.
Il token ha durata di 600 secondi (10 minuti).
"""
response = requests.post(
CDSE_TOKEN_URL,
data={
"grant_type": "password",
"client_id": CDSE_CLIENT_ID,
"username": username,
"password": password,
},
headers={"Content-Type": "application/x-www-form-urlencoded"},
timeout=30,
)
response.raise_for_status()
token_data = response.json()
return token_data["access_token"]
class CDSESession:
"""Sessione autenticata con gestione automatica del refresh token."""
def __init__(self, username: str, password: str):
self._username = username
self._password = password
self._token: str | None = None
self._token_expiry: datetime | None = None
def _refresh_token(self) -> None:
self._token = get_cdse_access_token(self._username, self._password)
# Margine di 60 secondi prima della scadenza reale (600s)
self._token_expiry = datetime.utcnow() + timedelta(seconds=540)
def get_headers(self) -> dict:
if self._token is None or datetime.utcnow() >= self._token_expiry:
self._refresh_token()
return {"Authorization": f"Bearer {self._token}"}
# Inizializza sessione
session = CDSESession(CDSE_USERNAME, CDSE_PASSWORD)Căutați produse Sentinel-2 prin API-ul OData
from shapely.geometry import box
import json
ODATA_BASE_URL = "https://catalogue.dataspace.copernicus.eu/odata/v1"
def search_sentinel2_products(
bbox: tuple[float, float, float, float], # (lon_min, lat_min, lon_max, lat_max)
start_date: str, # formato: "2024-05-01T00:00:00.000Z"
end_date: str, # formato: "2024-05-31T23:59:59.000Z"
cloud_cover_max: float = 20.0,
product_type: str = "S2MSI2A", # L2A = Bottom of Atmosphere, preferire per agricoltura
max_results: int = 20,
) -> list[dict]:
"""
Cerca prodotti Sentinel-2 nel catalogo CDSE.
Args:
bbox: Bounding box nell'ordine (lon_min, lat_min, lon_max, lat_max)
start_date: Data inizio (formato ISO 8601)
end_date: Data fine (formato ISO 8601)
cloud_cover_max: Percentuale massima di copertura nuvolosa (0-100)
product_type: S2MSI2A (L2A, consigliato) o S2MSI1C (L1C)
max_results: Numero massimo di risultati
Returns:
Lista di dizionari con metadata prodotto
"""
lon_min, lat_min, lon_max, lat_max = bbox
aoi_wkt = f"POLYGON(( {lon_min} {lat_min}, {lon_max} {lat_min}, {lon_max} {lat_max}, {lon_min} {lat_max}, {lon_min} {lat_min} ))"
filter_query = (
f"Collection/Name eq 'SENTINEL-2' "
f"and Attributes/OData.CSC.DoubleAttribute/any(att:att/Name eq 'cloudCover' "
f" and att/OData.CSC.DoubleAttribute/Value le {cloud_cover_max}) "
f"and Attributes/OData.CSC.StringAttribute/any(att:att/Name eq 'productType' "
f" and att/OData.CSC.StringAttribute/Value eq '{product_type}') "
f"and ContentDate/Start ge {start_date} "
f"and ContentDate/Start lt {end_date} "
f"and OData.CSC.Intersects(area=geography'SRID=4326;{aoi_wkt}')"
)
params = {
"$filter": filter_query,
"$orderby": "ContentDate/Start desc",
"$top": max_results,
"$expand": "Attributes",
}
response = requests.get(
f"{ODATA_BASE_URL}/Products",
params=params,
timeout=60,
)
response.raise_for_status()
data = response.json()
return data.get("value", [])
# Esempio: cerca immagini per la vigna di Manduria (Puglia)
# Campagna estiva: maggio-settembre 2024
bbox_manduria = (17.4, 40.3, 17.7, 40.5) # Zona DOC Primitivo di Manduria
products = search_sentinel2_products(
bbox=bbox_manduria,
start_date="2024-05-01T00:00:00.000Z",
end_date="2024-09-30T23:59:59.000Z",
cloud_cover_max=10.0,
product_type="S2MSI2A",
)
print(f"Trovati {len(products)} prodotti Sentinel-2 L2A con copertura nuvolosa <= 10%")
for p in products[:5]:
cloud = next(
(a["Value"] for a in p.get("Attributes", []) if a["Name"] == "cloudCover"),
"N/A",
)
print(f" {p['Name']} | Cloud: {cloud:.1f}% | Date: {p['ContentDate']['Start'][:10]}")Descărcați și accesați benzi cu Sentinel Hub Python
Pentru procesarea benzii fără descărcarea întregii scene (care cântărește 800 MB - 1,2 GB per produs L2A), este mai eficient să utilizați API-ul Sentinel Hub Process, care returnează numai benzi necesare în regiunea de interes:
from sentinelhub import (
SHConfig,
SentinelHubRequest,
DataCollection,
MimeType,
BBox,
CRS,
bbox_to_dimensions,
)
import numpy as np
# Configurazione SentinelHub via CDSE
config = SHConfig()
config.sh_client_id = os.getenv("SH_CLIENT_ID") # da apps.sentinel-hub.com
config.sh_client_secret = os.getenv("SH_CLIENT_SECRET")
config.sh_base_url = "https://sh.dataspace.copernicus.eu" # endpoint CDSE
# Definizione Area of Interest (AOI) - Vigna Manduria 100 ha circa
aoi_bbox = BBox(
bbox=[17.42, 40.34, 17.52, 40.42],
crs=CRS.WGS84,
)
# Calcola dimensioni raster a 10m di risoluzione
size = bbox_to_dimensions(aoi_bbox, resolution=10)
print(f"Dimensioni raster: {size[0]} x {size[1]} pixel a 10m")
# Evalscript per scaricare B04 (Red) e B08 (NIR) per calcolo NDVI
evalscript_ndvi_bands = """
//VERSION=3
function setup() {
return {
input: [{
bands: ["B04", "B08", "CLM"], // Red, NIR, Cloud Mask
units: "REFLECTANCE"
}],
output: {
bands: 3,
sampleType: "FLOAT32"
}
};
}
function evaluatePixel(sample) {
// Restituisce [Red, NIR, CloudMask]
return [sample.B04, sample.B08, sample.CLM];
}
"""
# Richiesta dati
request = SentinelHubRequest(
evalscript=evalscript_ndvi_bands,
input_data=[
SentinelHubRequest.input_data(
data_collection=DataCollection.SENTINEL2_L2A.define_from(
"s2l2a",
service_url=config.sh_base_url,
),
time_interval=("2024-07-15", "2024-07-20"),
other_args={"dataFilter": {"maxCloudCoverage": 10}},
)
],
responses=[SentinelHubRequest.output_response("default", MimeType.TIFF)],
bbox=aoi_bbox,
size=size,
config=config,
)
# Download dati (restituisce array numpy [H, W, Bande])
images = request.get_data()
band_data = images[0] # shape: (height, width, 3)
red_band = band_data[:, :, 0].astype(np.float32) # B04 - Red
nir_band = band_data[:, :, 1].astype(np.float32) # B08 - NIR
cloud_mask = band_data[:, :, 2] # CLM - 0=clear, 1=cloud
print(f"Bande scaricate - Shape: {red_band.shape}")
print(f"Red B04 - Min: {red_band.min():.4f}, Max: {red_band.max():.4f}, Mean: {red_band.mean():.4f}")
print(f"NIR B08 - Min: {nir_band.min():.4f}, Max: {nir_band.max():.4f}, Mean: {nir_band.mean():.4f}")
print(f"Pixel nuvolosi: {cloud_mask.sum()} su {cloud_mask.size} totali ({100*cloud_mask.mean():.1f}%)")Calcul NDVI și salvarea GeoTIFF
import rasterio
from rasterio.transform import from_bounds
from rasterio.crs import CRS as RasterioCRS
import matplotlib.pyplot as plt
import matplotlib.colors as mcolors
def calculate_ndvi(
red: np.ndarray,
nir: np.ndarray,
cloud_mask: np.ndarray | None = None,
) -> np.ndarray:
"""
Calcola NDVI dalla banda rossa e NIR.
Gestisce divisione per zero con np.errstate.
Maschera i pixel nuvolosi con np.nan.
Returns:
Array NDVI con float32, range [-1, 1], NaN per pixel invalidi
"""
with np.errstate(divide="ignore", invalid="ignore"):
ndvi = np.where(
(nir + red) == 0,
np.nan,
(nir - red) / (nir + red),
).astype(np.float32)
# Maschera nuvole
if cloud_mask is not None:
ndvi = np.where(cloud_mask == 1, np.nan, ndvi)
# Clip valori fisicamente impossibili (artefatti strumentali)
ndvi = np.clip(ndvi, -1.0, 1.0)
return ndvi
def save_ndvi_geotiff(
ndvi: np.ndarray,
bbox_wgs84: tuple[float, float, float, float],
output_path: str,
) -> None:
"""
Salva l'array NDVI come GeoTIFF georeferenziato in WGS84.
Args:
ndvi: Array NDVI 2D (H, W)
bbox_wgs84: (lon_min, lat_min, lon_max, lat_max)
output_path: Path del file output
"""
height, width = ndvi.shape
lon_min, lat_min, lon_max, lat_max = bbox_wgs84
transform = from_bounds(lon_min, lat_min, lon_max, lat_max, width, height)
with rasterio.open(
output_path,
"w",
driver="GTiff",
height=height,
width=width,
count=1,
dtype=rasterio.float32,
crs=RasterioCRS.from_epsg(4326),
transform=transform,
compress="lzw",
nodata=np.nan,
) as dst:
dst.write(ndvi, 1)
dst.update_tags(
NDVI_COMPUTED_AT=datetime.utcnow().isoformat(),
SENTINEL2_PRODUCT_TYPE="S2MSI2A",
FORMULA="(B08-B04)/(B08+B04)",
)
# Calcola e salva NDVI
ndvi_map = calculate_ndvi(red_band, nir_band, cloud_mask)
# Statistiche NDVI per l'AOI (escluso NaN)
valid_pixels = ndvi_map[~np.isnan(ndvi_map)]
print(f"\nStatistiche NDVI - Vigna Manduria (luglio 2024)")
print(f" Pixel validi: {len(valid_pixels)} su {ndvi_map.size}")
print(f" NDVI medio: {valid_pixels.mean():.3f}")
print(f" NDVI mediano: {np.median(valid_pixels):.3f}")
print(f" NDVI p10: {np.percentile(valid_pixels, 10):.3f} (zone a stress)")
print(f" NDVI p90: {np.percentile(valid_pixels, 90):.3f} (zone ottimali)")
bbox_wgs84 = (17.42, 40.34, 17.52, 40.42)
save_ndvi_geotiff(ndvi_map, bbox_wgs84, "ndvi_manduria_20240717.tif")
print("GeoTIFF NDVI salvato: ndvi_manduria_20240717.tif")API-uri meteo pentru AgriTech: comparație și implementare
Datele meteorologice sunt a doua sursă critică de date pentru agricultura de precizie. Integrate cu datele NDVI și IoT de pe teren, acestea vă permit să construiți modele predictive pentru stresul hidric, riscul de îmbolnăvire, evapotranspirația și prognoza randamentului. Panorama dintre API-urile meteo în 2025 este împărțită în furnizori gratuiti cu sursă deschisă, furnizori comerciali cu niveluri gratuite și servicii regionale ale ARPA italiană.
API-uri meteo pentru AgriTech - Comparație 2025
| Furnizorii | Preţ | Rezoluție spațială | Prognoza | Variabile agricole | Arhiva istorică | API-uri |
|---|---|---|---|---|---|---|
| Vreme deschisă | Gratuit (necomercial); Reclamă de ~15 USD/lună | 1-11 km (multimodel) | 16 zile | ET0, temperatura solului, umiditatea solului, presiunea vaporilor | 1940-prezent (ERA5) | REST JSON, client Python |
| OpenWeatherMap | Gratuit până la 60 de apeluri/min; 40 USD/lună pro | 2,5 km (ICON) | 5 zile (gratuit), 16 zile (pro) | Limitat, fără ET0 | Doar istoric plătit | REST JSON, SDK Python |
| Mâine.io | Gratuit 500 apeluri/zi; 199 USD/lună de bază | 1 km | 21 de zile | Umiditatea solului, ET0, condițiile de pulverizare, riscul dăunătorilor | 6 ani | REST, WebSocket, gRPC |
| ARPA Lombardia / Piemont / Veneto | Gratuit (date regionale deschise) | Posturi punctuale (rețea ~1500 de posturi în Italia) | Nu (doar observat) | Toate variabilele agro-meteorologice | Arhivă completă (decenii) | Descărcare WMS, REST, CSV/JSON |
| Meteoam / CFS/ECMWF | date deschise gratuite ale ECMWF; CFS NOAA gratuit | 9-25 km | 10-45 zile | Variabile bazate pe vreme | 1940-prezent prin ERA5 | API-ul ECMWF (Python), API-ul CDS |
| Încrucișare vizuală | Gratuit 1000 de înregistrări/zi; 35 USD/luna standard | Interpolat din stații | 15 zile | ET0, grade-zile de creștere, risc de îngheț | Plătit nelimitat | REST JSON, CSV, asemănător SQL |
Pentru aplicațiile AgriTech în producție, recomandarea este utilizarea Vreme deschisă cum de bază (gratuit, open-source, date ERA5 pentru istorici, 16 zile de prognoză) integrându-l cu rețele de stații ARPA pentru calibrare locală. Open-Meteo include variabile agrometeorologice probleme critice, cum ar fi ET0 (evapotranspirația de referință), umiditatea solului la adâncimi multiple și aburul apoase, care nu sunt întotdeauna disponibile gratuit de la furnizorii comerciali.
Implementarea Open-Meteo Client cu Cache și Retry
import openmeteo_requests
import requests_cache
import pandas as pd
from retry_requests import retry
from dataclasses import dataclass
from typing import Optional
@dataclass
class WeatherForecast:
"""Dati meteo giornalieri per decisioni agronomiche."""
date: pd.DatetimeIndex
temperature_max: np.ndarray # Celsius
temperature_min: np.ndarray # Celsius
precipitation: np.ndarray # mm
et0: np.ndarray # mm/giorno - evapotraspirazione di riferimento (FAO-56)
wind_speed_max: np.ndarray # km/h
solar_radiation: np.ndarray # MJ/m2/giorno
soil_moisture_0_7cm: np.ndarray # m3/m3 (0-7 cm profondità)
soil_moisture_7_28cm: np.ndarray # m3/m3 (7-28 cm profondità)
soil_temp_0_7cm: np.ndarray # Celsius
def get_weather_forecast(
latitude: float,
longitude: float,
start_date: Optional[str] = None, # "YYYY-MM-DD" per archivio
end_date: Optional[str] = None,
forecast_days: int = 16,
) -> WeatherForecast:
"""
Recupera dati meteo da Open-Meteo con cache locale (1 ora) e retry automatico.
Combina forecast (16 giorni) con archivio storico ERA5 se start_date specificato.
Args:
latitude: Latitudine WGS84
longitude: Longitudine WGS84
start_date: Se fornito, richiede dati storici (non forecast)
end_date: Fine periodo storico
forecast_days: Giorni di forecast (1-16)
Returns:
WeatherForecast con tutti i parametri agrometeorologici
"""
# Cache HTTP locale (1 ora di validita) + retry su errori transitori
cache_session = requests_cache.CachedSession(".cache/open_meteo", expire_after=3600)
retry_session = retry(cache_session, retries=5, backoff_factor=0.2)
client = openmeteo_requests.Client(session=retry_session)
base_url = (
"https://archive-api.open-meteo.com/v1/archive"
if start_date
else "https://api.open-meteo.com/v1/forecast"
)
params = {
"latitude": latitude,
"longitude": longitude,
"daily": [
"temperature_2m_max",
"temperature_2m_min",
"precipitation_sum",
"et0_fao_evapotranspiration",
"wind_speed_10m_max",
"shortwave_radiation_sum",
"soil_moisture_0_to_7cm",
"soil_moisture_7_to_28cm",
"soil_temperature_0_to_7cm",
],
"timezone": "Europe/Rome", # Fuso orario italiano
"wind_speed_unit": "kmh",
}
if start_date:
params["start_date"] = start_date
params["end_date"] = end_date
else:
params["forecast_days"] = forecast_days
# Open-Meteo usa ERA5 per archivio storico (1940-oggi), eccellente per agricoltura
if start_date:
# Archivio ERA5-Land: alta risoluzione 9 km per Italia
params["models"] = "era5_land"
responses = client.weather_api(base_url, params=params)
r = responses[0] # primo (e unico) punto
daily = r.Daily()
return WeatherForecast(
date=pd.date_range(
start=pd.to_datetime(daily.Time(), unit="s", utc=True),
end=pd.to_datetime(daily.TimeEnd(), unit="s", utc=True),
freq=pd.Timedelta(seconds=daily.Interval()),
inclusive="left",
),
temperature_max = daily.Variables(0).ValuesAsNumpy(),
temperature_min = daily.Variables(1).ValuesAsNumpy(),
precipitation = daily.Variables(2).ValuesAsNumpy(),
et0 = daily.Variables(3).ValuesAsNumpy(),
wind_speed_max = daily.Variables(4).ValuesAsNumpy(),
solar_radiation = daily.Variables(5).ValuesAsNumpy(),
soil_moisture_0_7cm = daily.Variables(6).ValuesAsNumpy(),
soil_moisture_7_28cm = daily.Variables(7).ValuesAsNumpy(),
soil_temp_0_7cm = daily.Variables(8).ValuesAsNumpy(),
)
# Esempio: meteo vigna Manduria (campagna 2024)
lat_manduria, lon_manduria = 40.38, 17.47
weather_storico = get_weather_forecast(
latitude=lat_manduria,
longitude=lon_manduria,
start_date="2024-04-01",
end_date="2024-09-30",
)
weather_forecast = get_weather_forecast(
latitude=lat_manduria,
longitude=lon_manduria,
forecast_days=16,
)
df_meteo = pd.DataFrame({
"data": weather_storico.date,
"t_max_C": weather_storico.temperature_max,
"t_min_C": weather_storico.temperature_min,
"pioggia_mm": weather_storico.precipitation,
"et0_mm": weather_storico.et0,
"rad_MJ": weather_storico.solar_radiation,
"sw_0_7cm": weather_storico.soil_moisture_0_7cm,
"sw_7_28cm": weather_storico.soil_moisture_7_28cm,
})
print(df_meteo.describe())
print(f"\nET0 totale campagna apr-set 2024: {df_meteo['et0_mm'].sum():.1f} mm")
print(f"Pioggia totale campagna apr-set 2024: {df_meteo['pioggia_mm'].sum():.1f} mm")
deficit_idrico = df_meteo["et0_mm"].sum() - df_meteo["pioggia_mm"].sum()
print(f"Deficit idrico stagionale (ET0 - pioggia): {deficit_idrico:.1f} mm")Conducta geospațială completă: GeoPandas, rasterio și PostGIS
Gestionarea profesională a datelor prin satelit necesită o infrastructură geospațială cuprinzătoare care integrează date raster (imagini din satelit NDVI, hărți meteorologice) cu date vectoriale (limite parcele, zone de management, puncte de prelevare). Conducta pe care o descriem aici adoptă stiva Python standard de facto pentru geospațial: rasterio pentru raster, GeoPandas pentru vectori, PostGIS ca bază de date spațială e GDAL ca motor de traducere a formatului de bază.
Arhitectura conductelor geospațiale
Tehnologia AgriTech Geospatial Pipeline Stack
┌─────────────────────────────────────────────────────────────────────────┐
│ DATI DI INPUT │
│ [Sentinel-2 GeoTIFF] [Shapefile Appezzamenti] [CSV Meteo/IoT] │
│ │ │ │ │
│ rasterio geopandas pandas │
└─────────┼──────────────────────┼───────────────────────┼────────────────┘
│ │ │
┌─────────▼──────────────────────▼───────────────────────▼────────────────┐
│ PROCESSING LAYER (Python) │
│ [NDVI Calculation] [Zonal Statistics] [Spatial Join] │
│ numpy/rasterio rasterstats geopandas │
│ │ │ │ │
└─────────┼──────────────────────┼───────────────────────┼────────────────┘
│ │ │
└──────────────────────▼───────────────────────┘
│
┌────────────────────────────────▼────────────────────────────────────────┐
│ STORAGE LAYER (PostGIS + S3) │
│ [PostGIS - geometrie + attributi] [S3/MinIO - GeoTIFF raster] │
│ - Tabella parcelle con NDVI medio - File raster originali │
│ - Serie storica per parcella - Archivio scene satellite │
│ - Spatial index GIST - Formato Cloud Optimized GeoTIFF │
└────────────────────────────────┬────────────────────────────────────────┘
│
┌────────────────────────────────▼────────────────────────────────────────┐
│ SERVING LAYER │
│ [REST API - FastAPI] [Dashboard - Grafana] [ML Models] │
│ NDVI per appezzamento Mappe NDVI tempo reale Predizione stress │
└─────────────────────────────────────────────────────────────────────────┘
Statistici zonale: NDVI mediu pe parcelă cu statistici raster
import geopandas as gpd
from rasterstats import zonal_stats
import rasterio
from sqlalchemy import create_engine, text
from geoalchemy2 import Geometry
import pandas as pd
# Carica shapefile appezzamenti vigna
gdf_parcelle = gpd.read_file("dati/parcelle_vigna_manduria.shp")
gdf_parcelle = gdf_parcelle.to_crs("EPSG:4326") # Riproietta in WGS84
print(f"Appezzamenti caricati: {len(gdf_parcelle)}")
print(gdf_parcelle[["id_parcella", "varieta", "ettari", "anno_impianto"]].head(10))
def compute_zonal_ndvi(
ndvi_raster_path: str,
parcelle_gdf: gpd.GeoDataFrame,
acquisition_date: str,
) -> gpd.GeoDataFrame:
"""
Calcola statistiche NDVI zonali per ogni appezzamento.
Per ogni parcella calcola:
- NDVI medio (indicatore di vigor generale)
- NDVI mediano (robusto agli outlier)
- NDVI std (indica variabilità intra-parcella)
- Percentile 10 (zone a stress all'interno della parcella)
- Percentile 90 (zone ottimali)
- Percentuale pixel validi (esclude NaN da nuvole)
Returns:
GeoDataFrame con colonne NDVI aggiunte
"""
stats = zonal_stats(
vectors=parcelle_gdf.geometry,
raster=ndvi_raster_path,
stats=["mean", "median", "std", "percentile_10", "percentile_90", "count", "nodata"],
nodata=np.nan,
geojson_out=False,
)
df_stats = pd.DataFrame(stats)
df_stats.columns = [
"ndvi_mean", "ndvi_median", "ndvi_std",
"ndvi_p10", "ndvi_p90", "pixel_count", "pixel_nodata",
]
df_stats["acquisition_date"] = pd.to_datetime(acquisition_date)
df_stats["pixel_valid_pct"] = (
df_stats["pixel_count"] /
(df_stats["pixel_count"] + df_stats["pixel_nodata"].fillna(0)) * 100
)
# Classificazione stress basata su NDVI medio (soglie per vite in estate)
df_stats["stress_category"] = pd.cut(
df_stats["ndvi_mean"],
bins=[-1.0, 0.25, 0.40, 0.55, 0.70, 1.0],
labels=["Stress Severo", "Stress Moderato", "Normale", "Buono", "Ottimale"],
)
return gpd.GeoDataFrame(
pd.concat([parcelle_gdf.reset_index(drop=True), df_stats], axis=1),
geometry="geometry",
crs=parcelle_gdf.crs,
)
# Calcola NDVI zonale per tutte le parcelle
gdf_ndvi = compute_zonal_ndvi(
ndvi_raster_path="ndvi_manduria_20240717.tif",
parcelle_gdf=gdf_parcelle,
acquisition_date="2024-07-17",
)
# Parcelle in stress - priorità irrigazione
parcelle_in_stress = gdf_ndvi[
gdf_ndvi["stress_category"].isin(["Stress Severo", "Stress Moderato"])
]
print(f"\nParcelle in stress ({len(parcelle_in_stress)}/{len(gdf_ndvi)}):")
print(parcelle_in_stress[["id_parcella", "varieta", "ndvi_mean", "stress_category", "ettari"]])
# Salva in PostGIS per persistenza e query spaziali
engine = create_engine("postgresql://agritech:pass@localhost:5432/vigna_db")
gdf_ndvi.to_postgis(
name="ndvi_storico",
con=engine,
if_exists="append",
index=False,
dtype={"geometry": Geometry("MULTIPOLYGON", srid=4326)},
)
print("\nDati NDVI salvati in PostGIS (tabella: ndvi_storico)")Interogări PostGIS pentru analiză spațială avansată
-- Schema PostGIS per sistema AgriTech
CREATE TABLE IF NOT EXISTS parcelle (
id_parcella SERIAL PRIMARY KEY,
codice_catasto VARCHAR(20) UNIQUE NOT NULL, -- codice catastale italiano
varieta VARCHAR(50),
ettari NUMERIC(8, 4),
anno_impianto INTEGER,
sistema_allevamento VARCHAR(30),
geom GEOMETRY(MULTIPOLYGON, 4326)
);
CREATE INDEX idx_parcelle_geom ON parcelle USING GIST(geom);
CREATE TABLE IF NOT EXISTS ndvi_storico (
id SERIAL PRIMARY KEY,
id_parcella INTEGER REFERENCES parcelle(id_parcella),
data_satellite DATE NOT NULL,
ndvi_mean NUMERIC(6, 4),
ndvi_median NUMERIC(6, 4),
ndvi_std NUMERIC(6, 4),
ndvi_p10 NUMERIC(6, 4),
ndvi_p90 NUMERIC(6, 4),
pixel_valid_pct NUMERIC(5, 2),
stress_category VARCHAR(20),
satellite VARCHAR(20) DEFAULT 'Sentinel-2',
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX idx_ndvi_parcella_data ON ndvi_storico(id_parcella, data_satellite);
-- Query: tendenza NDVI ultime 4 acquisizioni per parcella
SELECT
p.codice_catasto,
p.varieta,
n.data_satellite,
n.ndvi_mean,
LAG(n.ndvi_mean, 1) OVER (PARTITION BY n.id_parcella ORDER BY n.data_satellite) AS ndvi_prev,
n.ndvi_mean - LAG(n.ndvi_mean, 1) OVER (PARTITION BY n.id_parcella ORDER BY n.data_satellite) AS ndvi_delta
FROM parcelle p
JOIN ndvi_storico n ON p.id_parcella = n.id_parcella
WHERE n.data_satellite >= NOW() - INTERVAL '60 days'
ORDER BY p.id_parcella, n.data_satellite;
-- Query: parcelle con NDVI in calo > 0.1 nell'ultimo mese (alert stress)
SELECT
p.codice_catasto,
p.varieta,
p.ettari,
latest.ndvi_mean AS ndvi_corrente,
prev.ndvi_mean AS ndvi_30gg_fa,
(latest.ndvi_mean - prev.ndvi_mean) AS variazione
FROM parcelle p
JOIN ndvi_storico latest ON p.id_parcella = latest.id_parcella
AND latest.data_satellite = (SELECT MAX(data_satellite) FROM ndvi_storico WHERE id_parcella = p.id_parcella)
JOIN ndvi_storico prev ON p.id_parcella = prev.id_parcella
AND prev.data_satellite = (SELECT MAX(data_satellite) FROM ndvi_storico
WHERE id_parcella = p.id_parcella AND data_satellite < NOW() - INTERVAL '25 days')
WHERE (latest.ndvi_mean - prev.ndvi_mean) < -0.10
ORDER BY variazione ASC;Modelul predictiv al stresului hidric cu ML
Combinarea satelitului NDVI cu datele meteorologice și IoT vă permite să construiți modele predictive de stres hidric cu 3-7 zile înainte, permițând planificarea proactivă a irigațiilor. Modelul descris aici integrează trei surse de date: seria temporală NDVI de la Sentinel-2, bilanțul apei calculat din ET0 și precipitații și umiditatea solului de la senzorii IoT (sau de la Open-Meteo ERA5-Land dacă senzorii fizici nu sunt disponibili).
Inginerie caracteristică pentru model
import pandas as pd
import numpy as np
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.metrics import classification_report
import joblib
from typing import Tuple
def build_stress_features(
df_ndvi: pd.DataFrame, # colonne: data, ndvi_mean, ndvi_std, ndvi_p10
df_meteo: pd.DataFrame, # colonne: data, t_max_C, t_min_C, pioggia_mm, et0_mm, sw_0_7cm
lookback_days: int = 14,
) -> pd.DataFrame:
"""
Costruisce feature set per modello predittivo stress idrico.
Combina NDVI satellite con bilancio idrico e meteo.
Features generate:
- NDVI corrente e variazione a 5/10 giorni
- Deficit idrico cumulato (ET0 - pioggia) a 7/14/21 giorni
- Growing Degree Days (GDD) da inizio stagione
- Soil moisture media e tendenza
- Variabilità termica (t_max - t_min)
"""
df = df_meteo.merge(df_ndvi, on="data", how="left").sort_values("data")
df["ndvi_mean"] = df["ndvi_mean"].interpolate(method="linear", limit=5)
# Bilancio idrico cumulato (mm) - positivo = surplus, negativo = deficit
df["bilancio_idrico"] = df["pioggia_mm"] - df["et0_mm"]
df["deficit_7gg"] = df["bilancio_idrico"].rolling(7).sum()
df["deficit_14gg"] = df["bilancio_idrico"].rolling(14).sum()
df["deficit_21gg"] = df["bilancio_idrico"].rolling(21).sum()
# NDVI trend (variazione puntuale e su finestra)
df["ndvi_delta_5gg"] = df["ndvi_mean"] - df["ndvi_mean"].shift(5)
df["ndvi_delta_10gg"] = df["ndvi_mean"] - df["ndvi_mean"].shift(10)
# Growing Degree Days (base 10 gradi - standard per vite)
df["t_media"] = (df["t_max_C"] + df["t_min_C"]) / 2
df["gdd_giornaliero"] = np.maximum(df["t_media"] - 10, 0)
df["gdd_cumulato"] = df["gdd_giornaliero"].cumsum()
# Variabilità termica (indicatore stress termico)
df["escursione_termica"] = df["t_max_C"] - df["t_min_C"]
# Soil moisture media e trend
df["sm_media_7gg"] = df["sw_0_7cm"].rolling(7).mean()
df["sm_trend"] = df["sw_0_7cm"] - df["sw_0_7cm"].shift(3)
feature_cols = [
"ndvi_mean", "ndvi_std", "ndvi_p10",
"ndvi_delta_5gg", "ndvi_delta_10gg",
"deficit_7gg", "deficit_14gg", "deficit_21gg",
"et0_mm", "pioggia_mm",
"t_max_C", "t_min_C", "escursione_termica",
"gdd_cumulato", "gdd_giornaliero",
"sm_media_7gg", "sm_trend",
]
return df[["data"] + feature_cols].dropna()
def train_stress_model(
features_df: pd.DataFrame,
labels: pd.Series, # 0=no stress, 1=stress moderato, 2=stress severo
) -> Tuple[Pipeline, dict]:
"""
Addestra modello GradientBoosting per classificazione stress idrico.
Usa pipeline sklearn con StandardScaler integrato.
Returns:
(pipeline_addestrata, metriche_validazione)
"""
feature_cols = [c for c in features_df.columns if c != "data"]
X = features_df[feature_cols].values
y = labels.values
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42, stratify=y
)
pipeline = Pipeline([
("scaler", StandardScaler()),
("clf", GradientBoostingClassifier(
n_estimators=200,
learning_rate=0.05,
max_depth=4,
subsample=0.8,
random_state=42,
)),
])
pipeline.fit(X_train, y_train)
y_pred = pipeline.predict(X_test)
cv_scores = cross_val_score(pipeline, X, y, cv=5, scoring="f1_weighted")
metriche = {
"f1_cv_mean": cv_scores.mean(),
"f1_cv_std": cv_scores.std(),
"classification_report": classification_report(
y_test, y_pred,
target_names=["No Stress", "Stress Moderato", "Stress Severo"],
),
}
return pipeline, metriche
# Salva modello per deployment
def save_model(pipeline: Pipeline, path: str) -> None:
joblib.dump(pipeline, path)
print(f"Modello salvato: {path}")
def predict_stress_7days(
pipeline: Pipeline,
features_forecast: pd.DataFrame, # features calcolate su dati forecast 7 giorni
) -> pd.DataFrame:
"""
Predici stress idrico per i prossimi 7 giorni.
Usa le features calcolate su dati forecast meteo (Open-Meteo, 16 gg).
Returns:
DataFrame con data, probabilità per classe, classe predetta
"""
feature_cols = [c for c in features_forecast.columns if c != "data"]
X_forecast = features_forecast[feature_cols].values
proba = pipeline.predict_proba(X_forecast)
pred_class = pipeline.predict(X_forecast)
labels = ["no_stress", "stress_moderato", "stress_severo"]
result_df = pd.DataFrame(proba, columns=[f"prob_{l}" for l in labels])
result_df["classe_predetta"] = pred_class
result_df["data"] = features_forecast["data"].values
result_df["label_predetto"] = [labels[c] for c in pred_class]
return result_df[["data", "label_predetto", "prob_no_stress",
"prob_stress_moderato", "prob_stress_severo"]]Studiu de caz: monitorizarea NDVI într-o podgorie Primitivo din Manduria
Pentru a concretiza conceptele descrise, prezentăm un studiu de caz inspirat dintr-o aplicație real pe teritoriul DOC Primitivo di Manduria (Taranto, Puglia). Zona are conditii tipice viticulturii mediteraneene: veri calde si uscate (ploi de vara < 50 mm), soluri argilo-calcaroase, sisteme de antrenament contra-spalier și puieți.
Parametrii studiului de caz - Vigna Primitivo Manduria
| Parametru | Valoare |
|---|---|
| Locaţie | Manduria (TA), Puglia - lat 40.38, lon 17.47 |
| Suprafata totala | 35 de hectare împărțite în 12 parcele |
| Principalele soiuri | Primitivo (80%), Negroamaro (20%) |
| Sistemul de reproducere | Puieți din Apulia (8 unități), Contraspalier (4 unități) |
| Anul plantei | 2003-2015 (pogorie mixtă tineri/adulti) |
| Irigare | Picătură cu picătură (nu toate parcelele) |
| Urmărirea satelitului | Sentinel-2A/B, L2A, rezoluție 10m |
| Frecvența achizițiilor utilizate | 28 de scene în mediul rural din 2024 (nor < 20%) |
| Senzori IoT integrati | 6 statii meteo, 18 senzori de umiditate a solului la 30/60 cm |
| Perioada de analiză | aprilie - octombrie 2024 |
Rezultate sezoniere NDVI și decizii de irigare
Analiza sezonieră NDVI a relevat modele semnificativ diferențiate de stres hidric între parcele, cu implicații directe pentru managementul irigațiilor și calitatea produsului:
Rezultate NDVI după parcelă - Campanie 2024
| Complot | Varietate | NDVI apr | NDVI jos | NDVI iulie (vârf) | NDVI aug | set NDVI | Alertă de irigare |
|---|---|---|---|---|---|---|---|
| Parcul A1 | Primitiv/sad | 0,28 | 0,42 | 0,58 | 0,41 | 0,31 | Stres moderat pe ac |
| Parcul A2 | Primitiv/sad | 0,22 | 0,36 | 0,47 | 0,29 | 0,24 | Stres sever prin acul |
| Parcul B1 | Primitiv/contraspalier | 0,31 | 0,53 | 0,66 | 0,57 | 0,44 | Fără alerte |
| Parcul B2 | Primitiv/contraspalier | 0,29 | 0,49 | 0,63 | 0,52 | 0,40 | Fără alerte |
| Parcul C1 | Negroamaro/copac | 0,25 | 0,44 | 0,55 | 0,38 | 0,28 | Stres moderat pe ac |
| Parcul C2 | Negroamaro/contraspalliera | 0,30 | 0,51 | 0,61 | 0,50 | 0,38 | Fără alerte |
Modelul care a apărut este clar și semnificativ din punct de vedere agronomic: parcelele de puieți fără irigarea de urgență arată un stres de apă mai marcat în august-septembrie comparativ cu parcelele contraspalier cu sistem de picurare. Plot A2, în special, are a arătat o scădere a NDVI de la 0,47 la 0,29 între iulie și august 2024, corelat temporal cu o secvență de 23 de zile fără precipitații și temperaturi maxime peste 36 de grade.
Script de alertă automată
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from typing import List
def check_ndvi_alerts(
gdf_ndvi: gpd.GeoDataFrame,
stress_threshold: float = 0.35,
min_valid_pixels_pct: float = 70.0,
) -> List[dict]:
"""
Controlla le parcelle con NDVI sotto soglia di stress.
Genera lista di alert per notifica agronomo.
Args:
gdf_ndvi: GeoDataFrame con colonna ndvi_mean e pixel_valid_pct
stress_threshold: Soglia NDVI sotto cui scattare alert (default 0.35 per vite)
min_valid_pixels_pct: Percentuale minima pixel validi per affidabilità
Returns:
Lista di dict con dettagli alert per parcella
"""
alerts = []
for _, row in gdf_ndvi.iterrows():
if row["pixel_valid_pct"] < min_valid_pixels_pct:
continue # Troppi pixel nuvolosi, dato inaffidabile
if row["ndvi_mean"] < stress_threshold:
severity = "SEVERO" if row["ndvi_mean"] < 0.25 else "MODERATO"
alerts.append({
"id_parcella": row["id_parcella"],
"varieta": row.get("varieta", "N/A"),
"ettari": row.get("ettari", 0),
"ndvi_corrente": round(row["ndvi_mean"], 3),
"ndvi_soglia": stress_threshold,
"severity": severity,
"data": row.get("acquisition_date", "N/A"),
"pixel_valid": round(row["pixel_valid_pct"], 1),
})
return sorted(alerts, key=lambda x: x["ndvi_corrente"])
def send_alert_email(
alerts: List[dict],
recipient: str,
sender: str,
smtp_host: str = "smtp.gmail.com",
smtp_port: int = 587,
) -> None:
"""Invia email di alert NDVI all'agronomo."""
if not alerts:
return
body_lines = [
f"Alert Monitoraggio NDVI Satellite - {alerts[0]['data']}\n",
f"Parcelle in stress idrico: {len(alerts)}\n",
"=" * 60,
"",
]
for a in alerts:
body_lines.append(
f" [{a['severity']}] Parcella {a['id_parcella']} ({a['varieta']}, {a['ettari']} ha)\n"
f" NDVI: {a['ndvi_corrente']} (soglia: {a['ndvi_soglia']}) | Pixel validi: {a['pixel_valid']}%\n"
)
msg = MIMEMultipart()
msg["Subject"] = f"[ALERT NDVI] {len(alerts)} parcelle in stress - {alerts[0]['data']}"
msg["From"] = sender
msg["To"] = recipient
msg.attach(MIMEText("\n".join(body_lines), "plain"))
smtp_password = os.getenv("SMTP_PASSWORD")
with smtplib.SMTP(smtp_host, smtp_port) as server:
server.starttls()
server.login(sender, smtp_password)
server.send_message(msg)
print(f"Alert inviato a {recipient} per {len(alerts)} parcelle in stress")ROI al sistemului de monitorizare prin satelit
Cuantificarea rentabilității investiției în monitorizarea prin satelit este fundamentală pentru justifica investitia tehnologica catre firmele viticole. Pentru via de la Manduria, analiza cost-beneficiu pentru campania 2024 arată:
Analiza ROI - Sistem de monitorizare prin satelit Vigna Manduria (35 ha, campanie 2024)
| Voce | Valoare | Note |
|---|---|---|
| COSTURI | ||
| Acces la date Sentinel-2 (CDSE) | 0 EUR | Date deschise gratuite |
| Open-Weather Weather API | 0 EUR | Gratuit non-comercial |
| Dezvoltarea conductei Python | 2.500 EUR | 30 de ore de dezvoltator x 83 EUR/h (o singură dată, amortizat pe 3 ani) |
| Găzduire în cloud (VM + stocare) | 600 EUR/an | VM 4 vCPU + 50 GB stocare S3 |
| Consultanta agronomica de integrare | 1.000 EUR | Calibrare unică a pragului specifică culturii |
| Costul total pentru anul 1 | 4.100 EUR | Anii următori: ~1.400 EUR/an |
| BENEFICII ESTIMATE (35 ha, 100.000 sticle/an) | ||
| Reducerea irigației (optimizarea timpului) | 2.800 EUR | 15% economie de apă, 8.500 m3 x 0,33 EUR/m3 |
| Reducerea tratamentelor fitosanitare | 1.750 EUR | Intervenție direcționată numai în zonele de risc (-20% tratamente) |
| Îmbunătățirea calității strugurilor | 8.500 EUR | +0,5 puncte grad mediu Brix pe 12 ha, +0,20 EUR/kg premium de calitate |
| Reducerea pierderilor din cauza stresului hidric | 3.200 EUR | Recuperare de 8% a randamentului în 2 parcele la stres sever așteptat |
| Beneficii totale anul 1 | 16.250 EUR | |
| Rentabilitatea investiției în anul 1 | 296% | Rambursare in aproximativ 3 luni |
Google Earth Engine: procesare în cloud pentru analiză la scară largă
Pentru analiză la scară regională sau națională (de exemplu, cartografierea varietăților la nivel de raion DOC, evaluarea daunelor cauzate de înghețurile de primăvară la scară provincială, monitorizare al PAC pentru agenții de plăți precum AGEA), Google Earth Engine (GEE) oferă capacitate de calcul cloud-native care altfel ar necesita infrastructuri HPC dedicate.
import ee
# Autenticazione Google Earth Engine (richiede account GEE)
ee.Authenticate()
ee.Initialize(project="your-gee-project-id")
def calculate_ndvi_gee(
aoi_geojson: dict,
start_date: str,
end_date: str,
cloud_threshold: int = 20,
) -> ee.ImageCollection:
"""
Calcola NDVI su larga scala usando Google Earth Engine.
Adatto per analisi su comprensori DOC o scala regionale.
Args:
aoi_geojson: GeoJSON dell'area di interesse
start_date: "YYYY-MM-DD"
end_date: "YYYY-MM-DD"
cloud_threshold: Percentuale max copertura nuvolosa (0-100)
Returns:
ImageCollection con NDVI calcolato per ogni scena
"""
aoi = ee.Geometry(aoi_geojson)
def mask_s2_clouds(image: ee.Image) -> ee.Image:
"""Maschera nuvole usando QA60 band di Sentinel-2."""
qa = image.select("QA60")
cloud_bit_mask = 1 << 10 # bit 10: nuvole opache
cirrus_bit_mask = 1 << 11 # bit 11: nuvole cirro
mask = qa.bitwiseAnd(cloud_bit_mask).eq(0).And(
qa.bitwiseAnd(cirrus_bit_mask).eq(0))
return image.updateMask(mask).divide(10000) # scala a [0,1]
def add_ndvi(image: ee.Image) -> ee.Image:
"""Aggiunge banda NDVI all'immagine Sentinel-2."""
ndvi = image.normalizedDifference(["B8", "B4"]).rename("NDVI")
return image.addBands(ndvi)
collection = (
ee.ImageCollection("COPERNICUS/S2_SR_HARMONIZED")
.filterBounds(aoi)
.filterDate(start_date, end_date)
.filter(ee.Filter.lt("CLOUDY_PIXEL_PERCENTAGE", cloud_threshold))
.map(mask_s2_clouds)
.map(add_ndvi)
)
return collection
# Calcola NDVI mediano per stagione (mosaico cloud-free)
def get_seasonal_ndvi_mosaic(
collection: ee.ImageCollection,
aoi: ee.Geometry,
) -> ee.Image:
"""Crea mosaico NDVI stagionale (mediano = robusto a outlier residui)."""
ndvi_mosaic = collection.select("NDVI").median().clip(aoi)
return ndvi_mosaic
# Esempio: Comprensorio DOC Primitivo di Manduria (~21.000 ha)
aoi_doc_manduria = {
"type": "Polygon",
"coordinates": [[
[17.2, 40.2], [17.8, 40.2], [17.8, 40.6],
[17.2, 40.6], [17.2, 40.2],
]],
}
collection_estate = calculate_ndvi_gee(
aoi_geojson=aoi_doc_manduria,
start_date="2024-07-01",
end_date="2024-08-31",
cloud_threshold=15,
)
print(f"Scene disponibili: {collection_estate.size().getInfo()}")
# Export NDVI mosaic su Google Drive (per analisi successive in Python locale)
ndvi_mosaic = get_seasonal_ndvi_mosaic(collection_estate, ee.Geometry(aoi_doc_manduria))
task = ee.batch.Export.image.toDrive(
image=ndvi_mosaic,
description="NDVI_DOC_Manduria_Estate2024",
folder="AgriTech_Export",
fileNamePrefix="ndvi_manduria_estate2024",
region=ee.Geometry(aoi_doc_manduria),
scale=10,
crs="EPSG:4326",
maxPixels=1e10,
)
task.start()
print(f"Export avviato - ID task: {task.id}")Reglementări și date deschise pentru AgriTech în Italia
Ecosistemul de date deschise agricole din Italia este structurat în jurul a trei actori principali: AGEA (Agenția de Plăți pentru Agricultură), the SIAN (Sistemul Național de Informații Agricole) și portalurile regionale conectate. În februarie 2025, AGEA a finalizat migrarea SIAN către Polul Strategic Național, actualizându-se infrastructura cloud și extinderea capacităților de procesare a datelor geospațiale.
Principalele surse de date agricole deschise italiene
| Sursă | URL | Date disponibile | Format | Actualizare |
|---|---|---|---|---|
| Date deschise SIAN | data.sian.it | Ortofotografii de înaltă rezoluție, Hartă de utilizare a terenurilor (pe baza AI), Planuri grafice corporative | GeoTIFF, Shapefile, WFS | Anual |
| Servicii AGEA EO | sian.it | Ortofotografii multispectrale, imagini radar în bandă X, DEM, DSM | GeoTIFF, ECW | Periodic (campanie) |
| ISTAT - Statistica Agricolă | istat.it/agricoltura | SAU, producție, structura companiei, recensământ agricol | CSV, JSON, SDMX | Anual |
| Geoportalul Național MATTM | geoportale.gov.it | Carta Natura, Corine Land Cover, IDT, DBT | WMS, WFS, Shapefile | Variabilă |
| Regional ARPA (15 regiuni) | harpă.[regiune].it | Datele statiilor meteoclimatice, modele regionale de prognoza | CSV, JSON, NetCDF | Aproape în timp real |
| Serviciul funciar Copernicus | teren.copernicus.eu | CORINE Acoperire Teren, Pajiști, Apă și Umiditate, Atlas Urban | GeoTIFF, Shapefile | Trienal/anual |
Tranziția 5.0 a creditelor fiscale PNRR pentru AgriTech
Investițiile în sisteme de monitorizare prin satelit și agricultura de precizie pot se califică pentru beneficiile Planului de tranziție 5.0 (Legea 207/2024 - Legea bugetului 2025). Afacerile agricole, admise din 2025, pot accesa credite fiscale pentru investiții în bunuri de capital corporale și necorporale 4.0, inclusiv software de management al afacerilor cu componente de inteligență artificială și analiza datelor geospațiale.
Tranziția 5.0 Ratele creditelor fiscale pentru întreprinderile agricole (2025)
| Gama de investiții | Economii de energie 3-6% | Economii 6-10% | Economii >10% |
|---|---|---|---|
| Până la 2,5 milioane EUR | 35% | 40% | 45% |
| De la 2,5 milioane la 10 milioane EUR | 15% | 20% | 25% |
| De la 10 milioane la 50 milioane EUR | 5% | 10% | 15% |
Sistemele de irigare de precizie conduse de NDVI și prognozele meteo se pot califica pentru banda „economisire energie >6%” dacă documentează reducerea consumului de apă (energie de pompare) în comparație cu linia de bază. Solicitare expertiza tehnica jurata.
Arhitectura de producție: bune practici și anti-modele
După descrierea componentelor, este important să rezumați cele mai bune practici pentru o implementare robustă în producție și să identifice anti-modele comune care duce la sisteme fragile sau inexacte.
Cele mai bune practici - Sistemul de monitorizare prin satelit AgriTech
- Filtrare progresivă în cloud: Nu utilizați doar procentul de acoperire în cloud al produsului (metadatele de nivel 1). Aplicați întotdeauna mascarea la nivelul pixelului cu QA60 (Sentinel-2) sau BQA (Landsat). Un produs cloud 5% poate avea un anumit complot complet ascuns.
- GeoTIFF optimizat pentru cloud (COG): Stocați rasterele NDVI în format COG pentru acces eficient la cererea de interval de la S3/MinIO. Evitați GeoTIFF-urile clasice care nu sunt optimizate pentru stocarea în cloud.
- Proiecție consistentă: Standardizați TOTUL la EPSG:4326 (WGS84) pentru datele finale sau la zona UTM 32N (EPG:32632) pentru Italia pentru a menține măsurători metrice corecte. Nu amestecați niciodată CRS fără reproiectare explicită.
- Validarea datelor cu adevărul de teren IoT: Calibrați pragurile NDVI cu măsurători pe teren (senzori de umiditate a solului, contor portabil LAI, date agronomice colectate). Numai NDVI fără calibrare locală poate da alerte fals pozitive/negative de 15-30%.
- Gestionarea arhivei timpului: Mențineți cel puțin 3-5 ani de serie de timp NDVI pentru fiecare parcelă. Anomalia NDVI comparativ cu media istorică (aceeași lună, aceeași cultură) este mult mai informativă decât valoarea absolută.
- API pentru limitarea ratei și stocarea în cache: CDSE și Open-Meteo au limite de rată. Implementați întotdeauna cache-ul local (bazat pe fișiere sau Redis) pentru a evita re-descărcările inutile. O descărcare Sentinel-2 durează 30-120 de secunde: nu solicitați aceleași date de două ori.
- Înregistrare și pistă de audit: Fiecare calcul NDVI trebuie să fie urmărit până la scena satelitului sursă, data achiziției, procentul de acoperire a norilor, versiunea algoritmului. Fundamental pentru auditurile PAC/AGEA și pentru depanarea anomaliilor.
Anti-modele comune - Sisteme de satelit AgriTech
- Ignorând corecția atmosferică: Folosirea Level-1C (top-of-atmosphere) în loc de Level-2A (bottom-of-atmosphere) introduce erori sistematice în NDVI de ordinul 10-20% în condiții de umiditate atmosferică ridicată (ceață din Valea Po, Marea Adriatică). ALWAYS use L2A for quantitative applications.
- NDVI ca singur indicator: Bazați-vă numai pe NDVI, ignorând EVI (pentru vegetație densă), NDWI (stresul de apă), Red Edge NDVI (stresul timpuriu înainte de a fi vizibil pe NDVI). Un sistem profesional folosește cel puțin 3-4 indici în combinație.
- Dimensiunea pixelilor și parcela: Aplicarea NDVI pe parcele mai mici de 0,3 ha cu pixeli la 10 m produce medii nesigure din punct de vedere statistic (mai puțin de 30 de pixeli). Pentru parcele mici, luați în considerare Planet Labs (3,7 m) sau senzori de drone.
- Lipsa compoziției temporale: Luarea unei singure achiziții disponibile pentru o dată, chiar și la 19% acoperire cu nori, duce la artefacte NDVI în zonele de nor parțial. Utilizați întotdeauna compoziția medoid pe o fereastră de timp de 10-15 zile.
- Conductă fără versiunea datelor: Recalcularea NDVI cu versiuni diferite de Python/rasterio/sentinelhub produce valori ușor diferite pentru aceleași date sursă. Versiune conductele de calcul cu DVC sau echivalent.
Conductă completă: orchestrație cu prefectul
Prin integrarea tuturor componentelor descrise, conducta completă de urmărire prin satelit poate fi orchestrat cu Prefect (vezi seria Pipeline Orchestration) să ruleze automat la fiecare 5 zile (la rata de revizuire Sentinel-2).
from prefect import flow, task
from prefect.schedules import CronSchedule
from datetime import datetime, timedelta
import logging
logger = logging.getLogger(__name__)
@task(retries=3, retry_delay_seconds=60, name="search-sentinel2-products")
def search_products_task(bbox: tuple, lookback_days: int = 7) -> list[dict]:
"""Task Prefect: ricerca prodotti Sentinel-2 recenti."""
end_date = datetime.utcnow()
start_date = end_date - timedelta(days=lookback_days)
products = search_sentinel2_products(
bbox=bbox,
start_date=start_date.strftime("%Y-%m-%dT00:00:00.000Z"),
end_date=end_date.strftime("%Y-%m-%dT23:59:59.000Z"),
cloud_cover_max=20.0,
)
logger.info(f"Trovati {len(products)} prodotti Sentinel-2")
return products
@task(retries=2, retry_delay_seconds=120, name="download-compute-ndvi")
def compute_ndvi_task(
product: dict,
aoi_bbox: tuple,
output_dir: str = "/data/ndvi",
) -> str:
"""Task Prefect: scarica bande e calcola NDVI per un prodotto."""
# Implementazione basata su funzioni definite nelle sezioni precedenti
acquisition_date = product["ContentDate"]["Start"][:10]
output_path = f"{output_dir}/ndvi_{acquisition_date}.tif"
# Download + calcolo (codice omesso per brevita, usa funzioni precedenti)
logger.info(f"NDVI calcolato per {acquisition_date}: {output_path}")
return output_path
@task(name="zonal-stats-postgis")
def zonal_stats_task(ndvi_path: str, parcelle_shapefile: str) -> dict:
"""Task Prefect: calcola statistiche zonali e salva in PostGIS."""
gdf_parcelle = gpd.read_file(parcelle_shapefile).to_crs("EPSG:4326")
acquisition_date = ndvi_path.split("ndvi_")[1].replace(".tif", "")
gdf_ndvi = compute_zonal_ndvi(ndvi_path, gdf_parcelle, acquisition_date)
engine = create_engine(os.getenv("POSTGIS_URL"))
gdf_ndvi.to_postgis("ndvi_storico", engine, if_exists="append", index=False)
alerts = check_ndvi_alerts(gdf_ndvi)
return {"parcelle_processate": len(gdf_ndvi), "alerts": len(alerts), "alert_details": alerts}
@task(name="send-alerts")
def alerts_task(stats: dict) -> None:
"""Task Prefect: invia alert email se presenti."""
if stats["alerts"] > 0:
send_alert_email(
alerts=stats["alert_details"],
recipient=os.getenv("AGRONOMO_EMAIL"),
sender=os.getenv("ALERT_EMAIL"),
)
logger.info(f"Alert inviati per {stats['alerts']} parcelle")
@flow(
name="satellite-monitoring-pipeline",
schedule=CronSchedule(cron="0 8 */5 * *"), # ogni 5 giorni alle 8:00
)
def satellite_monitoring_flow(
bbox: tuple = (17.35, 40.28, 17.60, 40.50),
parcelle_shapefile: str = "/data/parcelle_vigna.shp",
) -> None:
"""
Pipeline completa di monitoraggio satellite per vigna.
Si esegue ogni 5 giorni, allineata alla frequenza Sentinel-2.
"""
products = search_products_task(bbox=bbox)
if not products:
logger.warning("Nessun prodotto trovato nell'ultimo periodo")
return
# Processo il prodotto più recente con minima copertura nuvolosa
best_product = sorted(
products,
key=lambda p: next(
(a["Value"] for a in p.get("Attributes", []) if a["Name"] == "cloudCover"), 100
),
)[0]
ndvi_path = compute_ndvi_task(best_product, bbox)
stats = zonal_stats_task(ndvi_path, parcelle_shapefile)
alerts_task(stats)
logger.info(
f"Pipeline completata: {stats['parcelle_processate']} parcelle, {stats['alerts']} alert"
)
if __name__ == "__main__":
satellite_monitoring_flow()Concluzii și pașii următori
Datele satelitului Sentinel-2, accesibile gratuit prin CDSE, reprezintă probabil cea mai subutilizată sursă de date din AgriTech italian. Cu Python, rasterio, sentinelhub și un cont CDSE, este posibil să construiți un Sistem de monitorizare NDVI care depășește multe soluții în calitate și granularitate comercial contra cost.
Cheia succesului nu este sofisticarea tehnologică, ci calibrare agronomică locală: praguri NDVI, ferestre de timp de alertă, greutăți dintre caracteristicile modelului predictiv trebuie calibrate la cultura specifică, asupra climatului local și practicilor agronomice ale companiei. Un sistem calibrat la viță de vie din Puglia nu funcționează la fel de bine și la grâul din Valea Po fără el intervenție de calibrare umană.
Punctele cheie de reținut pentru cei care doresc să implementeze un sistem similar:
- Utilizați întotdeauna Sentinel-2 Nivel-2A (corecția atmosferică aplicată) pentru comparații cantitative între date diferite.
- Combinați NDVI cu cel puțin un indice complementar (EVI pentru podgorii dense, NDWI pentru stresul de apă timpurie, Red Edge NDVI pentru monitorizarea clorofilei).
- Integra Open-Weather ERA5-Land pentru bilanțul apei: ET0 + precipitații + umiditatea solului din model și un proxy excelent când nu ai senzori IoT în teren.
- Păstrează unul serie istorică de minim 3 ani pentru fiecare parcelă: anomalia față de media istorică este mult mai utilă decât valoarea absolută NDVI.
- Scala cu Google Earth Engine numai atunci când trebuie să procesați la scară regională sau națională: pentru companiile individuale, conducta Python locală pe CDSE este mai controlabilă și fără blocare.
Seria FoodTech continuă
Ați construit conducta prin satelit. Acum explorați celelalte articole din serie pentru completați arhitectura AgriTech:
- Articolul 1: Conductă IoT pentru agricultura de precizie cu Python și MQTT - Cum se integrează senzorii de sol cu conducta de satelit.
- Articolul 2: ML Edge pentru monitorizarea culturilor: computer Vision in the Fields - Modele de computer vision pentru depistarea precoce a bolilor plantelor.
- Articolul 4: Trasabilitatea blockchain în alimente: de la câmp la supermarket - Cum datele NDVI alimentează certificatele de calitate în lanț.
Pentru a afla mai multe despre MLOps și implementarea modelelor predictive descrise în acest articol, vezi serialul MLOps pentru afaceri cu MLflow si serialul LLM în afaceri: RAG Enterprise.