AgriTech용 위성 및 날씨 API: Sentinel-2 및 Python을 사용한 예측 데이터
2024년에 유럽우주국은 1.5페타바이트의 데이터 매일 코페르니쿠스 프로그램의 센티넬 위성에서. 이 중 농업 모니터링에 전념하는 부분 유럽에서는 가장 크고 전략적으로 관련성이 높은 부문을 나타냅니다. 그래도 대다수는 가장 기술적으로 진보된 기업을 포함하여 이탈리아 농업 기업 중 의 이 공개적이고 무료이며 과학적으로 검증된 인프라입니다.
그 이유는 데이터가 부족해서가 아니라 데이터에 액세스하고 처리하고 변환하는 과정이 복잡하기 때문입니다. 구체적인 농업적 결정. Sentinel-2 위성은 5일마다 포도밭 위를 지나갑니다. 픽셀당 10미터의 해상도. 각 패스는 건강을 인코딩하는 13개의 스펙트럼 밴드를 생성합니다. 식물, 수분 스트레스, 엽록소 함량, 기생충의 존재. 데이터와 결합 날씨, IoT 지상 센서 및 예측 모델 등 이러한 데이터는 근본적으로 변화할 수 있습니다. 농업적 결정의 질을 높여 투입 비용을 15-25% 줄이고 수확량을 높입니다. 최대 10-20%.
전 세계 지구 관측 시장은 그만한 가치가 있습니다 2025년에는 100억 7천만 달러 Straits Research에 따르면 2033년까지 172억 달러(CAGR 6.92%)의 성장이 예상됩니다. 농업 부문은 애플리케이션의 21%를 차지하며 가장 높은 비율로 성장하고 있습니다. 정밀 농업, 수확량 모니터링 및 관개 최적화에 대한 수요에 의해 주도됩니다. 이 기사에서는 Sentinel-2에 액세스하기 위한 완전한 Python 파이프라인을 단계별로 구축합니다. CDSE(Copernicus Data Space Ecosystem)를 통해 NDVI 및 기타 식생 지수를 계산하고 통합 Open-Meteo의 기상 데이터를 활용하고 물 스트레스에 대한 예측 모델을 제공합니다.
이 기사에서 배울 내용
- Copernicus 프로그램의 아키텍처 및 CDSE를 통해 Sentinel-2에 무료로 액세스하는 방법
- 식생 지수: NDVI, EVI, SAVI, LAI - 이탈리아 작물에 대한 수학 공식, 해석 및 임계값
- 완전한 Python 구현: CDSE 인증, 밴드 다운로드, rasterio 및 numpy를 사용한 NDVI 계산
- 비교된 날씨 API: Open-Meteo, OpenWeatherMap, Tomorrow.io 및 위성 데이터 통합
- 벡터 및 래스터 데이터용 GeoPandas, rasterio 및 PostGIS를 사용한 지리공간 파이프라인
- 물 스트레스 예측 모델: Sentinel-2 + 날씨 + IoT를 scikit-learn과 결합
- 실제 사례 연구: Puglia의 포도원, 계절별 NDVI 모니터링, 관개 최적화
- 이탈리아 법률: AGEA, SIAN, 개방형 데이터 CAP 및 디지털 CAP
FoodTech 시리즈 - 모든 기사
| # | Articolo | 수준 | 상태 |
|---|---|---|---|
| 1 | Python 및 MQTT를 사용한 정밀 농업용 IoT 파이프라인 | 고급의 | 사용 가능 |
| 2 | 작물 모니터링을 위한 ML Edge: 현장의 컴퓨터 비전 | 고급의 | 사용 가능 |
| 3 | AgriTech용 위성 및 날씨 API: 예측 데이터(현재 위치) | 고급의 | 현재의 |
| 4 | 식품의 블록체인 추적성: 현장에서 슈퍼마켓까지 | 중급 | 곧 출시 예정 |
| 5 | 식품 산업의 품질 관리를 위한 컴퓨터 비전 | 고급의 | 곧 출시 예정 |
| 6 | FSMA 및 디지털 규정 준수: 규제 프로세스 자동화 | 중급 | 곧 출시 예정 |
| 7 | 수직 농업: IoT 및 ML을 통한 환경 제어 | 고급의 | 곧 출시 예정 |
| 8 | Prophet 및 LightGBM을 사용한 식품 소매 수요 예측 | 중급 | 곧 출시 예정 |
| 9 | Farm Intelligence 대시보드: Grafana를 사용한 실시간 분석 | 중급 | 곧 출시 예정 |
| 10 | 공급망 식품 최적화: 폐기물 감소를 위한 ML | 중급 | 곧 출시 예정 |
코페르니쿠스 프로그램: 지구 관측을 위한 유럽 인프라
유럽연합의 코페르니쿠스 프로그램은 세계 최대의 관측 인프라입니다. 땅은 건설되지 않았습니다. ESA(유럽 우주국)와 유럽 위원회가 관리합니다. EUMETSAT의 운영 지원, Copernicus는 명명된 위성 집합을 운영합니다. "Sentinel"은 각각 특정 모니터링 임무를 위해 설계되었습니다. 농업의 경우, 절대 참조 위성 e 센티넬-2.
Sentinel-2 별자리는 두 개의 쌍둥이 위성(Sentinel-2A 및 Sentinel-2B)으로 구성됩니다. 고도 786km의 태양 동기 궤도. 두 개의 위성 구성은 다음을 보장합니다. 시간 5일 리뷰 적도에서는 2~3일, 유럽 위도에서는 2~3일(이탈리아) 포함). 각 위성에는 푸시브룸인 MSI(MultiSpectral Instrument) 센서가 탑재되어 있습니다. 스와이프 폭 290km, 13개 스펙트럼 밴드로 데이터를 수집하는 스캐너 10미터, 20미터, 60미터의 세 가지 공간 해상도로 분산됩니다.
Sentinel-2의 가장 전략적인 특징은 완전 무료: 2014년부터 모든 코페르니쿠스 데이터는 상업적 또는 비상업적 용도에 관계없이 공개적으로 액세스할 수 있습니다. 상업용. 2023년부터 새로운 액세스 포털과 코페르니쿠스 데이터 공간 생태계 (CDSE), 이는 2009년에 폐기된 이전 Copernicus Open Access Hub(SciHub)를 대체했습니다. 2023. CDSE는 획득 대기 시간을 포함하여 전체 Sentinel-2 아카이브에 대한 즉각적인 액세스를 제공합니다. 가장 최근 이미지의 경우 약 3시간 정도 소요됩니다.
위성 Sentinel-2: MSI 센서 기술 사양
| Banda | Lunghezza d'onda (nm) | Risoluzione | Applicazione Principale |
|---|---|---|---|
| B1 - Coastal aerosol | 443 | 60 m | qualità aria, correzione atmosferica |
| B2 - Blue | 490 | 10 m | Discriminazione vegetazione/suolo, mappatura |
| B3 - Green | 560 | 10 m | Vigor vegetazione, colore fogliare |
| B4 - Red | 665 | 10 m | Clorofilla, NDVI (banda rossa) |
| B5 - Vegetation Red Edge | 705 | 20 m | Stress vegetazione, Red Edge NDVI |
| B6 - Vegetation Red Edge | 740 | 20 m | Contenuto clorofilla, classificazione specie |
| B7 - Vegetation Red Edge | 783 | 20 m | Biomassa, LAI |
| B8 - NIR | 842 | 10 m | NDVI (banda NIR), biomassa, LAI |
| B8a - Vegetation Red Edge | 865 | 20 m | Canopy structure, NDVI narrow |
| B9 - Water vapour | 940 | 60 m | Correzione vapore acqueo atmosferico |
| B10 - SWIR - Cirrus | 1375 | 60 m | Rilevamento nubi cirro |
| B11 - SWIR | 1610 | 20m | 수분 스트레스, 잎 수분 함량 |
| B12 - SWIR | 2190 | 20m | 고도의 수분 스트레스, 노화 |
농업과 관련된 Sentinel-2 데이터 처리에는 주로 두 가지 수준이 있습니다. 레벨-1C(L1C) - 기하학적 보정을 통한 최고 대기 복사량, e 레벨-2A(L2A) - 대기층의 반사율 (Bottom of Atmosphere, BOA) Sen2Cor 알고리즘을 통해 대기 보정이 적용됩니다. 직접 농업에 적용하려면, Level-2A는 대기의 영향이 이미 보정되어 있어 권장되는 수준입니다. 서로 다른 날짜와 서로 다른 지리적 위치 간의 비교 가능한 반사율 값.
AgriTech용 위성 플랫폼: 전체 비교 2025
Sentinel-2가 사용 가능한 유일한 옵션은 아닙니다. 농업용 위성 데이터 시장 Planet Labs와 같은 상용 제공업체, Landsat(NASA/USGS)과 같은 기존 업체 및 Google Earth Engine과 같은 통합 클라우드 플랫폼. 선택은 복잡한 균형에 달려 있습니다 공간 해상도, 획득 빈도, 관리되는 클라우드 범위 및 사용 가능한 예산 간.
정밀 농업을 위한 위성 플랫폼 비교 - 2025년
| 플랫폼 | 공간 해상도 | 재방문 빈도 | 스펙트럼 밴드 | 비용 | 데이터 대기 시간 | 아피스 |
|---|---|---|---|---|---|---|
| 센티넬-2(ESA/코페르니쿠스) | 10m(가시/NIR), 20m(SWIR) | 5일(EU의 경우 2~3일) | MSI 밴드 13개 | 무료(오픈데이터) | 획득 후 ~3시간 | CDSE OData, STAC, SentinelHub, OpenEO |
| 행성 PlanetScope | 3.7m | 매일(준일일) | 8밴드(PS2.SD) | 상업용 구독; 연구를 위해 무료 | 24시간 | 플래닛 API v1, 구독 API |
| Landsat 8/9(NASA/USGS) | 30m(다중 스펙트럼), 15m(팬) | 16일 | 11개 밴드 OLI/TIRS | 무료(오픈데이터) | 24~48시간 | USGS EarthExplorer, Google Earth Engine |
| MODIS(NASA 지구/물) | 250m / 500m / 1km | 1~2일 | 36개 밴드 | 무료(오픈데이터) | 24~48시간 | NASA EarthData STAC, Google Earth Engine |
| Google 어스 엔진 | 데이터 세트에 따라 다름(10m-1km) | 데이터 세트에 따라 다릅니다. | 통합된 다중 데이터 세트 | 무료 비상업적; 유료 상업용 | 즉각적인 클라우드 처리 | 파이썬 ee, 자바스크립트 API |
| 맥사 월드뷰-3 | 0.3m(팬), 1.24m(멀티) | 1~4일 | 29 밴드(CAVIS) | ~25-40 USD/km2 | 4~8시간 | Maxar 스트리밍 API |
| 에어버스 플레이아데스 네오 | 0.3m | 1~2일 | 6개의 다중 스펙트럼 대역 | ~15-30 EUR/km2 | 24~48시간 | 원아틀라스 API |
이탈리아의 대다수 농업 응용 분야에서는 Sentinel-2와 선택 최적의: 0.5헥타르보다 큰 부지에는 10미터의 해상도로 충분합니다. (픽셀 통계가 신뢰할 수 없게 되는 임계값 미만), 재방문 빈도 이탈리아에서는 2~3일 정도 소요되며 계절별 모니터링에 적합하며 완전 무료입니다. 채택에 대한 경제적 장벽을 제거합니다. Planet Labs는 다음과 같은 경우에만 관련됩니다. 거의 매일 모니터링이 필요한 애플리케이션(예: 조기 발병 감지) 집약적인 과일 및 채소와 같은 고부가가치 작물의 수분 스트레스) 또는 10m 미만의 해상도. MODIS와 Landsat는 넓은 지역과 수십 년 단위의 시계열 분석에 여전히 유용합니다.
식생 지수: 이탈리아 작물에 대한 NDVI, EVI, SAVI 및 LAI
식생 지수(VI)는 다음에서 수학적으로 파생된 측정항목입니다. 특정 식생 특성을 정량화하는 스펙트럼 밴드의 조합. 그들은 착취한다 엽록소는 적색밴드(665 nm)에서 강하게 흡수하고 강하게 반사한다는 사실 근적외선(842nm): 이 대역 사이의 비율은 에너지의 밀도와 활력을 나타냅니다. 수학적 우아함을 지닌 식물.
NDVI - 정규화된 차이 식생 지수
NDVI는 Rouse 등이 소개한 세계에서 가장 많이 사용되는 식생 지수입니다. 1973년에. 수학 공식은 다음과 같습니다.
NDVI 공식
NDVI = (NIR - Red) / (NIR + Red)
Su Sentinel-2:
NDVI = (B8 - B4) / (B8 + B4)
Range: da -1.0 a +1.0
NDVI 값은 표준화된 척도에 따라 해석되지만 최적의 임계값은 다양합니다. 작물 및 생리학적 단계별. 다음 표에는 다음에 대한 참조 임계값이 나와 있습니다. 과학 문헌 및 경험적 검증을 통해 얻은 주요 이탈리아 작물 일반적인 지역(Pianura Padana, Puglia, Sicily, Tuscany):
이탈리아 작물에 대한 NDVI 해석
| NDVI 범위 | 일반통역 | 밀(삼) | 포도나무(Vitis) | 토마토 (Solanum) | 옥수수(제아 메이스) | 올리브(올리아) |
|---|---|---|---|---|---|---|
| < 0.1 | 맨땅, 바위, 눈 | 심지 않은 땅 | 겨울, 싹트기 전 | 수확 후 | 파종 전 | 줄 사이의 맨땅 |
| 0.1 - 0.2 | 매우 드물거나 건조한 초목 | 초기 비상사태 | 휴면/심한 스트레스 | 최근 이식 | 비상(VE) | 심각한 물/열 스트레스 |
| 0.2 - 0.4 | 드문 식물, 적당한 스트레스 | 초기 경작 | 개화 전, 가벼운 스트레스 | 초기 영양 성장 | V1-V3 단계 | 열악한 식물, 스트레스 |
| 0.4 - 0.6 | 적당한 식물, 좋은 건강 | 가득 차있는 상승 | 개화전/과일 세트 | 활발한 식물 발달 | V4-V8 단계 | 정상적인 잎, 물이 잘 공급됨 |
| 0.6 - 0.8 | 식물이 빽빽하고 활력이 뛰어남 | 방향(계절성수기) | 전체 잎, 베레종 | 최대 적용 범위(개화) | V10-VT 단계(개화) | 잎이 빽빽하고 상태가 양호함 |
| > 0.8 | 매우 빽빽한 초목, 숲 | 희귀(필드 가장자리 숲) | 울타리와 국경 나무 | 전체 적용 범위를 갖춘 온실 | 드문 최적 조건 | 밀도가 높은 올리브 과수원 |
EVI - 식생 강화 지수
Huete 등이 개발한 EVI. MODIS 센서의 경우 해당 영역의 NDVI 한계를 수정합니다. 식생 밀도가 높고(NDVI 포화도가 0.7 이상) 토양이 노출된 지역 (지면 반사로 인해 발생하는 NDVI 오류). 공식은 다음과 같습니다.
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 - 토양 조정 식생 지수
1988년 Huete가 제안한 SAVI는 다음과 같은 반건조 환경에서 특히 유용합니다. 노출된 토양이 스펙트럼 반응에 큰 영향을 미치는 풀리아(Puglia) 또는 시칠리아(Sicily). 보정 계수 L은 지반의 영향을 줄입니다.
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 - 잎 면적 지수
LAI(잎 면적 지수)는 단위 표면적당 전체 잎 표면을 정량화합니다. 토양. 이는 잠재적인 생산성과 관련된 기본적인 농업적 매개변수입니다. 땀. Sentinel-2를 사용하면 경험적 또는 물리적 알고리즘을 사용하여 LAI를 추정할 수 있습니다. (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 massima위성 식생 지수의 한계
- 흐림: Sentinel-2 및 광학 - 클라우드로 인해 이미지를 사용할 수 없게 됩니다. 이탈리아의 평균 구름량은 겨울에 48%, 여름에 15%입니다. QA60 밴드는 자동 클라우드 마스킹을 허용합니다.
- NDVI 포화도: NDVI ~0.7-0.8 이상에서는 감도가 급격하게 감소합니다. 고밀도 작물의 경우 EVI를 사용하십시오.
- 픽셀 혼합: 10m 해상도에서 픽셀은 식물과 토양을 모두 포함할 수 있습니다. 작은 토지(< 0.5ha)의 경우 평균 NDVI는 대표성이 떨어집니다.
- 생리학적 계절성: 서로 다른 날짜 간의 NDVI를 비교하려면 절대 값뿐만 아니라 현상학적 단계도 고려해야 합니다.
- 연간 변동성: NDVI는 또한 농업 관행과 관계없이 계절적 기상 조건(온도, 강수량)에 따라 달라집니다.
Python 구현: CDSE 액세스 및 Sentinel-2 다운로드
Copernicus Data Space Ecosystem(CDSE)은 Sentinel 데이터에 액세스하기 위한 단일 포털이 되었습니다. 2023년부터. 네 가지 주요 API를 제공합니다. O데이터 API (상품 검색 및 다운로드), STAC API (SpatioTemporal 자산 카탈로그), 오픈EO API (처리 표준화) 전자 센티넬 허브 API (서비스로 처리). 시작하려면, dataspace.copernicus.eu에서 무료 계정을 만들고 OAuth2 자격 증명을 생성하세요.
Python 환경 설정 및 종속성
# 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 idricoOAuth2를 사용한 CDSE 인증
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)OData API를 통해 Sentinel-2 제품 검색
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]}")Sentinel Hub Python을 사용하여 밴드 다운로드 및 액세스
전체 장면을 다운로드하지 않고 밴드를 처리하는 경우(무게는 800MB - 1.2GB L2A 제품), 해당 API만 반환하는 Sentinel Hub Process API를 사용하는 것이 더 효율적입니다. 관심 영역에 필요한 밴드:
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}%)")NDVI 계산 및 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")AgriTech용 Weather API: 비교 및 구현
날씨 데이터는 정밀 농업의 두 번째 중요한 데이터 소스입니다. 현장의 NDVI 및 IoT 데이터와 통합되어 다음에 대한 예측 모델을 구축할 수 있습니다. 물 스트레스, 질병 위험, 증발산 및 수확량 예측. 파노라마 2025년 날씨 API 중 무료 오픈소스 제공업체와 상용 제공업체로 나뉜다. 이탈리아 ARPA의 무료 등급 및 지역 서비스를 제공합니다.
AgriTech용 날씨 API - 비교 2025
| 공급자 | 가격 | 공간 해상도 | 예측 | 농업 변수 | 역사 기록 보관소 | 아피스 |
|---|---|---|---|---|---|---|
| 개방형 날씨 | 무료(비상업) ~$15/월 상업용 | 1-11km(다중 모델) | 16일 | ET0, 토양온도, 토양수분, 증기압 | 1940년~현재(ERA5) | REST JSON, Python 클라이언트 |
| OpenWeather지도 | 분당 최대 60통화까지 무료입니다. $40/월 프로 | 2.5km(아이콘) | 5일(무료), 16일(프로) | 제한적, ET0 없음 | 유료 내역만 | REST JSON, Python SDK |
| 내일.io | 하루 500통화 무료; $199/월 코어 | 1km | 21일 | 토양수분, ET0, 분무 조건, 해충 위험 | 6년 | REST, 웹소켓, gRPC |
| ARPA 롬바르디아 / 피에몬테 / 베네토 | 무료(지역 오픈 데이터) | 시간 엄수 역(이탈리아 내 최대 1500개 역 네트워크) | 아니요(관찰만 가능) | 모든 농기상 변수 | 전체 아카이브(수십 년) | WMS, REST, CSV/JSON 다운로드 |
| 메테오암 / CFS/ECMWF | ECMWF 무료 공개 데이터; 무료 NOAA CFS | 9-25km | 10~45일 | 날씨 기반 변수 | ERA5를 통해 1940년부터 현재까지 | ECMWF API(파이썬), CDS API |
| 시각적 교차 | 무료 1000개 레코드/일; $35/월 표준 | 스테이션에서 보간됨 | 15일 | ET0, 성장온도일, 서리 위험 | 무제한 유료 | REST JSON, CSV, SQL과 유사 |
생산 중인 AgriTech 애플리케이션의 경우 다음을 사용하는 것이 좋습니다. 개방형 날씨 어떻게 기반(역사가를 위한 무료, 오픈 소스, ERA5 데이터, 16일 예측)을 로컬 교정을 위한 ARPA 스테이션 네트워크. Open-Meteo에는 농업기상학적 변수가 포함되어 있습니다. ET0(기준 증발산량), 여러 깊이의 토양 수분 및 증기와 같은 중요한 문제 상용 공급자로부터 항상 무료로 제공되는 것은 아닙니다.
캐시 및 재시도를 통한 Open-Meteo 클라이언트 구현
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")완전한 지리공간 파이프라인: GeoPandas, rasterio 및 PostGIS
전문적인 위성 데이터 관리에는 포괄적인 지리공간 인프라가 필요합니다. 래스터 데이터(NDVI 위성 이미지, 날씨 지도)를 벡터 데이터(경계)와 통합합니다. 플롯, 관리 영역, 샘플링 지점). 여기서 설명하는 파이프라인은 다음을 채택합니다. 지리 공간에 대한 사실상의 표준 Python 스택: 래스테리오 래스터의 경우, GeoPandas 벡터의 경우, PostGIS 공간 데이터베이스로 e 그달 기본 형식 번역 엔진으로 사용됩니다.
지리공간 파이프라인 아키텍처
AgriTech 지리공간 파이프라인 기술 스택
┌─────────────────────────────────────────────────────────────────────────┐
│ 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 │
└─────────────────────────────────────────────────────────────────────────┘
구역 통계: rasterstats를 사용한 플롯당 평균 NDVI
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)")고급 공간 분석을 위한 PostGIS 쿼리
-- 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;ML을 사용한 물 스트레스 예측 모델
위성 NDVI를 기상 및 IoT 데이터와 결합하면 예측 모델을 구축할 수 있습니다. 3~7일 전에 물 부족을 예방하여 적극적인 관개 계획을 세울 수 있습니다. 여기에 설명된 모델은 Sentinel-2의 NDVI 시계열, ET0와 강수량, IoT 센서(또는 물리적 센서를 사용할 수 없는 경우 Open-Meteo ERA5-Land).
모델의 특성 엔지니어링
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"]]사례 연구: Manduria의 Primitivo 포도원에서 NDVI 모니터링
설명된 개념을 구체적으로 만들기 위해 애플리케이션에서 영감을 얻은 사례 연구를 제시합니다. DOC Primitivo di Manduria (Taranto, Puglia)의 영토에 있습니다. 해당 지역에는 조건이 있습니다. 전형적인 지중해 포도재배: 덥고 건조한 여름(여름 강수량 < 50mm), 점토-석회암 토양, 카운터 에스팔리어 및 묘목 훈련 시스템.
사례 연구 매개변수 - Vigna Primitivo Manduria
| 매개변수 | Valore |
|---|---|
| 위치 | 만두리아(TA), 풀리아 - 위도 40.38, 경도 17.47 |
| 총 표면적 | 12개 부지로 나누어진 35헥타르 |
| 주요 품종 | 프리미티보(80%), 네그로아마로(20%) |
| 사육 시스템 | Apulian 묘목(8개), Counter-espalier(4개) |
| 공장 연도 | 2003-2015 (젊은/성인 혼합 포도원) |
| 관개 | 한 방울씩(모든 플롯이 아님) |
| 추적 위성 | Sentinel-2A/B, L2A, 10m 분해능 |
| 사용된 획득 빈도 | 2024년 시골 풍경 28개(구름 < 20%) |
| 통합 IoT 센서 | 6개의 기상 관측소, 30/60cm의 토양 습도 센서 18개 |
| 분석기간 | 2024년 4월~10월 |
계절별 NDVI 결과 및 관개 결정
계절별 NDVI 분석을 통해 상당히 차별화된 물 스트레스 패턴이 밝혀졌습니다. 관개 관리 및 제품 품질에 직접적인 영향을 미치는 플롯 간:
플롯 별 NDVI 결과 - 2024 캠페인
| 구성 | 다양성 | NDVI 4월 | NDVI 다운 | NDVI 7월(성수기) | NDVI 8월 | NDVI 세트 | 관개 경보 |
|---|---|---|---|---|---|---|---|
| 파크A1 | 원시/묘목 | 0.28 | 0.42 | 0.58 | 0.41 | 0.31 | 보통의 바늘 스트레스 |
| 파크 A2 | 원시/묘목 | 0.22 | 0.36 | 0.47 | 0.29 | 0.24 | 심한 바늘 세트 스트레스 |
| 파크B1 | 기본/카운터스페리어 | 0.31 | 0.53 | 0.66 | 0.57 | 0.44 | 알림 없음 |
| 파크B2 | 기본/카운터스페리어 | 0.29 | 0.49 | 0.63 | 0.52 | 0.40 | 알림 없음 |
| 파크C1 | 네그로아마로/나무 | 0.25 | 0.44 | 0.55 | 0.38 | 0.28 | 보통의 바늘 스트레스 |
| 파크 C2 | 네그로아마로/카운터팔리에라 | 0.30 | 0.51 | 0.61 | 0.50 | 0.38 | 알림 없음 |
나타난 패턴은 명확하고 농업학적으로 중요합니다. 비상 관개 시설은 8~9월에 더 심한 물 스트레스를 나타냅니다. 드립 시스템을 사용한 카운터 에스팔리어 플롯. 특히 플롯 A2에는 2024년 7월과 8월 사이에 NDVI가 0.47에서 0.29로 감소한 것으로 나타났습니다. 23일 동안 강수량이 없고 최고 기온이 36도를 넘었습니다.
자동 경고 스크립트
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
위성 모니터링의 투자 수익을 정량화하는 것은 기본입니다. 와인 재배 회사에 대한 기술 투자를 정당화합니다. 만두리아 포도원의 경우, 2024년 캠페인의 비용-이익 분석은 다음을 보여줍니다.
ROI 분석 - 위성 모니터링 시스템 Vigna Manduria(35ha, 2024년 캠페인)
| 목소리 | Valore | 메모 |
|---|---|---|
| 소송 비용 | ||
| Sentinel-2 데이터 액세스(CDSE) | 0유로 | 무료 오픈 데이터 |
| 개방형 날씨 API | 0유로 | 비상업적 무료 |
| Python 파이프라인 개발 | 2,500유로 | 30 개발자 시간 x 83 EUR/h(일회성, 3년에 걸쳐 분할 상환) |
| 클라우드 호스팅(VM + 스토리지) | 600유로/년 | VM 4 vCPU + 50GB S3 스토리지 |
| 통합 농업 컨설팅 | 1,000유로 | 일회성 작물별 임계값 교정 |
| 1년차의 총 비용 | 4,100유로 | 이후 연도: ~1,400 EUR/년 |
| 예상 이익(35ha, 100,000병/년) | ||
| 관개 감소(타이밍 최적화) | 2,800유로 | 15% 물 절약, 8,500m3 x 0.33 EUR/m3 |
| 식물위생 처리의 감소 | 1,750유로 | 위험 지역에만 표적 개입(-20% 치료) |
| 포도 품질 향상 | 8,500유로 | 12ha의 평균 Brix 등급 +0.5포인트, 품질 프리미엄 +0.20 EUR/kg |
| 물 스트레스로 인한 손실 감소 | 3,200유로 | 예상되는 심각한 스트레스에서 2개 플롯의 8% 수율 회복 |
| 1년차 총 혜택 | 16,250유로 | |
| 1년차 ROI | 296% | 약 3개월 만에 회수 |
Google Earth Engine: 대규모 분석을 위한 클라우드 처리
지역 또는 국가 규모의 분석(예: 품종 수준의 품종 매핑) DOC 지구, 지방 규모의 봄 서리 피해 평가, 모니터링 AGEA와 같은 대행사에 대한 CAP), Google Earth Engine(GEE)은 컴퓨팅 용량을 제공합니다. 전용 HPC 인프라가 필요한 클라우드 기반입니다.
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}")이탈리아 농업 기술에 대한 규정 및 공개 데이터
이탈리아의 농업 개방형 데이터 생태계는 세 가지 주요 주체를 중심으로 구성됩니다. 아게아 (농산물결제대행) 시안 (국가농업정보시스템)과 연계된 지역포털입니다. 2025년 2월, AGEA는 SIAN을 국가 전략 극점으로 마이그레이션하는 작업을 완료했습니다. 클라우드 인프라 및 지리공간 데이터 처리 기능 확장.
주요 이탈리아 농업 공개 데이터 소스
| 원천 | URL | 사용 가능한 데이터 | 체재 | 업데이트 |
|---|---|---|---|---|
| 시안 오픈 데이터 | data.sian.it | 고해상도 정사사진, 토지이용도(AI 기반), 기업 그래픽 계획 | GeoTIFF, Shapefile, WFS | 연간 |
| AGEA EO 서비스 | 시안잇 | 다중 스펙트럼 정사사진, X-밴드 레이더 이미지, DEM, DSM | GeoTIFF, ECW | 정기 간행물(캠페인) |
| ISTAT - 농업 통계 | istat.it/agricoltura | UAA, 생산, 회사 구조, 농업 인구 조사 | CSV, JSON, SDMX | 연간 |
| MATTM 국립지질포탈 | geoportale.gov.it | 카르타 나투라(Carta Natura), 코린 랜드 커버(Corine Land Cover), IDT, DBT | WMS, WFS, 셰이프파일 | 변하기 쉬운 |
| ARPA 지역(15개 지역) | 하프.[지역].it | 기상 관측소 데이터, 지역 예측 모델 | CSV, JSON, NetCDF | 거의 실시간 |
| 코페르니쿠스 랜드 서비스 | land.copernicus.eu | CORINE 토지 피복, 초원, 물과 습기, 도시 아틀라스 | GeoTIFF, 셰이프파일 | 3년마다/연간 |
AgriTech을 위한 PNRR 세금 공제 전환 5.0
위성 모니터링 시스템과 정밀 농업에 대한 투자는 전환 계획 5.0(법률 207/2024 - 예산법 2025)의 혜택을 받을 자격이 있습니다. 2025년부터 허용되는 농업 기업은 투자에 대한 세액공제를 받을 수 있습니다. 비즈니스 관리 소프트웨어를 포함한 유형 및 무형 자본재 4.0 인공 지능 구성 요소 및 지리 공간 데이터 분석.
전환 5.0 농업 기업에 대한 세금 공제율(2025)
| 투자 범위 | 에너지 절감 3~6% | 6~10% 절감 | 절감액 >10% |
|---|---|---|---|
| 최대 250만 유로 | 35% | 40% | 45% |
| 250만 유로에서 1000만 유로까지 | 15% | 20% | 25% |
| 1천만 유로에서 5천만 유로까지 | 5% | 10% | 15% |
NDVI 및 일기 예보에 의해 구동되는 정밀 관개 시스템이 적합할 수 있습니다. 물 소비량 감소를 기록한 경우 "에너지 절약 >6%" 범위 (펌핑 에너지)를 기준선과 비교합니다. 선서된 기술 전문 지식을 요청합니다.
프로덕션 아키텍처: 모범 사례 및 안티패턴
구성요소를 설명한 후에는 다음에 대한 모범 사례를 요약하는 것이 중요합니다. 프로덕션 환경에서 강력한 구현을 수행하고 일반적인 안티 패턴을 식별합니다. 취약하거나 부정확한 시스템을 초래합니다.
모범 사례 - AgriTech 위성 모니터링 시스템
- 점진적인 클라우드 필터링: 제품의 클라우드 커버율(레벨 1 메타데이터)만 사용하지 마세요. 항상 QA60(Sentinel-2) 또는 BQA(Landsat)를 사용하여 픽셀 수준에서 마스킹을 적용하세요. 5% 클라우드 제품에는 특정 플롯이 완전히 가려질 수 있습니다.
- 클라우드 최적화 GeoTIFF(COG): S3/MinIO에서 효율적인 범위 요청 액세스를 위해 NDVI 래스터를 COG 형식으로 저장합니다. 클라우드 스토리지에 최적화되지 않은 기존 GeoTIFF를 피하세요.
- 일관된 투영: 모든 것을 최종 데이터의 경우 EPSG:4326(WGS84)으로 표준화하고 이탈리아의 경우 올바른 미터법 측정을 유지하려면 UTM 구역 32N(EPSG:32632)으로 표준화하세요. 명시적인 재투영 없이 CRS를 혼합하지 마십시오.
- IoT Ground Truth를 통한 데이터 검증: 현장 측정(토양 습도 센서, 휴대용 LAI 측정기, 수집된 농업 데이터)을 통해 NDVI 임계값을 보정합니다. 로컬 교정 없이 NDVI만 사용하면 15~30%의 거짓 양성/음성 경고가 발생할 수 있습니다.
- 시간 아카이브 관리: 각 플롯에 대해 최소 3~5년의 NDVI 시계열을 유지합니다. 과거 평균(동일 월, 동일 작물)과 비교한 NDVI 이상은 절대값보다 훨씬 더 많은 정보를 제공합니다.
- 속도 제한 및 캐싱 API: CDSE와 Open-Meteo에는 속도 제한이 있습니다. 불필요한 재다운로드를 방지하려면 항상 로컬 캐시(파일 기반 또는 Redis)를 구현하십시오. Sentinel-2 다운로드에는 30~120초가 소요됩니다. 동일한 데이터를 두 번 요청하지 마십시오.
- 로깅 및 감사 추적: 각 NDVI 계산은 소스 위성 장면, 획득 날짜, 구름 비율, 알고리즘 버전을 추적할 수 있어야 합니다. PAC/AGEA 감사 및 이상 현상 디버깅을 위한 기본입니다.
일반적인 안티 패턴 - AgriTech 위성 시스템
- 대기 보정 무시: 레벨 2A(대기 최하위) 대신 레벨 1C(대기 최상부)를 사용하면 대기 습도가 높은 조건(포 계곡 안개, 아드리아 해)에서 NDVI에 10~20% 정도의 체계적인 오류가 발생합니다. 정량적 애플리케이션에는 항상 L2A를 사용하십시오.
- 유일한 지표인 NDVI: EVI(밀집된 초목의 경우), NDWI(물 스트레스), Red Edge NDVI(NDVI에 표시되기 전의 초기 스트레스)를 무시하고 NDVI에만 의존하십시오. 전문 시스템은 최소 3~4개의 지표를 조합하여 사용합니다.
- 픽셀 크기 및 소포: 10m 픽셀을 사용하여 0.3ha보다 작은 플롯에 NDVI를 적용하면 통계적으로 신뢰할 수 없는 평균(30픽셀 미만)이 생성됩니다. 작은 부지의 경우 Planet Labs(3.7m) 또는 드론 센서를 고려하세요.
- 시간적 합성 부족: 19% 구름 덮개에서도 특정 날짜에 사용할 수 있는 단일 획득을 취하면 부분 구름 영역에서 NDVI 아티팩트가 발생합니다. 항상 10-15일 단위로 메도이드 합성을 사용하십시오.
- 데이터 버전 관리가 없는 파이프라인: 다른 버전의 Python/rasterio/sentinelhub를 사용하여 NDVI를 다시 계산하면 동일한 소스 데이터에 대해 약간 다른 값이 생성됩니다. DVC 또는 이에 상응하는 버전으로 컴퓨팅 파이프라인의 버전을 지정하세요.
완전한 파이프라인: Prefect와의 오케스트레이션
설명된 모든 구성요소를 통합함으로써 완전한 위성 추적 파이프라인이 완성됩니다. Prefect와 함께 조율할 수 있습니다(시리즈 참조). 파이프라인 오케스트레이션) 5일마다 자동으로 실행됩니다(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()결론 및 다음 단계
CDSE를 통해 무료로 액세스할 수 있는 Sentinel-2 위성 데이터는 아마도 이탈리아 AgriTech에서 가장 많이 사용되지 않는 데이터 소스일 것입니다. 파이썬으로, rasterio, sentinelhub 및 CDSE 계정을 사용하면 품질과 세분성 측면에서 많은 솔루션을 능가하는 NDVI 모니터링 시스템 유료로 상업.
성공의 열쇠는 기술의 정교함이 아니라 교정 지역 농업: NDVI 임계값, 경고 시간 창, 가중치 예측 모델의 기능 중 일부를 특정 작물에 맞게 보정해야 합니다. 지역 기후와 회사의 농업 관행에 관한 것입니다. 보정된 시스템 Puglia의 포도나무에서는 그것이 없으면 Po Valley의 밀에서는 똑같이 잘 작동하지 않습니다. 인간 교정 개입.
유사한 시스템을 구현하려는 사람들이 기억해야 할 핵심 사항은 다음과 같습니다.
- 항상 사용 센티넬-2 레벨-2A (대기 보정 적용) 서로 다른 날짜 간의 정량적 비교를 위한 것입니다.
- NDVI를 하나 이상의 보완 지수(밀집된 포도원의 경우 EVI, 초기 수분 스트레스의 경우 NDWI, 엽록소 모니터링의 경우 Red Edge NDVI)와 결합합니다.
- 통합 개방형 ERA5-랜드 물 균형의 경우: ET0 + 강수량 + 모델의 토양 수분 및 현장에 IoT 센서가 없을 때 탁월한 프록시입니다.
- 하나를 유지 최소 3년 이상의 역사 시리즈 각 구획에 대해 과거 평균과 비교한 이상치가 절대 NDVI 값보다 훨씬 더 유용합니다.
- 다음으로 확장 Google 어스 엔진 지역 또는 국가 규모를 처리해야 하는 경우에만: 개별 회사의 경우 CDSE의 로컬 Python 파이프라인이 더 쉽게 제어 가능하고 종속되지 않습니다.
FoodTech 시리즈는 계속됩니다
위성 파이프라인을 구축했습니다. 이제 시리즈의 다른 기사를 살펴보세요. AgriTech 아키텍처를 완성합니다.
- 제1조: Python 및 MQTT를 사용한 정밀 농업용 IoT 파이프라인 - 지상 센서를 위성 파이프라인과 통합하는 방법.
- 제2조: 작물 모니터링을 위한 ML Edge: 현장의 컴퓨터 비전 - 식물 질병의 조기 발견을 위한 컴퓨터 비전 모델.
- 제4조: 식품의 블록체인 추적성: 현장에서 슈퍼마켓까지 - NDVI 데이터가 온체인 품질 인증서를 강화하는 방법.
이 문서에 설명된 MLOps 및 예측 모델 배포에 대해 자세히 알아보려면 시리즈를 참조하세요 MLflow를 사용한 비즈니스용 MLOps 그리고 시리즈 비즈니스 LLM: RAG Enterprise.