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.failure_contract import RAGServiceError 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, irrigation_recommendation: dict[str, Any] | None = None, fertilization_recommendation: dict[str, Any] | None = None, ) -> 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 "" farm_context["irrigation_recommendation"] = irrigation_recommendation or {} farm_context["fertilization_recommendation"] = fertilization_recommendation 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, irrigation_recommendation=irrigation_recommendation, fertilization_recommendation=fertilization_recommendation, ) 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, irrigation_recommendation=irrigation_recommendation, fertilization_recommendation=fertilization_recommendation, ) 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, irrigation_recommendation=irrigation_recommendation, fertilization_recommendation=fertilization_recommendation, ) 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 RAGServiceError as exc: logger.warning( "Yield harvest narrative generation failed for farm_uuid=%s: %s", farm_uuid, exc, ) narrative_data = { "status": "error", "source": "llm", "narrative_error": exc.to_dict(), } 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], irrigation_recommendation: dict[str, Any] | None = None, fertilization_recommendation: dict[str, Any] | None = None, ) -> 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, irrigation_recommendation=irrigation_recommendation, fertilization_recommendation=fertilization_recommendation, ) 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], irrigation_recommendation: dict[str, Any] | None = None, fertilization_recommendation: dict[str, Any] | None = None, ) -> 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, irrigation_recommendation=irrigation_recommendation, fertilization_recommendation=fertilization_recommendation, ) 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], irrigation_recommendation: dict[str, Any] | None = None, fertilization_recommendation: dict[str, Any] | None = None, ) -> 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, irrigation_recommendation=irrigation_recommendation, fertilization_recommendation=fertilization_recommendation, ) 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", "plant_assignments__plant") .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, ) merged["narrative_status"] = narratives.get("status", "success") merged["narrative_source"] = narratives.get("source", "deterministic") if isinstance(narratives.get("narrative_error"), dict): merged["narrative_error"] = narratives["narrative_error"] 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}" )