Датчик уже умеет передавать показания с участка. Теперь дадим ему постоянного слушателя: Raspberry Pi примет радиопакет, сохранит его на диске и передаст в InfluxDB. В Grafana появится история, к которой можно вернуться вечером. Это лабораторная сборка для формата FZ1 из проекта передатчика ESP32; работу на участке проверим после первого запуска на столе.
Выбираем место для базы и собираем комплект
В примере используются Raspberry Pi 4 с 64-битной Raspberry Pi OS и Adafruit RFM96W для 433 МГц. Подготовьте питание, накопитель, антенну и сетевое подключение. Если берёте другую Pi или плату SX127x, заранее сверьте диапазон, распиновку, питание и поддержку драйвера.
RFM96W breakout → Raspberry Pi 4
VIN → 3,3 В, физический контакт 1 (сверьте схему платы)
GND → GND, контакт 6
SCK → GPIO11, контакт 23
MISO → GPIO9, контакт 21
MOSI → GPIO10, контакт 19
CS → GPIO8 / CE0, контакт 24
RST → GPIO25, контакт 22
DIO0 → не подключён: здесь используется опрос состоянияПередатчик ESP32 с SX1278 и приёмник используют одинаковые параметры: SF10, полосу 125 кГц, кодирование 4/8, sync word 0x12, явный заголовок и CRC. Это собственный радиоканал, не LoRaWAN; один модуль не гарантирует приём перекрывающихся передач. GPIO Pi не предназначены для сигналов 5 В.
Как пакет становится записью на диске
Каждое сообщение проходит проверку ключа AES-GCM, ID, полей и номера. Новый номер — новое измерение, даже если температура осталась прежней. Повторные и опоздавшие пакеты отбрасываются. SQLite сохраняет запись и счётчик вместе, одной транзакцией: так они остаются согласованными при ошибке записи.
Представьте короткое отключение сети: принятые данные остаются в очереди на диске и переживают перезапуск. Отправка повторяет ту же строку с прежним временем, пока InfluxDB не ответит HTTP 204. В очереди помещаются 10 000 записей; при заполнении новые пакеты отклоняются с сообщением full. Радио работает без ACK, поэтому проверим отдельно пропуски в эфире и поведение хранилища.
Сохраните файлы protocol.py и gateway.py. Драйвер RFM9x 2.2.25 вызывается с with_header=True и with_ack=False: пакет FZ1 возвращается целиком, без удаления байтов и без подтверждения другого формата. RegSyncWord задаётся через низкоуровневый API с проверкой чтением.
"""FZ1 decoder for the matching Raspberry Pi gateway; no radio dependency."""
import struct
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
def decode(packet: bytes, keys: dict):
if len(packet) != 40 or packet[:3] != b'FZ1':
raise ValueError('packet length/version')
device, seq = struct.unpack('>IQ', packet[3:15])
if not device or not seq or device not in keys:
raise ValueError('unknown device or zero sequence')
plain = AESGCM(keys[device]).decrypt(packet[3:15], packet[15:], packet[:15])
flags, temp, humidity, soil, battery = struct.unpack('>BhHHH', plain)
if not flags or flags & 240:
raise ValueError('flags')
for mask, value, low, high in [(1,temp,-4000,8000),(2,humidity,0,10000),
(4,soil,0,4095),(8,battery,1,6000)]:
if (flags & mask and not low <= value <= high) or (not flags & mask and value != 0):
raise ValueError('field range')
fields = {}
for mask, name, value in [(1,'temperature',temp/100),(2,'humidity',humidity/100),
(4,'soil_raw',soil),(8,'battery',battery/1000)]:
if flags & mask:
fields[name] = value
return device, seq, fields
#!/usr/bin/env python3
import argparse, hashlib, json, logging, math, re, sqlite3, subprocess
import threading, time, urllib.error, urllib.parse, urllib.request
from pathlib import Path
from cryptography.exceptions import InvalidTag
from protocol import decode
log = logging.getLogger('lora')
LIMIT = 10000
class Store:
def __init__(self, path, keys, gateway, create=False):
path = Path(path)
if create:
# Exclusive create: never overwrite a replay ledger.
with path.open('xb'):
pass
elif not path.is_file():
raise RuntimeError('Missing ledger: restore current state or rotate ALL keys')
self.db = sqlite3.connect(path, timeout=1, isolation_level=None)
self.db.execute('PRAGMA journal_mode=DELETE')
self.db.execute('PRAGMA synchronous=FULL')
self.gateway = gateway
if not re.fullmatch(r'[A-Za-z0-9_-]{1,32}', gateway):
raise ValueError('Invalid gateway ID')
if create:
self.db.executescript('''BEGIN IMMEDIATE;
CREATE TABLE nodes(device INTEGER PRIMARY KEY, keyhash TEXT NOT NULL,
seq TEXT NOT NULL, seen INTEGER NOT NULL);
CREATE TABLE outbox(id INTEGER PRIMARY KEY, line TEXT NOT NULL);
CREATE TABLE settings(gateway TEXT NOT NULL);
COMMIT;''')
self.db.execute('BEGIN IMMEDIATE')
try:
self.db.executemany('INSERT INTO nodes VALUES(?,?,?,0)',
[(d, hashlib.sha256(k).hexdigest(), '0') for d,k in keys.items()])
self.db.execute('INSERT INTO settings VALUES(?)', (gateway,))
self.db.commit()
except Exception:
self.db.rollback(); raise
actual = dict(self.db.execute('SELECT device,keyhash FROM nodes'))
expected = {d:hashlib.sha256(k).hexdigest() for d,k in keys.items()}
if actual != expected or self.db.execute('SELECT gateway FROM settings').fetchone() != (gateway,):
raise RuntimeError('Configuration does not match provisioned ledger')
def accept(self, packet, keys, rssi, snr, now):
device, seq, fields = decode(packet, keys)
if not math.isfinite(rssi) or not math.isfinite(snr):
raise ValueError('Invalid radio statistics')
if not 1700000000000000000 < now < 9000000000000000000:
raise ValueError('Invalid clock')
self.db.execute('BEGIN IMMEDIATE')
try:
previous, seen = self.db.execute('SELECT seq,seen FROM nodes WHERE device=?', (device,)).fetchone()
if seq <= int(previous):
self.db.rollback(); return 'replay'
if self.db.execute('SELECT count(*) FROM outbox').fetchone()[0] >= LIMIT:
self.db.rollback(); return 'full'
# Unique point for this receiver even if its wall clock moves backwards.
stamp = max(now, seen + 1)
values = [f'sequence={seq}u']
for name, value in fields.items():
values.append(f'{name}={value}'+('i' if name=='soil_raw' else ''))
values += [f'rssi={round(rssi)}i', f'snr={float(snr)}']
line = f'lora,sensor={device},gateway={self.gateway} '+','.join(values)+f' {stamp}'
self.db.execute('INSERT INTO outbox(line) VALUES(?)', (line,))
self.db.execute('UPDATE nodes SET seq=?,seen=? WHERE device=?', (str(seq),stamp,device))
self.db.commit(); return 'queued'
except Exception:
self.db.rollback(); raise
def pending(self):
return self.db.execute('SELECT id,line FROM outbox ORDER BY id LIMIT 1').fetchone()
def ack(self, row_id):
self.db.execute('DELETE FROM outbox WHERE id=?', (row_id,))
def health(self, now, stale_seconds):
lines = []
for device, seen in self.db.execute('SELECT device,seen FROM nodes ORDER BY device'):
age = (now-seen)//1000000000 if seen else -1
healthy = int(seen != 0 and 0 <= age < stale_seconds)
lines.append(f'lora_health,sensor={device},gateway={self.gateway} '
f'healthy={healthy}i,age_seconds={age}i {now}')
return '\n'.join(lines)
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
return None
def post(url, token, body):
req = urllib.request.Request(url, data=body.encode(), method='POST', headers={
'Authorization':'Token '+token, 'Content-Type':'text/plain; charset=utf-8'})
try:
with urllib.request.build_opener(NoRedirect).open(req, timeout=5) as response:
return response.status == 204
except (urllib.error.URLError, TimeoutError, OSError):
return False
def deliver_one(store, url, token):
row = store.pending()
if row is None:
return True
if not post(url, token, row[1]):
return False
store.ack(row[0]); return True
def upload(config, keys, url, stop):
store = Store(config['database'], keys, config['gateway'])
next_health = 0.0
try:
while not stop.is_set():
ok = deliver_one(store, url, config['token'])
if time.monotonic() >= next_health:
healthy_sent = post(url, config['token'], store.health(time.time_ns(), config['stale_seconds']))
if not healthy_sent:
log.warning('Health upload pending; check network/server/token')
next_health = time.monotonic()+60
if not ok:
log.warning('Data upload pending; record retained')
stop.wait(0.2 if ok else 30)
finally:
store.db.close()
def load_config(path):
config = json.loads(Path(path).read_text())
keys = {int(d):bytes.fromhex(k) for d,k in config['keys'].items()}
if not keys or any(not 0 < d < 2**32 or len(k)!=32 for d,k in keys.items()):
raise ValueError('Device IDs and 32-byte keys required')
if len(set(keys.values())) != len(keys):
raise ValueError('Use a different key for each device')
if not 60 <= config['stale_seconds'] <= 86400:
raise ValueError('Choose stale_seconds between 60 and 86400')
base = urllib.parse.urlsplit(config['influx_url'])
if base.username or base.password or base.query or base.fragment or base.path not in ('','/'):
raise ValueError('Use an Influx base URL without credentials/path/query')
if base.scheme != 'https' and not (base.scheme=='http' and base.hostname in ('127.0.0.1','::1','localhost')):
raise ValueError('Remote Influx requires HTTPS')
query = urllib.parse.urlencode({'org':config['org'],'bucket':config['bucket'],'precision':'ns'})
return config, keys, config['influx_url'].rstrip('/')+'/api/v2/write?'+query
def make_radio(config):
import board, busio, digitalio, adafruit_rfm9x
spi = busio.SPI(board.SCK, MOSI=board.MOSI, MISO=board.MISO)
cs = digitalio.DigitalInOut(board.CE0)
reset = digitalio.DigitalInOut(board.D25)
radio = adafruit_rfm9x.RFM9x(spi,cs,reset,config['frequency_mhz'],preamble_length=8,agc=True,crc=True)
radio.signal_bandwidth=125000; radio.coding_rate=8; radio.spreading_factor=10
radio.node=0xff
# RegSyncWord; pinned driver exposes this through its low-level register API.
radio._write_u8(0x39,0x12)
if radio._read_u8(0x39)!=0x12:
raise RuntimeError('Sync word readback failed')
return radio
def main():
parser=argparse.ArgumentParser()
parser.add_argument('config'); parser.add_argument('--init',action='store_true')
args=parser.parse_args()
config,keys,url=load_config(args.config)
if args.init:
store=Store(config['database'],keys,config['gateway'],create=True)
store.db.close(); print('Provisioned; start normally without --init'); return
if config.get('radio_config_verified') is not True:
raise RuntimeError('Verify radio/module/antenna configuration first')
synced=subprocess.run(['timedatectl','show','-p','NTPSynchronized','--value'],
capture_output=True,text=True,check=True,timeout=5).stdout.strip()
if synced!='yes':
raise RuntimeError('Clock is not synchronized; radio not started')
store=Store(config['database'],keys,config['gateway'])
radio=make_radio(config)
stop=threading.Event()
worker=threading.Thread(target=upload,args=(config,keys,url,stop),daemon=True)
worker.start()
try:
while worker.is_alive():
packet=radio.receive(timeout=0.2,with_header=True,with_ack=False,keep_listening=True)
if packet is None:
continue
try:
state=store.accept(bytes(packet),keys,radio.last_rssi,radio.last_snr,time.time_ns())
log.info('Packet %s',state)
except (ValueError,InvalidTag):
log.warning('Rejected packet')
raise RuntimeError('Upload worker stopped')
finally:
stop.set(); radio.sleep(); store.db.close(); worker.join(timeout=12)
if __name__=='__main__':
logging.basicConfig(level=logging.INFO,format='%(asctime)s %(levelname)s %(message)s')
main()
Знакомим шлюз со своими датчиками
{
"gateway": "barn",
"database": "/var/lib/lora-gateway/queue.sqlite",
"radio_config_verified": false,
"frequency_mhz": 433.0,
"keys": {"1": "REPLACE_WITH_THE_TRANSMITTER_KEY_64_HEX_CHARACTERS"},
"influx_url": "http://127.0.0.1:8086",
"org": "farm",
"bucket": "sensors",
"token": "REPLACE_WITH_WRITE_ONLY_TOKEN",
"stale_seconds": 900
}
- В keys внесите ID и тот же ключ из 64 шестнадцатеричных символов, что установлен в ESP32. Для каждого передатчика нужен отдельный случайный ключ; доступ к файлу ограничен служебным пользователем и администратором.
- Заполните адрес InfluxDB, организацию, bucket и токен записи. HTTP разрешён только через локальный адрес Pi; удалённому серверу нужен HTTPS. Перенаправления запрещены, проверка сертификата включена.
- radio_config_verified включают после проверки оборудования и условий эксплуатации. Значение 433 МГц не разрешает передачу само по себе; передатчик должен соответствовать действующим правилам.
- Время записи — время приёма на шлюзе. До подтверждения синхронизации часов через timedatectl радио не запускается. После запуска потеря сети не останавливает приём; для автономной установки отдельно проверяют точное время и запуск без внешнего интернета.
Запускаем базу сначала в терминале, затем как службу
Включите SPI через raspi-config и проверьте /dev/spidev0.0; выполните предложенную системой перезагрузку. Подготовку Blinka сверяйте с инструкцией Adafruit. Группы spi и gpio должны существовать; существующего пользователя lora повторно создавать не нужно.
sudo raspi-config
# Interface Options → SPI → Enable
ls /dev/spidev0.0
sudo apt update
sudo apt install python3-venv python3-dev
sudo useradd --system --user-group --home /var/lib/lora-gateway --shell /usr/sbin/nologin --groups spi,gpio lora
sudo install -d -o root -g lora -m 0750 /opt/lora-gateway
sudo python3 -m venv /opt/lora-gateway/.venv
sudo /opt/lora-gateway/.venv/bin/pip install Adafruit-Blinka adafruit-circuitpython-rfm9x==2.2.25 cryptography
sudo install -o root -g lora -m 0640 gateway.py protocol.py /opt/lora-gateway/
sudo install -o root -g lora -m 0640 config.json /etc/lora-gateway.json
sudo install -d -o lora -g lora -m 0700 /var/lib/lora-gateway
# После заполнения конфигурации, только для нового хранилища:
sudo -u lora /opt/lora-gateway/.venv/bin/python /opt/lora-gateway/gateway.py /etc/lora-gateway.json --init
# Проверка времени и пробный запуск:
timedatectl show -p NTPSynchronized --value
sudo -u lora /opt/lora-gateway/.venv/bin/python /opt/lora-gateway/gateway.py /etc/lora-gateway.jsonЗаполните конфигурацию и создайте хранилище флагом --init. Уже существующее хранилище эта команда сохраняет. Первый запуск удобнее наблюдать в терминале: остановите службу, чтобы с одним радиомодулем работал только один экземпляр.
Запросы ниже рассчитаны на InfluxDB OSS 2.x и Flux, которого нет в InfluxDB 3.x. Используйте инструкцию установки InfluxData и инструкцию Grafana для Debian, выбирая актуальный выпуск нужной ветки и архитектуры. Создайте farm/sensors, токен записи для шлюза и отдельный токен чтения для Grafana.
Сохраните lora-gateway.service и включите службу:
[Unit]
Description=FZ1 LoRa gateway
Wants=network-online.target
After=network-online.target
[Service]
Type=simple
User=lora
SupplementaryGroups=spi gpio
WorkingDirectory=/opt/lora-gateway
ExecStart=/opt/lora-gateway/.venv/bin/python /opt/lora-gateway/gateway.py /etc/lora-gateway.json
Restart=on-failure
RestartSec=10
StateDirectory=lora-gateway
StateDirectoryMode=0700
UMask=0077
NoNewPrivileges=true
StandardOutput=journal
StandardError=journal
[Install]
WantedBy=multi-user.target
sudo install -m 0644 lora-gateway.service /etc/systemd/system/lora-gateway.service
sudo systemctl daemon-reload
sudo systemctl enable --now lora-gateway
systemctl status lora-gateway
journalctl -u lora-gateway -fТеперь проверьте запуск при недоступной базе данных: показания должны ждать в очереди. After задаёт порядок запуска служб, а Restart=on-failure повторяет запуск завершившегося с ошибкой процесса через 10 секунд. Для зависаний понадобится отдельно испытанный watchdog. Также ограничьте размер systemd journal и следите за свободным местом на накопителе.
Рисуем первый график и замечаем замолчавший датчик
Добавьте в Grafana источник InfluxDB: язык Flux, адрес, организацию, bucket и токен чтения. Нажмите Save & test. Если Grafana находится на другом компьютере, укажите адрес Pi: localhost означает сам компьютер с Grafana. Начнём с RSSI для шлюза barn:
from(bucket: "sensors")
|> range(start: -24h)
|> filter(fn: (r) => r._measurement == "lora" and r._field == "rssi" and r.gateway == "barn")
|> aggregateWindow(every: 1h, fn: mean, createEmpty: false)После первого графика добавьте temperature, humidity, battery или snr. Сравните изменения температуры с качеством связи. Поле battery появится после включения измерения на передатчике; его пороги выбираются под аккумулятор и питание. Для soil_raw сначала нужна калибровка: пока график показывает отсчёты АЦП.
Важно заметить и тишину. Примерно раз в минуту шлюз сообщает состояние настроенных датчиков: healthy=1 при недавнем пакете и 0, когда данных давно нет. До самого первого пакета age_seconds равно −1. В примере ожидание составляет 900 секунд, или 15 минут; согласуйте его с интервалом своего передатчика.
from(bucket: "sensors")
|> range(start: -5m)
|> filter(fn: (r) => r._measurement == "lora_health" and r._field == "healthy" and r.gateway == "barn")
|> last()Создайте правило Reduce → Last с условием ниже 1 и, для первого опыта, подтверждением в две минуты. Переведите No Data и Error в тревогу. Затем по очереди проверьте три ситуации: датчик ещё ни разу не передавал, работавший датчик замолчал, шлюз остановлен. Так станет понятно, какое сообщение придёт в каждом случае.
Создайте Telegram-бота через BotFather, получите токен и chat ID, добавьте контакт Telegram в Grafana и привяжите к правилу или политике уведомлений. Отправьте тестовое уведомление по инструкции Grafana. После теста выключите передатчик и дождитесь настоящего срабатывания правила.
Устраиваем базе небольшое испытание
- Сверьте ключи и радиопараметры. Сравните показания с контрольным прибором и записями в базе; queued подтверждает лишь локальное сохранение.
- Отключите сеть или InfluxDB, примите данные, перезапустите процесс и восстановите связь. Очередь должна выгрузиться с прежними метками времени; старые записи могут быть отклонены политикой хранения bucket.
- Проверьте повреждённые и повторные пакеты, неверный токен, заполнение очереди и накопителя. Испытайте питание, нагрев и долю доставки на участке.
- Перед обслуживанием остановите службу. SQLite и конфигурацию копируют при остановке или штатным механизмом резервного копирования SQLite; архивы требуют ограниченного доступа.
Изменение набора ID, ключей или имени шлюза без согласованного изменения хранилища останавливает запуск; автоматической миграции нет. Перед переинициализацией сохраните архив и выгрузите очередь, затем замените ключи всех зарегистрированных передатчиков и шлюза. Старая копия счётчика с прежним ключом может снова допустить повторные пакеты.
Если позже поставите ещё один шлюз, назначьте ему другое имя gateway. Один пакет могут услышать оба приёмника: получите два наблюдения с разными радиопоказателями. Учитывайте это в графиках и расчётах; для объединения в одно измерение потребуется удаление повторов по паре «ID — номер».
Перед постоянной установкой ограничьте локальную базу локальным интерфейсом, защитите доступ к Grafana и администрированию. Наружную антенну и грозозащиту подберите для всей установки. Затем повторите проверку приёма из тех мест, где будут работать датчики.
Самый наглядный опыт — отключить доступ к InfluxDB, принять несколько пакетов и вернуть соединение. Проследите, как очередь убывает и история появляется на графике. После этого отдельно проверьте перезапуск шлюза с накопленными данными: так вы познакомитесь с поведением своей базы до установки в хозяйстве.
Документация: RFM9x, SQLite в Python, ответы InfluxDB, No Data и Error.
Для полевой стороны проекта — Передатчик ESP32 и формат FZ1.
Для датчика вдали от розетки — Питание удалённых датчиков от аккумулятора и солнечной панели.
