[TTN] Add TTS (The Things Stack) / TTN (The Things Network) decoder

A basic decoder for TTS/TTN v3 Uplink Messages [1], submitted using HTTP
Webhooks [2].

[1] https://www.thethingsindustries.com/docs/the-things-stack/concepts/data-formats/#uplink-messages
[2] https://www.thethingsindustries.com/docs/integrations/webhooks/

CHANGES.rst

@@ -11,6 +11,7 @@ in progress
   Other types will make the panel croak like ``InfluxDB Error: unsupported
   mean iterator type: *query.stringInterruptIterator`` or ``InfluxDB Error:
   not executed``.
+- Add TTS (The Things Stack) / TTN (The Things Network) decoder
 
 
 .. _kotori-0.27.0:

kotori/daq/decoder/__init__.py

@@ -3,6 +3,7 @@
 from kotori.daq.decoder.airrohr import AirrohrDecoder
 from kotori.daq.decoder.tasmota import TasmotaSensorDecoder, TasmotaStateDecoder
 from kotori.daq.decoder.schema import MessageType
+from kotori.daq.decoder.tts_ttn import TheThingsStackDecoder
 
 
 class DecoderInfo:
@@ -18,7 +19,7 @@ def __init__(self, topology):
         self.topology = topology
         self.info = DecoderInfo()
 
-    def probe(self):
+    def probe(self, payload: str = None):
 
         if 'slot' not in self.topology:
             return False
@@ -41,4 +42,13 @@ def probe(self):
             self.info.decoder = TasmotaStateDecoder
             return True
 
+        # TTS/TTN: The Things Stack / The Things Network
+        if self.topology.slot.endswith('data.json') \
+                and payload is not None \
+                and "uplink_message" in payload \
+                and "decoded_payload" in payload:
+            self.info.message_type = MessageType.DATA_CONTAINER
+            self.info.decoder = TheThingsStackDecoder
+            return True
+
         return False

kotori/daq/decoder/tts_ttn.py

@@ -0,0 +1,41 @@
+# -*- coding: utf-8 -*-
+# (c) 2022 Andreas Motl <[email protected]>
+import json
+from collections import OrderedDict
+
+
+class TheThingsStackDecoder:
+    """
+    Decode JSON payloads in TTS-/TTN-webhook JSON format.
+    TTS/TTN means "The Things Stack" / "The Things Network".
+
+    Documentation
+    =============
+    - https://getkotori.org/docs/handbook/decoders/tts-ttn.html
+
+    References
+    ==========
+    - https://www.thethingsindustries.com/docs/the-things-stack/concepts/data-formats/#uplink-messages
+    - https://www.thethingsindustries.com/docs/integrations/webhooks/
+    - https://community.hiveeyes.org/t/more-data-acquisition-payload-formats-for-kotori/1421
+    - https://community.hiveeyes.org/t/tts-ttn-daten-an-kotori-weiterleiten/1422/34
+    """
+
+    @staticmethod
+    def decode(payload: str):
+
+        # Decode from JSON.
+        message = json.loads(payload)
+
+        # Use timestamp and decoded payload.
+        data = OrderedDict()
+        if "received_at" in message:
+            data["timestamp"] = message["received_at"]
+        data.update(message["uplink_message"]["decoded_payload"])
+
+        # TODO: Add more data / metadata.
+        #       This is an implementation from scratch. It can be improved by
+        #       cherry-picking more specific decoding routines from `ttnlogger`.
+        #       https://github.com/mqtt-tools/ttnlogger
+
+        return data

kotori/daq/services/mig.py

@@ -191,7 +191,7 @@ def decode_message(self, topic, payload):
 
         # Message can be handled by one of the device-specific decoders.
         decoder_manager = DecoderManager(topology)
-        if decoder_manager.probe():
+        if decoder_manager.probe(payload):
             message.type = decoder_manager.info.message_type
             message.data = decoder_manager.info.decoder.decode(payload)
             return message

pytest.ini

@@ -38,5 +38,7 @@ markers =
     homie: Tests specific to the Homie firmware framework.
     airrohr: Tests specific to Airrohr devices.
     weewx: Tests for WeeWX integration.
+    tts: Tests for TTS/TTN adapter.
+    ttn: Tests for TTS/TTN adapter.
     hiveeyes: Tests for vendor hiveeyes.
     legacy: Tests for legacy endpoints and such.

test/test_tts_ttn.py

@@ -0,0 +1,97 @@
+# -*- coding: utf-8 -*-
+# (c) 2022 Andreas Motl <[email protected]>
+"""
+Test the TTS-/TTN-webhook receiver implementation.
+TTS/TTN means "The Things Stack" / "The Things Network".
+
+Usage
+=====
+::
+
+    source .venv/bin/activate
+    pytest -m ttn
+
+References
+==========
+- https://www.thethingsindustries.com/docs/the-things-stack/concepts/data-formats/#uplink-messages
+- https://www.thethingsindustries.com/docs/integrations/webhooks/
+- https://community.hiveeyes.org/t/more-data-acquisition-payload-formats-for-kotori/1421
+- https://community.hiveeyes.org/t/tts-ttn-daten-an-kotori-weiterleiten/1422/34
+"""
+import logging
+from test.settings.basic import (
+    settings,
+    influx_sensors,
+    create_influxdb,
+    reset_influxdb,
+    reset_grafana,
+    PROCESS_DELAY_MQTT,
+)
+from test.util import http_json_sensor, read_jsonfile, sleep
+
+import pytest
+import pytest_twisted
+from twisted.internet import threads
+
+logger = logging.getLogger(__name__)
+
+
+def make_testcases():
+    """
+    Define different test cases with in/out pairs.
+    """
+    return [
+        {
+            "in": read_jsonfile("test_tts_ttn_full.json"),
+            "out": {
+                "time": "2022-01-19T19:02:34.007345Z",
+                "analog_in_1": 59.04,
+                "analog_in_2": 58.69,
+                "analog_in_3": 3.49,
+                "relative_humidity_2": 78.5,
+                "temperature_2": 4.2,
+                "temperature_3": 3.4,
+            },
+        },
+        {
+            "in": read_jsonfile("test_tts_ttn_minimal.json"),
+            "__delete__": ["time"],
+            "out": {"temperature_1": 53.3, "voltage_4": 3.3},
+        },
+    ]
+
+
+@pytest_twisted.inlineCallbacks
+@pytest.mark.http
+@pytest.mark.tts
+@pytest.mark.ttn
+@pytest.mark.parametrize("testcase", make_testcases())
+def test_tts_ttn_http_json_full(
+    testcase, machinery_basic, create_influxdb, reset_influxdb
+):
+    """
+    Submit single reading in TTS/TTN webhook JSON format to HTTP API,
+    and verify it was correctly stored in the InfluxDB database.
+    """
+
+    data_in = testcase["in"]
+    data_out = testcase["out"]
+
+    # Submit a single measurement, without timestamp.
+    yield threads.deferToThread(http_json_sensor, settings.channel_path_data, data_in)
+
+    # Wait for some time to process the message.
+    yield sleep(PROCESS_DELAY_MQTT)
+    yield sleep(PROCESS_DELAY_MQTT)
+
+    # Proof that data arrived in InfluxDB properly.
+    record = influx_sensors.get_first_record()
+
+    # Optionally delete specific fields, to make comparison work.
+    if "__delete__" in testcase:
+        for delete_field in testcase["__delete__"]:
+            del record[delete_field]
+
+    # Verify the records looks like expected.
+    assert record == data_out
+    yield record

test/test_tts_ttn_full.json

@@ -0,0 +1,99 @@
+{
+  "end_device_ids": {
+    "device_id": "testdrive-foo-bar-baz",
+    "application_ids": {
+      "application_id": "acme"
+    },
+    "dev_eui": "030E202BD5504B52",
+    "join_eui": "FB81CCAAE6D7B2F7",
+    "dev_addr": "4406826C"
+  },
+  "correlation_ids": [
+    "as:up:01FSSVG88HVN4WVXT5X75KG545",
+    "gs:conn:01FSRW4G4EK6JFFREN8CF04YAW",
+    "gs:up:host:01FSRW4G4NF8PN3HZP2MT482ED",
+    "gs:uplink:01FSSVG822Q4KP03T951ZFD9HZ",
+    "ns:uplink:01FSSVG8230BA2CGX4B5M6BQDJ",
+    "rpc:/ttn.lorawan.v3.GsNs/HandleUplink:01FSSVG823BPTEQDJR7EQWA63E",
+    "rpc:/ttn.lorawan.v3.NsAs/HandleUplink:01FSSVG88GDGNTYP2X7MV3DXC9"
+  ],
+  "received_at": "2022-01-19T19:02:34.007345025Z",
+  "uplink_message": {
+    "session_key_id": "QR5vq9sPi2tZPtH8rDSjpA==",
+    "f_port": 1,
+    "f_cnt": 2289,
+    "frm_payload": "AQIXEAICFu0CZwAqAmidA2cAIgMCAV0=",
+    "decoded_payload": {
+      "analog_in_1": 59.04,
+      "analog_in_2": 58.69,
+      "analog_in_3": 3.49,
+      "relative_humidity_2": 78.5,
+      "temperature_2": 4.2,
+      "temperature_3": 3.4
+    },
+    "rx_metadata": [
+      {
+        "gateway_ids": {
+          "gateway_id": "somewhere-ffp",
+          "eui": "FEA9BE611914F97D"
+        },
+        "timestamp": 208623723,
+        "rssi": -107,
+        "channel_rssi": -107,
+        "snr": -6.5,
+        "location": {
+          "latitude": 52.21,
+          "longitude": 12.96,
+          "source": "SOURCE_REGISTRY"
+        },
+        "uplink_token": "amLzqHY2UL1HQfYlwc+vByXUwkZ6FiwUwKh/5T29axuAfl+XOXnpMKzFBVPy9bAmyqvv6qxOsotx8qw=",
+        "channel_index": 2
+      },
+      {
+        "gateway_ids": {
+          "gateway_id": "elsewhere-ffp",
+          "eui": "3398CD3B764F4A14"
+        },
+        "time": "2022-01-19T19:02:33.764357Z",
+        "timestamp": 26499475,
+        "rssi": -90,
+        "channel_rssi": -90,
+        "snr": 7,
+        "location": {
+          "latitude": 52.20,
+          "longitude": 12.99,
+          "altitude": 35,
+          "source": "SOURCE_REGISTRY"
+        },
+        "uplink_token": "AaVQj9emrquiRtdWhDB9+PyZLJeCzJ6zB2ZDGahlF47vhW0vYdFuqUpVj4eqW1OlFwKZUYdH1ArOW6w=",
+        "channel_index": 2
+      }
+    ],
+    "settings": {
+      "data_rate": {
+        "lora": {
+          "bandwidth": 125000,
+          "spreading_factor": 7
+        }
+      },
+      "coding_rate": "4/5",
+      "frequency": "868500000",
+      "timestamp": 208623723
+    },
+    "received_at": "2022-01-19T19:02:33.795669934Z",
+    "consumed_airtime": "0.077056s",
+    "locations": {
+      "user": {
+        "latitude": 52.2134,
+        "longitude": 12.9812,
+        "altitude": 35,
+        "source": "SOURCE_REGISTRY"
+      }
+    },
+    "network_ids": {
+      "net_id": "000013",
+      "tenant_id": "ttn",
+      "cluster_id": "ttn-eu1"
+    }
+  }
+}

test/test_tts_ttn_minimal.json

@@ -0,0 +1,8 @@
+{
+  "uplink_message": {
+    "decoded_payload": {
+      "temperature_1": 53.3,
+      "voltage_4": 3.3
+    }
+  }
+}

test/util.py

@@ -6,6 +6,7 @@
 import random
 import string
 import sys
+import typing as t
 
 import pytest
 import requests
@@ -275,3 +276,9 @@ def idgen(size=6, chars=string.ascii_uppercase + string.digits):
     - https://stackoverflow.com/questions/2257441/random-string-generation-with-upper-case-letters-and-digits/2257449#2257449
     """
     return ''.join(random.choice(chars) for _ in range(size))
+
+
+def read_jsonfile(name: str) -> t.Dict[str, t.Any]:
+    path = os.path.join(os.path.dirname(__file__), name)
+    with open(path, mode="r") as f:
+        return json.load(f)