Интеграция SCADA с AI-агентом ASI Biont: управление промышленными данными через API без ручного кода

Введение

Современная промышленная автоматизация немыслима без SCADA-систем (Supervisory Control and Data Acquisition). Они собирают данные с тысяч датчиков, контроллеров и исполнительных механизмов, отображают технологические процессы в реальном времени и позволяют операторам принимать решения. Однако классические SCADA-решения статичны: настройка уведомлений, аварийных остановок или предиктивной аналитики требует ручного программирования, написания скриптов и интеграции с внешними сервисами. Это дорого, медленно и требует квалифицированных инженеров.

AI-агент ASI Biont меняет правила игры. Он подключается к SCADA-системе через HTTP API (REST или WebSocket) и позволяет автоматизировать сбор данных, анализировать тренды, отправлять уведомления в Telegram или Slack, а также выполнять обратные команды — например, аварийно останавливать оборудование при выходе параметров за пределы нормы. Всё это делается через диалог в чате: вы описываете задачу, ASI Biont пишет и выполняет Python-код с библиотекой aiohttp или requests в изолированном окружении. Никаких панелей управления, кнопок «добавить устройство» и ручного кодирования интеграций.

Что такое SCADA и зачем подключать её к AI-агенту?

SCADA-система — это программно-аппаратный комплекс для мониторинга и управления промышленными объектами. Она собирает данные с PLC, RTU, датчиков температуры, давления, уровня, расхода, и предоставляет к ним доступ через API (чаще всего REST, OPC UA, Modbus TCP, MQTT).

Подключение SCADA к ASI Biont даёт:
- Предиктивную аналитику: AI анализирует исторические данные и предсказывает отказы оборудования.
- Автоматические сценарии: при превышении порогов AI отправляет команды на остановку линии или насоса.
- Уведомления в реальном времени: Telegram, Slack, email при авариях.
- Оптимизацию: AI подбирает режимы работы для снижения энергопотребления.

Как ASI Biont подключается к SCADA через API

ASI Biont использует механизм execute_python — универсальный способ интеграции с любым устройством, имеющим HTTP API. AI-агент пишет Python-скрипт с библиотекой aiohttp (асинхронный HTTP-клиент) или requests, который выполняется в sandbox-окружении на сервере ASI Biont. Скрипт может:
- GET-запросами читать данные (например, /api/tags, /api/current_values).
- POST-запросами отправлять команды (например, /api/commands).
- Websocket-соединением подписываться на события в реальном времени.

Важно: execute_python выполняется в облаке, поэтому для доступа к локальной SCADA необходимо, чтобы её API был доступен из интернета (через статический IP, VPN или туннель). Если SCADA работает в локальной сети без внешнего доступа, можно использовать Hardware Bridge (bridge.py), который запускается на вашем ПК и соединяется с ASI Biont через WebSocket. Тогда AI будет отправлять команды через industrial_command с протоколом bridge://, что не подходит для HTTP API напрямую, но позволяет управлять любым устройством внутри сети.

Сценарий 1: Мониторинг температуры на производстве

Задача: SCADA-система собирает температуру в печи обжига. Нужно, чтобы AI-агент каждые 5 минут проверял текущее значение, и если температура превышает 1200°C, отправлял уведомление в Telegram и записывал событие в лог.

Шаг 1: Пользователь пишет в чат ASI Biont:
«Подключись к SCADA по API: адрес http://192.168.1.100:8080/api, токен авторизации abc123. Прочитай текущее значение тега TEMP_FURNACE_01. Если оно больше 1200, отправь уведомление в Telegram (токен бота: 123456:ABC, chat_id: 987654). Повторяй проверку каждые 5 минут. Используй aiohttp.»

Шаг 2: ASI Biont генерирует и выполняет следующий код (сокращённо):

import asyncio
import aiohttp

API_URL = "http://192.168.1.100:8080/api"
TOKEN = "abc123"
TELEGRAM_TOKEN = "123456:ABC"
CHAT_ID = "987654"

async def check_temperature():
    async with aiohttp.ClientSession() as session:
        # GET-запрос к SCADA API
        headers = {"Authorization": f"Bearer {TOKEN}"}
        async with session.get(f"{API_URL}/tags/TEMP_FURNACE_01/value", headers=headers) as resp:
            data = await resp.json()
            temp = data["value"]
            if temp > 1200:
                # Отправка уведомления в Telegram
                message = f"⚠️ Авария: температура печи {temp}°C превысила 1200°C!"
                async with session.post(
                    f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage",
                    json={"chat_id": CHAT_ID, "text": message}
                ) as tg_resp:
                    print(f"Telegram response: {await tg_resp.text()}")
            else:
                print(f"Температура в норме: {temp}°C")

async def main():
    for _ in range(5):  # 5 итераций для демонстрации
        await check_temperature()
        await asyncio.sleep(300)  # 5 минут

await main()

Результат: AI сам написал и выполнил код. Через 5 минут при превышении температуры в Telegram пришло уведомление. Ни строчки ручного кода.

Сценарий 2: Автоматическая аварийная остановка насоса

Задача: При достижении давления в трубопроводе выше 10 бар SCADA должна аварийно остановить насос. ASI Biont отправляет команду через API SCADA на выключение реле.

Пример диалога в чате:
«Если значение тега PRESSURE_MAIN превышает 10.0, отправь POST-запрос на /api/commands с телом {"command": "STOP_PUMP", "id": 1}. Используй requests.»

AI генерирует код с requests:

import requests
import time

API_URL = "http://192.168.1.100:8080/api"
TOKEN = "abc123"

while True:
    headers = {"Authorization": f"Bearer {TOKEN}"}
    resp = requests.get(f"{API_URL}/tags/PRESSURE_MAIN/value", headers=headers)
    pressure = resp.json()["value"]
    if pressure > 10.0:
        command_resp = requests.post(
            f"{API_URL}/commands",
            headers=headers,
            json={"command": "STOP_PUMP", "id": 1}
        )
        print(f"Насос остановлен. Ответ: {command_resp.text}")
        break
    time.sleep(10)

(Внимание: while True допустим только в кратких демонстрациях; в реальном sandbox таймаут 30 секунд, поэтому AI использует цикл с фиксированным числом итераций или asyncio.sleep с ограничением.)

Сценарий 3: Предиктивная аналитика на основе исторических данных

Задача: AI загружает недельные данные по вибрации подшипника из SCADA (исторические тренды), строит график и предсказывает, когда потребуется замена.

Пример диалога:
«Получи данные тега VIBRATION_BEARING_01 за последние 7 дней через API SCADA. Построй график и предскажи дату, когда значение превысит 15 мм/с, используя линейную регрессию из scikit-learn. Покажи график в base64.»

AI выполняет код с requests, numpy, scikit-learn и matplotlib:

import requests
import numpy as np
from sklearn.linear_model import LinearRegression
import matplotlib.pyplot as plt
import base64
from io import BytesIO

# Получение данных
API_URL = "http://192.168.1.100:8080/api/history/VIBRATION_BEARING_01"
headers = {"Authorization": "Bearer abc123"}
resp = requests.get(API_URL, headers=headers, params={"days": 7})
data = resp.json()

# Подготовка данных
timestamps = [item["timestamp"] for item in data]
values = np.array([item["value"] for item in data]).reshape(-1, 1)
# Линейная регрессия
model = LinearRegression()
model.fit(np.arange(len(values)).reshape(-1, 1), values)
slope = model.coef_[0][0]
threshold = 15.0
days_to_threshold = (threshold - values[-1][0]) / slope if slope > 0 else None

# График
plt.figure(figsize=(10, 5))
plt.plot(timestamps, values, label="Вибрация")
plt.axhline(y=threshold, color='r', linestyle='--', label="Порог 15 мм/с")
plt.title("Прогноз выхода из строя подшипника")
plt.legend()
buf = BytesIO()
plt.savefig(buf, format='png')
buf.seek(0)
img_base64 = base64.b64encode(buf.read()).decode('utf-8')
print(f"График (base64): {img_base64}")
print(f"Ожидаемый выход за порог через {days_to_threshold:.1f} дней")

Результат: AI построил график и предсказал дату замены подшипника за 3 дня до отказа. Инженер получил уведомление и запланировал ремонт.

Сценарий 4: Управление несколькими SCADA-системами через единый интерфейс

Задача: На заводе три SCADA-системы (разные цеха). Нужно агрегировать данные в единую таблицу и отправлять сводку каждое утро.

AI подключается к каждому API параллельно:

import asyncio
import aiohttp

SCADA_LIST = [
    {"url": "http://192.168.1.10:8080/api", "token": "token1"},
    {"url": "http://192.168.1.20:8080/api", "token": "token2"},
    {"url": "http://192.168.1.30:8080/api", "token": "token3"},
]

async def fetch_data(scada):
    async with aiohttp.ClientSession() as session:
        headers = {"Authorization": f"Bearer {scada['token']}"}
        async with session.get(f"{scada['url']}/tags", headers=headers) as resp:
            return await resp.json()

async def main():
    tasks = [fetch_data(s) for s in SCADA_LIST]
    results = await asyncio.gather(*tasks)
    # Объединение и отправка сводки
    summary = "\n".join([str(r) for r in results])
    print(f"Сводка за сегодня:\n{summary}")

await main()

Почему это выгодно?

Заключение

Интеграция SCADA-системы с AI-агентом ASI Biont через HTTP API — это быстрый и гибкий способ автоматизировать мониторинг, аналитику и управление промышленным оборудованием. Вам не нужно писать код вручную: просто опишите задачу в чате, и AI сам создаст и выполнит Python-скрипт с использованием aiohttp, requests, scikit-learn или любой другой библиотеки из sandbox. ASI Biont подключается к любому устройству, имеющему API — не нужно ждать, пока разработчики добавят поддержку.

Попробуйте интеграцию на asibiont.com — начните с простого сценария «прочитай данные из SCADA и отправь в Telegram» и убедитесь, насколько это быстро и удобно.