sajad-dev/Ai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

UPDATE

Browse Source
This commit is contained in:
Mohammad Sajad Pourajam
2026-04-29 01:27:29 +03:30
parent cb60254c81
commit 27bdad0111
12 changed files with 1443 additions and 155 deletions
Download Patch File Download Diff File Expand all files Collapse all files
location_data/apps.py
+19
View File
@@ -1,6 +1,7 @@
from functools import cached_property
from django.apps import AppConfig
from django.conf import settings
class SoilDataConfig(AppConfig):
@@ -14,5 +15,23 @@ class SoilDataConfig(AppConfig):
return NdviHealthService()
@cached_property
def soil_data_adapter(self):
from .soil_adapters import MockSoilDataAdapter, SoilGridsAdapter
provider = getattr(settings, "SOIL_DATA_PROVIDER", "mock")
if provider == "soilgrids":
return SoilGridsAdapter(
timeout=getattr(settings, "SOILGRIDS_TIMEOUT_SECONDS", 60)
)
if provider == "mock":
return MockSoilDataAdapter(
delay_seconds=getattr(settings, "SOIL_MOCK_DELAY_SECONDS", 0.8)
)
raise ValueError(f"Unsupported soil data provider: {provider}")
def get_ndvi_health_service(self):
return self.ndvi_health_service
def get_soil_data_adapter(self):
return self.soil_data_adapter
location_data/serializers.py
+1 -2
View File
@@ -1,6 +1,7 @@
from rest_framework import serializers
from .models import SoilDepthData, SoilLocation
from .soil_adapters import DEPTHS
class SoilDataRequestSerializer(serializers.Serializer):
@@ -56,8 +57,6 @@ class SoilLocationResponseSerializer(serializers.ModelSerializer):
fields = ["id", "lon", "lat", "depths"]
def get_depths(self, obj):
from .tasks import DEPTHS
depth_qs = obj.depths.all()
order = {d: i for i, d in enumerate(DEPTHS)}
sorted_depths = sorted(
location_data/soil_adapters.py
+286
View File
@@ -0,0 +1,286 @@
from __future__ import annotations
import hashlib
import math
import random
import time
from abc import ABC, abstractmethod
try:
import requests
except ImportError: # pragma: no cover - handled when live adapter is used
requests = None
SOILGRIDS_BASE = "https://rest.isric.org/soilgrids/v2.0/properties/query"
PROPERTIES = [
"bdod",
"cec",
"cfvo",
"clay",
"nitrogen",
"ocd",
"ocs",
"phh2o",
"sand",
"silt",
"soc",
"wv0010",
"wv0033",
"wv1500",
]
VALUES = ["Q0.5", "Q0.05", "Q0.95", "mean", "uncertainty"]
DEPTHS = ["0-5cm", "5-15cm", "15-30cm"]
DEPTH_INDEX = {depth: index for index, depth in enumerate(DEPTHS)}
def _clamp(value: float, lower: float, upper: float) -> float:
return max(lower, min(upper, value))
def _round_field(name: str, value: float) -> float:
if name in {"nitrogen", "soc", "ocs", "wv0010", "wv0033", "wv1500"}:
return round(value, 3)
return round(value, 2)
class BaseSoilDataAdapter(ABC):
source_name = "base"
@abstractmethod
def fetch_depth_fields(self, lon: float, lat: float, depth: str) -> dict:
"""Return normalized field values for a single soil depth."""
class SoilGridsAdapter(BaseSoilDataAdapter):
source_name = "soilgrids"
def __init__(self, base_url: str = SOILGRIDS_BASE, timeout: float = 60):
self.base_url = base_url
self.timeout = timeout
def fetch_depth_fields(self, lon: float, lat: float, depth: str) -> dict:
if requests is None:
raise RuntimeError("requests package is required for SoilGridsAdapter")
params = {
"lon": lon,
"lat": lat,
"depth": depth,
}
for prop in PROPERTIES:
params.setdefault("property", []).append(prop)
for value in VALUES:
params.setdefault("value", []).append(value)
response = requests.get(
self.base_url,
params=params,
headers={"accept": "application/json"},
timeout=self.timeout,
)
response.raise_for_status()
return self._parse_response_to_fields(response.json())
def _parse_response_to_fields(self, data: dict) -> dict:
fields = {prop: None for prop in PROPERTIES}
layers = data.get("properties", {}).get("layers", [])
for layer in layers:
name = layer.get("name")
if name not in fields:
continue
depths_list = layer.get("depths", [])
if not depths_list:
continue
values = depths_list[0].get("values", {})
mean_value = values.get("mean")
if mean_value is not None:
fields[name] = float(mean_value)
return fields
class MockSoilDataAdapter(BaseSoilDataAdapter):
source_name = "mock"
def __init__(
self,
delay_seconds: float = 0.8,
seed_namespace: str = "croplogic-soil",
):
self.delay_seconds = max(0.0, delay_seconds)
self.seed_namespace = seed_namespace
def fetch_depth_fields(self, lon: float, lat: float, depth: str) -> dict:
if depth not in DEPTH_INDEX:
raise ValueError(f"Unsupported soil depth: {depth}")
if self.delay_seconds:
time.sleep(self.delay_seconds)
depth_index = DEPTH_INDEX[depth]
texture_score = self._layered_noise(lon, lat, "texture")
organic_score = self._layered_noise(lon, lat, "organic")
moisture_score = self._layered_noise(lon, lat, "moisture")
mineral_score = self._layered_noise(lon, lat, "mineral")
stone_score = self._layered_noise(lon, lat, "stone")
ph_score = self._layered_noise(lon, lat, "ph")
sand, clay, silt = self._build_texture(
texture_score=texture_score,
organic_score=organic_score,
depth_index=depth_index,
)
soc = _clamp(
0.7
+ (organic_score * 1.9)
+ (clay * 0.012)
- (depth_index * 0.28)
+ ((1 - moisture_score) * 0.08),
0.45,
4.2,
)
nitrogen = _clamp(
0.04
+ (soc * 0.085)
+ ((1 - (sand / 100.0)) * 0.025)
+ ((2 - depth_index) * 0.008),
0.03,
0.42,
)
ocd = _clamp(
10.0 + (soc * 8.5) + (organic_score * 4.0) - (depth_index * 2.6),
7.0,
46.0,
)
ocs = _clamp(
1.0 + (soc * 1.55) - (depth_index * 0.28) + (organic_score * 0.12),
0.5,
8.5,
)
cec = _clamp(
7.0
+ (clay * 0.33)
+ (soc * 1.7)
+ ((1 - (sand / 100.0)) * 2.6)
+ (mineral_score * 1.4),
5.0,
38.0,
)
cfvo = _clamp(1.0 + (stone_score * 12.0) + (depth_index * 2.4), 0.0, 35.0)
bdod = _clamp(
1.06
+ (sand * 0.0038)
+ (depth_index * 0.06)
- (soc * 0.035)
+ (stone_score * 0.03),
0.95,
1.62,
)
phh2o = _clamp(
6.2
+ ((ph_score - 0.5) * 1.1)
+ (depth_index * 0.08)
- (organic_score * 0.12),
5.6,
8.1,
)
wv1500 = _clamp(
0.05
+ (clay * 0.0016)
+ (soc * 0.012)
- (sand * 0.0003)
+ (depth_index * 0.004),
0.05,
0.22,
)
wv0033 = _clamp(
wv1500 + 0.07 + (clay * 0.0015) + (soc * 0.01) - (sand * 0.0002),
wv1500 + 0.04,
0.38,
)
wv0010 = _clamp(
wv0033 + 0.03 + (soc * 0.006) + (moisture_score * 0.01),
wv0033 + 0.015,
0.48,
)
fields = {
"bdod": bdod,
"cec": cec,
"cfvo": cfvo,
"clay": clay,
"nitrogen": nitrogen,
"ocd": ocd,
"ocs": ocs,
"phh2o": phh2o,
"sand": sand,
"silt": silt,
"soc": soc,
"wv0010": wv0010,
"wv0033": wv0033,
"wv1500": wv1500,
}
return {name: _round_field(name, value) for name, value in fields.items()}
def _build_texture(
self,
texture_score: float,
organic_score: float,
depth_index: int,
) -> tuple[float, float, float]:
sand = _clamp(
30.0
+ (texture_score * 28.0)
+ ((organic_score - 0.5) * 3.5)
- (depth_index * 2.5),
18.0,
72.0,
)
clay = _clamp(
13.0
+ ((1 - texture_score) * 18.0)
+ (depth_index * 5.5)
+ ((organic_score - 0.5) * 2.0),
8.0,
42.0,
)
minimum_silt = 12.0
total = sand + clay
if total > 100.0 - minimum_silt:
excess = total - (100.0 - minimum_silt)
sand -= excess * 0.65
clay -= excess * 0.35
silt = 100.0 - sand - clay
return sand, clay, silt
def _layered_noise(self, lon: float, lat: float, key: str) -> float:
regional = self._smooth_noise(lon, lat, f"{key}:regional", scale=1.7)
local = self._smooth_noise(lon, lat, f"{key}:local", scale=0.32)
micro = self._smooth_noise(lon, lat, f"{key}:micro", scale=0.08)
return _clamp((regional * 0.55) + (local * 0.3) + (micro * 0.15), 0.0, 1.0)
def _smooth_noise(self, lon: float, lat: float, key: str, scale: float) -> float:
grid_x = lon / scale
grid_y = lat / scale
x0 = math.floor(grid_x)
y0 = math.floor(grid_y)
tx = grid_x - x0
ty = grid_y - y0
v00 = self._cell_noise(key, x0, y0)
v10 = self._cell_noise(key, x0 + 1, y0)
v01 = self._cell_noise(key, x0, y0 + 1)
v11 = self._cell_noise(key, x0 + 1, y0 + 1)
tx = tx * tx * (3.0 - (2.0 * tx))
ty = ty * ty * (3.0 - (2.0 * ty))
top = (v00 * (1 - tx)) + (v10 * tx)
bottom = (v01 * (1 - tx)) + (v11 * tx)
return (top * (1 - ty)) + (bottom * ty)
def _cell_noise(self, key: str, grid_x: int, grid_y: int) -> float:
seed_input = f"{self.seed_namespace}:{key}:{grid_x}:{grid_y}"
digest = hashlib.sha256(seed_input.encode("ascii")).digest()
seed = int.from_bytes(digest[:8], "big", signed=False)
return random.Random(seed).random()
location_data/tasks.py
+17 -58
View File
@@ -1,63 +1,22 @@
"""
تسک‌های Celery برای واکشی داده‌های خاک از API SoilGrids.
تسک‌های Celery برای واکشی داده‌های خاک.
"""
from decimal import Decimal
import requests
from config.celery import app
from django.apps import apps
from django.db import transaction
from .models import SoilDepthData, SoilLocation
from .soil_adapters import DEPTHS
SOILGRIDS_BASE = "https://rest.isric.org/soilgrids/v2.0/properties/query"
PROPERTIES = [
"bdod", "cec", "cfvo", "clay", "nitrogen", "ocd", "ocs",
"phh2o", "sand", "silt", "soc", "wv0010", "wv0033", "wv1500",
]
VALUES = ["Q0.5", "Q0.05", "Q0.95", "mean", "uncertainty"]
DEPTHS = ["0-5cm", "5-15cm", "15-30cm"]
def _fetch_soilgrids(lon: float, lat: float, depth: str) -> dict | None:
"""درخواست به API SoilGrids برای یک عمق."""
params = {
"lon": lon,
"lat": lat,
"depth": depth,
}
for p in PROPERTIES:
params.setdefault("property", []).append(p)
for v in VALUES:
params.setdefault("value", []).append(v)
resp = requests.get(
SOILGRIDS_BASE,
params=params,
headers={"accept": "application/json"},
timeout=60,
)
resp.raise_for_status()
return resp.json()
def _parse_response_to_fields(data: dict) -> dict:
"""
استخراج مقادیر mean از response و ساخت dict مناسب برای SoilDepthData.
"""
fields = {p: None for p in PROPERTIES}
layers = data.get("properties", {}).get("layers", [])
for layer in layers:
name = layer.get("name")
if name not in fields:
continue
depths_list = layer.get("depths", [])
if depths_list:
values = depths_list[0].get("values", {})
mean_val = values.get("mean")
if mean_val is not None:
fields[name] = float(mean_val)
return fields
try:
import requests
except ImportError: # pragma: no cover - handled in stripped envs
RequestException = Exception
else:
RequestException = requests.RequestException
def fetch_soil_data_for_coordinates(
@@ -72,6 +31,7 @@ def fetch_soil_data_for_coordinates(
"""
lat = Decimal(str(round(float(latitude), 6)))
lon = Decimal(str(round(float(longitude), 6)))
adapter = apps.get_app_config("location_data").get_soil_data_adapter()
with transaction.atomic():
location, created = SoilLocation.objects.select_for_update().get_or_create(
@@ -83,18 +43,17 @@ def fetch_soil_data_for_coordinates(
location.task_id = task_id
location.save(update_fields=["task_id"])
for i, depth in enumerate(DEPTHS):
for index, depth in enumerate(DEPTHS):
if progress_callback is not None:
progress_callback(
state="PROGRESS",
meta={
"current": i + 1,
"current": index + 1,
"total": len(DEPTHS),
"message": f"در حال واکشی عمق {depth}...",
},
)
data = _fetch_soilgrids(float(lon), float(lat), depth)
fields = _parse_response_to_fields(data)
fields = adapter.fetch_depth_fields(float(lon), float(lat), depth)
with transaction.atomic():
SoilDepthData.objects.update_or_create(
soil_location=location,
@@ -117,8 +76,8 @@ def fetch_soil_data_for_coordinates(
@app.task(bind=True)
def fetch_soil_data_task(self, latitude: float, longitude: float):
"""
واکشی داده‌های خاک برای مختصات داده‌شده از SoilGrids و ذخیره در DB.
برای هر عمق (0-5cm, 5-15cm, 15-30cm) یک ریکوئست جدا زده می‌شود.
واکشی داده‌های خاک برای مختصات داده‌شده و ذخیره در DB.
برای هر عمق (0-5cm, 5-15cm, 15-30cm) یک ریکوئست/شبیه‌سازی جدا انجام می‌شود.
"""
try:
return fetch_soil_data_for_coordinates(
@@ -127,12 +86,12 @@ def fetch_soil_data_task(self, latitude: float, longitude: float):
task_id=self.request.id,
progress_callback=self.update_state,
)
except requests.RequestException as e:
except RequestException as exc:
lat = Decimal(str(round(float(latitude), 6)))
lon = Decimal(str(round(float(longitude), 6)))
location = SoilLocation.objects.filter(latitude=lat, longitude=lon).first()
return {
"status": "error",
"location_id": getattr(location, "id", None),
"error": str(e),
"error": str(exc),
}
location_data/test_soil_adapters.py
+92
View File
@@ -0,0 +1,92 @@
from __future__ import annotations
from django.apps import apps
from django.test import SimpleTestCase, TestCase, override_settings
from location_data.models import SoilDepthData, SoilLocation
from location_data.soil_adapters import (
DEPTHS,
MockSoilDataAdapter,
SoilGridsAdapter,
)
from location_data.tasks import fetch_soil_data_for_coordinates
class MockSoilDataAdapterTests(SimpleTestCase):
def setUp(self):
self.adapter = MockSoilDataAdapter(delay_seconds=0)
def test_same_coordinate_returns_same_values(self):
first = self.adapter.fetch_depth_fields(51.4, 35.71, "0-5cm")
second = self.adapter.fetch_depth_fields(51.4, 35.71, "0-5cm")
self.assertEqual(first, second)
def test_nearby_coordinates_produce_nearby_values(self):
first = self.adapter.fetch_depth_fields(51.4, 35.71, "0-5cm")
second = self.adapter.fetch_depth_fields(51.405, 35.715, "0-5cm")
self.assertLess(abs(first["sand"] - second["sand"]), 4.5)
self.assertLess(abs(first["clay"] - second["clay"]), 4.5)
self.assertLess(abs(first["phh2o"] - second["phh2o"]), 0.35)
self.assertLess(abs(first["wv1500"] - second["wv1500"]), 0.03)
def test_depth_profiles_follow_expected_trend(self):
shallow = self.adapter.fetch_depth_fields(51.4, 35.71, "0-5cm")
medium = self.adapter.fetch_depth_fields(51.4, 35.71, "5-15cm")
deep = self.adapter.fetch_depth_fields(51.4, 35.71, "15-30cm")
self.assertGreaterEqual(deep["bdod"], medium["bdod"])
self.assertGreaterEqual(medium["bdod"], shallow["bdod"])
self.assertLessEqual(deep["soc"], medium["soc"])
self.assertLessEqual(medium["soc"], shallow["soc"])
class SoilDataAdapterSelectionTests(SimpleTestCase):
def tearDown(self):
apps.get_app_config("location_data").__dict__.pop("soil_data_adapter", None)
@override_settings(SOIL_DATA_PROVIDER="mock", SOIL_MOCK_DELAY_SECONDS=0)
def test_app_config_returns_mock_adapter(self):
config = apps.get_app_config("location_data")
config.__dict__.pop("soil_data_adapter", None)
adapter = config.get_soil_data_adapter()
self.assertIsInstance(adapter, MockSoilDataAdapter)
@override_settings(SOIL_DATA_PROVIDER="soilgrids", SOILGRIDS_TIMEOUT_SECONDS=12)
def test_app_config_returns_live_adapter(self):
config = apps.get_app_config("location_data")
config.__dict__.pop("soil_data_adapter", None)
adapter = config.get_soil_data_adapter()
self.assertIsInstance(adapter, SoilGridsAdapter)
self.assertEqual(adapter.timeout, 12)
@override_settings(SOIL_DATA_PROVIDER="mock", SOIL_MOCK_DELAY_SECONDS=0)
class SoilDataFetchTests(TestCase):
def test_fetch_soil_data_for_coordinates_persists_three_depths(self):
result = fetch_soil_data_for_coordinates(latitude=35.71, longitude=51.4)
self.assertEqual(result["status"], "completed")
self.assertEqual(result["depths"], DEPTHS)
location = SoilLocation.objects.get(latitude="35.710000", longitude="51.400000")
self.assertEqual(location.depths.count(), 3)
self.assertTrue(location.is_complete)
self.assertCountEqual(
list(location.depths.values_list("depth_label", flat=True)),
DEPTHS,
)
self.assertTrue(
SoilDepthData.objects.filter(
soil_location=location,
depth_label="0-5cm",
sand__isnull=False,
clay__isnull=False,
wv1500__isnull=False,
).exists()
)