from __future__ import annotations from typing import Any from dashboard_data.card_utils import average, safe_number DEFAULT_HEALTH_PROFILE = { "moisture": {"ideal_value": 65.0, "min_range": 45.0, "max_range": 75.0, "weight": 0.45}, "ph": {"ideal_value": 6.6, "min_range": 6.0, "max_range": 7.5, "weight": 0.30}, "ec": {"ideal_value": 1.2, "min_range": 0.2, "max_range": 3.0, "weight": 0.25}, } METRIC_SPECS = { "moisture": { "sensor_field": "soil_moisture", "label": "رطوبت خاک", "unit": "%", }, "ph": { "sensor_field": "soil_ph", "label": "pH خاک", "unit": "pH", }, "ec": { "sensor_field": "electrical_conductivity", "label": "هدایت الکتریکی", "unit": "dS/m", }, } def _normalize_metric(value: float, ideal_value: float, min_range: float, max_range: float) -> float: if max_range <= min_range: return 0.0 if value <= min_range or value >= max_range: return 0.0 if value == ideal_value: return 1.0 if value < ideal_value: span = ideal_value - min_range if span <= 0: return 0.0 return max(0.0, min(1.0, (value - min_range) / span)) span = max_range - ideal_value if span <= 0: return 0.0 return max(0.0, min(1.0, (max_range - value) / span)) def _resolve_plant_profile(context: dict[str, Any]) -> tuple[dict[str, dict[str, float]], str]: plants = context.get("plants", []) for plant in plants: profile = getattr(plant, "health_profile", None) or {} if profile: merged = { metric: { **DEFAULT_HEALTH_PROFILE.get(metric, {}), **profile.get(metric, {}), } for metric in set(DEFAULT_HEALTH_PROFILE) | set(profile) } return merged, getattr(plant, "name", "گیاه") return DEFAULT_HEALTH_PROFILE, (plants[0].name if plants else "پروفایل پیش‌فرض") def _compute_health_score(sensor: Any, profile: dict[str, dict[str, float]]) -> tuple[int, list[dict[str, Any]]]: weighted_sum = 0.0 total_weight = 0.0 components: list[dict[str, Any]] = [] for metric_type, config in profile.items(): spec = METRIC_SPECS.get(metric_type) if spec is None: continue sensor_value = getattr(sensor, spec["sensor_field"], None) if sensor_value is None: continue current_value = float(safe_number(sensor_value, 0)) ideal_value = float(config.get("ideal_value", DEFAULT_HEALTH_PROFILE.get(metric_type, {}).get("ideal_value", 0))) min_range = float(config.get("min_range", DEFAULT_HEALTH_PROFILE.get(metric_type, {}).get("min_range", 0))) max_range = float(config.get("max_range", DEFAULT_HEALTH_PROFILE.get(metric_type, {}).get("max_range", 0))) weight = float(config.get("weight", DEFAULT_HEALTH_PROFILE.get(metric_type, {}).get("weight", 0))) if weight <= 0: continue normalized_value = _normalize_metric( value=current_value, ideal_value=ideal_value, min_range=min_range, max_range=max_range, ) weighted_sum += weight * normalized_value total_weight += weight components.append( { "metricType": metric_type, "label": spec["label"], "unit": spec["unit"], "currentValue": round(current_value, 2), "idealValue": round(ideal_value, 2), "minRange": round(min_range, 2), "maxRange": round(max_range, 2), "weight": round(weight, 3), "normalizedValue": round(normalized_value, 4), "weightedContribution": round(weight * normalized_value, 4), } ) if total_weight <= 0: return 0, components score = round((weighted_sum / total_weight) * 100) return max(0, min(100, score)), components def _health_language(health_score: int, ai_bundle: dict | None = None) -> dict[str, str]: ai_bundle = ai_bundle or {} ai_health = ai_bundle.get("farmOverviewKpis", {}) if isinstance(ai_bundle, dict) else {} short_chip_text = ai_health.get("short_chip_text") action_hint = ai_health.get("action_hint") explanation = ai_health.get("explanation") if isinstance(short_chip_text, str) and short_chip_text.strip() and isinstance(action_hint, str) and action_hint.strip() and isinstance(explanation, str) and explanation.strip(): return { "short_chip_text": short_chip_text.strip(), "action_hint": action_hint.strip(), "explanation": explanation.strip(), } if health_score >= 85: return { "short_chip_text": "بسیار خوب", "action_hint": "برنامه فعلی پایش و نگهداری حفظ شود.", "explanation": "شاخص سلامت مزرعه به محدوده بسیار خوب رسیده و بیشتر پارامترهای کلیدی نزدیک به پروفایل ایده‌آل گیاه هستند.", } if health_score >= 70: return { "short_chip_text": "پایدار", "action_hint": "تنظیمات فعلی حفظ و فقط شاخص‌های مرزی پایش شوند.", "explanation": "سلامت مزرعه در محدوده قابل قبول است، اما برخی پارامترها هنوز با مقدار ایده‌آل فاصله دارند.", } if health_score >= 50: return { "short_chip_text": "نیازمند تنظیم", "action_hint": "پارامترهای دور از محدوده ایده‌آل در اولویت اصلاح قرار گیرند.", "explanation": "امتیاز سلامت نشان می‌دهد بخشی از شرایط محیطی از پروفایل مطلوب گیاه فاصله گرفته و باید تنظیم شود.", } return { "short_chip_text": "تنش بالا", "action_hint": "اصلاح فوری رطوبت، تغذیه یا شوری بر اساس اجزای امتیاز انجام شود.", "explanation": "سلامت مزرعه در محدوده ضعیف قرار دارد و چند شاخص اصلی خارج از بازه قابل قبول گیاه هستند.", } def build_farm_overview_kpis(sensor_id: str, context: dict | None = None, ai_bundle: dict | None = None) -> dict: context = context or {} sensor = context.get("sensor") forecasts = context.get("forecasts", []) if sensor is None: return {"kpis": []} profile, profile_source = _resolve_plant_profile(context) health_score, health_components = _compute_health_score(sensor, profile) health_language = _health_language(health_score, ai_bundle=ai_bundle) moisture = safe_number(sensor.soil_moisture, 0) ph = safe_number(sensor.soil_ph, 7) humidity = average([forecast.humidity_mean for forecast in forecasts[:3]], default=45) water_stress = max(0, min(100, round(35 - (moisture / 2)))) disease_risk = max(0, min(100, round((humidity * 0.4) + (safe_number(sensor.soil_temperature, 0) * 0.6) - 20))) yield_prediction = round(max(5, (health_score / 2.1)), 1) primary_gap = min(health_components, key=lambda item: item["normalizedValue"], default=None) return { "kpis": [ { "id": "farm_health_score", "title": "امتیاز سلامت مزرعه", "subtitle": f"پروفایل {profile_source}", "stats": f"{health_score}%", "avatarColor": "success" if health_score >= 70 else "warning" if health_score >= 50 else "error", "avatarIcon": "tabler-heartbeat", "chipText": health_language["short_chip_text"], "chipColor": "success" if health_score >= 70 else "warning" if health_score >= 50 else "error", "actionHint": health_language["action_hint"], "explanation": health_language["explanation"], "healthScoreDetails": { "method": "normalized_weighted_average", "profileSource": profile_source, "components": health_components, }, }, { "id": "water_stress_index", "title": "شاخص تنش آبی", "subtitle": "فعلی", "stats": f"{water_stress}%", "avatarColor": "info", "avatarIcon": "tabler-droplet", "chipText": "پایین" if water_stress <= 20 else "متوسط", "chipColor": "success" if water_stress <= 20 else "warning", }, { "id": "disease_risk", "title": "ریسک بیماری", "subtitle": "۷ روز اخیر", "stats": "پایین" if disease_risk < 30 else "متوسط", "avatarColor": "success" if disease_risk < 30 else "warning", "avatarIcon": "tabler-bug", "chipText": f"{disease_risk}%", "chipColor": "success" if disease_risk < 30 else "warning", }, { "id": "avg_soil_moisture", "title": "میانگین رطوبت خاک", "subtitle": "کل مزرعه", "stats": f"{round(moisture)}%", "avatarColor": "primary", "avatarIcon": "tabler-plant-2", "chipText": "بهینه" if 45 <= moisture <= 75 else "نیازمند بررسی", "chipColor": "success" if 45 <= moisture <= 75 else "warning", }, { "id": "yield_prediction", "title": "پیش‌بینی عملکرد", "subtitle": "این فصل", "stats": f"{yield_prediction} تن", "avatarColor": "secondary", "avatarIcon": "tabler-chart-bar", "chipText": ( primary_gap["label"] if primary_gap else "پایدار" ), "chipColor": "warning" if primary_gap and primary_gap["normalizedValue"] < 0.6 else "success", }, { "id": "pest_risk", "title": "ریسک آفات", "subtitle": "پیش‌بینی هوشمند", "stats": f"{max(5, round(disease_risk * 0.7))}%", "avatarColor": "warning", "avatarIcon": "tabler-bug-off", "chipText": "تحت نظر", "chipColor": "warning", }, ] }