31 lines
1.1 KiB
Python
31 lines
1.1 KiB
Python
from __future__ import annotations
|
|
|
|
from django.apps import apps
|
|
|
|
from farm_data.models import SensorData
|
|
|
|
class WaterStressService:
|
|
def get_water_stress(
|
|
self,
|
|
*,
|
|
farm_uuid: str,
|
|
plant_name: str | None = None,
|
|
irrigation_recommendation: dict | None = None,
|
|
fertilization_recommendation: dict | None = None,
|
|
) -> dict[str, Any]:
|
|
sensor = SensorData.objects.filter(farm_uuid=farm_uuid).first()
|
|
if sensor is None:
|
|
raise ValueError("Farm not found.")
|
|
simulation_service = apps.get_app_config("crop_simulation").get_water_stress_service()
|
|
try:
|
|
return simulation_service.get_water_stress(
|
|
farm_uuid=str(sensor.farm_uuid),
|
|
plant_name=plant_name,
|
|
irrigation_recommendation=irrigation_recommendation,
|
|
fertilization_recommendation=fertilization_recommendation,
|
|
)
|
|
except Exception as exc:
|
|
raise RuntimeError(
|
|
f"Water stress simulation failed for farm {sensor.farm_uuid}: {exc}"
|
|
) from exc
|