¶ Написание модулей к хабу
¶ PS ссылки могут быть неактуальны, в этом случае обратитесь к руководителям проекта #19102 Телецентры
Ссылка на репозиторий
В хаб можно добавлять программные модули для считывания данных с устройств или сервисов и отправки команд в другие сервисы. Есть 2 вида модулей:
- listener: для постоянного считывания данных.
- publisher: для отправки данных в другие сервисы
Взаимодействие между ними (передача данных) с помощью каналов redis репозиторий в соответствии с конфигурацией, которую задает пользователь в последнем изменении для этого использовалась таблица конфигурации, инструкция к ней представлена на листе "инструкция". С помощью скрипта (который так же поставляется вместе с хабом скрипт)
Диаграмма, визуализирующая работу хаба ссылка
¶ Требования для классов listener'ов
¶ Структура
Класс должен поставляться в виде модуля с такой структурой
— <listener_name>/
- __main__.py
— <listener_name>.py
— settings.py
¶ <listener_name>.py
Содержит код класса
- На вход класс принимает пути конфигурационных файлов с расширением YAML. Должны быть конфиги по умолчанию, но они будут использоваться в основном в тестовых режимах, при "боевой" работе хаба будут использоваться конфиги пользователя.
- Функция job - основная функция всего класса, она не должна быть блокирующей (будет вызываться в цикле while), то есть просто раз в некоторый момент времени будет проверять состояние пульта
- Функция echo - функция для теста подключения пульта, предусмотрен только вывод сырых данных с пульта
- Если ваш listener должен совершить какие-то действия перед тем как будет запущен (имеет конструктор), то все нужные операции кладем в функцию prepare(вызывается в конструкторе родительского класса)
from listener.base_listener import BaseListener
from utils.logger import logger
logger = logger(__name__)
class TestListener(BaseListener):
redis_channels = [] # список каналов, на сообщения в которых следует реагировать (включить подсветку и тд) инициализировать до вызова конструктора родительского класса
def __init__(self, google_sheet):
super().__init__(google_sheet)
def prepare(self):
self.device = connect_device(port) # ПРИМЕР
. . .
def job(self):
data = <your_protocol>.read() # Для разных протоколов - разные функции, это просто пример
if data in self.conf:
return self.conf[button]
def redis_handler(self, topic, message):
pass
# Обрабатывать сообщения, выполнять какие-то действия
def echo(self):
while True:
raw = <your_protocol>.read() # Для разных протоколов - разные функции, это просто пример
if raw:
print(raw)
¶ Если пульт возвращает численные значения
Если ваш пульт возвращает числинные значения (например координаты джойстика или значение на котором сейчас фейдер), то в функцию job должно быть включено типовое сообщение, в которое кладется нужное значение и возвращается (в этом случае лучше всего задать типовые сообщения прямо в файле настроек)
Пример:
from listener.base_listener import BaseListener
from utils.logger import logger
from listener.joystick.settings import config, stop_msg, z_msg, x_msg, y_msg
logger = logger(__name__)
class TestListener(BaseListener):
def __init__(self, google_sheet):
super().__init__(google_sheet)
def prepare(self):
# Standard messages
self.stop_msg = stop_msg
self.z_msg = z_msg
self.x_msg = x_msg
self.y_msg = y_msg
self.joystick = None
self.connect_joystick()
def job(self):
data = <your_protocol>.read() # Для разных протоколов - разные функции, это просто пример
if data.startswith('Joystick'): # Нужно отловить, что данные пришли с джойстика
msg = copy.deepcopy(self.x_msg) # Если движение по оси x, то нам нужно типовое сообщение для x
msg = copy.deepcopy(self.y_msg) # Если движение по оси y, то нам нужно типовое сообщение для y
msg = copy.deepcopy(self.z_msg) # Если движение по оси z(прим. zoom), то нам нужно типовое сообщение для z
msg['value'] = data['x/y/z']
return [msg]
if data in config: # Если же пришла информация о нажатой кнопке, то возвращаем значение из конфига
return config[button]
def echo(self):
while True:
data = <your_protocol>.read() # Для разных протоколов - разные функции, это просто пример
if data.startswith('Joystick'): # Нужно отловить, что данные пришли с джойстика
msg = copy.deepcopy(self.x_msg) # Если движение по оси x, то нам нужно типовое сообщение для x
msg = copy.deepcopy(self.y_msg) # Если движение по оси y, то нам нужно типовое сообщение для y
msg = copy.deepcopy(self.z_msg) # Если движение по оси z(прим. zoom), то нам нужно типовое сообщение для z
msg['value'] = data['x/y/z']
print(msg)
if data in config: # Если же пришла информация о нажатой кнопке, то возвращаем значение из конфига
print(config[button])
¶ main.py
- Файл должен выглядеть строго таким образом
- Секция
if __name__ == '__main__':предусмотрена для запуска listener'а
from listener.<listener_name>.<listener_name>.py import TestListener
import argparse
def main():
parser = argparse.ArgumentParser(description='Run onvif publisher on your module')
parser.add_argument('google_sheet', type=str, help='Type of API')
args = parser.parse_args()
TestListener(args.google_sheet)
if __name__ == '__main__':
main()
¶ settings.py
Файл с настройками listener’а, это могут быть настройки redis сервера либо порты подключения пульта, всё это должно быть собрано в этом файле. Также в этом файле должны открываться все конфиги:
listener_port = os.environ.get('<LISTENER_NAME>_PORT', '/dev/cu.usbserial-14230')
PS Потом это импортится в <listener_name>.py:
from listener.<listener_name>.settings import listener_conf, listener_port
¶ button identifier
От программиста требуется список разных button_identifier, которые используются в config.yaml и их соответствие физическим кнопкам на пульте
Файл config.yaml содержит конфиг для кнопок такой структуры (генерируется автоматически утилитой google_sheet)
<button-identifier>:
- publisher: <publisher1_name>
data:
. . .
- publisher: <publisher2_name>
data:
. . .
Здесь publisher1_name, publisher2_name - названия pulisher'ов которым будет передаваться data (data для разных publisher'ов может быть разная)
PS Структура любого сообщения -- {"publisher":<publisher_name>, "data":<your_data>}
PS2 Конфиги могут меняться в зависимости от программ, с которыми планируется использовать ваш listener, поэтому ваша задача - сделать его максимально универсальным. Нужно предоставить тестовый конфиг.
¶ Требования для классов publisher'ов
¶ Структура
Класс должен поставляться в виде модуля с такой структурой:
— <publisher_name>/
— __init__.py
— <publisher_name>.py
— settings.py
¶ <publisher_name>.py
Содержит код класса:
- Функция job - основная функция всего класса, она должна парсить data и в зависимости от содержимого вызывать те или иные функции (для примера расписано условие изменения активного пресета камеры)
Программист должен предоставить требования структуры данных, которые должен будет обрабатывать его publisher
from publisher.base_publisher import BasePublisher
class TestPublisher(BasePublisher):
def __init__(self):
super().__init__()
. . .
def job(self, data):
if data[‘change’] == ‘preset’:
self.ptz_change_preset(data[‘preset_token’])
¶ settings.py
Файл с настройками publisher'а, это могут быть порт и хост сервера, куда отправляются request'ы, всё это должно быть собрано в этом файле:
server_host = os.environ.get('<PUBLISHER_NAME>SERVER_HOST', '172.18.130.50')
server_port = int(os.environ.get('<PUBLISHER_NAME>SERVER_PORT', 5119))
PS Потом это импортится в <publisher_name>.py:
from publisher.<pulisher_name>.publisher_settings import server_host, server_port
¶ Дополнительно
¶ env.sh
Также желательно создать файл env.sh в котором вы будете обозначать ваши переменные окружения, такие как порт или хост сервера:
<PUBLISHER/LISTENER_NAME>_SERVER_PORT=5119
<PUBLISHER/LISTENER_NAME>_SERVER_HOST=172.18.130.50
Вместо <PUBLISHER_NAME> или <publisher_name> вставь название своего publisher'а, например:
- Если название: onvif, то <PUBLISHER_NAME> ——> ONVIF, <publisher_name> ——> onvif
¶ Utils
В проект включены некоторые утилиты, они будут описаны в этом разделе
- google_sheet_conf
Используется для преобразования листа конфигурационной google таблицы в yaml файл
Для запуска отдельно от listener'а выполнить команду (находясь в репозитории universal-hub):
python3 utils/google_sheet_conf <sheet_name>
Файл будет сохранен по пути universal-hub/data/<sheet_name>.yaml
- logger
Используется для централизации настройки формата и уровня логгирования, все логгирование должно быть осуществлено с использованием этой утилиты:
from utils.logger import logger
logger = logger(__name__)
- redis
Используется для взаимодействия с redis'ом.
¶ Запуск на модуле
¶ Перед запуском
Перед запуском нужно убедиться, что на рабочем ПК с ОС Windows установлен плагин для OBS obs-websocket-5 актуальной версии для установленного OBS и установить его в случае отсутствия.
¶ Требования к модулю
Нужен компьютер с установленной на него ОС Debian Bullseye или Ubuntu 20.04 LTS.
¶ Установка необходимых пакетов, создание venv, активация udev-правил
sudo apt install libhidapi-dev libhidapi-hidraw0 libhidapi-libusb0 libportmidi-dev libsdl2-dev python3 python3-dev python3-pip python3-venv -y
git clone https://git.miem.hse.ru/19102/universal-hub.git
cd universal-hub
git checkout develop
python3 -m venv venv
source venv/bin/activate
pip3 install -r requirements.txt
sudo sh udev_rules/udev.sh
¶ Настройка Redis
- Установить актуальную версию redis согласно инструкции
- Подтянуть и запустить образ redis (с модулем RedisJSON)
sudo docker run -d -p 6379:6379 --name redis --restart always redis/redis-stack-server:latest
Или с помощью docker compose
sudo docker compose up -d
¶ Простой запуск листенеров и паблишеров
- Задать переменные среды в
src/env.sh - Ввести команду(листенер):
python3 listener/<listener_name>
- Ввести команду(паблишер)
python3 publisher/<publisher_name>
¶ Запуск в качестве сервиса
- Скопировать содержимое папки
service_filesв /etc/systemd/system
cp service_files/* /etc/systemd/system
- Daemon reload
sudo systemctl daemon-reload
- Включить автозагрузку сервисов при старте системы
sudo systemctl enable service-name.service
- Запуск сервиса вручную
sudo systemctl start service-name.service
¶ Прочитать логи сервиса
sudo journalctl -eu service-name.service
¶ Таблица с конфигурацией
¶ Дополнительно
-
¶ HID листенеры
Должны запускаться с sudo и указанием PYTHONPATH вручную.
(sudo env PYTHONPATH=*your/path* python3 ...)