UPDATE

soile/health_summary.py

@@ -0,0 +1,147 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+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 _safe_number(value: Any, default: float = 0.0) -> float:
+    return default if value is None else float(value)
+
+
+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(plants: list[Any]) -> tuple[dict[str, dict[str, float]], str]:
+    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 = _safe_number(sensor_value, 0)
+        defaults = DEFAULT_HEALTH_PROFILE.get(metric_type, {})
+        ideal_value = float(config.get("ideal_value", defaults.get("ideal_value", 0)))
+        min_range = float(config.get("min_range", defaults.get("min_range", 0)))
+        max_range = float(config.get("max_range", defaults.get("max_range", 0)))
+        weight = float(config.get("weight", defaults.get("weight", 0)))
+        if weight <= 0:
+            continue
+
+        normalized_value = _normalize_metric(current_value, ideal_value, min_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) -> dict[str, str]:
+    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_soil_health_summary(sensor: Any, plants: list[Any]) -> dict[str, Any]:
+    profile, profile_source = resolve_plant_profile(plants)
+    health_score, health_components = compute_health_score(sensor, profile)
+    moisture = _safe_number(getattr(sensor, "soil_moisture", None), 0)
+    language = health_language(health_score)
+    return {
+        "healthScore": health_score,
+        "profileSource": profile_source,
+        "healthScoreDetails": {
+            "method": "normalized_weighted_average",
+            "profileSource": profile_source,
+            "components": health_components,
+        },
+        "healthLanguage": language,
+        "avgSoilMoisture": round(moisture),
+        "avgSoilMoistureRaw": round(moisture, 2),
+        "avgSoilMoistureStatus": "بهینه" if 45 <= moisture <= 75 else "نیازمند بررسی",
+    }