xGjdZddlZddlmZddlmZddlmZddlm Z ddl m Z ddl m Z eeZddlZd ZeZd Zed d Zdededededeef dZde de dedededeeeff dZy)u Aggregation layer. Runs all 6 models concurrently using a thread pool (model inference via joblib/xgboost/statsmodels is CPU-bound and releases the GIL during the heavy numeric work, so threads — not full async — give real parallelism here without the complexity of multiprocessing). This module is called from the Celery task (app/workers/tasks.py), NOT directly from an API request handler — see the architecture decision in docs/architecture.md. That keeps slow ML work off the web-serving threads entirely. N)ThreadPoolExecutor)Any) get_logger) ModelRegistry) model_runners)FeatureBuilderc 6ddl}|jj|jj|jj t ddddd} t |5}tj|cdddS#1swYyxYw#icYSxYw)Nrz..datalookupz crops.json) ospathnormpathjoindirname__file__open_jsonload)r r fs DC:\Crop_Prediction\Backend\crop-ai-system\app\services\aggregator.py _load_cropsrsy 77  BGGLL)BD$PVX`bno pD $Z !1::a= ! ! ! s* B'B< BB BBBc`tj|j}|r|ddz SdS)N water_req_mmg(@)_CROPSgetlower)croppropss r_monthly_water_reqr&s- JJtzz| $E-2E. !D (<<z model-infer) max_workersthread_name_prefixclimate feasibility yield_predtrendreturncg}|jddk(r%|jddk(r|jd|jddk(r7|jdddkr"|jd s|jd |jd d k(r%|jddk(r|jd |S)Nlabelsuitable anomaly_labelhighuqClimate anomaly is high but feasibility model says suitable — treat the suitable label with caution this month.expected_yield_ton_harg?degradedzGYield prediction is unusually low despite a suitable feasibility label.r'downwarduMulti-year yield trend is downward even though this month looks suitable — consider the longer-term pattern, not just this month.)rappend)r$r%r&r' conflictss r_detect_conflictsr3.sIw:-'++o2NRX2X @  J. NN2A 6 <z* U  yyZ'KOOG,D ,R E r registryfeature_builderdistrictrmonthc tj}|j|||}tj t j |||tj t j||||tj t j||tj t j||tj t j||||j|tj t j|||d}|jD cic]\}} || j} }} t| d| d| d| d} | jD cgc]*\} } t| t s| j#ds)| ,}} } |rt$j'd|t)tj|z dz}| d| d| d| d| d | d | ||d Scc} }wcc} } w) u Synchronous, blocking function — this is exactly what we want inside a Celery worker task. Building it as a plain function (not async) keeps it trivially testable and avoids mixing asyncio with the thread-pool model calls. )r$r%yield_predictionr'recommendationmarketr$r%r9r'r/z)Analysis completed with degraded models: ir:r;) r$r%r9r'r:r;r2degraded_models processing_ms)time perf_counterbuild _executorsubmitrrun_climate_modelrun_feasibility_modelrun_yield_modelrun_trend_modelrun_recommendation_modelrun_market_modelitemsresultr3 isinstancedictrloggerwarningint)r4r5r6rr7startfeature_vectorfutureskeyfutureresultsr2kvr< elapsed_mss rrun_full_analysisrYKs    E$**8T5AN##M$C$CXxY^_ '' (K(KXWegkmuv%,,]-J-JHVde!!-"?"?>Z#**  2 2 h'<'concurrent.futuresrtypingrapp.core.logging_configrapp.models_loader.registryr app.servicesrapp.services.feature_builderr__name__rMjsonrrrrrArLliststrr3rOrYr rrfs  1.4&7 H  = 1 O   $26?C #Y:44#44  4  4  #s(^ 4r