В данном руководстве рассмотрено создание чат-бот помощника в Viber на языке программирования Python. Чат-бот имеет доступ к базе данных MySQL, которая в свою очередь связана с АСУ ТП (автоматизированной системой управления технологическим процессом), разработанной на базе логического контроллера Siemens серии S7-1500.
Чат-бот автоматизирует передачу информации о течение технологического процесса конечному пользователю, позволяет получать доступ к собранным данным с любого устройства (на котором, конечно, установлен Viber) и в любом месте.
Руководство рассчитано на инженеров-программистов, работающих в сфере автоматизации (АСУ ТП или АСУП).
Стек технологий
Для реализации данной идеи необходимо знать:
Dim TableName, Uname, DBPassword, DataBaseName, SQLServer
Dim SQLConnection
Dim SQLConnectionString
Dim SQLFields,SQLValues
Dim SQLQueryString
Dim SQLCommand
'Конфигурация для подключения к базе данных, все данные - строго в соотвествии с ODBC connector
DataBaseName = "siemens"
Uname = "ilya"
DBPassword = "qwerty123"
SQLServer = "127.0.0.1"
TableName = "data"
'Конфигурационная строка
SQLConnectionString = "Provider=MSDASQL;Initial Catalog=siemens; DSN=siemens"
'Создание объектов (команда set создаёт объект)
Set SQLConnection = CreateObject("ADODB.Connection")
SQLConnection.ConnectionString = SQLConnectionString
'Открытие соединения
SQLConnection.Open
Set SQLCommand = CreateObject("ADODB.Command")
SQLCommand.ActiveConnection = SQLConnection
'Формирование запроса
SQLFields = " (Value)"
SQLValues = " Values('" & SmartTags("test_test") & "') "
SQLQueryString = "INSERT INTO " & TableName & SQLFields & SQLValues
'Выполнение SQL запроса
SQLCommand.CommandText = SQLQueryString
SQLCommand.Execute
'Закрытие соединения с базой данных
Set SQLCommand = Nothing
SQLConnection.Close
Set SQLConnection = Nothing
Данный код открывает соединение с базой данных, созданной выше. Записывает переменную "test_test" в таблицу "data". Также стоит обратить внимание, что обращение к тэгам HMI происходит через нотацию SmartTags():
VBS скрипт создаётся в дереве HMI в разделе Scripts:
Чтобы скрипт периодически исполнялся, необходимо создать для него триггер. В качестве триггера выбираем изменение переменной test_test. Во вкладке HMI находим тэг test_test и во вкладке Properties -> Events по параметру Value change добавляем вызов VBS скрипта:
Для тестирования работоспособности кода необходимо запустить PLC-SIM, выгрузить в него проект с конфигурированным контроллером, перейти в онлайн режим, активировать симуляцию HMI. Затем изменить переменную test_test. Чтобы убедиться, что запись в базу данных работает можно отправить к ней SQL запрос вручную с помощью командной строки:
use siemens
SELECT * FROM data;
viberbot==1.0.11
flask
mysql-connector-python
Ядро чат-бота - приложение Flask, работающее с API Viber. Для локального тестирования приложения необходимо использовать ngrok. После его запуска вводим команду:
ngrok http 8443
В окне ngrok копируем ссылку https и сохраняем её, она нам скоро пригодится. Ngrok создаёт туннель, связывающий наш порт, в данном случае 8443, с https ссылкой, доступной по всему интернету. Этот https будет служить вебхуком для нашего приложения. Также пригодится токен viber чат-бота, который был получен в результате регистрации на Viber Admin Panel.
А вот и сам код чат-бота на Python 3:
#Импорт всех необходимых библиотек
from flask import Flask, request, Response, session
from viberbot import Api
from viberbot.api.bot_configuration import BotConfiguration
from viberbot.api.messages import (
TextMessage,
KeyboardMessage
)
from viberbot.api.viber_requests import ViberMessageRequest, ViberConversationStartedRequest
import sched
import threading
import time
import mysql.connector
token = 'ВАШ ТОКЕН'
class UseDataBase:
'''Менеджер контекста для подключения к базе данных'''
def __init__(self, config: dict) -> None:
self.configuration = config
def __enter__(self):
self.conn = mysql.connector.connect(**self.configuration)
self.cursor = self.conn.cursor()
return self.cursor
def __exit__(self, exc_type, exc_val, exc_tb):
self.conn.commit()
self.cursor.close()
self.conn.close()
class ViberSay():
'''Чат-бот'''
def __init__(self, token):
#Инициализация чат-бота данными, полученными при регистрации на Admin Panel
self.viber = Api(BotConfiguration(
name='TestChatBotByIlya',
avatar='http://viber.com/avatar.jpg',
auth_token=token))
self.app = Flask(__name__) #Создание экземпляра приложения Flask
self.scheduler = sched.scheduler(time.time, time.sleep)
self.scheduler.enter(5, 1, self.set_webhook, ())
self.t = threading.Thread(target=self.scheduler.run)
self.t.start() #запуск потока
self.app.secret_key = 'YouWillNeverGuess'
#конфигурация базы данных
self.app.config['dbconfig'] = {'host': '127.0.0.1',
'user': 'ilya',
'password': 'qwerty123',
'database': 'siemens', }
self.index = 0
self.app.add_url_rule('/', 'home', self.poll, methods=['POST', 'GET'])
self.app.run(host='0.0.0.0', port=8443, debug=True) #пуск приложения
def set_webhook(self):
#установка вебхука
self.viber.set_webhook('https://06841b02dbf3.ngrok.io')
def poll(self):
# приём входящих сообщений, обработка и декодирование
viber_request = self.viber.parse_request(request.get_data().decode('utf8'))
if isinstance(viber_request, ViberConversationStartedRequest):
#Определяем начало диалога с ботом
session['uid'] = viber_request.user.id
self.viber.send_messages(session['uid'], [
TextMessage(text='Добро пожаловать!')
])
#отправляем кнопку с предложением получить данные
self.send_buttons(uid=session['uid'], text=['Получить данные'], btns=1)
elif isinstance(viber_request, ViberMessageRequest):
#если не начало диалога, то ждём запроса данных
session['uid'] = viber_request.sender.id
if viber_request.message.text == 'Получить данные':
with UseDataBase(self.app.config['dbconfig']) as cursor:
#ыполнение запроса к базу данных
_SQL = """select Id, Value from data"""
cursor.execute(_SQL)
contents = cursor.fetchall()
#Выбираем последнее значение из массива чисел и отправляем пользователю
if self.index != contents[-1][0]:
self.index = contents[-1][0]
text_mes = 'Значение = {}'.format(contents[-1][1])
self.viber.send_messages(to=session['uid'], messages=[TextMessage(text=text_mes)])
#снова отправляем кнопку с предложением получить данные
self.send_buttons(uid=session['uid'], text=['Получить данные'], btns=1)
return Response(status=200)
def send_buttons(self, uid, text, btns):
#метод формирует кнопку
KEYBOARD = {"Type": "keyboard", "Buttons": []}
for i in range(btns):
KEYBOARD["Buttons"].append(
{
"Columns": 6,
"Rows": 1,
"BgColor": '#4169E1',
"BgMedia": None,
"BgMediaType": None,
"BgLoop": True,
"ActionType": "reply",
"ActionBody": text,
"ReplyType": "message",
"Text": text
}
)
message = KeyboardMessage(tracking_data='tracking_data', keyboard=KEYBOARD)
self.viber.send_messages(session['uid'], [message])
if __name__ == "__main__":
ViberSay(token)
Для лучшего понимая кода можно прочитать документацию по Flask и Viber API. Для того, чтобы веб-приложение заработало необходимо:
Источник статьи: https://habr.com/ru/post/569548/
Чат-бот автоматизирует передачу информации о течение технологического процесса конечному пользователю, позволяет получать доступ к собранным данным с любого устройства (на котором, конечно, установлен Viber) и в любом месте.
Руководство рассчитано на инженеров-программистов, работающих в сфере автоматизации (АСУ ТП или АСУП).
Для реализации данной идеи необходимо знать:
- Основы программирования контроллеров Siemens серии S7-1200/1500
- Базовые принципы работы баз данных и языка запросов SQL
- Язык программирования Python
- Установка и конфигурация всех необходимых приложений
- Написание программ
Установка и конфигурация приложений
- Установить TIA Portal v16 из официального источника можно по ссылке.
В разделе TRIAL Download STEP 7 Basic/Professional, STEP 7 Safety Basic/Advanced and WinCC Basic/Comfort/Advanced and WinCC Unified скачиваем DVD 1 Setup и при установке выбираем Step 7 Professional и WinCC Advanced. А также в разделе TRIAL Download STEP 7 PLCSIM скачиваем и устанавливаем DVD 1 Setup - это симулятор, который позволяет тестировать ПО без физического контроллера.
После установки в новом проекте TIA Portal необходимо сконфигурировать ПЛК серии S7-1200/1500 (можно выбрать любой, главное не резервируемый) и панель оператора. Конфигурация, используемая мной на рисунке ниже.
Конфигурация "железа"
Далее создаётся глобальный дата блок, например, "test" с переменной "test". Переменную добавляем в HMI Tags, так как связь с базой данных MySQL будет реализована на стороне HMI клиента, в данном примере это панель оператора TP1200 Comfort. Программное обеспечение HMI разрабатывается в WinCC Advanced, интегрированной в TIA Portal (да-да, немного запутано). Средства WinCC позволяют писать код на VBScrpt, который и будет реализовывать обмен данными между ПЛК и БД MySQL.
Создание переменных
- Установка сервера MySQL подробно описана на других ресурсах (например, здесь). Однако стоит отметить, что необходимо помимо MySQL Server 8.0.XX также установить драйвер для Python и ODBC.
Установка MySQL
Для подключения БД к TIA Portal необходимо дополнительно скачать и установить 32-ух разрядный ODBC драйвер по ссылке.
После установки открываем командную строку (cmd) и запускаем MySQL:
mysql -u root -p
После ввода пароля и успешного входа необходимо создать и настроить базу данных с таблицей:
CREATE DATABASE Siemens;
USE Siemens;
CREATE TABLE Data
(
Id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
Value INT
);
Для настройки драйвера ODBC запускаем приложение "Источник данных ODBC" -32-ух разрядную версию и по кнопке "Добавить" необходимо выбрать MySQL ODBS 8.0 Unicode Driver.
Конфигурация ODBC коннектора
В настройках вводим данные пользователя, сконфигурированные при установке MySQL и выбираем созданную базу данных siemens. В качестве TCP/IP Server необходимо выбрать localhost. Кнопкой Test можно проверить корректность соединения. DSN ( Data Sourse Name) вводим также, как на скриншоте.
Настройка DSN пользователя
- Установка Python версии 3 осуществляется с официального сайта.
- Пожалуй, финальный подготовительный шаг - создание чат-бота и получение токена, который мы будем использовать в нашем python-приложении.
Создание чат-бота производится через панель администрирования по ссылке. регистрация и получение токена бесплатны.
ПО обмена данными между БД и Siemens
'Объявление переменныхDim TableName, Uname, DBPassword, DataBaseName, SQLServer
Dim SQLConnection
Dim SQLConnectionString
Dim SQLFields,SQLValues
Dim SQLQueryString
Dim SQLCommand
'Конфигурация для подключения к базе данных, все данные - строго в соотвествии с ODBC connector
DataBaseName = "siemens"
Uname = "ilya"
DBPassword = "qwerty123"
SQLServer = "127.0.0.1"
TableName = "data"
'Конфигурационная строка
SQLConnectionString = "Provider=MSDASQL;Initial Catalog=siemens; DSN=siemens"
'Создание объектов (команда set создаёт объект)
Set SQLConnection = CreateObject("ADODB.Connection")
SQLConnection.ConnectionString = SQLConnectionString
'Открытие соединения
SQLConnection.Open
Set SQLCommand = CreateObject("ADODB.Command")
SQLCommand.ActiveConnection = SQLConnection
'Формирование запроса
SQLFields = " (Value)"
SQLValues = " Values('" & SmartTags("test_test") & "') "
SQLQueryString = "INSERT INTO " & TableName & SQLFields & SQLValues
'Выполнение SQL запроса
SQLCommand.CommandText = SQLQueryString
SQLCommand.Execute
'Закрытие соединения с базой данных
Set SQLCommand = Nothing
SQLConnection.Close
Set SQLConnection = Nothing
Данный код открывает соединение с базой данных, созданной выше. Записывает переменную "test_test" в таблицу "data". Также стоит обратить внимание, что обращение к тэгам HMI происходит через нотацию SmartTags():
VBS скрипт создаётся в дереве HMI в разделе Scripts:
Чтобы скрипт периодически исполнялся, необходимо создать для него триггер. В качестве триггера выбираем изменение переменной test_test. Во вкладке HMI находим тэг test_test и во вкладке Properties -> Events по параметру Value change добавляем вызов VBS скрипта:
Для тестирования работоспособности кода необходимо запустить PLC-SIM, выгрузить в него проект с конфигурированным контроллером, перейти в онлайн режим, активировать симуляцию HMI. Затем изменить переменную test_test. Чтобы убедиться, что запись в базу данных работает можно отправить к ней SQL запрос вручную с помощью командной строки:
use siemens
SELECT * FROM data;
ПО Viber чат-бота
Для обеспечения работоспособности кода необходимо установить следующие модули:viberbot==1.0.11
flask
mysql-connector-python
Ядро чат-бота - приложение Flask, работающее с API Viber. Для локального тестирования приложения необходимо использовать ngrok. После его запуска вводим команду:
ngrok http 8443
В окне ngrok копируем ссылку https и сохраняем её, она нам скоро пригодится. Ngrok создаёт туннель, связывающий наш порт, в данном случае 8443, с https ссылкой, доступной по всему интернету. Этот https будет служить вебхуком для нашего приложения. Также пригодится токен viber чат-бота, который был получен в результате регистрации на Viber Admin Panel.
А вот и сам код чат-бота на Python 3:
#Импорт всех необходимых библиотек
from flask import Flask, request, Response, session
from viberbot import Api
from viberbot.api.bot_configuration import BotConfiguration
from viberbot.api.messages import (
TextMessage,
KeyboardMessage
)
from viberbot.api.viber_requests import ViberMessageRequest, ViberConversationStartedRequest
import sched
import threading
import time
import mysql.connector
token = 'ВАШ ТОКЕН'
class UseDataBase:
'''Менеджер контекста для подключения к базе данных'''
def __init__(self, config: dict) -> None:
self.configuration = config
def __enter__(self):
self.conn = mysql.connector.connect(**self.configuration)
self.cursor = self.conn.cursor()
return self.cursor
def __exit__(self, exc_type, exc_val, exc_tb):
self.conn.commit()
self.cursor.close()
self.conn.close()
class ViberSay():
'''Чат-бот'''
def __init__(self, token):
#Инициализация чат-бота данными, полученными при регистрации на Admin Panel
self.viber = Api(BotConfiguration(
name='TestChatBotByIlya',
avatar='http://viber.com/avatar.jpg',
auth_token=token))
self.app = Flask(__name__) #Создание экземпляра приложения Flask
self.scheduler = sched.scheduler(time.time, time.sleep)
self.scheduler.enter(5, 1, self.set_webhook, ())
self.t = threading.Thread(target=self.scheduler.run)
self.t.start() #запуск потока
self.app.secret_key = 'YouWillNeverGuess'
#конфигурация базы данных
self.app.config['dbconfig'] = {'host': '127.0.0.1',
'user': 'ilya',
'password': 'qwerty123',
'database': 'siemens', }
self.index = 0
self.app.add_url_rule('/', 'home', self.poll, methods=['POST', 'GET'])
self.app.run(host='0.0.0.0', port=8443, debug=True) #пуск приложения
def set_webhook(self):
#установка вебхука
self.viber.set_webhook('https://06841b02dbf3.ngrok.io')
def poll(self):
# приём входящих сообщений, обработка и декодирование
viber_request = self.viber.parse_request(request.get_data().decode('utf8'))
if isinstance(viber_request, ViberConversationStartedRequest):
#Определяем начало диалога с ботом
session['uid'] = viber_request.user.id
self.viber.send_messages(session['uid'], [
TextMessage(text='Добро пожаловать!')
])
#отправляем кнопку с предложением получить данные
self.send_buttons(uid=session['uid'], text=['Получить данные'], btns=1)
elif isinstance(viber_request, ViberMessageRequest):
#если не начало диалога, то ждём запроса данных
session['uid'] = viber_request.sender.id
if viber_request.message.text == 'Получить данные':
with UseDataBase(self.app.config['dbconfig']) as cursor:
#ыполнение запроса к базу данных
_SQL = """select Id, Value from data"""
cursor.execute(_SQL)
contents = cursor.fetchall()
#Выбираем последнее значение из массива чисел и отправляем пользователю
if self.index != contents[-1][0]:
self.index = contents[-1][0]
text_mes = 'Значение = {}'.format(contents[-1][1])
self.viber.send_messages(to=session['uid'], messages=[TextMessage(text=text_mes)])
#снова отправляем кнопку с предложением получить данные
self.send_buttons(uid=session['uid'], text=['Получить данные'], btns=1)
return Response(status=200)
def send_buttons(self, uid, text, btns):
#метод формирует кнопку
KEYBOARD = {"Type": "keyboard", "Buttons": []}
for i in range(btns):
KEYBOARD["Buttons"].append(
{
"Columns": 6,
"Rows": 1,
"BgColor": '#4169E1',
"BgMedia": None,
"BgMediaType": None,
"BgLoop": True,
"ActionType": "reply",
"ActionBody": text,
"ReplyType": "message",
"Text": text
}
)
message = KeyboardMessage(tracking_data='tracking_data', keyboard=KEYBOARD)
self.viber.send_messages(session['uid'], [message])
if __name__ == "__main__":
ViberSay(token)
Для лучшего понимая кода можно прочитать документацию по Flask и Viber API. Для того, чтобы веб-приложение заработало необходимо:
- Cкопировать и вставить в код ваш токен от чат-бота
- Правильно проинициализировать конфигурацию чат-бота
- Указать верные настройки для подключения к базе данных
- Установить вебхук (метод set_webhook) - та самая https ссылка из ngrok
Заключение
Таким образом, данный проект реализует удалённый доступ к данным, которые контролируются АСУ ТП на базе ПЛК Siemens, через привычные интерфейс повседневного мессенджера. Доступность данных позволяет качественнее управлять технологическим процессом и анализировать его на более высоком уровне.Источник статьи: https://habr.com/ru/post/569548/