bobaos.pub — KNX TP/UART, Raspberry Pi и Redis

Нет предела совершенству. Казалось бы, все работало хорошо, исправлялись мелкие баги и прочее.

Сейчас я расскажу, во-первых, о проблемах, с которыми столкнулся за все время, прошедшее с момента предыдущей статьи, и, во-вторых, о решениях, которые способствовали текущему статусу проекта.

→ Статья про предыдущую версию

Обозначения

bobaos — npm модуль для взаимодействия с BAOS 83x с помощью UART. Возвращает сырые данные. Используется во всех остальных модулях, перечисленных ниже.

bdsd.sock — скрипт для работы с KNX объектами. Хранит список датапоинтов, при отправке/получании преобразует значения. Из DPT1 в true/false, из DPT9 — во float. Также слушает Unix Socket для приема запросов от других процессов.

bobaos.pub — новая версия, использующая redis для межпроцессного взаимодействия.
KNX объект/датапоинт — настроенный в ETS объект связи модуля BAOS 83x, которому соответствует(или нет) групповой адрес(а). В текущих версиях железа максимальное количество 1000.

Задача

Основная задача та же, что решала и предыдущая версия. Соединение с серийным портом можно открыть только одно. Скриптов, работающих с KNX, хочется запустить много. В довесок к этому, захотелось реализовать межпроцессное взаимодействие. Т.е. чтобы не только один процесс bdsd.sock слушал сокет, а каждый запущенный скрипт мог как отправлять, так и принимать запросы.

Идея

В голове родилась идея сделать свой брокер сообщений на node.js поверх юникс сокетов, к которому подключались бы клиенты, подписывались на топики и получали/отправляли сообщения в соответствии с прописанным в них кодом. Я знал, что есть уже готовые решения, про которые в последнее время только ленивый не слышал, изучал, но идея сделать решение свое была навязчивой.

И вот в итоге сервис запущен.

Написал логгер, который сообщения отправляет в топик. Подписчики получают и вольны делать все что угодно, а точнее — что прописано. Удобно — логи с нескольких источников можно смотреть в одном консольном выводе.

Написал, опубликовал в npm пакет bobaos.pub, который, в отличие от bdsd.sock, уже не создает сокет файл, а подключается к брокеру. На первый взгляд все работает как надо.

Проблема

Дальше запустил скрипт, периодически отправляющий запросы в шину KNX, на ночь. Проснувшись утром, по миганию светодиодов, сигнализирующих отправку/передачу данных, понял, что что-то не так. Сообщения не доходили вовремя. Обнаружил, что самописный брокер сообщений занял практически всю доступную из 512МБ(у BeagleBoard Black) оперативной памяти. Дальнейшая работа с nodejs подтвердила, что память — слабое место место js скриптов.

Решение

В результате было принято решение перейти с самописных Unix сокетов на Redis(к слову, он тоже умеет работать с ними). Возможно, стоило разобраться с памятью, найти утечки, но хотелось быстрее запустить.

bobaos подразумевает общение по UART с оборачиванием сообщений в FT1.2, коммуникация у нас синхронная. Т.е. <..сообщение-подтверждение-ответ..>. Модуль bobaos, ответственный за коммуникацию, хранит все запросы в массиве, выдергивает оттуда по очереди, отправляет в UART, и при входящем ответе разрешает промис, ответственный за этот запрос.

Можно пойти следующим путем: сервис слушает PUB/SUB канал redis, принимает запросы, отправляет в KNX. В этом случае нагрузка на очередь запросов ложится на js модуль bobaos. Для реализации надо написать простой модуль, подписанный на канал и преобразующий сообщения методом JSON.parse(). Далее этот модуль можно использовать в остальных скриптах.

Другой вариант, на котором в итоге остановился: использовать существующий менеджер задач поверх redis. Существует несколько, выбор сделал на bee-queue.

→ Под капотом

Тут описано каким образом работает очередь bee-queue. Если реализовать эту библиотеку для других языков программирования, то можно таким образом сделать клиентские библиотеки и для bobaos.

Во втором варианте все запросы хранятся в списках redis, выдергиваются по очереди и отправляются на серийный порт.

Далее, следует переписывание предыдущей версии, но уже все данные по датапоинтам храню в redis базе данных. Единственное неудобство, которое испытываю — все запросы асинхронны, соответственно, получить массив значений уже чуть тяжелее, чем просто обратиться к массиву.

Были сделаны небольшие оптимизации.

Если раньше были раздельные методы getValue/getValues/readValue/readValues/setValue/setValues, то теперь getValue/readValue/setValues принимают как одно значение, так и массив.

Метод getValue([id1, id2, ...]) в прошлой версии отправлял для каждого датапоинта запрос в серийный порт. Но ведь есть возможность отправить запрос для нескольких значений. Ограничения — размер ответа должен быть равен BufferSize, максимальный — 250 байт; также нельзя выходить за пределы количества объектов, для текущих версий модулей BAOS 83x — 1000.

Длины значений известны, заголовка тоже. Далее достаточно простой алгоритм с циклами while и await-ами =)

1. Сортируем массив, удаляем повторяющиеся элементы, если есть. получаем массив idUniq.
2. Запускаем цикл i < idUniq.length, в котором делаем следующее:
   а) Берем start: idUniq[i], для него считаем максимальное количество значений, которое можем получить. Для примера, если все объекты имеют тип DPT1/DPT5(1 байт), то можем получить значения в количестве 48. Тут есть одно замечание: если, для примера, у нас в ETS настроены объекты #[1, 2, 3, 10, 20], то при запросе GetDatapointValue.Req(1, 30) в ответ будут возвращены нулевые однобайтные значения даже для несуществующих датапоинтов [4, 5, 6, ..., 30].
   б) Подсчет происходит в новом цикле j < i + max(где max — 50, либо, если близко к 1000, то максимально 1000 - id + 1, для 990 это будет 11, для 999 — 2), если в процессе подсчета встречаем элементы массива из исходного запроса, то переменной i присваеваем его индекс.
   в) Если в цикле j расчетная длина выходит больше максимальной длины буффера, то формируем элемент "карты" запросов {start: start, number: number}, закидываем в отдельный массив, увеличиваем переменную i(либо присваиваем индекс найденной в массиве idUniq), прерываем цикл j, оба цикла запускаются заново.

Таким образом, формируем несколько запросов для bobaos. Для примера, если отправить запрос getValue([1, 2, 3, 40, 48, 49, 50, 100, 998, 999, 1000]), то запросы могут быть для частного случая следующие:

{start: 1, number: 48}, // 1, 2, 3, 40, 48
{start: 49, number: 48}, // 49, 50
{start: 100, number: 48}, // 100
{start: 998, number: 3} // 998, 999, 1000

Можно было бы сделать по другому:

{start: 1, number: 48}, // 1, 2, 3, 40, 48
{start: 49, number: 2}, // 49, 50
{start: 100, number: 1}, // 100
{start: 998, number: 3} // 998, 999, 1000

Запросов было бы столько же, данных меньше. Но остановился на первом варианте, поскольку полученные значения сохраняются в базу redis, соответственно, их можно получить методом getStoredValue, который я стараюсь использовать чаще, чем getValue, который отправляет данные по серийному порту.

Для сервисных методов ping/get sdk state/reset создается отдельная очередь. Таким образом, если с коммуникацией по серийному порту что-то не так(сбился счетчик фреймов, и т.д.) и выполнение остановилось на какой-то задаче, можно послать запрос reset в другой очереди, и соответственно, перезапустить sdk.

Клиентская часть — bobaos.sub

Для управления датапоинтами KNX в пользовательских скриптах может быть использован bobaos.sub модуль.

Следующий пример покрывает все функции модуля:

const BobaosSub = require("bobaos.sub");

// в параметры можно передать опционально:
// redis: объект либо url
// request_channel: "bobaos_req" по умолчанию, 
// service_channel: "bobaos_service" по умолчанию,
// broadcast_channel: "bobaos_bcast" по умолчанию
let my = BobaosSub();

my.on("connect", _ => {
  console.log("connected to ipc, still not subscribed to channels");
});

my.on("ready", async _ => {
  try {
    console.log("hello, friend");
    console.log("ping:", await my.ping());
    console.log("get sdk state:", await my.getSdkState());
    console.log("get value:", await my.getValue([1, 107, 106]));
    console.log("get stored value:", await my.getValue([1, 107, 106]));
    console.log("get server item:", await my.getServerItem([1, 2, 3]));
    console.log("set value:", await my.setValue({id: 103, value: 0}));
    console.log("read value:", await my.readValue([1, 103, 104, 105]));
    console.log("get programming mode:", await my.getProgrammingMode());
    console.log("set programming mode:", await my.setProgrammingMode(true));
    console.log("get parameter byte", await my.getParameterByte([1, 2, 3, 4]));
    console.log("reset", await my.reset());
  } catch(e) {
    console.log("err", e.message);
  }
});

my.on("datapoint value", payload => {
  // Имейте в виду, что payload может быть как объектом, так и массивом
  // так что необходима проверка Array.isArray(payload)
  console.log("broadcasted datapoint value: ", payload);
});

my.on("server item", payload => {
  // Имейте в виду, что payload может быть как объектом, так и массивом
  // так что необходима проверка Array.isArray(payload)
  console.log("broadcasted server item: ", payload);
});

my.on("sdk state", payload => {
  console.log("broadcasted sdk state: ", payload);
});

bobaos.tool

Был переписан интерфейс командной строки. О тот как я его реализовывал следующая статья:

→ Пишем CLI на NodeJS

Команды стали короче, понятнее и функциональнее.

bobaos> progmode ?
BAOS module in programming mode: false
bobaos> progmode 1
BAOS module in programming mode: true
bobaos> progmode false
BAOS module in programming mode: false
bobaos> description 1 2 3
#1: length = 2, dpt = dpt9, prio: low, flags: [C-WTU]
#2: length = 1, dpt = dpt1, prio: low, flags: [C-WT-]
#3: length = 1, dpt = dpt1, prio: low, flags: [C-WT-]
bobaos> set 2: 0
20:27:06:239,    id: 2, value: false, raw: [AA==]
bobaos> set [2: 0, 3: false]
20:28:48:586,    id: 2, value: false, raw: [AA==]
20:28:48:592,    id: 3, value: false, raw: [AA==]

Послесловие

Получилась рабочая стабильная система. Redis в качестве бэкенда работает стабильно хорошо. В ходе разработки было набито немало шишек. Но процесс обучения таков, что иногда это неизбежно. Из полученного опыта отмечу, что nodejs процессы потребляют довольно много оперативной памяти(20МБ на старте) и могут быть утечки. Для домашней автоматизации это критично — потому как скрипт должен работать постоянно, а если он растет с течением времени все больше, то к определенному моменту он может занять все пространство. Поэтому надо внимательно писать скрипты, понимать как работает сборщик мусора и держать все под контролем.

Обновляющаяся документация может быть найдена здесь.

В следующей статье расскажу о том, как с помощью redis и bee-queue сделал сервис для программных аксессуаров.

UPD: bobaoskit — аксессуары, dnssd и WebSocket

Буду рад любой обратной связи.