Протоколы интеграции оборудования

1. Спецификация сетевого интерфейса приёма телеметрии

Входящий протокол телеметрии платформы «UrFU HeatBalance» стандартизован на базе протокола HTTPS (версии TLS 1.2 / TLS 1.3) и формата JSON:

POST /api/v1/iot/telemetry HTTP/1.1
Host: heatbalance.online
Content-Type: application/json
X-Device-Token: 7f9a2b8c4d1e0f3a6b5c7d8e9f0a1b2c
User-Agent: HeatBalance-EdgeClient/2.4.0

{
  "dryMaterialMass": 12500.0,
  "initialMoisture": 14.8,
  "finalMoisture": 1.2,
  "airTemperature": 22.5,
  "recirculateTemp": 235.4
}

Коды возврата и семантика ответов:

• 202 Accepted: пакет успешно принят, валидирован и поставлен в очередь RabbitMQ на расчет. Тело ответа содержит идентификатор лога:

  {
    "logId": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
    "status": "Pending"
  }

• 401 Unauthorized: заголовок X-Device-Token отсутствует, либо указанный токен не зарегистрирован в базе данных IoTDbContext.Devices.
• 400 Bad Request: синтаксическая ошибка в JSON-документе.

---

2. Пример интеграции 1: Python-агент опроса ПЛК по Modbus TCP

Данный скрипт предназначен для развертывания на цеховом шлюзе (Moxa, Advantech, Raspberry Pi CM4). Он непрерывно опрашивает Holding Registers контроллера печи (Siemens S7-1200 / ОВЕН ПЛК-210) по протоколу Modbus TCP, преобразует пары 16-битных регистров в числа с плавающей точкой стандарта IEEE 754 (Float32) и отправляет их в платформу.

#!/usr/bin/env python3
"""
HeatBalance Edge Agent: Modbus TCP to Cloud Gateway
Лицензия: Промышленная лицензия UrFU HeatBalance
"""

import time
import struct
import logging
import requests
from pymodbus.client import ModbusTcpClient

# Настройка логирования
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("ModbusGateway")

# Конфигурация цехового ПЛК (Level 1)
PLC_IP = "192.168.10.50"
PLC_PORT = 502
POLL_INTERVAL_SEC = 5.0

# Конфигурация платформы HeatBalance (Level 4)
CLOUD_URL = "https://heatbalance.online/api/v1/iot/telemetry"
DEVICE_TOKEN = "7f9a2b8c4d1e0f3a6b5c7d8e9f0a1b2c"

def decode_float32(registers, offset):
    """Декодирование двух 16-битных регистров Modbus в float32 (Big-Endian)."""
    r1, r2 = registers[offset], registers[offset + 1]
    packed = struct.pack(">HH", r1, r2)
    return round(struct.unpack(">f", packed)[0], 2)

def main():
    logger.info("Запуск шлюза Modbus TCP -> HeatBalance Cloud...")
    client = ModbusTcpClient(PLC_IP, port=PLC_PORT, timeout=3.0)

    while True:
        try:
            if not client.is_socket_open():
                if not client.connect():
                    logger.warning(f"Не удалось подключиться к ПЛК {PLC_IP}:{PLC_PORT}. Повтор через 5с...")
                    time.sleep(5)
                    continue

            # Считывание 10 регистров начиная с адреса 40001 (offset 0)
            # 40001-40002: Температура рабочей камеры (°C)
            # 40003-40004: Температура рециркулята (°C)
            # 40005-40006: Начальная влажность материала (%)
            # 40007-40008: Конечная влажность материала (%)
            # 40009-40010: Массовый расход материала (кг/ч)
            rr = client.read_holding_registers(address=0, count=10)
            if rr.isError():
                logger.error(f"Ошибка протокола Modbus: {rr}")
                time.sleep(POLL_INTERVAL_SEC)
                continue

            regs = rr.registers
            payload = {
                "airTemperature": decode_float32(regs, 0),
                "recirculateTemp": decode_float32(regs, 2),
                "initialMoisture": decode_float32(regs, 4),
                "finalMoisture": decode_float32(regs, 6),
                "dryMaterialMass": decode_float32(regs, 8)
            }

            headers = {
                "X-Device-Token": DEVICE_TOKEN,
                "Content-Type": "application/json"
            }

            resp = requests.post(CLOUD_URL, json=payload, headers=headers, timeout=5.0)
            if resp.status_code == 202:
                logger.info(f"Телеметрия успешно отправлена: {payload}, Ответ: {resp.json()}")
            else:
                logger.warning(f"Платформа вернула статус {resp.status_code}: {resp.text}")

        except requests.exceptions.RequestException as net_err:
            logger.error(f"Ошибка связи с сервером HeatBalance: {net_err}")
        except Exception as ex:
            logger.error(f"Внутренний сбой итерации: {ex}", exc_info=True)

        time.sleep(POLL_INTERVAL_SEC)

if __name__ == "__main__":
    main()

---

3. Пример интеграции 2: Прошивка C++ для промышленного микроконтроллера (ESP32 / W5500)

Для автономных измерительных узлов на базе контроллеров ESP32 с аппаратным проводным интерфейсом Ethernet (чип WIZnet W5500) или защищенным Wi-Fi:

#include <Arduino.h>
#include <WiFi.h>
#include <HTTPClient.h>
#include <ArduinoJson.h>

// Сетевые настройки
const char* WIFI_SSID = "PROD_FURNACE_ZONE_A";
const char* WIFI_PASS = "IndustrialSecurityKey2026";
const char* SERVER_URL = "https://heatbalance.online/api/v1/iot/telemetry";
const char* DEVICE_TOKEN = "7f9a2b8c4d1e0f3a6b5c7d8e9f0a1b2c";

// АЦП и калибровка первичных преобразователей 4-20 мА
const int PIN_TEMP_SENSOR = 34;   // Термопара (через нормирующий усилитель 4-20 мА -> 0-3.3 В)
const int PIN_PRESSURE_SENSOR = 35; // Датчик давления газа (0-3.3 В)

float readTemperatureCelsius() {
    int rawValue = analogRead(PIN_TEMP_SENSOR);
    // Калибровка: 0..4095 дискрет АЦП соответствует 0..1200 °C
    return (float)rawValue * (1200.0f / 4095.0f);
}

float readGasPressureKPa() {
    int rawValue = analogRead(PIN_PRESSURE_SENSOR);
    // Калибровка: 0..4095 дискрет АЦП соответствует 0..250 кПа
    return (float)rawValue * (250.0f / 4095.0f);
}

void setup() {
    Serial.begin(115200);
    analogReadResolution(12); // 12-битное разрешение АЦП

    WiFi.begin(WIFI_SSID, WIFI_PASS);
    Serial.print("Подключение к промышленной сети WiFi...");
    while (WiFi.status() != WL_CONNECTED) {
        delay(500);
        Serial.print(".");
    }
    Serial.println("\nWiFi подключен. IP: " + WiFi.localIP().toString());
}

void loop() {
    if (WiFi.status() == WL_CONNECTED) {
        HTTPClient http;
        http.begin(SERVER_URL);
        http.addHeader("Content-Type", "application/json");
        http.addHeader("X-Device-Token", DEVICE_TOKEN);

        StaticJsonDocument<256> doc;
        doc["airTemperature"] = round(readTemperatureCelsius() * 10.0f) / 10.0f;
        doc["gasPressureKPa"] = round(readGasPressureKPa() * 10.0f) / 10.0f;

        String requestBody;
        serializeJson(doc, requestBody);

        int httpResponseCode = http.POST(requestBody);
        if (httpResponseCode == 202) {
            String response = http.getString();
            Serial.println("Телеметрия принята сервером: " + response);
        } else {
            Serial.printf("Ошибка отправки HTTP: %d\n", httpResponseCode);
        }
        http.end();
    }

    // Такт опроса датчиков: 5000 мс
    delay(5000);
}

---

4. Пример интеграции 3: Edge-демон с локальным буфером (Store-and-Forward на SQLite)

Для обеспечения непрерывности технологического учета при авариях цеховой сети демон сохраняет неотправленные пакеты в локальную легковесную базу данных SQLite и автоматически выгружает их при восстановлении внешнего соединения:

#!/usr/bin/env python3
"""
HeatBalance Resilient Store-and-Forward Edge Daemon
"""

import sqlite3
import time
import json
import requests
import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("StoreAndForward")

DB_PATH = "/var/lib/heatbalance/edge_buffer.db"
CLOUD_URL = "https://heatbalance.online/api/v1/iot/telemetry"
DEVICE_TOKEN = "7f9a2b8c4d1e0f3a6b5c7d8e9f0a1b2c"

def init_db():
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS telemetry_queue (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
            payload_json TEXT NOT NULL,
            sent_flag INTEGER DEFAULT 0
        )
    """)
    conn.commit()
    conn.close()

def enqueue_reading(reading_dict):
    """Фиксация телеметрического замера в локальном отказоустойчивом буфере."""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute("INSERT INTO telemetry_queue (payload_json) VALUES (?)", (json.dumps(reading_dict),))
    conn.commit()
    conn.close()

def flush_queue():
    """Выгрузка накопленных данных на сервер HeatBalance."""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute("SELECT id, payload_json FROM telemetry_queue WHERE sent_flag = 0 ORDER BY id ASC LIMIT 50")
    rows = cursor.fetchall()
    
    if not rows:
        conn.close()
        return

    headers = {
        "X-Device-Token": DEVICE_TOKEN,
        "Content-Type": "application/json"
    }

    delivered_ids = []
    for row_id, payload_str in rows:
        try:
            payload = json.loads(payload_str)
            resp = requests.post(CLOUD_URL, json=payload, headers=headers, timeout=3.0)
            if resp.status_code == 202:
                delivered_ids.append(row_id)
            else:
                logger.warning(f"Сервер отклонил пакет {row_id}: {resp.status_code}")
                break
        except requests.exceptions.RequestException:
            logger.warning("Сеть недоступна. Выгрузка приостановлена, данные сохраняются локально.")
            break

    if delivered_ids:
        placeholders = ",".join("?" * len(delivered_ids))
        cursor.execute(f"DELETE FROM telemetry_queue WHERE id IN ({placeholders})", delivered_ids)
        conn.commit()
        logger.info(f"Успешно выгружено {len(delivered_ids)} исторических пакетов.")

    conn.close()

def main():
    init_db()
    logger.info("Служба Store-and-Forward запущена.")
    while True:
        # 1. Сбор реальных данных (здесь эмулируется технологический замер)
        sample = {
            "airTemperature": 23.1,
            "dryMaterialMass": 10500.0,
            "initialMoisture": 12.0
        }
        enqueue_reading(sample)

        # 2. Попытка выгрузки очереди
        flush_queue()
        time.sleep(5)

if __name__ == "__main__":
    main()

---

5. Пример интеграции 4: Опрос по протоколу OPC UA (IEC 62541)

В современных АСУ ТП на базе Siemens Simatic S7-1500 или распределенных систем управления (РСУ) стандартом де-факто является серверный протокол OPC UA. Пример асинхронного клиента на Python:

import asyncio
import requests
from asyncua import Client

OPC_SERVER_URL = "opc.tcp://192.168.10.100:4840"
CLOUD_URL = "https://heatbalance.online/api/v1/iot/telemetry"
DEVICE_TOKEN = "7f9a2b8c4d1e0f3a6b5c7d8e9f0a1b2c"

async def opc_collector():
    async with Client(url=OPC_SERVER_URL) as client:
        print("Подключено к OPC UA серверу металлургической печи.")
        
        # Получение узлов пространства переменных ПЛК
        node_temp = client.get_node("ns=2;s=DB_Furnace.Zone1.Temperature")
        node_o2 = client.get_node("ns=2;s=DB_Furnace.FlueGas.OxygenPercent")
        node_mass = client.get_node("ns=2;s=DB_Furnace.Production.MaterialMassPerHour")

        while True:
            temp_val = await node_temp.read_value()
            o2_val = await node_o2.read_value()
            mass_val = await node_mass.read_value()

            payload = {
                "airTemperature": round(temp_val, 1),
                "o2Percentage": round(o2_val, 2),
                "dryMaterialMass": round(mass_val, 0)
            }

            try:
                headers = {"X-Device-Token": DEVICE_TOKEN, "Content-Type": "application/json"}
                resp = requests.post(CLOUD_URL, json=payload, headers=headers, timeout=3.0)
                print(f"[OPC UA] Отправка {payload} -> Статус {resp.status_code}")
            except Exception as e:
                print(f"[OPC UA] Ошибка отправки: {e}")

            await asyncio.sleep(5.0)

if __name__ == "__main__":
    asyncio.run(opc_collector())

---

6. Отладка и нагрузочное тестирование с помощью cURL и PowerShell

cURL (Linux / macOS / Windows Bash):

curl -X POST "https://heatbalance.online/api/v1/iot/telemetry" \
  -H "X-Device-Token: 7f9a2b8c4d1e0f3a6b5c7d8e9f0a1b2c" \
  -H "Content-Type: application/json" \
  -d '{
    "dryMaterialMass": 12500.0,
    "initialMoisture": 14.8,
    "finalMoisture": 1.2,
    "airTemperature": 22.5,
    "recirculateTemp": 235.4
  }' \
  -w "\nHTTP Status: %{http_code}\nTime Total: %{time_total}s\n"

PowerShell (Windows Server):

$headers = @{
    "X-Device-Token" = "7f9a2b8c4d1e0f3a6b5c7d8e9f0a1b2c"
    "Content-Type"   = "application/json"
}

$body = @{
    dryMaterialMass = 12500.0
    initialMoisture = 14.8
    finalMoisture   = 1.2
    airTemperature  = 22.5
    recirculateTemp = 235.4
} | ConvertTo-Json

$response = Invoke-RestMethod -Uri "https://heatbalance.online/api/v1/iot/telemetry" `
                              -Method Post `
                              -Headers $headers `
                              -Body $body

Write-Host "Ответ платформы:" ($response | Format-List | Out-String)