Ingestion pipelines — Harvestree Integration Guide

Generic pipeline

flowchart LR
  LNS["Network server"] -->|"MQTT or HTTP"| RX["Receive"]
  RX --> VAL["Validate"]
  VAL --> RAW["Store raw"]
  RAW --> NORM["Normalize"]
  NORM --> DB[("Database")]
  NORM --> PUB["Optional publish"]

Typical LNS output (after codec)

{
  "devEui": "70B3D57ED0001234",
  "fPort": 1,
  "fCnt": 42,
  "object": { "serial_number": 23118, "status": 1, "pt_1": 25.0 },
  "rxInfo": [{ "rssi": -95, "snr": 9.5 }],
  "time": "2026-07-06T08:00:00Z"
}

Exact JSON shape depends on your LNS (ChirpStack, The Things Stack, Actility, etc.). Map devEui / deviceInfo.devEui and object or decoded_payload accordingly.

ChirpStack v4 integration endpoints

Transport	ChirpStack setting	Example consumer
MQTT	Application integration → MQTT broker	Subscribe to `application/+/device/+/event/up`
HTTP	Application integration → HTTP endpoint	`POST` JSON to your URL (same body shape as MQTT)

Runnable examples

Pick one script matching your language and LNS transport. All share the same normalization logic per language. Full comparison: examples/README.md.

Example	When to use	Path
Python — MQTT	Python stack; LNS publishes to a shared MQTT broker (fan-out, several subscribers).	python/mqtt_ingest.py
Python — HTTP	Python stack; LNS POSTs uplinks to your service (ingress, serverless, API behind reverse proxy).	python/http_webhook.py
Node.js — MQTT	Node/TypeScript stack; MQTT from the LNS. Optional local codec via `DECODER_PATH` for tests.	node/mqtt_ingest.js
Node.js — HTTP	Node/TypeScript stack; HTTP webhook from the LNS. Same optional `DECODER_PATH`.	node/http_webhook.js
ChirpStack setup	Region, codec, OTAA, and integration wiring before running a consumer.	chirpstack/device-profile.md

Shared modules: python/ingest_common.py, node/ingest_common.js (decode → normalize → flat rows). Replace stdout with inserts described in Data model.

Idempotency and timestamps

Deduplicate on (deveui, f_cnt) or message UUID when your LNS provides one.
Store server receive time as primary timestamp; keep LNS time as metadata.
Expect irregular intervals (configurable measurement period).