94 lines
3.6 KiB
Python
94 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from farm_data.services import build_ai_farm_snapshot, clone_snapshot_as_runtime_plant, get_primary_plant_snapshot, get_ai_snapshot_weather
|
|
from irrigation.evapotranspiration import calculate_forecast_water_needs, resolve_crop_profile
|
|
|
|
from farm_data.models import SensorData
|
|
from rag.services import get_water_need_prediction_insight
|
|
|
|
from .services import get_forecast_for_location
|
|
|
|
|
|
def build_water_need_prediction_payload(*, sensor: Any, forecasts: list[Any], ai_snapshot: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
location = getattr(sensor, "center_location", None)
|
|
plant = clone_snapshot_as_runtime_plant(get_primary_plant_snapshot(sensor))
|
|
irrigation_method = getattr(sensor, "irrigation_method", None)
|
|
|
|
if not forecasts or location is None:
|
|
return {
|
|
"totalNext7Days": 0,
|
|
"unit": "mm",
|
|
"categories": [],
|
|
"series": [],
|
|
"dailyBreakdown": [],
|
|
"cropProfile": {},
|
|
"irrigationEfficiencyPercent": None,
|
|
}
|
|
|
|
crop_profile = resolve_crop_profile(plant)
|
|
efficiency = getattr(irrigation_method, "water_efficiency_percent", None) if irrigation_method else None
|
|
daily = calculate_forecast_water_needs(
|
|
forecasts=forecasts[:7],
|
|
latitude_deg=float(location.latitude),
|
|
crop_profile=crop_profile,
|
|
growth_stage=crop_profile.get("current_stage"),
|
|
irrigation_efficiency_percent=efficiency,
|
|
)
|
|
daily_requirements = [round(item["gross_irrigation_mm"], 2) for item in daily]
|
|
|
|
return {
|
|
"totalNext7Days": round(sum(daily_requirements), 2),
|
|
"unit": "mm",
|
|
"categories": [f"روز {index}" for index in range(1, len(daily_requirements) + 1)],
|
|
"series": [{"name": "نیاز آبی تعدیلشده", "data": daily_requirements}],
|
|
"dailyBreakdown": daily,
|
|
"cropProfile": crop_profile,
|
|
"irrigationEfficiencyPercent": efficiency,
|
|
"source_metadata": {
|
|
"agronomic_metrics": {
|
|
"source": "build_ai_farm_snapshot",
|
|
"policy": "cluster_block_farm_aggregated",
|
|
"available": bool(ai_snapshot),
|
|
},
|
|
"weather": {
|
|
"source": "center_location_forecast",
|
|
"policy": "center_location_latest_forecast",
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
class WaterNeedPredictionService:
|
|
def get_water_need_prediction(self, *, farm_uuid: str) -> dict[str, Any]:
|
|
sensor = (
|
|
SensorData.objects.select_related("center_location", "irrigation_method")
|
|
.prefetch_related("plant_assignments__plant")
|
|
.filter(farm_uuid=farm_uuid)
|
|
.first()
|
|
)
|
|
if sensor is None:
|
|
raise ValueError("Farm not found.")
|
|
|
|
forecasts = get_forecast_for_location(sensor.center_location, days=7)
|
|
ai_snapshot = build_ai_farm_snapshot(farm_uuid)
|
|
payload = build_water_need_prediction_payload(sensor=sensor, forecasts=forecasts, ai_snapshot=ai_snapshot)
|
|
insight = get_water_need_prediction_insight(
|
|
farm_uuid=farm_uuid,
|
|
prediction_payload=payload,
|
|
)
|
|
|
|
return {
|
|
"farm_uuid": farm_uuid,
|
|
**payload,
|
|
"insight": {
|
|
"summary": insight.get("summary"),
|
|
"irrigation_outlook": insight.get("irrigation_outlook"),
|
|
"recommended_action": insight.get("recommended_action"),
|
|
"risk_note": insight.get("risk_note"),
|
|
"confidence": insight.get("confidence"),
|
|
},
|
|
"raw_response": insight.get("raw_response"),
|
|
}
|