Lab: publish and receive a sensor reading

export PYTHONPATH=/absolute/path/to/sdk/python

import argparse
import json
import os
import threading

from telequick.data import (
    Data,
    decode_data_frame,
    encode_data_frame,
    topic_matches,
)


def check_locally():
    topic = "academy/alice/temperature"
    payload = b'{"sequence": 1, "celsius": 23.5}'
    frame = encode_data_frame("academy-publisher", topic, payload)
    decoded = decode_data_frame(frame)
    assert decoded.from_client_id == "academy-publisher"
    assert decoded.topic == topic
    assert decoded.payload == payload
    print("frame round trip: OK")
    print("matching filter:", topic_matches(topic, "academy/+/temperature"))
    print("wrong filter:", topic_matches(topic, "academy/alice/humidity"))


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("mode", choices=("check", "publish", "subscribe"))
    parser.add_argument("--topic", default="academy/alice/temperature")
    args = parser.parse_args()
    if args.mode == "check":
        check_locally()
        return

    token = os.environ.get("ACADEMY_DATA_TOKEN")
    relay = os.environ.get("ACADEMY_RELAY_HOST")
    if not token or not relay:
        parser.error("set ACADEMY_DATA_TOKEN and ACADEMY_RELAY_HOST")

    client = Data(
        token=token,
        client_id=f"academy-{args.mode}-{os.getpid()}",
        relay_host=relay,
    )
    subscription = None
    stop = threading.Event()

    def receive(message):
        # Callback data comes from another client. Handle invalid JSON here.
        try:
            reading = json.loads(message.payload)
            if not isinstance(reading, dict):
                raise ValueError("reading must be an object")
            print("RECEIVED", message.topic, reading, flush=True)
        except (ValueError, UnicodeError):
            print("RECEIVED invalid reading", flush=True)

    try:
        if args.mode == "subscribe":
            subscription = client.subscribe(
                topic_filter=args.topic, on_message=receive
            )
            print("Subscription requested:", args.topic, flush=True)
            stop.wait()
        else:
            sequence = 0
            while not stop.is_set():
                sequence += 1
                payload = json.dumps({
                    "sequence": sequence, "celsius": 23.5
                }).encode("utf-8")
                client.publish(topic=args.topic, payload=payload)
                print("PUBLISH CALLED", sequence, args.topic, flush=True)
                stop.wait(1)
    except KeyboardInterrupt:
        pass
    finally:
        if subscription is not None:
            subscription.close()
        client.close()


if __name__ == "__main__":
    main()

frame round trip: OK
matching filter: True
wrong filter: False

python3 data_lab.py subscribe --topic academy/alice/temperature

python3 data_lab.py publish --topic academy/alice/temperature

RECEIVED academy/alice/temperature {'sequence': 3, 'celsius': 23.5}
RECEIVED academy/alice/temperature {'sequence': 4, 'celsius': 23.5}