Select most recent `pax` measurement values from ESP32-Paxcounter devices recording data using the Kotori data historian.

example_jmespath_paxcounter_kotori.py

# About
# =====
# Select most recent `pax` measurement values from ESP32-Paxcounter
# devices recording data using the Kotori data historian.
#
# Synopsis
# ========
#
#   pip install jmespath requests
#   python3 example_jmespath_paxcounter_kotori.py "https://swarm.hiveeyes.org/api/hiveeyes/zku/paxcounter/bauschilderung/data.json?from=now-5m"
#
# Resources
# =========
# - https://cyberman54.github.io/ESP32-Paxcounter/
# - https://kotori.readthedocs.io/
# - https://community.hiveeyes.org/t/pax-counter-mit-lopy4/4838
# - https://community.hiveeyes.org/t/anleitung-pax-conter-per-ttn-webhook-an-swarm-anschliessen/4940
# - https://community.hiveeyes.org/t/den-aktuellsten-publizierten-datensatz-oder-einzelwert-eines-bestimmten-datenkanals-abrufen/4966
import sys

import jmespath
import requests


# =============
# Demonstration
# =============

paxcounter_demo_data = [
    {
        "ble": 0.0,
        "pax": 2.0,
        "time": "2023-06-08T18:52:39.502Z",
        "timesync_seqno": None,
        "wifi": 2.0
    },
    {
        "ble": 0.0,
        "pax": 42.0,
        "time": "2023-06-08T18:53:39.501Z",
        "timesync_seqno": None,
        "wifi": 2.0
    }
]

def demo():
    """
    Run a small demonstration.

    Both variants produce the same result.
    """
    # path = jmespath.search("[] | sort_by(@, &time) | reverse(@) | [0].pax", paxcounter_data)
    path = jmespath.search("reverse(sort_by([], &time)) | [0].pax", paxcounter_demo_data)
    print(path)



# =========
# Live data
# =========

def live(url):
    """
    Acquire live data, and run transformation.
    """
    data = requests.get(url).json()
    most_recent_pax_count = select_most_recent_scalar(data, "pax")
    print(most_recent_pax_count)


def select_most_recent_scalar(data, fieldname):
    """
    Decode most recent scalar value for specific field from list of potentially unsorted list of dictionaries.
    """
    expression = f"reverse(sort_by([], &time)) | [0].{fieldname}"
    return jmespath.search(expression, data)


if __name__ == "__main__":
    # demo()
    url = sys.argv[1]
    live(url)

Update

Kotori can do it natively now, using the new HTTP export API query parameters sort, direction, limit, and scalar.

Example

http "https://swarm.hiveeyes.org/api/hiveeyes/zku/paxcounter/bauschilderung/data.json?from=now-5m&sort=time&direction=desc&limit=1&scalar=pax"

References

• daq-tools/kotori#146

Kotori can do it natively now, using the new HTTP export API query parameters

Excellent!