Created
September 8, 2025 16:26
-
-
Save johncalistro/17d75e1a2e1e3810eaba5c2736bce979 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # LoRaWAN AU915 Starter (iMasters) | |
| Pequeno kit para testar *uplinks* LoRaWAN com **AU915** e consumir mensagens via **MQTT**. | |
| ## Conteúdo | |
| - `decoders/ttn_decoder.js` — Decoder para The Things Stack (TTN) | |
| - `clients/mqtt_consumer.py` — Cliente MQTT (Python) | |
| - `.env.example` — Variáveis para o cliente MQTT | |
| - `LICENSE` — MIT | |
| ## Requisitos | |
| - TTN (ou outro network server) com device **OTAA** registrado | |
| - Python 3.10+ | |
| ## Uso rápido | |
| ```bash | |
| # 1) Instale dependências | |
| pip install paho-mqtt python-dotenv | |
| # 2) Configure variáveis | |
| cp .env.example .env | |
| # edite .env com APP_ID e API KEY do TTN | |
| # 3) Rode o consumidor | |
| python clients/mqtt_consumer.py | |
| --- | |
| `decoders/ttn_decoder.js` | |
| ```javascript | |
| // Decoder simples: temp/umidade em 2 bytes cada (c/100) | |
| function decodeUplink(input) { | |
| const b = input.bytes || []; | |
| if (b.length < 4) { | |
| return { errors: ["payload curto"] }; | |
| } | |
| const temp = (b[0] << 8 | b[1]) / 100; | |
| const rh = (b[2] << 8 | b[3]) / 100; | |
| return { | |
| data: { temperature_c: temp, humidity_pct: rh }, | |
| warnings: [] | |
| }; | |
| } | |
| --- | |
| clients/mqtt_consumer.py | |
| import os, json, base64 | |
| from dotenv import load_dotenv | |
| import paho.mqtt.client as mqtt | |
| load_dotenv() | |
| TTN_HOST = os.getenv("TTN_HOST", "YOUR_REGION.cloud.thethings.network") | |
| TTN_PORT = int(os.getenv("TTN_PORT", "1883")) | |
| TTN_APP_ID = os.getenv("TTN_APP_ID", "your-app-id") | |
| TTN_API_KEY = os.getenv("TTN_API_KEY", "NNSXS.YOUR.TOKEN") | |
| TTN_TOPIC = os.getenv("TTN_TOPIC", "#") | |
| print(f"Conectando em {TTN_HOST}:{TTN_PORT} como {TTN_APP_ID}…") | |
| def on_connect(client, userdata, flags, rc): | |
| print("Conectado, código:", rc) | |
| client.subscribe(TTN_TOPIC) | |
| def on_message(_, __, msg): | |
| try: | |
| payload = json.loads(msg.payload) | |
| uplink = payload.get("uplink_message", {}) | |
| frm = uplink.get("frm_payload") | |
| if not frm: | |
| return | |
| b = base64.b64decode(frm) | |
| temp = int.from_bytes(b[0:2], "big") / 100 | |
| rh = int.from_bytes(b[2:4], "big") / 100 | |
| print({ | |
| "device": payload.get("end_device_ids", {}).get("device_id"), | |
| "temp_c": temp, | |
| "humidity_pct": rh | |
| }) | |
| except Exception as e: | |
| print("Erro ao processar mensagem:", e) | |
| client = mqtt.Client() | |
| client.username_pw_set(TTN_APP_ID, password=TTN_API_KEY) | |
| client.on_connect = on_connect | |
| client.on_message = on_message | |
| client.connect(TTN_HOST, TTN_PORT) | |
| client.loop_forever() | |
| --- | |
| .env.example | |
| # Host do cluster (ex.: eu1.cloud.thethings.network) | |
| TTN_HOST= | |
| TTN_PORT=1883 | |
| TTN_APP_ID= | |
| TTN_API_KEY= | |
| # Tópico — use # para tudo, ou um específico | |
| TTN_TOPIC=# | |
| --- | |
| LICENSE | |
| MIT License | |
| Copyright (c) 2025 John Calistro | |
| Permission is hereby granted, free of charge, to any person obtaining a copy | |
| of this software and associated documentation files (the "Software"), to deal | |
| in the Software without restriction... |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment