Parsing The Things Network Uplinks into PostGIS-Ready Rows

A The Things Network (TTN) v3 uplink is a deeply nested JSON document, and turning it into a row you can insert into PostGIS means reaching into three separate branches: end_device_ids for identity, uplink_message.decoded_payload (or the base64 frm_payload) for the measurement, and uplink_message.rx_metadata for the gateway coordinates that stand in for the sensor’s location. The parse function has to be defensive — gateways drop out of rx_metadata, decoded_payload is absent when no formatter is configured, and received_at is the only reliable event time because the node has no clock. This page builds one function that flattens a TTN uplink into a PostGIS-ready dict, and it is the storage-facing companion to the LoRaWAN payload parsing workflow.

The TTN Application Server publishes each uplink to MQTT under a topic ending in /up, with a JSON body whose relevant structure is:

  • end_device_ids — device_id, dev_eui, and the application_ids.application_id. This is your sensor_id; the dev_eui is the globally unique hardware identifier and the safest join key.
  • received_at — a top-level RFC 3339 UTC timestamp stamped by the server. Because most environmental nodes lack a real-time clock, this receipt time is the event time you store.
  • uplink_message.f_port and frm_payload — the fPort selects the payload format, and frm_payload is the base64 measurement. When an uplink formatter is configured, uplink_message.decoded_payload also carries the server-decoded fields.
  • uplink_message.rx_metadata — an array, one entry per gateway that received the transmission, each with gateway_ids, rssi, snr, and often a location object with latitude, longitude, and altitude. A node with no onboard GPS inherits its position from the strongest-signal gateway here.

The single most important habit is treating every one of these as optional. Gateways come and go between uplinks, formatters get misconfigured, and a frame can arrive with an empty rx_metadata. Parse with .get() and explicit fallbacks, never with chained bracket indexing that raises deep in the tree.

The gateway-location fallback deserves particular care because it is a genuine approximation, not a measurement. A LoRaWAN node without onboard GPS has no idea where it is; the only spatial signal available is which gateways heard it and how strongly. Taking the strongest-signal gateway’s coordinate is a defensible proxy — a strong signal usually means proximity — but it can be tens of kilometres off in open terrain, and it will jump between uplinks as atmospheric conditions change which gateway wins. For a fixed environmental installation, a better long-term strategy is to record the gateway-derived position on the first few uplinks, then pin the device to a surveyed static coordinate in a device-metadata table and stop trusting the per-uplink gateway fix entirely. The parser below still emits the gateway position and tags it with location_source so you can distinguish a real device fix from an inherited one downstream.

There is also a subtle multiplicity to be aware of: the same physical uplink appears once in TTN’s stream but was received by every gateway in range, and all of those receptions are folded into the single rx_metadata array. You are not seeing duplicate messages — you are seeing one message with several independent signal reports. Deduplication across gateways is already handled by the network server before the message reaches you, so your parser’s job is only to choose among the reported locations, never to dedupe the reading itself.

Production-Ready Implementation

The function below flattens one TTN v3 uplink into a dict whose keys map to PostGIS columns. It prefers a device-carried GPS fix, falls back to the best gateway location, builds a WKT POINT for the geometry column, and parses received_at into a timezone-aware datetime. It raises on the two conditions that make a row unstorable — no coordinates and no timestamp — so callers can route those to a dead-letter queue.

# python 3.11 · pandas==2.2.2
from __future__ import annotations

from datetime import datetime, timezone
from typing import Any


def ttn_uplink_to_postgis_row(uplink: dict[str, Any]) -> dict[str, Any]:
    """Flatten a TTN v3 uplink JSON object into a PostGIS-ready row.

    Returns a dict with keys:
        device_id, dev_eui, f_port, received_at (aware datetime),
        latitude, longitude, geom_wkt (SRID 4326 POINT),
        location_source ('device' | 'gateway'),
        best_gateway, rssi, measurements (dict).

    Raises
    ------
    ValueError
        If neither a device nor a gateway coordinate is available,
        or if received_at is missing/unparseable.
    """
    ids = uplink.get("end_device_ids", {})
    device_id = ids.get("device_id")
    dev_eui = ids.get("dev_eui")

    msg = uplink.get("uplink_message", {})
    f_port = msg.get("f_port")
    decoded = msg.get("decoded_payload") or {}

    # --- event time: server receipt in UTC ---
    received_raw = uplink.get("received_at") or msg.get("received_at")
    if not received_raw:
        raise ValueError(f"uplink for {device_id!r} has no received_at")
    # RFC 3339 with trailing Z -> aware UTC datetime
    received_at = datetime.fromisoformat(received_raw.replace("Z", "+00:00"))
    received_at = received_at.astimezone(timezone.utc)

    # --- location: prefer device GPS, else strongest-signal gateway ---
    lat = decoded.get("latitude")
    lon = decoded.get("longitude")
    location_source = "device"
    best_gw: str | None = None
    best_rssi: float | None = None

    if lat is None or lon is None:
        location_source = "gateway"
        best = _strongest_gateway(msg.get("rx_metadata", []))
        if best is None:
            raise ValueError(f"uplink for {device_id!r} has no usable coordinates")
        lat, lon, best_gw, best_rssi = best

    measurements = {k: v for k, v in decoded.items()
                    if k not in ("latitude", "longitude", "altitude")}

    return {
        "device_id": device_id,
        "dev_eui": dev_eui,
        "f_port": f_port,
        "received_at": received_at,
        "latitude": float(lat),
        "longitude": float(lon),
        "geom_wkt": f"SRID=4326;POINT({float(lon)} {float(lat)})",
        "location_source": location_source,
        "best_gateway": best_gw,
        "rssi": best_rssi,
        "measurements": measurements,
    }


def _strongest_gateway(
    rx_metadata: list[dict[str, Any]],
) -> tuple[float, float, str, float] | None:
    """Return (lat, lon, gateway_id, rssi) for the located gateway with the best RSSI."""
    best: tuple[float, float, str, float] | None = None
    for gw in rx_metadata:
        loc = gw.get("location") or {}
        lat, lon = loc.get("latitude"), loc.get("longitude")
        if lat is None or lon is None:
            continue
        rssi = gw.get("rssi", float("-inf"))
        gw_id = gw.get("gateway_ids", {}).get("gateway_id", "unknown")
        if best is None or rssi > best[3]:
            best = (float(lat), float(lon), gw_id, float(rssi))
    return best

The emitted geom_wkt is an EWKT string PostGIS accepts directly via ST_GeomFromEWKT, with longitude first — the ordering mistake that silently plots every sensor in the wrong place. Rows produced here are ready for bulk insertion into the spatial store described in PostGIS storage and spatial indexing for sensor networks.

A few implementation decisions are worth calling out. The parser raises only on the two conditions that make a row genuinely unstorable — no coordinate of any kind, and no timestamp — and returns everything else even when partially populated, because a reading with a valid measurement but a missing rssi is still worth keeping. Deciding what is fatal versus what is merely degraded is a domain judgement: for a spatial pipeline, geometry and time are non-negotiable, so their absence is a hard failure that belongs in a dead-letter queue rather than a NULL row that will break spatial indexes or time-window joins later. Keeping measurements as a nested dict rather than exploding it into columns at this stage lets one parser serve every device profile in a mixed fleet; the explosion into typed columns, if you want it, happens at the storage boundary where you know which metrics each table indexes.

The location_source field is small but load-bearing for data quality. Six months after deployment, an analyst looking at a suspiciously mobile “fixed” sensor needs to know instantly whether the movement is real device GPS or gateway-fix jitter. Carrying the provenance on every row answers that without re-parsing the raw archive, and it lets you filter gateway-located readings out of any analysis that assumes precise coordinates.

PostGIS Row Field Map

Each key the parser emits maps to a column in a sensor readings table.

Row key Source in TTN JSON PostGIS column type Notes
device_id end_device_ids.device_id text Human-readable device name
dev_eui end_device_ids.dev_eui text Globally unique; preferred join key
received_at top-level received_at timestamptz UTC server receipt time
f_port uplink_message.f_port smallint Payload format selector
geom_wkt decoded GPS or gateway location geometry(Point,4326) Insert via ST_GeomFromEWKT
location_source derived text device or gateway provenance
rssi strongest rx_metadata[].rssi real Signal quality of the chosen fix
measurements decoded_payload (minus GPS) jsonb Or expand to typed columns

Storing measurements as jsonb keeps heterogeneous device profiles in one table; expand to typed columns only for the metrics you query and index heavily.

Verification and Testing

The test feeds a representative TTN v3 uplink with no device GPS and two gateways, then asserts the parser picks the stronger gateway and builds correct EWKT.

import pytest
from datetime import timezone


def _sample_uplink() -> dict:
    return {
        "end_device_ids": {"device_id": "air-node-07", "dev_eui": "70B3D57ED0001A2B"},
        "received_at": "2026-07-10T08:15:30.512Z",
        "uplink_message": {
            "f_port": 1,
            "decoded_payload": {"temperature": 24.6, "humidity": 58.0},
            "rx_metadata": [
                {"gateway_ids": {"gateway_id": "gw-far"},
                 "rssi": -119, "location": {"latitude": 52.10, "longitude": 4.30}},
                {"gateway_ids": {"gateway_id": "gw-near"},
                 "rssi": -78, "location": {"latitude": 52.37, "longitude": 4.90}},
            ],
        },
    }


def test_parses_uplink_with_gateway_location():
    row = ttn_uplink_to_postgis_row(_sample_uplink())

    assert row["device_id"] == "air-node-07"
    assert row["location_source"] == "gateway"
    assert row["best_gateway"] == "gw-near"          # stronger RSSI wins
    assert row["latitude"] == 52.37 and row["longitude"] == 4.90
    assert row["geom_wkt"] == "SRID=4326;POINT(4.9 52.37)"  # lon first
    assert row["received_at"].tzinfo == timezone.utc
    assert row["measurements"] == {"temperature": 24.6, "humidity": 58.0}


def test_missing_coordinates_raises():
    uplink = _sample_uplink()
    uplink["uplink_message"]["rx_metadata"] = []      # no located gateway, no device GPS
    with pytest.raises(ValueError, match="no usable coordinates"):
        ttn_uplink_to_postgis_row(uplink)

The EWKT assertion is the load-bearing one: POINT(4.9 52.37) puts longitude before latitude. Swapping them passes every type check and stores a sensor 48 degrees away from where it sits.

Gotchas

Longitude/latitude order in WKT. PostGIS POINT(x y) is POINT(longitude latitude). Latitude-first WKT is syntactically valid and geographically wrong, and no insert error will warn you.

Empty rx_metadata. A frame can arrive with no located gateway — every entry missing its location. Guard for it and dead-letter rather than inserting a null geometry that breaks spatial indexes downstream.

Trusting decoded_payload unconditionally. If no uplink formatter is configured on the application, decoded_payload is absent entirely. Always keep frm_payload and be ready to decode locally, as in decoding Cayenne LPP payloads into a pandas DataFrame.

Naive timestamp parsing. received_at ends in Z; datetime.fromisoformat on older assumptions chokes on it. Replace Z with +00:00 (or use Python 3.11’s native handling) and store as timestamptz, never a naive datetime.