208 lines
8.2 KiB
Python
208 lines
8.2 KiB
Python
|
from datetime import date
|
||
|
|
from unittest.mock import patch
|
||
|
|
|
||
|
|
from django.test import TestCase, override_settings
|
||
|
|
from rest_framework.test import APIClient
|
||
|
|
|
||
|
|
from location_data.models import (
|
||
|
|
AnalysisGridCell,
|
||
|
|
BlockSubdivision,
|
||
|
|
RemoteSensingSubdivisionOption,
|
||
|
|
RemoteSensingSubdivisionOptionAssignment,
|
||
|
|
RemoteSensingSubdivisionOptionBlock,
|
||
|
|
RemoteSensingRun,
|
||
|
|
RemoteSensingSubdivisionResult,
|
||
|
|
SoilLocation,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@override_settings(ROOT_URLCONF="location_data.urls")
|
||
|
|
class RemoteSensingSubdivisionOptionApiTests(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.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=1,
|
||
|
|
selected_features=["ndvi", "ndwi", "soil_vv_db"],
|
||
|
|
metadata={"recommended_requested_k": 2, "active_requested_k": 1},
|
||
|
|
)
|
||
|
|
self.cells = [
|
||
|
|
AnalysisGridCell.objects.create(
|
||
|
|
soil_location=self.location,
|
||
|
|
block_subdivision=self.subdivision,
|
||
|
|
block_code="block-1",
|
||
|
|
cell_code=f"cell-{index}",
|
||
|
|
chunk_size_sqm=900,
|
||
|
|
geometry={
|
||
|
|
"type": "Polygon",
|
||
|
|
"coordinates": [[
|
||
|
|
[51.3890 + (index * 0.0001), 35.6890],
|
||
|
|
[51.3891 + (index * 0.0001), 35.6890],
|
||
|
|
[51.3891 + (index * 0.0001), 35.6891],
|
||
|
|
[51.3890 + (index * 0.0001), 35.6891],
|
||
|
|
[51.3890 + (index * 0.0001), 35.6890],
|
||
|
|
]],
|
||
|
|
},
|
||
|
|
centroid_lat=f"{35.68905 + (index * 0.0001):.6f}",
|
||
|
|
centroid_lon=f"{51.38905 + (index * 0.0001):.6f}",
|
||
|
|
)
|
||
|
|
for index in range(2)
|
||
|
|
]
|
||
|
|
self.option_k1 = RemoteSensingSubdivisionOption.objects.create(
|
||
|
|
result=self.result,
|
||
|
|
requested_k=1,
|
||
|
|
effective_cluster_count=1,
|
||
|
|
is_active=True,
|
||
|
|
is_recommended=False,
|
||
|
|
selection_source="system",
|
||
|
|
metadata={"cluster_summaries": []},
|
||
|
|
)
|
||
|
|
self.option_k2 = RemoteSensingSubdivisionOption.objects.create(
|
||
|
|
result=self.result,
|
||
|
|
requested_k=2,
|
||
|
|
effective_cluster_count=2,
|
||
|
|
is_active=False,
|
||
|
|
is_recommended=True,
|
||
|
|
selection_source="system",
|
||
|
|
metadata={"cluster_summaries": []},
|
||
|
|
)
|
||
|
|
for cell in self.cells:
|
||
|
|
RemoteSensingSubdivisionOptionAssignment.objects.create(
|
||
|
|
option=self.option_k1,
|
||
|
|
cell=cell,
|
||
|
|
cluster_label=0,
|
||
|
|
raw_feature_values={"ndvi": 0.4},
|
||
|
|
scaled_feature_values={"ndvi": 0.0},
|
||
|
|
)
|
||
|
|
RemoteSensingSubdivisionOptionBlock.objects.create(
|
||
|
|
option=self.option_k1,
|
||
|
|
cluster_label=0,
|
||
|
|
sub_block_code="cluster-0",
|
||
|
|
chunk_size_sqm=900,
|
||
|
|
centroid_lat="35.689100",
|
||
|
|
centroid_lon="51.389100",
|
||
|
|
center_cell_code="cell-0",
|
||
|
|
center_cell_lat="35.689050",
|
||
|
|
center_cell_lon="51.389050",
|
||
|
|
cell_count=2,
|
||
|
|
cell_codes=[cell.cell_code for cell in self.cells],
|
||
|
|
geometry=self.boundary,
|
||
|
|
metadata={
|
||
|
|
"source": "analysis_grid_cells",
|
||
|
|
"center_selection": {"strategy": "coordinate_1_center", "center_cell_code": "cell-0"},
|
||
|
|
},
|
||
|
|
)
|
||
|
|
for index, cell in enumerate(self.cells):
|
||
|
|
RemoteSensingSubdivisionOptionAssignment.objects.create(
|
||
|
|
option=self.option_k2,
|
||
|
|
cell=cell,
|
||
|
|
cluster_label=index,
|
||
|
|
raw_feature_values={"ndvi": 0.4 + index},
|
||
|
|
scaled_feature_values={"ndvi": float(index)},
|
||
|
|
)
|
||
|
|
RemoteSensingSubdivisionOptionBlock.objects.create(
|
||
|
|
option=self.option_k2,
|
||
|
|
cluster_label=index,
|
||
|
|
sub_block_code=f"cluster-{index}",
|
||
|
|
chunk_size_sqm=900,
|
||
|
|
centroid_lat=f"{35.68905 + (index * 0.0001):.6f}",
|
||
|
|
centroid_lon=f"{51.38905 + (index * 0.0001):.6f}",
|
||
|
|
center_cell_code=cell.cell_code,
|
||
|
|
center_cell_lat=f"{35.68905 + (index * 0.0001):.6f}",
|
||
|
|
center_cell_lon=f"{51.38905 + (index * 0.0001):.6f}",
|
||
|
|
cell_count=1,
|
||
|
|
cell_codes=[cell.cell_code],
|
||
|
|
geometry=cell.geometry,
|
||
|
|
metadata={
|
||
|
|
"source": "analysis_grid_cells",
|
||
|
|
"center_selection": {"strategy": "coordinate_1_center", "center_cell_code": cell.cell_code},
|
||
|
|
},
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_get_k_options_returns_all_persisted_options(self):
|
||
|
|
response = self.client.get(
|
||
|
|
f"/remote-sensing/results/{self.result.id}/k-options/"
|
||
|
|
)
|
||
|
|
|
||
|
|
self.assertEqual(response.status_code, 200)
|
||
|
|
payload = response.json()["data"]
|
||
|
|
self.assertEqual(payload["result_id"], self.result.id)
|
||
|
|
self.assertEqual(payload["active_requested_k"], 1)
|
||
|
|
self.assertEqual(payload["recommended_requested_k"], 2)
|
||
|
|
self.assertEqual([item["requested_k"] for item in payload["options"]], [1, 2])
|
||
|
|
self.assertEqual(payload["options"][0]["cluster_blocks"][0]["center_cell_code"], "cell-0")
|
||
|
|
|
||
|
|
@patch("location_data.data_driven_subdivision.render_elbow_plot", return_value=None)
|
||
|
|
def test_post_activate_k_marks_selected_option_active_and_syncs_result(self, _mock_plot):
|
||
|
|
response = self.client.post(
|
||
|
|
f"/remote-sensing/results/{self.result.id}/k-options/activate/",
|
||
|
|
data={"requested_k": 2},
|
||
|
|
format="json",
|
||
|
|
)
|
||
|
|
|
||
|
|
self.assertEqual(response.status_code, 200)
|
||
|
|
payload = response.json()["data"]
|
||
|
|
self.assertEqual(payload["activated_requested_k"], 2)
|
||
|
|
self.assertEqual(payload["subdivision_result"]["cluster_count"], 2)
|
||
|
|
self.assertEqual(
|
||
|
|
payload["subdivision_result"]["metadata"]["active_requested_k"],
|
||
|
|
2,
|
||
|
|
)
|
||
|
|
self.assertEqual(len(payload["subdivision_result"]["cluster_blocks"]), 2)
|
||
|
|
self.assertEqual(
|
||
|
|
payload["subdivision_result"]["cluster_blocks"][0]["center_cell_code"],
|
||
|
|
"cell-0",
|
||
|
|
)
|
||
|
|
|
||
|
|
self.option_k1.refresh_from_db()
|
||
|
|
self.option_k2.refresh_from_db()
|
||
|
|
self.assertFalse(self.option_k1.is_active)
|
||
|
|
self.assertTrue(self.option_k2.is_active)
|
||
|
|
self.assertEqual(self.option_k2.selection_source, "user")
|
||
|
|
|
||
|
|
self.result.refresh_from_db()
|
||
|
|
self.assertEqual(self.result.cluster_count, 2)
|
||
|
|
self.assertEqual(self.result.assignments.count(), 2)
|
||
|
|
self.assertEqual(self.result.cluster_blocks.count(), 2)
|
||
|
|
self.assertEqual(self.result.cluster_blocks.order_by("cluster_label").first().center_cell_code, "cell-0")
|