993 lines
40 KiB
Python
993 lines
40 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import logging
|
|
import math
|
|
from datetime import date, datetime
|
|
from typing import Any, Callable
|
|
|
|
from django.apps import apps
|
|
from django.conf import settings
|
|
|
|
from farm_data.models import SensorData
|
|
from farm_data.services import get_farm_details
|
|
from location_data.models import NdviObservation, SoilLocation
|
|
|
|
from rag.services.yield_harvest import YieldHarvestRAGService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
READINESS_STATUS_FA = {
|
|
"ready": "آماده",
|
|
"approaching": "نزدیک به آمادگی",
|
|
"monitoring": "نیازمند پایش",
|
|
"not_ready": "آماده نیست",
|
|
}
|
|
|
|
|
|
class YieldHarvestSummaryService:
|
|
def get_summary(
|
|
self,
|
|
farm_uuid: str,
|
|
season_year: str,
|
|
crop_name: str,
|
|
include_narrative: bool = True,
|
|
) -> dict[str, Any]:
|
|
farm_context = self._get_farm_context(farm_uuid)
|
|
farm_context["season_year"] = season_year
|
|
farm_context["crop_name"] = crop_name or farm_context.get("crop_name") or ""
|
|
yield_prediction = self._build_yield_prediction(
|
|
farm_uuid=farm_uuid,
|
|
season_year=season_year,
|
|
crop_name=crop_name,
|
|
include_narrative=include_narrative,
|
|
farm_context=farm_context,
|
|
)
|
|
harvest_prediction_card = self._build_harvest_prediction_card(
|
|
farm_uuid=farm_uuid,
|
|
season_year=season_year,
|
|
crop_name=crop_name,
|
|
include_narrative=include_narrative,
|
|
farm_context=farm_context,
|
|
)
|
|
harvest_readiness_zones = self._build_harvest_readiness_zones(
|
|
farm_uuid=farm_uuid,
|
|
season_year=season_year,
|
|
crop_name=crop_name,
|
|
include_narrative=include_narrative,
|
|
farm_context=farm_context,
|
|
)
|
|
yield_quality_bands = self._build_yield_quality_bands(
|
|
farm_uuid=farm_uuid,
|
|
season_year=season_year,
|
|
crop_name=crop_name,
|
|
include_narrative=include_narrative,
|
|
farm_context=farm_context,
|
|
)
|
|
harvest_operations_card = self._build_harvest_operations_card(
|
|
farm_context=farm_context,
|
|
harvest_prediction_card=harvest_prediction_card,
|
|
pcse_dvs_stage=self._extract_pcse_dvs_stage(harvest_prediction_card),
|
|
)
|
|
yield_prediction_chart = self._build_yield_prediction_chart(
|
|
farm_uuid=farm_uuid,
|
|
season_year=season_year,
|
|
crop_name=crop_name,
|
|
include_narrative=include_narrative,
|
|
farm_context=farm_context,
|
|
)
|
|
season_highlights_card = self._build_season_highlights_card(
|
|
farm_uuid=farm_uuid,
|
|
season_year=season_year,
|
|
crop_name=crop_name,
|
|
include_narrative=include_narrative,
|
|
farm_context=farm_context,
|
|
yield_prediction=yield_prediction,
|
|
harvest_prediction_card=harvest_prediction_card,
|
|
harvest_readiness_zones=harvest_readiness_zones,
|
|
yield_quality_bands=yield_quality_bands,
|
|
)
|
|
|
|
deterministic_payload = {
|
|
"farm_uuid": farm_uuid,
|
|
"season_highlights_card": season_highlights_card,
|
|
"yield_prediction": yield_prediction,
|
|
"harvest_prediction_card": harvest_prediction_card,
|
|
"harvest_readiness_zones": harvest_readiness_zones,
|
|
"yield_quality_bands": yield_quality_bands,
|
|
"harvest_operations_card": harvest_operations_card,
|
|
"yield_prediction_chart": yield_prediction_chart,
|
|
}
|
|
context_payload = {
|
|
**copy.deepcopy(deterministic_payload),
|
|
"farm_context": farm_context,
|
|
}
|
|
|
|
if not include_narrative:
|
|
return deterministic_payload
|
|
|
|
try:
|
|
rag_service = YieldHarvestRAGService()
|
|
narrative_data = rag_service.generate_narrative(context_payload)
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"Yield harvest narrative generation failed for farm_uuid=%s: %s",
|
|
farm_uuid,
|
|
exc,
|
|
)
|
|
narrative_data = {}
|
|
return self._merge_narrative(deterministic_payload, narrative_data)
|
|
|
|
def _build_yield_prediction(
|
|
self,
|
|
*,
|
|
farm_uuid: str,
|
|
season_year: str,
|
|
crop_name: str,
|
|
include_narrative: bool,
|
|
farm_context: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
service = apps.get_app_config("crop_simulation").get_yield_prediction_service()
|
|
result = service.get_yield_prediction(
|
|
farm_uuid=farm_uuid,
|
|
plant_name=crop_name or None,
|
|
)
|
|
supporting_metrics = dict(result.get("supportingMetrics") or {})
|
|
|
|
# Secondary KPIs are placeholders until dedicated deterministic formulas land.
|
|
supporting_metrics.setdefault(
|
|
"estimatedKpis",
|
|
{
|
|
"season_year": season_year,
|
|
"applied_rule": "simple_placeholder_rules",
|
|
"is_estimated": True,
|
|
},
|
|
)
|
|
|
|
return {
|
|
"farm_uuid": result.get("farm_uuid", farm_uuid),
|
|
"crop_name": result.get("plant_name") or crop_name,
|
|
"season_year": season_year,
|
|
"predicted_yield_tons": result.get("predictedYieldTons"),
|
|
"predicted_yield_raw": result.get("predictedYieldRaw"),
|
|
"unit": result.get("unit"),
|
|
"source_unit": result.get("sourceUnit"),
|
|
"simulation_engine": result.get("simulationEngine"),
|
|
"simulation_model": result.get("simulationModel"),
|
|
"scenario_id": result.get("scenarioId"),
|
|
"simulation_warning": result.get("simulationWarning"),
|
|
"secondary_kpis_estimated": True,
|
|
"descriptionSource": "deterministic",
|
|
"farm_context": {
|
|
"soil_type": farm_context.get("soil", {}).get("soil_type"),
|
|
"soil_data_provider": farm_context.get("soil", {}).get("provider"),
|
|
},
|
|
"supporting_metrics": supporting_metrics,
|
|
}
|
|
|
|
def _build_harvest_prediction_card(
|
|
self,
|
|
*,
|
|
farm_uuid: str,
|
|
season_year: str,
|
|
crop_name: str,
|
|
include_narrative: bool,
|
|
farm_context: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
service = apps.get_app_config("crop_simulation").get_harvest_prediction_service()
|
|
result = service.get_harvest_prediction(
|
|
farm_uuid=farm_uuid,
|
|
plant_name=crop_name or None,
|
|
)
|
|
|
|
fallback_description = (
|
|
f"پیش بینی قطعی برداشت برای {crop_name or 'محصول انتخاب شده'} "
|
|
f"در فصل زراعی {season_year}."
|
|
)
|
|
|
|
return {
|
|
"farm_uuid": farm_uuid,
|
|
"crop_name": crop_name,
|
|
"season_year": season_year,
|
|
"harvest_date": result.get("date"),
|
|
"harvest_date_formatted": result.get("dateFormatted"),
|
|
"days_until": result.get("daysUntil"),
|
|
"optimal_window_start": result.get("optimalWindowStart"),
|
|
"optimal_window_end": result.get("optimalWindowEnd"),
|
|
"description": result.get("description") or fallback_description,
|
|
"descriptionSource": "قطعی",
|
|
"field_conditions": {
|
|
"soil_moisture": farm_context.get("recent_sensor_averages", {}).get("soil_moisture"),
|
|
"soil_temperature": farm_context.get("recent_sensor_averages", {}).get("soil_temperature"),
|
|
},
|
|
"readiness_metrics": result.get("gddDetails") or {},
|
|
}
|
|
|
|
def _build_yield_prediction_chart(
|
|
self,
|
|
*,
|
|
farm_uuid: str,
|
|
season_year: str,
|
|
crop_name: str,
|
|
include_narrative: bool,
|
|
farm_context: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
simulator = apps.get_app_config("crop_simulation").get_current_farm_chart_simulator()
|
|
result = simulator.simulate(
|
|
farm_uuid=farm_uuid,
|
|
plant_name=crop_name or None,
|
|
)
|
|
pcse_timeseries = list(result.get("daily_output") or [])
|
|
|
|
yield_series: list[list[float]] = []
|
|
biomass_series: list[list[float]] = []
|
|
for item in pcse_timeseries:
|
|
timestamp = self._to_unix_timestamp(item.get("DAY"))
|
|
if timestamp is None:
|
|
continue
|
|
|
|
twso = self._safe_chart_value(item.get("TWSO"))
|
|
if twso is not None:
|
|
yield_series.append([timestamp, twso])
|
|
|
|
tagp = self._safe_chart_value(item.get("TAGP"))
|
|
if tagp is not None:
|
|
biomass_series.append([timestamp, tagp])
|
|
|
|
return {
|
|
"farm_uuid": farm_uuid,
|
|
"crop_name": result.get("plant_name") or crop_name,
|
|
"season_year": season_year,
|
|
"series": [
|
|
{
|
|
"name": "عملکرد پیش بینی شده",
|
|
"type": "line",
|
|
"data": yield_series,
|
|
},
|
|
{
|
|
"name": "بیوماس",
|
|
"type": "area",
|
|
"data": biomass_series,
|
|
},
|
|
],
|
|
"xAxis": {"type": "datetime", "label": "تاریخ"},
|
|
"meta": {
|
|
"unit": "کیلوگرم در هکتار",
|
|
"simulation_engine": result.get("engine"),
|
|
"simulation_model": result.get("model_name"),
|
|
"scenario_id": result.get("scenario_id"),
|
|
"simulation_warning": result.get("simulation_warning"),
|
|
"field_context": {
|
|
"soil_type": farm_context.get("soil", {}).get("soil_type"),
|
|
"center_coordinates": farm_context.get("center_coordinates"),
|
|
},
|
|
},
|
|
}
|
|
|
|
def _build_harvest_operations_card(
|
|
self,
|
|
*,
|
|
farm_context: dict[str, Any],
|
|
harvest_prediction_card: dict[str, Any],
|
|
pcse_dvs_stage: float,
|
|
) -> dict[str, Any]:
|
|
days_until = int(harvest_prediction_card.get("days_until") or 0)
|
|
stage_label, phase_name = self._map_dvs_to_phase(pcse_dvs_stage)
|
|
steps = self._build_operations_steps(
|
|
phase_name=phase_name,
|
|
days_until=days_until,
|
|
soil_moisture=farm_context.get("recent_sensor_averages", {}).get("soil_moisture"),
|
|
)
|
|
|
|
return {
|
|
"farm_uuid": farm_context.get("farm_uuid"),
|
|
"crop_name": farm_context.get("crop_name"),
|
|
"season_year": farm_context.get("season_year"),
|
|
"stage_label": stage_label,
|
|
"phase_name": phase_name,
|
|
"days_until_harvest": days_until,
|
|
"current_dvs": round(pcse_dvs_stage, 4),
|
|
"summary": (
|
|
f"عملیات برداشت برای {farm_context.get('crop_name') or 'محصول انتخاب شده'} "
|
|
f"با توجه به {days_until} روز باقی مانده تا بازه پیش بینی شده برداشت اولویت بندی شده است."
|
|
),
|
|
"rules_source": "قواعد_قطعی_DVS",
|
|
"field_context": {
|
|
"soil_type": farm_context.get("soil", {}).get("soil_type"),
|
|
"soil_moisture": farm_context.get("recent_sensor_averages", {}).get("soil_moisture"),
|
|
"soil_temperature": farm_context.get("recent_sensor_averages", {}).get("soil_temperature"),
|
|
},
|
|
"steps": steps,
|
|
}
|
|
|
|
def _build_season_highlights_card(
|
|
self,
|
|
*,
|
|
farm_uuid: str,
|
|
season_year: str,
|
|
crop_name: str,
|
|
include_narrative: bool,
|
|
farm_context: dict[str, Any],
|
|
yield_prediction: dict[str, Any],
|
|
harvest_prediction_card: dict[str, Any],
|
|
harvest_readiness_zones: dict[str, Any],
|
|
yield_quality_bands: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
primary_quality_grade = (
|
|
yield_quality_bands.get("primary_quality_grade")
|
|
or yield_quality_bands.get("top_band")
|
|
or yield_quality_bands.get("summary")
|
|
)
|
|
average_readiness = harvest_readiness_zones.get("averageReadiness")
|
|
total_predicted_yield = yield_prediction.get("predicted_yield_tons")
|
|
target_harvest_date = (
|
|
harvest_prediction_card.get("harvest_date_formatted")
|
|
or harvest_prediction_card.get("harvest_date")
|
|
)
|
|
estimated_revenue = self._get_estimated_revenue(
|
|
farm_uuid=farm_uuid,
|
|
total_predicted_yield=total_predicted_yield,
|
|
)
|
|
return {
|
|
"farm_uuid": farm_uuid,
|
|
"crop_name": crop_name,
|
|
"season_year": season_year,
|
|
"title": "خلاصه فصل",
|
|
# Left blank for narrative merge unless a non-LLM fallback is needed later.
|
|
"subtitle": "",
|
|
"total_predicted_yield": total_predicted_yield,
|
|
"yield_unit": yield_prediction.get("unit"),
|
|
"target_harvest_date": target_harvest_date,
|
|
"days_until_harvest": harvest_prediction_card.get("days_until"),
|
|
"average_readiness": average_readiness,
|
|
"primary_quality_grade": primary_quality_grade,
|
|
"estimated_revenue": estimated_revenue,
|
|
"soil_type": farm_context.get("soil", {}).get("soil_type"),
|
|
}
|
|
|
|
def _build_harvest_readiness_zones(
|
|
self,
|
|
*,
|
|
farm_uuid: str,
|
|
season_year: str,
|
|
crop_name: str,
|
|
include_narrative: bool,
|
|
farm_context: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
sensor = (
|
|
SensorData.objects.select_related("center_location")
|
|
.filter(farm_uuid=farm_uuid)
|
|
.first()
|
|
)
|
|
if sensor is None or sensor.center_location is None:
|
|
return {
|
|
"farm_uuid": farm_uuid,
|
|
"averageReadiness": None,
|
|
"zones": [],
|
|
"source": "سرویس_سلامت_NDVI",
|
|
}
|
|
|
|
location = sensor.center_location
|
|
ndvi_service = apps.get_app_config("location_data").get_ndvi_health_service()
|
|
health_card = ndvi_service.get_ndvi_health(farm_uuid=farm_uuid)
|
|
|
|
observations = list(
|
|
location.ndvi_observations.order_by("-observation_date", "-created_at")[:2]
|
|
)
|
|
latest_observation = observations[0] if observations else None
|
|
previous_observation = observations[1] if len(observations) > 1 else None
|
|
|
|
latest_ndvi = self._safe_float(health_card.get("mean_ndvi"), None)
|
|
previous_ndvi = self._safe_float(
|
|
previous_observation.mean_ndvi if previous_observation else None,
|
|
None,
|
|
)
|
|
ndvi_trend = None
|
|
if latest_ndvi is not None and previous_ndvi is not None:
|
|
ndvi_trend = round(latest_ndvi - previous_ndvi, 4)
|
|
|
|
grid = {}
|
|
if latest_observation and isinstance(latest_observation.ndvi_map, dict):
|
|
grid = latest_observation.ndvi_map
|
|
ndvi_grid = grid.get("grid") if isinstance(grid, dict) else None
|
|
|
|
zones: list[dict[str, Any]] = []
|
|
if isinstance(ndvi_grid, list) and ndvi_grid:
|
|
zone_index = 1
|
|
for row_index, row in enumerate(ndvi_grid):
|
|
if not isinstance(row, list):
|
|
continue
|
|
for col_index, cell in enumerate(row):
|
|
cell_ndvi = self._safe_chart_value(cell)
|
|
if cell_ndvi is None:
|
|
continue
|
|
readiness = self._ndvi_to_readiness(cell_ndvi, ndvi_trend)
|
|
zones.append(
|
|
{
|
|
"zoneId": f"zone-{zone_index}",
|
|
"zoneLabel": f"ناحیه {zone_index}",
|
|
"gridPosition": {"row": row_index, "col": col_index},
|
|
"meanNdvi": cell_ndvi,
|
|
"readiness": readiness,
|
|
"daysUntil": self._estimate_days_until_from_readiness(readiness),
|
|
"status": self._readiness_status(readiness),
|
|
}
|
|
)
|
|
zone_index += 1
|
|
|
|
if not zones and latest_ndvi is not None:
|
|
readiness = self._ndvi_to_readiness(latest_ndvi, ndvi_trend)
|
|
zones.append(
|
|
{
|
|
"zoneId": "zone-center",
|
|
"zoneLabel": "ناحیه مرکزی مزرعه",
|
|
"gridPosition": None,
|
|
"meanNdvi": latest_ndvi,
|
|
"readiness": readiness,
|
|
"daysUntil": self._estimate_days_until_from_readiness(readiness),
|
|
"status": self._readiness_status(readiness),
|
|
}
|
|
)
|
|
|
|
average_readiness = None
|
|
if zones:
|
|
average_readiness = round(
|
|
sum(zone["readiness"] for zone in zones) / len(zones),
|
|
2,
|
|
)
|
|
|
|
return {
|
|
"farm_uuid": farm_uuid,
|
|
"observationDate": (
|
|
latest_observation.observation_date.isoformat()
|
|
if latest_observation
|
|
else health_card.get("observation_date")
|
|
),
|
|
"vegetationHealthClass": health_card.get("vegetation_health_class"),
|
|
"meanNdvi": latest_ndvi,
|
|
"ndviTrend": ndvi_trend,
|
|
"averageReadiness": average_readiness,
|
|
"zones": zones,
|
|
"source": "سرویس_سلامت_NDVI",
|
|
}
|
|
|
|
def _to_unix_timestamp(self, value: Any) -> int | None:
|
|
if isinstance(value, datetime):
|
|
return int(value.timestamp() * 1000)
|
|
if isinstance(value, date):
|
|
return int(datetime.combine(value, datetime.min.time()).timestamp() * 1000)
|
|
if isinstance(value, str):
|
|
try:
|
|
if "T" in value:
|
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
else:
|
|
parsed = datetime.combine(date.fromisoformat(value), datetime.min.time())
|
|
return int(parsed.timestamp() * 1000)
|
|
except ValueError:
|
|
return None
|
|
return None
|
|
|
|
def _safe_chart_value(self, value: Any) -> float | None:
|
|
parsed = self._safe_float(value, None)
|
|
if parsed is None or math.isnan(parsed) or math.isinf(parsed):
|
|
return None
|
|
return round(parsed, 4)
|
|
|
|
def _ndvi_to_readiness(self, mean_ndvi: float, trend_delta: float | None) -> int:
|
|
base_score = ((0.75 - mean_ndvi) / 0.55) * 100.0
|
|
if trend_delta is not None and trend_delta < 0:
|
|
# A falling NDVI near season end suggests drying and harvest readiness.
|
|
base_score += min(abs(trend_delta) * 120.0, 18.0)
|
|
if trend_delta is not None and trend_delta > 0.05:
|
|
base_score -= min(trend_delta * 80.0, 12.0)
|
|
return int(round(max(0.0, min(base_score, 100.0))))
|
|
|
|
def _estimate_days_until_from_readiness(self, readiness: int) -> int:
|
|
return max(int(round((100 - readiness) / 12.0)), 0)
|
|
|
|
def _readiness_status(self, readiness: int) -> str:
|
|
if readiness >= 80:
|
|
return READINESS_STATUS_FA["ready"]
|
|
if readiness >= 55:
|
|
return READINESS_STATUS_FA["approaching"]
|
|
if readiness >= 30:
|
|
return READINESS_STATUS_FA["monitoring"]
|
|
return READINESS_STATUS_FA["not_ready"]
|
|
|
|
def _build_yield_quality_bands(
|
|
self,
|
|
*,
|
|
farm_uuid: str,
|
|
season_year: str,
|
|
crop_name: str,
|
|
include_narrative: bool,
|
|
farm_context: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
crop_key = (crop_name or farm_context.get("crop_name") or "").strip().lower()
|
|
yield_service = apps.get_app_config("crop_simulation").get_yield_prediction_service()
|
|
yield_payload = yield_service.get_yield_prediction(
|
|
farm_uuid=farm_uuid,
|
|
plant_name=crop_name or None,
|
|
)
|
|
predicted_yield_raw = self._safe_float(yield_payload.get("predictedYieldRaw"), 0.0) or 0.0
|
|
soil_metrics = farm_context.get("soil", {}).get("resolved_metrics") or {}
|
|
sensor_metrics = farm_context.get("recent_sensor_averages") or {}
|
|
|
|
try:
|
|
service = self._resolve_service(
|
|
getter_names=(
|
|
"get_yield_quality_bands_service",
|
|
"get_quality_grading_service",
|
|
"get_quality_model_service",
|
|
)
|
|
)
|
|
method = self._resolve_service_method(
|
|
service,
|
|
method_names=(
|
|
"get_yield_quality_bands",
|
|
"get_quality_bands",
|
|
"grade_yield_quality",
|
|
),
|
|
)
|
|
return method(
|
|
farm_uuid=farm_uuid,
|
|
season_year=season_year,
|
|
crop_name=crop_name or None,
|
|
include_narrative=include_narrative,
|
|
)
|
|
except AttributeError:
|
|
pass
|
|
|
|
protein_content = self._estimate_protein_content(
|
|
crop_key=crop_key,
|
|
nitrogen_value=self._safe_float(soil_metrics.get("nitrogen"), None),
|
|
predicted_yield_raw=predicted_yield_raw,
|
|
)
|
|
moisture_percent = self._estimate_moisture_percent(
|
|
crop_key=crop_key,
|
|
soil_moisture=sensor_metrics.get("soil_moisture"),
|
|
)
|
|
quality_score = self._estimate_quality_score(
|
|
protein_content=protein_content,
|
|
moisture_percent=moisture_percent,
|
|
predicted_yield_raw=predicted_yield_raw,
|
|
)
|
|
grade_distribution = self._build_grade_distribution(quality_score)
|
|
primary_quality_grade = max(
|
|
grade_distribution,
|
|
key=lambda item: item.get("share_percent", 0),
|
|
)["grade"]
|
|
|
|
return {
|
|
"farm_uuid": farm_uuid,
|
|
"crop_name": crop_name,
|
|
"season_year": season_year,
|
|
"source": "قواعد_قطعی_درجه_بندی",
|
|
"is_estimated": True,
|
|
"protein_content": {
|
|
"value": protein_content,
|
|
"unit": "%",
|
|
},
|
|
"moisture_percentage": {
|
|
"value": moisture_percent,
|
|
"unit": "%",
|
|
},
|
|
"grade_distribution": grade_distribution,
|
|
"primary_quality_grade": primary_quality_grade,
|
|
"quality_score": quality_score,
|
|
"summary": f"درجه کیفیت غالب محصول {primary_quality_grade} است.",
|
|
}
|
|
|
|
def _get_estimated_revenue(
|
|
self,
|
|
*,
|
|
farm_uuid: str,
|
|
total_predicted_yield: float | None,
|
|
) -> float | None:
|
|
try:
|
|
service = apps.get_app_config("economy").get_economic_overview_service()
|
|
overview = service.get_economic_overview(farm_uuid=farm_uuid)
|
|
except Exception:
|
|
return None
|
|
|
|
if not isinstance(overview, dict):
|
|
return None
|
|
|
|
price_per_ton = None
|
|
for item in overview.get("economicData") or []:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
title = str(item.get("title") or "").lower()
|
|
value = item.get("value")
|
|
if "price" in title or "قیمت" in title:
|
|
price_per_ton = self._extract_numeric(value)
|
|
break
|
|
|
|
if price_per_ton is None or total_predicted_yield is None:
|
|
return None
|
|
return round(total_predicted_yield * price_per_ton, 2)
|
|
|
|
def _estimate_protein_content(
|
|
self,
|
|
*,
|
|
crop_key: str,
|
|
nitrogen_value: float | None,
|
|
predicted_yield_raw: float,
|
|
) -> float:
|
|
nitrogen_factor = 0.0 if nitrogen_value is None else min(nitrogen_value / 2500.0, 2.0)
|
|
yield_factor = min(predicted_yield_raw / 10000.0, 1.5)
|
|
if "wheat" in crop_key or "گندم" in crop_key:
|
|
base = 11.8
|
|
return round(base + (nitrogen_factor * 1.2) - (yield_factor * 0.35), 2)
|
|
if "barley" in crop_key or "جو" in crop_key:
|
|
base = 10.4
|
|
return round(base + (nitrogen_factor * 0.9) - (yield_factor * 0.25), 2)
|
|
return round(9.5 + (nitrogen_factor * 0.8), 2)
|
|
|
|
def _estimate_moisture_percent(
|
|
self,
|
|
*,
|
|
crop_key: str,
|
|
soil_moisture: float | None,
|
|
) -> float:
|
|
soil_component = 0.0 if soil_moisture is None else min(max((soil_moisture - 20.0) / 10.0, -2.0), 4.0)
|
|
if "wheat" in crop_key or "barley" in crop_key or "گندم" in crop_key or "جو" in crop_key:
|
|
return round(12.6 + soil_component, 2)
|
|
return round(11.8 + soil_component, 2)
|
|
|
|
def _estimate_quality_score(
|
|
self,
|
|
*,
|
|
protein_content: float,
|
|
moisture_percent: float,
|
|
predicted_yield_raw: float,
|
|
) -> int:
|
|
protein_score = min(max((protein_content / 14.0) * 50.0, 0.0), 50.0)
|
|
moisture_penalty = min(abs(moisture_percent - 12.5) * 4.5, 22.0)
|
|
yield_bonus = min(predicted_yield_raw / 1500.0, 18.0)
|
|
score = protein_score + yield_bonus + 32.0 - moisture_penalty
|
|
return int(round(max(0.0, min(score, 100.0))))
|
|
|
|
def _build_grade_distribution(self, quality_score: int) -> list[dict[str, Any]]:
|
|
if quality_score >= 85:
|
|
return [
|
|
{"grade": "A", "share_percent": 62},
|
|
{"grade": "B", "share_percent": 28},
|
|
{"grade": "C", "share_percent": 10},
|
|
]
|
|
if quality_score >= 70:
|
|
return [
|
|
{"grade": "A", "share_percent": 38},
|
|
{"grade": "B", "share_percent": 44},
|
|
{"grade": "C", "share_percent": 18},
|
|
]
|
|
return [
|
|
{"grade": "A", "share_percent": 16},
|
|
{"grade": "B", "share_percent": 41},
|
|
{"grade": "C", "share_percent": 43},
|
|
]
|
|
|
|
def _extract_numeric(self, value: Any) -> float | None:
|
|
if isinstance(value, (int, float)):
|
|
return float(value)
|
|
if not isinstance(value, str):
|
|
return None
|
|
cleaned = "".join(ch for ch in value if ch.isdigit() or ch in {".", "-"})
|
|
return self._safe_float(cleaned, None)
|
|
|
|
def _get_farm_context(
|
|
self,
|
|
farm_uuid: str,
|
|
) -> dict[str, Any]:
|
|
farm = (
|
|
SensorData.objects.select_related("center_location", "weather_forecast")
|
|
.prefetch_related("center_location__depths", "plants")
|
|
.filter(farm_uuid=farm_uuid)
|
|
.first()
|
|
)
|
|
if farm is None:
|
|
return {
|
|
"farm_uuid": farm_uuid,
|
|
"center_coordinates": None,
|
|
"soil": {"provider": getattr(settings, "SOIL_DATA_PROVIDER", "نامشخص")},
|
|
"recent_sensor_averages": {},
|
|
}
|
|
|
|
farm_details = get_farm_details(str(farm_uuid)) or {}
|
|
center_location = farm.center_location
|
|
soil_details = (farm_details.get("soil") or {}).get("resolved_metrics") or {}
|
|
weather_details = farm_details.get("weather") or {}
|
|
recent_sensor_averages = {
|
|
"soil_moisture": self._safe_float(soil_details.get("soil_moisture", farm.soil_moisture), None),
|
|
"soil_temperature": self._safe_float(soil_details.get("soil_temperature", farm.soil_temperature), None),
|
|
"air_temperature_mean": self._safe_float(weather_details.get("temperature_mean"), None),
|
|
}
|
|
|
|
crop_name = ""
|
|
plant_names = farm_details.get("plants") or []
|
|
if plant_names:
|
|
first_plant = plant_names[0]
|
|
if isinstance(first_plant, dict):
|
|
crop_name = str(first_plant.get("name") or "")
|
|
|
|
return {
|
|
"farm_uuid": farm_uuid,
|
|
"crop_name": crop_name,
|
|
"center_coordinates": {
|
|
"lat": float(center_location.latitude),
|
|
"lon": float(center_location.longitude),
|
|
},
|
|
"farm_boundary": farm_details.get("center_location", {}).get("farm_boundary"),
|
|
"soil": {
|
|
"provider": getattr(settings, "SOIL_DATA_PROVIDER", "نامشخص"),
|
|
"soil_type": self._infer_soil_type(soil_details),
|
|
"resolved_metrics": soil_details,
|
|
},
|
|
"recent_sensor_averages": recent_sensor_averages,
|
|
"weather": {
|
|
"temperature_mean": self._safe_float(weather_details.get("temperature_mean"), None),
|
|
"temperature_min": self._safe_float(weather_details.get("temperature_min"), None),
|
|
"temperature_max": self._safe_float(weather_details.get("temperature_max"), None),
|
|
},
|
|
"source_models": {
|
|
"sensor_data": SensorData.__name__,
|
|
"soil_location": SoilLocation.__name__,
|
|
},
|
|
}
|
|
|
|
def _extract_pcse_dvs_stage(self, harvest_prediction_card: dict[str, Any]) -> float:
|
|
readiness_metrics = harvest_prediction_card.get("readiness_metrics") or {}
|
|
forecast = readiness_metrics.get("daily_gdd_forecast") or [{}]
|
|
return self._safe_float(forecast[-1].get("development_stage"), 0.0) or 0.0
|
|
|
|
def _map_dvs_to_phase(self, dvs: float) -> tuple[str, str]:
|
|
if dvs >= 2.0:
|
|
return "آماده", "رسیدگی"
|
|
if dvs >= 1.7:
|
|
return "پیش_برداشت_نهایی", "زایشی_پایانی"
|
|
if dvs >= 1.2:
|
|
return "پیش_برداشت_میانی", "پرشدن_دانه"
|
|
if dvs >= 0.8:
|
|
return "پایش", "گذار_زایشی"
|
|
return "پیش_برداشت_ابتدایی", "رشد_رویشی"
|
|
|
|
def _build_operations_steps(
|
|
self,
|
|
*,
|
|
phase_name: str,
|
|
days_until: int,
|
|
soil_moisture: float | None,
|
|
) -> list[dict[str, Any]]:
|
|
field_ready = soil_moisture is None or soil_moisture <= 35.0
|
|
|
|
if phase_name == "رسیدگی":
|
|
return [
|
|
{
|
|
"key": "desiccation",
|
|
"title": "بررسی خشک شدن محصول",
|
|
"status": "آماده",
|
|
"is_completed": False,
|
|
"estimated_days": 0,
|
|
},
|
|
{
|
|
"key": "harvesting",
|
|
"title": "برداشت",
|
|
"status": "آماده" if field_ready else "نیازمند بررسی شرایط مزرعه",
|
|
"is_completed": False,
|
|
"estimated_days": max(min(days_until, 2), 0),
|
|
},
|
|
{
|
|
"key": "transportation",
|
|
"title": "انتقال محصول",
|
|
"status": "آماده",
|
|
"is_completed": False,
|
|
"estimated_days": max(min(days_until + 1, 3), 1),
|
|
},
|
|
]
|
|
if phase_name == "زایشی_پایانی":
|
|
return [
|
|
{
|
|
"key": "equipment_check",
|
|
"title": "بازبینی تجهیزات برداشت",
|
|
"status": "اولویت بالا",
|
|
"is_completed": False,
|
|
"estimated_days": 1,
|
|
},
|
|
{
|
|
"key": "labor_plan",
|
|
"title": "نهایی کردن برنامه نیروی کار و حمل",
|
|
"status": "اولویت بالا",
|
|
"is_completed": False,
|
|
"estimated_days": 2,
|
|
},
|
|
{
|
|
"key": "field_entry",
|
|
"title": "بررسی امکان ورود به مزرعه و بازه های خشک",
|
|
"status": "آماده" if field_ready else "پایش",
|
|
"is_completed": False,
|
|
"estimated_days": max(min(days_until, 5), 1),
|
|
},
|
|
]
|
|
if phase_name == "پرشدن_دانه":
|
|
return [
|
|
{
|
|
"key": "monitor_maturity",
|
|
"title": "پایش رسیدگی و رشد اندام ذخیره ای",
|
|
"status": "در حال انجام",
|
|
"is_completed": False,
|
|
"estimated_days": 7,
|
|
},
|
|
{
|
|
"key": "review_readiness",
|
|
"title": "بررسی اختلاف آمادگی بین ناحیه ها",
|
|
"status": "در حال انجام",
|
|
"is_completed": False,
|
|
"estimated_days": 10,
|
|
},
|
|
{
|
|
"key": "prepare_logistics",
|
|
"title": "آماده سازی برنامه لجستیک برداشت",
|
|
"status": "پیش رو",
|
|
"is_completed": False,
|
|
"estimated_days": 14,
|
|
},
|
|
]
|
|
return [
|
|
{
|
|
"key": "weekly_monitoring",
|
|
"title": "پایش هفتگی رسیدگی محصول",
|
|
"status": "در حال انجام",
|
|
"is_completed": False,
|
|
"estimated_days": 14,
|
|
},
|
|
{
|
|
"key": "update_forecast",
|
|
"title": "به روزرسانی پیش بینی زمان برداشت",
|
|
"status": "در حال انجام",
|
|
"is_completed": False,
|
|
"estimated_days": 10,
|
|
},
|
|
{
|
|
"key": "draft_operations",
|
|
"title": "تهیه چک لیست عملیات برداشت",
|
|
"status": "پیش رو",
|
|
"is_completed": False,
|
|
"estimated_days": 21,
|
|
},
|
|
]
|
|
|
|
def _infer_soil_type(self, soil_metrics: dict[str, Any]) -> str | None:
|
|
sand = self._safe_float(soil_metrics.get("sand"), None)
|
|
clay = self._safe_float(soil_metrics.get("clay"), None)
|
|
silt = self._safe_float(soil_metrics.get("silt"), None)
|
|
if sand is None or clay is None or silt is None:
|
|
return None
|
|
if clay >= 40:
|
|
return "رسی"
|
|
if sand >= 70 and clay <= 15:
|
|
return "شنی"
|
|
if silt >= 50 and clay < 27:
|
|
return "سیلتی لوم"
|
|
return "لوم"
|
|
|
|
def _safe_float(self, value: Any, default: float | None = 0.0) -> float | None:
|
|
try:
|
|
if value in (None, ""):
|
|
return default
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
def _merge_narrative(
|
|
self,
|
|
final_payload: dict[str, Any],
|
|
narratives: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
merged = copy.deepcopy(final_payload)
|
|
if not isinstance(narratives, dict):
|
|
narratives = {}
|
|
|
|
season_card = merged.setdefault("season_highlights_card", {})
|
|
fallback_subtitle = self._default_season_highlights_subtitle(merged)
|
|
season_card["subtitle"] = self._coalesce_text(
|
|
narratives.get("season_highlights_subtitle"),
|
|
season_card.get("subtitle"),
|
|
fallback_subtitle,
|
|
)
|
|
|
|
yield_card = merged.setdefault("yield_prediction", {})
|
|
fallback_yield_explanation = self._default_yield_prediction_explanation(merged)
|
|
yield_card["explanation"] = self._coalesce_text(
|
|
narratives.get("yield_prediction_explanation"),
|
|
yield_card.get("explanation"),
|
|
fallback_yield_explanation,
|
|
)
|
|
|
|
readiness_card = merged.setdefault("harvest_readiness_zones", {})
|
|
fallback_readiness_summary = self._default_harvest_readiness_summary(merged)
|
|
readiness_card["summary"] = self._coalesce_text(
|
|
narratives.get("harvest_readiness_summary"),
|
|
readiness_card.get("summary"),
|
|
fallback_readiness_summary,
|
|
)
|
|
|
|
operations_card = merged.setdefault("harvest_operations_card", {})
|
|
deterministic_steps = operations_card.get("steps")
|
|
operation_notes = narratives.get("operation_notes")
|
|
if isinstance(deterministic_steps, list):
|
|
note_items = operation_notes if isinstance(operation_notes, list) else []
|
|
for index, step in enumerate(deterministic_steps):
|
|
if not isinstance(step, dict):
|
|
continue
|
|
fallback_note = self._default_operation_note(step)
|
|
candidate_note = note_items[index] if index < len(note_items) else None
|
|
step["note"] = self._coalesce_text(
|
|
candidate_note,
|
|
step.get("note"),
|
|
fallback_note,
|
|
)
|
|
|
|
return merged
|
|
|
|
def _coalesce_text(self, *values: Any) -> str:
|
|
for value in values:
|
|
if isinstance(value, str) and value.strip():
|
|
return value.strip()
|
|
return ""
|
|
|
|
def _default_season_highlights_subtitle(self, payload: dict[str, Any]) -> str:
|
|
highlights = payload.get("season_highlights_card") or {}
|
|
total_yield = highlights.get("total_predicted_yield")
|
|
unit = highlights.get("yield_unit") or ""
|
|
harvest_date = highlights.get("target_harvest_date") or "بازه پیش بینی شده برداشت"
|
|
if total_yield is None:
|
|
return f"بر اساس چشم انداز قطعی فصل، برداشت برای {harvest_date} هدف گذاری شده است."
|
|
return f"عملکرد پیش بینی شده {total_yield} {unit} است و برداشت برای {harvest_date} هدف گذاری شده است.".strip()
|
|
|
|
def _default_yield_prediction_explanation(self, payload: dict[str, Any]) -> str:
|
|
yield_card = payload.get("yield_prediction") or {}
|
|
predicted = yield_card.get("predicted_yield_tons")
|
|
unit = yield_card.get("unit") or ""
|
|
if predicted is None:
|
|
return "پیش بینی عملکرد بر پایه خروجی قطعی شبیه سازی محصول محاسبه شده است."
|
|
return f"پیش بینی عملکرد بر پایه شبیه سازی قطعی محصول انجام شده و در حال حاضر مقدار {predicted} {unit} را نشان می دهد.".strip()
|
|
|
|
def _default_harvest_readiness_summary(self, payload: dict[str, Any]) -> str:
|
|
readiness = payload.get("harvest_readiness_zones") or {}
|
|
average = readiness.get("averageReadiness")
|
|
if average is None:
|
|
return "آمادگی برداشت از آخرین سیگنال های قطعی ناحیه ای استخراج شده است."
|
|
return f"میانگین آمادگی برداشت بر اساس آخرین سیگنال های قطعی ناحیه ای، {average} است.".strip()
|
|
|
|
def _default_operation_note(self, step: dict[str, Any]) -> str:
|
|
title = step.get("title") or "این عملیات"
|
|
status = step.get("status") or "برنامه ریزی شده"
|
|
estimate = step.get("estimated_days")
|
|
if estimate is None:
|
|
return f"وضعیت {title} در حال حاضر «{status}» ثبت شده است."
|
|
return f"{title} با وضعیت «{status}» و زمان بندی تقریبی {estimate} روز ثبت شده است.".strip()
|
|
|
|
def _resolve_service(self, *, getter_names: tuple[str, ...]) -> Any:
|
|
app_config = apps.get_app_config("crop_simulation")
|
|
for getter_name in getter_names:
|
|
getter = getattr(app_config, getter_name, None)
|
|
if callable(getter):
|
|
return getter()
|
|
raise AttributeError(
|
|
f"None of the expected service getters were found on crop_simulation app config: {getter_names}"
|
|
)
|
|
|
|
def _resolve_service_method(
|
|
self,
|
|
service: Any,
|
|
*,
|
|
method_names: tuple[str, ...],
|
|
) -> Callable[..., dict[str, Any]]:
|
|
for method_name in method_names:
|
|
method = getattr(service, method_name, None)
|
|
if callable(method):
|
|
return method
|
|
raise AttributeError(
|
|
f"None of the expected service methods were found on {service.__class__.__name__}: {method_names}"
|
|
)
|