Surqo Data Flow: From Sensor Reading to AI Recommendation

Every piece of data in Surqo originates at the edge — a $15 sensor node buried in a Colombian field — and travels through a carefully orchestrated chain before it becomes an actionable AI recommendation on a farmer’s dashboard. This page walks through each segment of that chain: the MQTT transport, the backend ingestion pipeline, the alert threshold engine, the AI analysis workflow, and the real-time WebSocket broadcast that keeps the dashboard live without polling.

Sensor → Cloud (MQTT Path)

The ESP32 node operates on a strict duty cycle to maximise battery life. The full sequence from wake to deep sleep takes roughly 30 seconds and consumes minimal energy.

Wake from deep sleep

The ESP32 RTC timer fires every 15 minutes, pulling the chip out of deep sleep (~10µA draw). GPIO 25 is driven HIGH to power the sensor array.

Read all sensors

The firmware reads five sensors in sequence:

DHT22 — air temperature (±0.5°C) and relative humidity (±2%)
DS18B20 (waterproof probe) — soil temperature via OneWire
Capacitive soil sensor v2.0 — volumetric soil moisture percentage (ADC, no corrosion)
ML8511 — UV index (ADC1 input-only pin)
Voltage divider — battery millivolts from the 18650 cell pair

Connect and publish

The firmware connects to WiFi, syncs NTP time, and publishes a JSON payload over MQTT TLS (port 8883) to HiveMQ Cloud. If the MQTT connection fails, it falls back to an HTTP POST to https://surqo-api.fly.dev/api/v1/sensors/readings.

Return to deep sleep

GPIO 25 is driven LOW (cuts sensor power), and the ESP32 enters deep sleep for another 15 minutes.

MQTT Topic and Payload Format

The firmware publishes to the device-scoped topic, which the backend subscribes to via wildcard:

surqo/devices/{device_mac}/sensors

The JSON payload carries all sensor readings plus device metadata:

{
  "device_id": "a4cf12a3f2c1",
  "device_mac": "a4cf12a3f2c1",
  "claim_code": "A3F2C1",
  "firmware_version": "2.0.0",
  "battery_mv": 3820,
  "rssi_dbm": -62,
  "sensors": {
    "soil_moisture_pct": 44.5,
    "soil_temp_c": 27.8,
    "air_temp_c": 31.2,
    "air_humidity_pct": 70.1,
    "light_uv_index": 7.4
  }
}

The legacy topic surqo/farms/{farm_id}/sensors is also subscribed for backward compatibility with earlier firmware versions and the IoT simulator.

MQTT → Backend Processing

SurqoMQTTService runs as a background thread inside the FastAPI process, bridged to the async event loop via asyncio.run_coroutine_threadsafe. It subscribes to both topic patterns on connect:

# From mqtt_service.py
TOPIC_DEVICES = "surqo/devices/+/sensors"
TOPIC_FARMS   = "surqo/farms/+/sensors"

When a message arrives, the service:

Parse the payload

Decodes the JSON body. If parsing fails, the error is logged and the message is silently dropped — the node will publish again in 15 minutes.

Resolve device → farm

For device-scoped topics (surqo/devices/+/sensors), the service queries the Device table by normalized MAC address to retrieve the associated farm_id. If the MAC is not yet registered, a new Device record is created with farm_id = NULL and the reading is discarded until the farmer claims the device through the frontend.

Call ingest_service

The resolved payload and farm_id are handed to ingest_reading() inside ingest_service.py, which:

Calculates VPD (Vapour Pressure Deficit) in real time using the Magnus equation: VPD = es − ea where es is saturation vapour pressure at air temperature and ea is actual vapour pressure derived from relative humidity.
Persists a new SensorReading row to PostgreSQL with the computed vpd_kpa field included.
Broadcasts the reading to all WebSocket clients subscribed to that farm_id.
Passes the reading to AlertService.check_thresholds().

Alert Pipeline

AlertService.check_thresholds() evaluates every incoming reading against a fixed set of agronomic thresholds. Violations at any severity level are always persisted as an Alert row in PostgreSQL — but email dispatch is subject to a Redis cooldown.

Threshold Table

Condition	Threshold	Severity
Soil moisture low	< 25%	warning
Soil moisture critical	< 15%	critical
VPD high	> 1.6 kPa	warning
VPD critical	> 2.5 kPa	critical
Air temperature high	> 38°C	warning
Air temperature critical	> 42°C	critical
Battery low	< 3,400 mV	warning
Battery critical	< 3,200 mV	critical

Email Cooldown Flow

Check Redis cooldown key

Before sending any email, the service checks for the key alert_email_cooldown:{farm_id} in Upstash Redis. If the key exists, the alert is saved to the database but no email is sent.

Send email via Resend

If the cooldown key is absent, an HTML alert email is dispatched via the Resend API to the alert_email address stored on the Farm record.

Set cooldown TTL

Immediately after dispatch, the cooldown key is written to Redis with a 30-minute TTL. This ensures at most one alert email per farm every 30 minutes, regardless of how many threshold violations occur in that window.

The cooldown only prevents duplicate emails. Every threshold violation is still written to the alerts table and visible in the Alerts page, even during the cooldown window.

AI Analysis Pipeline

Unlike the automatic MQTT-driven flows above, AI analysis is explicitly triggered by a user. The endpoint is POST /api/v1/analysis/analyze and is subject to plan-based quotas (4 analyses lifetime on the free tier).

Fetch sensor history

The backend retrieves the last 48 hours of SensorReading rows for the requested farm_id from PostgreSQL.

Fetch weather forecast (Redis-cached)

ClimateService calls Open-Meteo’s 7-day forecast API for the farm’s latitude and longitude. The response is cached in Redis for 1 hour using a coordinate-derived key, reducing latency from ~800ms to ~5ms on cache hits.

Build LLM context

LLMService.analyze_farm() loads the versioned prompt file farm_analysis_v1.0.yaml and assembles the user content string by injecting:

build_sensor_context() — formatted summary of the 48h sensor history including averages and extremes
get_crop_knowledge() — crop-specific agronomic parameters from the YAML knowledge base (Kc coefficients, VPD tolerances, known pest risks by crop type)
Computed KPIs: ETc daily (ET₀ × Kc), 7-day water deficit (ETc − rainfall), water stress index

Call LLM provider

The assembled prompt is sent to the active provider (Groq by default) with json_mode=True to guarantee a parseable JSON response. The service requests at least max(max_output_tokens, 1500) tokens to prevent mid-JSON truncation.

Parse and persist

_parse_json_response() extracts the JSON object from the response, tolerating markdown code fences and extra surrounding text. The result is saved as an Analysis row in PostgreSQL and the user’s analyses_used counter is incremented atomically.

Analysis Response Structure

The LLM returns a structured JSON object that maps directly to the AnalysisResult dataclass:

{
  "alert_level": "warning",
  "summary_for_farmer": "Tu cultivo de maíz muestra estrés hídrico moderado...",
  "irrigation_needed": true,
  "water_stress_index": 0.67,
  "recommendations": [
    {
      "action": "Aplicar riego por goteo — 25mm en las próximas 6 horas",
      "time_window": "0-6h",
      "priority": 1,
      "category": "irrigation"
    }
  ],
  "model_used": "groq/llama-3.3-70b-versatile",
  "input_tokens": 624,
  "output_tokens": 318,
  "cost_usd": 0.0
}

Real-Time WebSocket

Every time a new sensor reading is ingested (whether from MQTT or direct HTTP), the backend broadcasts it to all WebSocket clients connected to that farm. Endpoint:

wss://surqo-api.fly.dev/api/v1/sensors/ws/live/{farm_id}

Authentication: A valid JWT must be passed as the token query parameter. The connection is rejected with 403 if the token is absent or invalid. Message types:

// On connection — last known reading
{ "type": "initial", "data": { /* SensorReading */ } }

// On every new MQTT or HTTP reading
{ "type": "sensor_reading", "data": { /* SensorReading */ } }

Client example:

const ws = new WebSocket(
  'wss://surqo-api.fly.dev/api/v1/sensors/ws/live/FARM_UUID?token=JWT'
)

ws.onmessage = (event) => {
  const { type, data } = JSON.parse(event.data)
  if (type === 'sensor_reading') {
    updateDashboard(data)
  }
}

The WebSocket manager stores connections in memory grouped by farm_id. A broadcast to one farm never touches connections for other farms, keeping per-message overhead O(1) relative to the number of farms.

Introducción

Arquitectura

Instalación y Configuración

Firmware ESP32

Despliegue

Surqo Data Flow: From Sensor Reading to AI Recommendation

Sensor → Cloud (MQTT Path)

MQTT Topic and Payload Format

MQTT → Backend Processing

Alert Pipeline

Threshold Table

Email Cooldown Flow

AI Analysis Pipeline

Analysis Response Structure

Real-Time WebSocket

Build docs developers (and LLMs) love

Introducción

Arquitectura

Instalación y Configuración

Firmware ESP32

Despliegue

Documentation Index

​Sensor → Cloud (MQTT Path)

​MQTT Topic and Payload Format

​MQTT → Backend Processing

​Alert Pipeline

​Threshold Table

​Email Cooldown Flow

​AI Analysis Pipeline

​Analysis Response Structure

​Real-Time WebSocket

Build docs developers (and LLMs) love

Sensor → Cloud (MQTT Path)

MQTT Topic and Payload Format

MQTT → Backend Processing

Alert Pipeline

Threshold Table

Email Cooldown Flow

AI Analysis Pipeline

Analysis Response Structure

Real-Time WebSocket