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", {}), }, )