# -*- coding: utf-8 -*- """ CropAI Solar-Powered Smart Farm IoT Sensor Hub & LoRaWAN Telemetry Engine Ingests continuous edge microclimate and soil moisture packets from low-power solar ESP32 / LoRaWAN Class A field nodes (865-867 MHz India band). Monitors: 1. Dual-depth Volumetric Water Content (VWC at 15cm & 30cm) 2. Permanent Wilting Point (PWP) Deficit & Irrigation Automation Alerts 3. Leaf Wetness Duration for Bio-Twin Spore Germination coupling 4. Solar LiPo Battery Voltage (V_bat) & Brownout Safeguards """ from typing import Dict, Any, Optional, List, Tuple import logging import time from datetime import datetime, timezone logger = logging.getLogger("cropai.iot_telemetry") Tuple_Status = Tuple[str, str, str] # ========================================================================= # In-Memory Telemetry & Device Registry Store # ========================================================================= ACTIVE_IOT_DEVICES: Dict[str, Dict[str, Any]] = { "NODE-SURGUJA-01": { "device_id": "NODE-SURGUJA-01", "name": "Paddy Field North Parcel (धान खेत उत्तरी खंड)", "location": "Lundra Block, Surguja (C.G.)", "crop": "Rice / धान (Swarna Sub-1)", "hardware_type": "ESP32-S3 + Semtech SX1262 LoRaWAN Node", "power_source": "5W Monocrystalline Solar Panel + 2500mAh 18650 Li-ion", "registered_at": "2026-08-15T06:00:00Z", "last_seen_utc": datetime.now(timezone.utc).isoformat(), "status": "ONLINE", "latest_packet": { "battery_voltage_v": 3.94, "solar_charging": True, "rssi_dbm": -82, "snr_db": 9.2, "soil_moisture_vwc_15cm": 28.5, "soil_moisture_vwc_30cm": 31.0, "canopy_temp_c": 26.8, "canopy_rh_pct": 84.0, "leaf_wetness_pct": 40.0 } }, "NODE-SURGUJA-02": { "device_id": "NODE-SURGUJA-02", "name": "Vegetable & Tomato Patch (टमाटर बाड़ी)", "location": "Ambikapur Rural, Surguja (C.G.)", "crop": "Tomato / टमाटर (Abhilash)", "hardware_type": "ESP32-C3 + SX1262 LoRaWAN Node", "power_source": "5W Solar Panel + 2500mAh 18650 Li-ion", "registered_at": "2026-09-01T08:30:00Z", "last_seen_utc": datetime.now(timezone.utc).isoformat(), "status": "ONLINE", "latest_packet": { "battery_voltage_v": 3.86, "solar_charging": True, "rssi_dbm": -89, "snr_db": 7.8, "soil_moisture_vwc_15cm": 22.0, "soil_moisture_vwc_30cm": 25.4, "canopy_temp_c": 27.5, "canopy_rh_pct": 76.0, "leaf_wetness_pct": 25.0 } } } def evaluate_moisture_status(vwc_pct: float) -> Tuple_Status: """Evaluates soil moisture status against agronomic thresholds.""" if vwc_pct <= 14.0: return "PERMANENT_WILTING_POINT", "अत्यधिक सूखा - पौधे मुरझा रहे हैं!", "Critical drought stress below permanent wilting point" elif vwc_pct < 20.0: return "MOISTURE_DEFICIT", "हल्की सिंचाई की आवश्यकता है", "Moisture deficit; schedule irrigation within 24h" elif vwc_pct <= 35.0: return "OPTIMAL_FIELD_CAPACITY", "मिट्टी में आदर्श नमी है", "Optimal field capacity; robust root uptake" else: return "SOIL_SATURATED", "खेत में पानी भरा है", "Soil saturation / flood conditions" def ingest_telemetry_packet(packet: Dict[str, Any]) -> Dict[str, Any]: """ Ingests and validates raw telemetry transmission from field node: 1. Validates physical sensor ranges. 2. Evaluates Permanent Wilting Point (PWP) deficit. 3. Checks battery voltage against brownout thresholds. 4. Triggers emergency agricultural alerts. """ device_id = packet.get("device_id", "NODE-SURGUJA-UNKNOWN") v_bat = float(packet.get("battery_voltage_v", 3.7)) solar_chg = bool(packet.get("solar_charging", False)) rssi = int(packet.get("rssi_dbm", -90)) snr = float(packet.get("snr_db", 6.0)) vwc_15 = float(packet.get("soil_moisture_vwc_15cm", 25.0)) vwc_30 = float(packet.get("soil_moisture_vwc_30cm", 28.0)) temp = float(packet.get("canopy_temp_c", 26.0)) rh = float(packet.get("canopy_rh_pct", 75.0)) wetness = float(packet.get("leaf_wetness_pct", 30.0)) # Evaluate moisture status moist_code, moist_hi, moist_en = evaluate_moisture_status(vwc_15) # Battery health if v_bat < 3.4: batt_status = "BROWNOUT_WARNING" batt_desc = "बैटरी कम (Low Battery < 3.4V) - रेडियो पावर सेव मोड सक्रिय" elif v_bat >= 4.0: batt_status = "FULLY_CHARGED" batt_desc = "सोलर से पूर्ण चार्ज (Fully Charged)" else: batt_status = "HEALTHY_NORMAL" batt_desc = "बैटरी स्वस्थ (Healthy Nominal)" # RF link health rf_health = "EXCELLENT" if rssi >= -80 else ("GOOD" if rssi >= -95 else "WEAK_MARGINAL") # Generate automated alerts active_alarms = [] if moist_code == "PERMANENT_WILTING_POINT": active_alarms.append({ "alarm_type": "CRITICAL_SOIL_DROUGHT", "message_hi": "🚨 तत्काल सिंचाई करें! 15 सेमी गहराई पर मिट्टी की नमी 14% से नीचे आ गई है।", "message_en": "Critical Soil Drought: VWC at 15cm is <= 14%. Activate irrigation immediately." }) if temp <= 2.0: active_alarms.append({ "alarm_type": "FROST_CANOPY_DROP", "message_hi": "❄️ पाला अलर्ट: कैनोपी का तापमान 2°C से नीचे गिर गया है। धुआं व रात्रि पानी चालू करें!", "message_en": "Nocturnal freezing drop: Canopy temp <= 2.0°C. Deploy thermal mitigation." }) if wetness > 80.0 and (24.0 <= temp <= 28.0): active_alarms.append({ "alarm_type": "BIOTWIN_FUNGAL_SPORULATION_WINDOW", "message_hi": "🍄 फंगस बीजाणु पनपने का अनुकूल मौसम: पत्तियां 80% से अधिक गीली हैं और तापमान 26°C है।", "message_en": "High disease risk: Leaf wetness > 80% at 26°C indicates active fungal infection window." }) timestamp = datetime.now(timezone.utc).isoformat() # Update device registry if device_id in ACTIVE_IOT_DEVICES: ACTIVE_IOT_DEVICES[device_id]["last_seen_utc"] = timestamp ACTIVE_IOT_DEVICES[device_id]["status"] = "ONLINE" ACTIVE_IOT_DEVICES[device_id]["latest_packet"] = { "battery_voltage_v": v_bat, "solar_charging": solar_chg, "rssi_dbm": rssi, "snr_db": snr, "soil_moisture_vwc_15cm": vwc_15, "soil_moisture_vwc_30cm": vwc_30, "canopy_temp_c": temp, "canopy_rh_pct": rh, "leaf_wetness_pct": wetness } return { "status": "success", "device_id": device_id, "packet_timestamp_utc": timestamp, "soil_moisture_assessment": { "vwc_15cm_percentage": vwc_15, "vwc_30cm_percentage": vwc_30, "classification": moist_code, "status_hi": moist_hi, "status_en": moist_en }, "microclimate": { "canopy_temp_c": temp, "canopy_rh_pct": rh, "leaf_wetness_pct": wetness }, "hardware_telemetry": { "battery_voltage_v": v_bat, "battery_health": batt_status, "battery_description_hi": batt_desc, "solar_charging": solar_chg, "rf_link_rssi_dbm": rssi, "rf_link_quality": rf_health }, "active_alarms": active_alarms } def get_registered_devices() -> List[Dict[str, Any]]: """Returns list of registered field IoT nodes and their operational status.""" return list(ACTIVE_IOT_DEVICES.values()) def get_live_device_telemetry(device_id: str) -> Dict[str, Any]: """Returns live telemetry and recent status for a specific IoT node.""" device = ACTIVE_IOT_DEVICES.get(device_id) if not device: # Fallback simulated dynamic node return { "status": "device_not_registered", "device_id": device_id, "message": "Device not found in active registry. Ingest first packet to auto-register." } return { "status": "success", "device": device }