from datetime import date from unittest.mock import patch from django.test import TestCase, override_settings from rest_framework.test import APIClient from farm_data.models import FarmPlantAssignment, PlantCatalogSnapshot, SensorData from location_data.models import ( AnalysisGridCell, AnalysisGridObservation, BlockSubdivision, RemoteSensingClusterAssignment, RemoteSensingClusterBlock, RemoteSensingRun, RemoteSensingSubdivisionResult, SoilLocation, ) from weather.models import WeatherForecast @override_settings(ROOT_URLCONF="location_data.urls") class RemoteSensingClusterRecommendationApiTests(TestCase): def setUp(self): self.client = APIClient() self.boundary = { "type": "Polygon", "coordinates": [ [ [51.3890, 35.6890], [51.3900, 35.6890], [51.3900, 35.6900], [51.3890, 35.6900], [51.3890, 35.6890], ] ], } self.location = SoilLocation.objects.create( latitude="35.689200", longitude="51.389000", farm_boundary=self.boundary, ) self.location.set_input_block_count(1) self.location.save(update_fields=["input_block_count", "block_layout", "updated_at"]) self.farm = SensorData.objects.create( farm_uuid="11111111-1111-1111-1111-111111111111", center_location=self.location, sensor_payload={}, ) for day_index in range(1, 5): WeatherForecast.objects.create( location=self.location, forecast_date=date(2025, 2, day_index), temperature_min=12.0, temperature_max=24.0, temperature_mean=18.0, precipitation=1.0, precipitation_probability=25.0, humidity_mean=55.0, wind_speed_max=10.0, et0=3.0, weather_code=1, ) self.subdivision = BlockSubdivision.objects.create( soil_location=self.location, block_code="block-1", source_boundary=self.boundary, chunk_size_sqm=900, status="subdivided", ) self.run = RemoteSensingRun.objects.create( soil_location=self.location, block_subdivision=self.subdivision, block_code="block-1", chunk_size_sqm=900, temporal_start=date(2025, 1, 1), temporal_end=date(2025, 1, 31), status=RemoteSensingRun.STATUS_SUCCESS, metadata={"stage": "completed"}, ) self.result = RemoteSensingSubdivisionResult.objects.create( soil_location=self.location, run=self.run, block_subdivision=self.subdivision, block_code="block-1", chunk_size_sqm=900, temporal_start=date(2025, 1, 1), temporal_end=date(2025, 1, 31), cluster_count=2, selected_features=["ndvi", "ndwi", "soil_vv_db"], metadata={"used_cell_count": 2, "skipped_cell_count": 0}, ) self.cell_1 = AnalysisGridCell.objects.create( soil_location=self.location, block_subdivision=self.subdivision, block_code="block-1", cell_code="cell-1", chunk_size_sqm=900, geometry=self.boundary, centroid_lat="35.689250", centroid_lon="51.389250", ) self.cell_2 = AnalysisGridCell.objects.create( soil_location=self.location, block_subdivision=self.subdivision, block_code="block-1", cell_code="cell-2", chunk_size_sqm=900, geometry=self.boundary, centroid_lat="35.689750", centroid_lon="51.389750", ) AnalysisGridObservation.objects.create( cell=self.cell_1, run=self.run, temporal_start=date(2025, 1, 1), temporal_end=date(2025, 1, 31), ndvi=0.51, ndwi=0.24, soil_vv=0.13, soil_vv_db=-10.0, metadata={"backend_name": "openeo"}, ) AnalysisGridObservation.objects.create( cell=self.cell_2, run=self.run, temporal_start=date(2025, 1, 1), temporal_end=date(2025, 1, 31), ndvi=0.71, ndwi=0.48, soil_vv=0.19, soil_vv_db=-7.5, metadata={"backend_name": "openeo"}, ) RemoteSensingClusterAssignment.objects.create( result=self.result, cell=self.cell_1, cluster_label=0, raw_feature_values={"ndvi": 0.51, "ndwi": 0.24, "soil_vv_db": -10.0}, scaled_feature_values={"ndvi": -1.0, "ndwi": -1.0, "soil_vv_db": -1.0}, ) RemoteSensingClusterAssignment.objects.create( result=self.result, cell=self.cell_2, cluster_label=1, raw_feature_values={"ndvi": 0.71, "ndwi": 0.48, "soil_vv_db": -7.5}, scaled_feature_values={"ndvi": 1.0, "ndwi": 1.0, "soil_vv_db": 1.0}, ) self.cluster_0 = RemoteSensingClusterBlock.objects.create( result=self.result, soil_location=self.location, block_subdivision=self.subdivision, block_code="block-1", sub_block_code="cluster-0", cluster_label=0, chunk_size_sqm=900, centroid_lat="35.689250", centroid_lon="51.389250", center_cell_code="cell-1", cell_count=1, cell_codes=["cell-1"], geometry=self.boundary, ) self.cluster_1 = RemoteSensingClusterBlock.objects.create( result=self.result, soil_location=self.location, block_subdivision=self.subdivision, block_code="block-1", sub_block_code="cluster-1", cluster_label=1, chunk_size_sqm=900, centroid_lat="35.689750", centroid_lon="51.389750", center_cell_code="cell-2", cell_count=1, cell_codes=["cell-2"], geometry=self.boundary, ) self.location.block_layout = { "input_block_count": 1, "default_full_farm": True, "algorithm_status": "completed", "blocks": [ { "block_code": "block-1", "order": 1, "source": "input", "boundary": self.boundary, "needs_subdivision": True, "sub_blocks": [ { "sub_block_code": "cluster-0", "cluster_label": 0, "cluster_uuid": str(self.cluster_0.uuid), }, { "sub_block_code": "cluster-1", "cluster_label": 1, "cluster_uuid": str(self.cluster_1.uuid), }, ], } ], } self.location.save(update_fields=["block_layout", "updated_at"]) self.tomato = PlantCatalogSnapshot.objects.create( backend_plant_id=101, name="Tomato", growth_profile={"simulation": {"crop_parameters": {"crop_name": "Tomato", "MAX_BIOMASS": 14000.0}}}, ) self.wheat = PlantCatalogSnapshot.objects.create( backend_plant_id=102, name="Wheat", growth_profile={"simulation": {"crop_parameters": {"crop_name": "Wheat", "MAX_BIOMASS": 11000.0}}}, ) FarmPlantAssignment.objects.create(farm=self.farm, plant=self.tomato, position=0, stage="vegetative") FarmPlantAssignment.objects.create(farm=self.farm, plant=self.wheat, position=1, stage="vegetative") @patch("location_data.cluster_recommendation._simulate_candidate") def test_cluster_recommendations_return_ranked_plants_for_each_cluster(self, simulate_mock): def fake_simulation(*, base_payload, soil_parameters, site_parameters): plant_name = base_payload["crop_parameters"]["crop_name"] smfcf = float(soil_parameters["SMFCF"]) if plant_name == "Tomato": yield_estimate = 150.0 if smfcf >= 0.4 else 80.0 else: yield_estimate = 110.0 if smfcf >= 0.4 else 120.0 return ( { "engine": "pcse", "model_name": "Wofost81_NWLP_CWB_CNB", "metrics": { "yield_estimate": yield_estimate, "biomass": yield_estimate * 2, "max_lai": 4.2, }, }, None, ) simulate_mock.side_effect = fake_simulation response = self.client.get( "/remote-sensing/cluster-recommendations/", data={"farm_uuid": str(self.farm.farm_uuid)}, ) self.assertEqual(response.status_code, 200) payload = response.json()["data"] self.assertEqual(payload["cluster_count"], 2) self.assertEqual(payload["evaluated_plant_count"], 2) self.assertEqual(len(payload["registered_plants"]), 2) clusters = {item["cluster_label"]: item for item in payload["clusters"]} self.assertEqual(clusters[0]["resolved_metrics"]["ndvi"], 0.51) self.assertEqual(clusters[0]["resolved_metrics"]["ndwi"], 0.24) self.assertEqual(clusters[0]["resolved_metrics"]["soil_vv"], 0.13) self.assertEqual(clusters[1]["resolved_metrics"]["ndwi"], 0.48) self.assertEqual(clusters[0]["suggested_plant"]["plant_name"], "Wheat") self.assertEqual(clusters[1]["suggested_plant"]["plant_name"], "Tomato") self.assertEqual(clusters[0]["candidate_plants"][0]["score"], 120.0) self.assertEqual(clusters[1]["candidate_plants"][0]["score"], 150.0) self.assertEqual(clusters[0]["cluster_block"]["uuid"], str(self.cluster_0.uuid)) self.assertEqual(clusters[1]["cluster_block"]["uuid"], str(self.cluster_1.uuid)) def test_cluster_recommendations_return_400_when_no_plants_registered(self): FarmPlantAssignment.objects.all().delete() response = self.client.get( "/remote-sensing/cluster-recommendations/", data={"farm_uuid": str(self.farm.farm_uuid)}, ) self.assertEqual(response.status_code, 400) self.assertEqual( response.json()["msg"], "برای این مزرعه هنوز هیچ گیاهی در farm_data ثبت نشده است.", ) @patch("location_data.cluster_recommendation._simulate_candidate") def test_cluster_recommendations_use_cached_payload_for_same_farm_assignments(self, simulate_mock): simulate_mock.return_value = ( { "engine": "pcse", "model_name": "Wofost81_NWLP_CWB_CNB", "metrics": { "yield_estimate": 100.0, "biomass": 200.0, "max_lai": 3.1, }, }, None, ) first_response = self.client.get( "/remote-sensing/cluster-recommendations/", data={"farm_uuid": str(self.farm.farm_uuid)}, ) self.assertEqual(first_response.status_code, 200) self.assertGreater(simulate_mock.call_count, 0) simulate_mock.reset_mock() simulate_mock.side_effect = AssertionError("cached recommendations should skip simulation") second_response = self.client.get( "/remote-sensing/cluster-recommendations/", data={"farm_uuid": str(self.farm.farm_uuid)}, ) self.assertEqual(second_response.status_code, 200) self.assertEqual(first_response.json()["data"], second_response.json()["data"]) simulate_mock.assert_not_called()