third / models /data_loader.py
twist-turn's picture
Upload 22 files
1c796e1 verified
Raw History Blame Contribute Delete
2.97 kB
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.",
}