# Harvestree ingestion examples

Illustrative consumers for **ChirpStack v4** uplink JSON. They print normalized metrics to stdout — replace that step with your database insert, message bus, or alarm rules.

Download from the [integration guide](https://portal.moiz-eh.com/harvestree/lorawan/integration/ingestion-pipelines.html) or use the copies bundled with your MOIZ delivery.

## Pick one example

You only need **one** runnable script. All four share the same processing logic within each language (`ingest_common.*`).

| Example | Path | Use when |
|---------|------|----------|
| **Python + MQTT** | `python/mqtt_ingest.py` | Your stack is Python; the LNS publishes to a **shared MQTT broker** (several subscribers, replay, IoT bus). |
| **Python + HTTP** | `python/http_webhook.py` | Your stack is Python; the LNS **POSTs** uplinks to your URL (simple ingress, reverse proxy, serverless). |
| **Node.js + MQTT** | `node/mqtt_ingest.js` | Your stack is Node/TypeScript; **MQTT** fan-out from the LNS. |
| **Node.js + HTTP** | `node/http_webhook.js` | Your stack is Node/TypeScript; **HTTP webhook** from the LNS. |

### MQTT vs HTTP

| | MQTT | HTTP webhook |
|---|------|----------------|
| **Typical fit** | Multiple consumers, durable bus, on-prem brokers | Single integration endpoint, cloud functions, API gateway |
| **ChirpStack** | Application integration → MQTT broker | Application integration → HTTP endpoint URL |
| **Ops** | Broker availability, topics, ACLs | HTTPS, auth on your endpoint, retries from LNS |
| **This repo** | Subscribe to `application/+/device/+/event/up` | `POST` JSON body (same shape as MQTT payload) |

Configure the LNS codec first — see [chirpstack/device-profile.md](chirpstack/device-profile.md).

### Python vs Node.js

| | Python | Node.js |
|---|--------|---------|
| **Pick if** | Data team, pandas/ML, Timescale ingest in Python | Existing Node services, same runtime as the ChirpStack codec |
| **Dependencies** | `paho-mqtt` (MQTT only); HTTP uses stdlib | `mqtt` package (MQTT only); HTTP uses stdlib |
| **Local raw decode** | Requires LNS `object` field (codec on LNS — **recommended**) | Optional `DECODER_PATH` to `harvestree_decoder.js` when `object` is empty |
| **Unit hints** | `guess_unit()` for DB `unit` columns | Same mapping in `ingest_common.js` |

**Recommendation:** configure `harvestree_decoder.js` on the LNS so every integration receives decoded `object` JSON. Use `DECODER_PATH` in Node only for local testing without ChirpStack.

## Prerequisites

1. ChirpStack (or compatible LNS) with Harvestree codec on the device profile.
2. At least one successful FPort 1 uplink with `serial_number` in `object`.
3. Application integration enabled (MQTT and/or HTTP).

## Quick start

### Python — MQTT

```bash
cd python
pip install -r requirements.txt
export MQTT_HOST=your-broker
python mqtt_ingest.py
```

### Python — HTTP

```bash
cd python
export HTTP_PORT=8080
python http_webhook.py
# ChirpStack HTTP integration URL: http://your-host:8080/webhook/chirpstack/up
```

### Node.js — MQTT

```bash
cd node
npm install
export MQTT_HOST=your-broker
node mqtt_ingest.js
```

### Node.js — HTTP

```bash
cd node
npm install
export HTTP_PORT=8080
node http_webhook.js
```

Optional local decode (Node, tests only):

```bash
export DECODER_PATH=/path/to/harvestree_decoder.js
```

## What each script does

1. Receive one ChirpStack **uplink event** (MQTT message or HTTP POST).
2. Read **`object`** (decoded metrics in engineering units).
3. **Normalize** to flat rows `(metric_key, value, unit_hint)`.
4. Print rows (stand in for `INSERT` into your time-series store).

Skipped automatically: non–FPort 1 events with no decoded object.

## Adapt for production

- Map flat rows to your schema — see [Data model](../data-model.html) (registry, raw, normalized EAV).
- Deduplicate on `(deveui, f_cnt)` — see [ingestion pipelines](../ingestion-pipelines.html).
- Store server receive time as primary timestamp; keep LNS `time` as metadata.
- Map `pot_*` / `fourtwenty_*` units from your asset commissioning data, not only `guess_unit()`.
- Add TLS, authentication, and rate limits on MQTT/HTTP.
- Keep raw envelope JSON before normalization for audit/replay.

## ChirpStack setup

See [chirpstack/device-profile.md](chirpstack/device-profile.md) (region, codec, OTAA, MQTT/HTTP integration).