from __future__ import annotations from typing import Any from dashboard_data.card_utils import average, safe_number DEFAULT_IDEAL_SENSOR_PROFILE = { "temperature": {"ideal": 24.0, "min": 18.0, "max": 30.0}, "moisture": {"ideal": 65.0, "min": 45.0, "max": 80.0}, "ph": {"ideal": 6.5, "min": 6.0, "max": 7.2}, "ec": {"ideal": 1.4, "min": 1.0, "max": 2.0}, "humidity": {"ideal": 60.0, "min": 45.0, "max": 75.0}, } METRIC_ORDER = [ ("temperature", "دما"), ("moisture", "رطوبت"), ("ph", "pH"), ("ec", "هدایت الکتریکی"), ("humidity", "رطوبت هوا"), ] def _normalize_to_ideal_score(value: float | None, minimum: float, ideal: float, maximum: float) -> int: if value is None or maximum <= minimum: return 0 if value <= minimum or value >= maximum: return 0 if value == ideal: return 100 if value < ideal: span = ideal - minimum if span <= 0: return 0 return round(max(0.0, min(1.0, (value - minimum) / span)) * 100) span = maximum - ideal if span <= 0: return 0 return round(max(0.0, min(1.0, (maximum - value) / span)) * 100) def _resolve_profile(context: dict[str, Any]) -> tuple[dict[str, dict[str, float]], str]: location = context.get("location") if location is not None: location_profile = getattr(location, "ideal_sensor_profile", None) or {} if location_profile: merged = { metric: {**DEFAULT_IDEAL_SENSOR_PROFILE.get(metric, {}), **location_profile.get(metric, {})} for metric in set(DEFAULT_IDEAL_SENSOR_PROFILE) | set(location_profile) } return merged, "location" plants = context.get("plants", []) for plant in plants: plant_profile = getattr(plant, "health_profile", None) or {} if plant_profile: translated: dict[str, dict[str, float]] = {} for metric in DEFAULT_IDEAL_SENSOR_PROFILE: metric_data = plant_profile.get(metric) if not metric_data: continue translated[metric] = { "ideal": float(metric_data.get("ideal_value", DEFAULT_IDEAL_SENSOR_PROFILE[metric]["ideal"])), "min": float(metric_data.get("min_range", DEFAULT_IDEAL_SENSOR_PROFILE[metric]["min"])), "max": float(metric_data.get("max_range", DEFAULT_IDEAL_SENSOR_PROFILE[metric]["max"])), } merged = { metric: {**DEFAULT_IDEAL_SENSOR_PROFILE.get(metric, {}), **translated.get(metric, {})} for metric in DEFAULT_IDEAL_SENSOR_PROFILE } return merged, f"plant:{getattr(plant, 'name', 'unknown')}" return DEFAULT_IDEAL_SENSOR_PROFILE, "default" def _current_metric_values(sensor: Any, forecasts: list[Any]) -> dict[str, float]: current_forecast = forecasts[0] if forecasts else None humidity = average( [getattr(forecast, "humidity_mean", None) for forecast in forecasts[:3]], default=safe_number(getattr(current_forecast, "humidity_mean", None), 0), ) return { "temperature": float(safe_number(getattr(sensor, "soil_temperature", None), 0)), "moisture": float(safe_number(getattr(sensor, "soil_moisture", None), 0)), "ph": float(safe_number(getattr(sensor, "soil_ph", None), 0)), "ec": float(safe_number(getattr(sensor, "electrical_conductivity", None), 0)), "humidity": float(safe_number(humidity, 0)), } def build_sensor_radar_chart(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 {"labels": [], "series": [], "profileSource": None, "profile": {}} profile, profile_source = _resolve_profile(context) current_values = _current_metric_values(sensor, forecasts) labels: list[str] = [] current_series: list[int] = [] ideal_series: list[int] = [] metric_details: list[dict[str, Any]] = [] for metric_key, label in METRIC_ORDER: metric_profile = profile.get(metric_key, DEFAULT_IDEAL_SENSOR_PROFILE.get(metric_key, {})) minimum = float(metric_profile.get("min", DEFAULT_IDEAL_SENSOR_PROFILE[metric_key]["min"])) ideal = float(metric_profile.get("ideal", DEFAULT_IDEAL_SENSOR_PROFILE[metric_key]["ideal"])) maximum = float(metric_profile.get("max", DEFAULT_IDEAL_SENSOR_PROFILE[metric_key]["max"])) current_value = current_values.get(metric_key) labels.append(label) current_score = _normalize_to_ideal_score(current_value, minimum, ideal, maximum) current_series.append(current_score) ideal_series.append(100) metric_details.append( { "metricType": metric_key, "label": label, "currentValue": round(current_value, 2) if current_value is not None else None, "idealValue": round(ideal, 2), "minRange": round(minimum, 2), "maxRange": round(maximum, 2), "currentScore": current_score, "idealScore": 100, } ) return { "labels": labels, "profileSource": profile_source, "profile": profile, "metricDetails": metric_details, "series": [ {"name": "امروز", "data": current_series}, {"name": "پروفایل ایده‌آل", "data": ideal_series}, ], }