from __future__ import annotations from datetime import date, timedelta from typing import Any from farm_data.services import get_canonical_farm_record, get_runtime_plant_for_farm from plant.gdd import resolve_growth_profile from .growth_simulation import ( DEFAULT_DYNAMIC_PARAMETERS, DEFAULT_PAGE_SIZE, GrowthSimulationError, _run_simulation, build_growth_context, ) def _safe_float(value: Any, default: float = 0.0) -> float: try: if value in (None, ""): return default return float(value) except (TypeError, ValueError): return default def _harvest_description( *, plant_name: str, current_gdd: float, required_gdd: float, remaining_gdd: float, estimated_days: int, maturity_reached_in_simulation: bool, ) -> str: if maturity_reached_in_simulation: return ( f"شبيه ساز رشد نشان مي دهد {plant_name} با روند فعلي در حدود {estimated_days} روز ديگر " f"به بازه برداشت مي رسد. تا امروز {round(current_gdd, 1)} واحد-روز رشد ثبت شده و " f"نياز باقيمانده تا بلوغ حدود {round(remaining_gdd, 1)} واحد-روز است." ) return ( f"شبيه ساز تا انتهاي forecast هنوز به رسيدگي کامل نرسيده، اما با ميانگين رشد فعلي " f"براورد مي شود {plant_name} حدود {estimated_days} روز ديگر به برداشت برسد. " f"تا امروز {round(current_gdd, 1)} از {round(required_gdd, 1)} واحد-روز مورد نياز طي شده است." ) def build_harvest_prediction_payload( *, farm_uuid: str, plant_name: str | None = None, irrigation_recommendation: dict[str, Any] | None = None, fertilization_recommendation: dict[str, Any] | None = None, ) -> dict[str, Any]: resolved_plant_name = plant_name if not resolved_plant_name: farm = get_canonical_farm_record(farm_uuid) if farm is None: raise GrowthSimulationError("مزرعه پیدا نشد.") plant = get_runtime_plant_for_farm(farm) if plant is None: raise GrowthSimulationError("گیاهی برای مزرعه انتخاب شده پیدا نشد.") resolved_plant_name = plant.name context = build_growth_context( { "farm_uuid": farm_uuid, "plant_name": resolved_plant_name, "dynamic_parameters": DEFAULT_DYNAMIC_PARAMETERS, "page_size": DEFAULT_PAGE_SIZE, } ) simulation_result, scenario_id, simulation_warning = _run_simulation( context, irrigation_recommendation=irrigation_recommendation, fertilization_recommendation=fertilization_recommendation, ) daily_output = simulation_result.get("daily_output") or [] if not daily_output: raise GrowthSimulationError("هیچ خروجی شبیه سازی در دسترس نیست.") profile = resolve_growth_profile(context.plant) required_gdd = _safe_float(profile.get("required_gdd_for_maturity"), 1200.0) current_gdd = _safe_float(profile.get("current_cumulative_gdd"), 0.0) cumulative_gdd = current_gdd maturity_date = None daily_gdd_forecast = [] for item in daily_output: day_gdd = _safe_float(item.get("GDD"), 0.0) cumulative_gdd += day_gdd day_value = item.get("DAY") iso_day = day_value.isoformat() if isinstance(day_value, date) else str(day_value) daily_gdd_forecast.append( { "date": iso_day, "gdd": round(day_gdd, 3), "cumulative_gdd": round(cumulative_gdd, 3), "development_stage": round(_safe_float(item.get("DVS"), 0.0), 4), } ) if _safe_float(item.get("DVS"), 0.0) >= 2.0 or cumulative_gdd >= required_gdd: maturity_date = date.fromisoformat(iso_day) break maturity_reached_in_simulation = maturity_date is not None if maturity_date is None: last_day = date.fromisoformat(str(daily_output[-1].get("DAY"))) simulated_days = max(len(daily_output), 1) avg_daily_gdd = max((cumulative_gdd - current_gdd) / simulated_days, 0.0) remaining_after_simulation = max(required_gdd - cumulative_gdd, 0.0) extra_days = 0 if avg_daily_gdd > 0 and remaining_after_simulation > 0: extra_days = int(remaining_after_simulation / avg_daily_gdd) if remaining_after_simulation % avg_daily_gdd: extra_days += 1 maturity_date = last_day + timedelta(days=max(extra_days, 0)) remaining_gdd = max(required_gdd - current_gdd, 0.0) days_until = max((maturity_date - date.today()).days, 0) window_start = maturity_date - timedelta(days=3) window_end = maturity_date + timedelta(days=3) return { "date": maturity_date.isoformat(), "dateFormatted": f"{maturity_date.day} {maturity_date.strftime('%B')} {maturity_date.year}", "daysUntil": days_until, "description": _harvest_description( plant_name=context.plant_name, current_gdd=current_gdd, required_gdd=required_gdd, remaining_gdd=remaining_gdd, estimated_days=days_until, maturity_reached_in_simulation=maturity_reached_in_simulation, ), "optimalWindowStart": window_start.isoformat(), "optimalWindowEnd": window_end.isoformat(), "gddDetails": { "current_cumulative_gdd": round(current_gdd, 3), "required_gdd_for_maturity": round(required_gdd, 3), "remaining_gdd": round(remaining_gdd, 3), "estimated_days_to_harvest": days_until, "predicted_harvest_date": maturity_date.isoformat(), "predicted_harvest_window": { "start": window_start.isoformat(), "end": window_end.isoformat(), }, "daily_gdd_forecast": daily_gdd_forecast, "simulation_engine": simulation_result.get("engine"), "simulation_model_name": simulation_result.get("model_name"), "simulation_warning": simulation_warning, "scenario_id": scenario_id, }, } class HarvestPredictionService: def get_harvest_prediction( self, *, farm_uuid: str, plant_name: str | None = None, irrigation_recommendation: dict[str, Any] | None = None, fertilization_recommendation: dict[str, Any] | None = None, ) -> dict[str, Any]: return build_harvest_prediction_payload( farm_uuid=farm_uuid, plant_name=plant_name, irrigation_recommendation=irrigation_recommendation, fertilization_recommendation=fertilization_recommendation, )