156 lines
4.6 KiB
Python
156 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from dataclasses import dataclass
|
|
from datetime import date, timedelta
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
from .models import NdviObservation
|
|
|
|
|
|
DEFAULT_SATELLITE_SOURCE = "sentinel-2"
|
|
DEFAULT_CLOUD_COVER = 20.0
|
|
|
|
|
|
def classify_ndvi(mean_ndvi: float) -> str:
|
|
if mean_ndvi < 0.2:
|
|
return "Bare soil"
|
|
if mean_ndvi < 0.4:
|
|
return "Weak vegetation"
|
|
if mean_ndvi < 0.6:
|
|
return "Moderate vegetation"
|
|
return "Healthy vegetation"
|
|
|
|
|
|
def calculate_ndvi(red: float, nir: float) -> float | None:
|
|
denominator = nir + red
|
|
if denominator == 0:
|
|
return None
|
|
return round((nir - red) / denominator, 4)
|
|
|
|
|
|
def calculate_ndvi_grid(red_band: list[list[float]], nir_band: list[list[float]]) -> list[list[float | None]]:
|
|
grid: list[list[float | None]] = []
|
|
for red_row, nir_row in zip(red_band, nir_band):
|
|
row: list[float | None] = []
|
|
for red, nir in zip(red_row, nir_row):
|
|
row.append(calculate_ndvi(float(red), float(nir)))
|
|
grid.append(row)
|
|
return grid
|
|
|
|
|
|
def mean_ndvi(grid: list[list[float | None]]) -> float:
|
|
values = [value for row in grid for value in row if value is not None]
|
|
if not values:
|
|
return 0.0
|
|
return round(sum(values) / len(values), 4)
|
|
|
|
|
|
def _default_bbox(location: Any, delta: float = 0.001) -> list[float]:
|
|
lat = float(location.latitude)
|
|
lon = float(location.longitude)
|
|
return [lon - delta, lat - delta, lon + delta, lat + delta]
|
|
|
|
|
|
def _geometry_payload(location: Any) -> dict:
|
|
boundary = getattr(location, "farm_boundary", None) or {}
|
|
if boundary:
|
|
return boundary
|
|
return {"bbox": _default_bbox(location)}
|
|
|
|
|
|
@dataclass
|
|
class SatelliteNdviResult:
|
|
observation_date: str
|
|
mean_ndvi: float
|
|
ndvi_map: list[list[float | None]]
|
|
vegetation_health_class: str
|
|
satellite_source: str
|
|
cloud_cover: float | None
|
|
metadata: dict[str, Any]
|
|
|
|
|
|
class SentinelCompatibleNdviClient:
|
|
def __init__(self) -> None:
|
|
self.endpoint = os.environ.get("SATELLITE_NDVI_ENDPOINT")
|
|
self.api_key = os.environ.get("SATELLITE_NDVI_API_KEY")
|
|
self.source = os.environ.get("SATELLITE_SOURCE", DEFAULT_SATELLITE_SOURCE)
|
|
|
|
@property
|
|
def is_configured(self) -> bool:
|
|
return bool(self.endpoint and self.api_key)
|
|
|
|
def fetch_red_nir(
|
|
self,
|
|
geometry: dict,
|
|
date_from: date,
|
|
date_to: date,
|
|
cloud_cover: float,
|
|
) -> dict[str, Any] | None:
|
|
if not self.is_configured:
|
|
return None
|
|
|
|
response = requests.post(
|
|
self.endpoint,
|
|
json={
|
|
"geometry": geometry,
|
|
"date_from": date_from.isoformat(),
|
|
"date_to": date_to.isoformat(),
|
|
"cloud_cover_max": cloud_cover,
|
|
"source": self.source,
|
|
"bands": ["B04", "B08"],
|
|
},
|
|
headers={
|
|
"Authorization": f"Bearer {self.api_key}",
|
|
"Content-Type": "application/json",
|
|
},
|
|
timeout=30,
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
def fetch_or_get_ndvi_observation(
|
|
location: Any,
|
|
days_back: int = 7,
|
|
cloud_cover: float = DEFAULT_CLOUD_COVER,
|
|
) -> NdviObservation | None:
|
|
observation = location.ndvi_observations.order_by("-observation_date", "-created_at").first()
|
|
if observation is not None:
|
|
return observation
|
|
|
|
client = SentinelCompatibleNdviClient()
|
|
payload = client.fetch_red_nir(
|
|
geometry=_geometry_payload(location),
|
|
date_from=date.today() - timedelta(days=days_back),
|
|
date_to=date.today(),
|
|
cloud_cover=cloud_cover,
|
|
)
|
|
if not payload:
|
|
return None
|
|
|
|
red_band = payload.get("red_band") or []
|
|
nir_band = payload.get("nir_band") or []
|
|
observation_date = payload.get("observation_date") or date.today().isoformat()
|
|
ndvi_grid = calculate_ndvi_grid(red_band=red_band, nir_band=nir_band)
|
|
ndvi_mean = mean_ndvi(ndvi_grid)
|
|
return NdviObservation.objects.create(
|
|
location=location,
|
|
observation_date=date.fromisoformat(observation_date),
|
|
mean_ndvi=ndvi_mean,
|
|
ndvi_map={
|
|
"grid": ndvi_grid,
|
|
"red_band_source": "B04",
|
|
"nir_band_source": "B08",
|
|
},
|
|
vegetation_health_class=classify_ndvi(ndvi_mean),
|
|
satellite_source=payload.get("satellite_source", client.source),
|
|
cloud_cover=payload.get("cloud_cover"),
|
|
metadata={
|
|
"geometry": _geometry_payload(location),
|
|
"raw_payload_meta": payload.get("metadata", {}),
|
|
},
|
|
)
|