File size: 2,968 Bytes
1c796e1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
import os
from datetime import datetime, timedelta
import random
from dataclasses import dataclass
from typing import List, Dict

@dataclass
class NDVIPoint:
    date: datetime
    ndvi: float
    evi: float

class DataManager:
    def __init__(self, config_cls):
        self.cfg = config_cls

    def load_ndvi_series(self, region: str) -> List[NDVIPoint]:
        days = self.cfg.NDVI_LENGTH_DAYS
        today = datetime.today()
        series = []
        base = 0.5 + (hash(region) % 20) / 100.0
        for i in range(days):
            d = today - timedelta(days=days - i)
            seasonal = 0.2 * (1 + (i / days))
            noise = random.uniform(-0.05, 0.05)
            ndvi = max(0.1, min(0.9, base + seasonal * 0.3 + noise))
            evi = max(0.05, min(0.8, ndvi - 0.05 + random.uniform(-0.02, 0.02)))
            series.append(NDVIPoint(date=d, ndvi=ndvi, evi=evi))
        return series

    def load_weather_forecast(self, region: str) -> Dict:
        today = datetime.today()
        forecast = []
        for i in range(30):
            d = today + timedelta(days=i)
            temp = 20 + (hash(region + "t") % 8) + (i * 0.1)
            rain = max(0.0, random.gauss(3.0, 2.0))
            forecast.append(
                {
                    "date": d.strftime("%Y-%m-%d"),
                    "tmax": round(temp + random.uniform(0, 4), 1),
                    "tmin": round(temp - random.uniform(4, 6), 1),
                    "rain": round(rain, 2),
                }
            )
        volatility = sum(abs(day["tmax"] - day["tmin"]) for day in forecast) / len(forecast)
        return {"region": region, "daily": forecast, "volatility": round(volatility, 2)}

    def load_soil_and_terrain(self, region: str) -> Dict:
        elevation = 100 + (hash(region + "elev") % 250)
        slope = round(1 + (hash(region + "slope") % 14), 1)
        texture = ["Sandy loam", "Clay loam", "Silty clay", "Loamy sand"][hash(region) % 4]
        organic = round(1.0 + (hash(region + "org") % 40) / 10.0, 1)
        drainage = ["Well drained", "Moderately drained", "Poorly drained"][hash(region) % 3]
        score = max(0.0, min(1.0, 0.2 + organic / 10.0 + (15 - abs(slope - 5)) / 40.0))
        return {
            "elevation_m": elevation,
            "slope_deg": slope,
            "texture": texture,
            "organic_matter_pct": organic,
            "drainage": drainage,
            "soil_score": round(score, 2),
        }

    def load_ag_commons_metadata(self) -> Dict:
        return {
            "sources": [
                "Landsat 8/9 NDVI (synthetic stub)",
                "MODIS NDVI/EVI (synthetic stub)",
                "Open-Meteo weather forecast (offline stub)",
                "Copernicus terrain + soil (offline stub)",
                "USDA Ag Data Commons (offline stub)",
            ],
            "note": "Replace synthetic loaders with real API and dataset integrations for production.",
        }