131 lines
5.1 KiB
Python
131 lines
5.1 KiB
Python
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]:
|
|
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},
|
|
],
|
|
}
|