Подключение с использованием собственного драйвера

Система КСИА.Шлюз предоставляет возможность добавлять кастомные драйвера для подключения оборудования с редкими протоколами или интеграции программного обеспечения.

Создание кастомного драйвера происходит в два шага:

1. Объявление драйвера. Создать и настроить конфигурационный файл в формате JSON.
2. Реализация драйвера (скрипт). Написание скрипта на языке Python или JavaScript.

Объявление драйвера

Создайте JSON-файл с описанием драйвера и поместите его в папку drivers (располагается в папке с установленной программой КСИА.Шлюз).

После этого драйвер отобразится в выпадающем списке Добавить устройство.

Важно: На этом этапе драйвер уже отображается в интерфейсе, но не работает — данные не передаются.

Добавление устройства с кастомным драйвером происходит также как и со встроенным драйвером.

Описание конфигурационного файла

{
  "id": "custom_device_mqtt",
  "title": "Пользовательское устройство (MQTT)",
  "deviceConfig": {
    "name": {
      "title": "Наименование",
      "type": "string",
      "default": "Устройство"
    },
    "path": {
      "title": "Путь к файлу",
      "type": "string",
      "default": ""
    },
    "command": {
      "title": "Команда для запуска",
      "type": "string",
      "default": "mqtt"
    },
    "url": {
      "title": "Адрес брокера",
      "type": "string",
      "default": "0.0.0.0:1883"
    }
  },
  "pointConfig": {
    "name": {
      "title": "Наименование точки",
      "type": "string",
      "default": ""
    },
    "param": {
      "title": "Topic",
      "type": "string",
      "default": ""
    }
  }
}

Файл JSON должен содержать следующие поля:

• id: string; обязательный параметр; уникальный идентификатор драйвера;
• title: string; обязательный параметр; отображаемое название драйвера в интерфейсе;
• deviceConfig: object; обязательный параметр; описание параметров устройства:
  • title: string; обязательный параметр; отображаемое название параметра в интерфейсе;
  • type: string; обязательный параметр; тип данных (string, number, boolean);
  • default: string; необязательный параметр; значение по умолчанию (опционально).
• pointConfig: object; обязательный параметр; описание параметров точки:
  • title: string; обязательный параметр; отображаемое название параметра в интерфейсе;
  • type: string; обязательный параметр; тип данных (string, number, boolean);
  • default: string; необязательный параметр; значение по умолчанию (опционально).

Реализация драйвера (скрипт на Python)

После объявления драйвера через JSON-конфиг необходимо обеспечить его работоспособность путём реализации скрипта на языке Python.

Важно: Без скрипта драйвер отображается в интерфейсе, но данные от устройств не принимаются.

Порядок работы скрипта:

1. КСИА.Шлюз запускает скрипт, передавая ему параметры устройства из deviceConfig в порядке объявления.
2. Скрипт устанавливает соединение с оборудованием в соответствии с полученными параметрами.
3. Скрипт считывает значения по адресам или топикам, заданным в настройках точек.
4. Скрипт выводит результат в стандартный поток вывода (stdout) в формате JSON.
5. КСИА.Шлюз принимает вывод скрипта и сохраняет значения в соответствующие точки.

Полный пример скрипта

Ниже представлен пример реализации скрипта на Python. Скрипт выполняет следующие функции:

• Принимает параметры устройства.
• Периодически генерирует и отправляет данные в КСИА.Шлюз.
• Корректно обрабатывает сигналы остановки (SIGTERM, SIGINT).

import sys
import json
import time
import signal
from typing import Dict, Any

# Флаг для остановки цикла
running = True

def send_message(data: Dict[str, Any]):
    """Отправляет JSON-сообщение в stdout"""
    print(json.dumps(data))
    sys.stdout.flush()

def signal_handler(signum, frame):
    """Обрабатывает SIGTERM/SIGINT"""
    global running
    running = False
    send_message({"status": "info", "message": "Получен сигнал остановки"})

# Регистрация обработчиков сигналов
signal.signal(signal.SIGINT, signal_handler)   # Ctrl+C
signal.signal(signal.SIGTERM, signal_handler)  # kill

def main():
    # Получение параметров устройства из командной строки
    # Порядок параметров соответствует объявлению в deviceConfig
    if len(sys.argv) < 3:
        send_message({"status": "error", "message": "Недостаточно параметров"})
        sys.exit(1)

    ip = sys.argv[1]
    port = int(sys.argv[2])

    # Параметры опроса
    interval = 5.0
    message_count = 0
    start_time = time.time()

    # Уведомление о готовности
    send_message({
        "status": "ready",
        "message": f"Драйвер запущен. Подключение к {ip}:{port}"
    })

    try:
        while running:
            # Здесь должна быть логика подключения к оборудованию
            # и получения реальных значений

            # Пример: получение значений
            temperature = 25.5
            pressure = 2.3

            # Формирование выходных данных
            result = [
                {"name": "температура", "value": temperature, "quality": "good"},
                {"name": "давление", "value": pressure, "quality": "good"}
            ]

            # Отправка данных в КСИА.Шлюз
            print(json.dumps(result))
            sys.stdout.flush()

            message_count += 1
            time.sleep(interval)

    except (KeyboardInterrupt, SystemExit):
        pass
    finally:
        send_message({
            "status": "shutdown",
            "message": "Драйвер остановлен",
            "total_messages": message_count
        })
        sys.exit(0)

if __name__ == "__main__":
    main()