Промышленные контроллеры, Linux и только C++. Часть 2

Как я и говорил в прошлой части, тут мы попробуем сделать самый простой обмен при помощи MQTT.

Что же из себя представляет MQTT? В первую очередь — это протокол обмена сообщениями. Данные сообщения могут группироваться по древовидному признаку. Мы можем как отправлять их, так и получать, подписываясь на определенные группы в дереве или отдельные сообщения.

Добавляем в репозитории контроллера системы управления пакетами сервера:

deb http://ppa.launchpad.net/mosquitto-dev/mosquitto-ppa/ubuntu precise main deb-src http://ppa.launchpad.net/mosquitto-dev/mosquitto-ppa/ubuntu precise main

Обновляем список пакетов и устанавливаем брокер MQTT Mosquitto на контроллере:

apt-get update apt-get install mosquitto

На системе для разработки устанавливаем файлы для сборки и исходники libmosquitto-dev:

sudo apt install libmosquitto-dev apt source libmosquitto-dev

Переходим в созданную папку с исходниками и собираем библиотеку под компилятор для arm-linux-gnueabihf:

make WITH_TLS=no WITH_CJSON=no WITH_STATIC_LIBRARIES=yes

В папке ./lib/cpp/вы найдете все собранные библиотеки чтобы подложить в проект. Нам потребуется файл libmosquitto.a.

Давайте разберем, что же мы хотим получить не выходе? Разложим все по пунктам:

• нам надо передавать адреса переменных;

• будем прикреплять их за топиками;

• можем подписываться на внешние сообщения, а можем и нет;

• периодически переменные должны обновляться на брокере.

Опишем структуру хранения самой переменной для контроля изменения переменной, времени изменения и точности, до которой надо контролировать числа с плавающей запятой:

template<typename T> struct Value {     Value(string topic,T old, T* current,int precision):         topic{topic},old{old},current{current},precision{precision} { }     ~Value() { }     string topic;     T old;     T* current;     uint64_t tFlag=timeSinceEpochMillisec();     int precision=2; };

Функция timeSinceEpochMillisec() была описана в предыдущей статье.

Мы будем разрабатывать класс, который необходимо наследовать от mosquittopp. Задаем Id для присоединения к брокеру на контроллере, хост, который у нас будет в нашем случае localhost, а также начальный топик device.

class MqttDevice:public mosquittopp::mosquittopp { public:     MqttDevice(string Id,string mqtt_host,string startTag="device");     ~MqttDevice();

Очередь сообщений  будем хранить списком:

    list< tuple<string,string> > messages;

Для добавления переменных в классе организуем необходимые функции. Мы передаем топик и переменную.

    void addSubscribed(string topic, volatile bool* value);     void addSubscribed(string topic, volatile float* value);     void addSubscribed(string topic, volatile uint16_t* value);     void add(string topic,volatile bool* value);     void add(string topic,volatile uint16_t* value);     void add(string topic,volatile float* value,int precision=1);     void add(string topic,volatile double* value,int precision=1);

Получим вот такой полный файл MqttDevice.h:

class MqttDevice:public mosquittopp::mosquittopp { public:     MqttDevice(string Id,string mqtt_host,string startTag="device");     ~MqttDevice();     void setTag(string topic,string value);     void message(string name, volatile double value);     void message(string name, volatile float value);     void message(string name, volatile bool value);     void message(string name, string value);     list< tuple<string,string> > messages;     void on_connect(int rc);     void on_message(const struct mosquitto_message* message);     void on_subscribe(int mid,int qos_count,const int* granted_qos);     void sendMessage(string topic, string& message);     void sendMessage(string& topic,char* message, int msgLength);     void addSubscribed(string topic, volatile bool* value);     void addSubscribed(string topic, volatile float* value);     void add(string topic,volatile bool* value);     void add(string topic,volatile float* value,int precision=1);     void add(string topic,volatile double* value,int precision=1); protected:     uint64_t tBool=0,tFloat=0,tDouble=0;     bool _exit; private:     string mqtt_host;     list<Value<volatile bool> > lwBool;     list<Value<volatile float> > lwFloat;     list<Value<volatile bool> > lrBool;     list<Value<volatile float> > lrFloat;     list<Value<volatile double> > lrDouble;     struct mosquitto *mosq;     string startTag;     void mosq_loop();     thread _threadMosquitto; }; 

В конструкторе и деструкторе мы организуем основные работы по созданию потока и его остановке.

MqttDevice::MqttDevice(string Id,string mqtt_host,string startTag) : startTag{startTag},mqtt_host{mqtt_host}, mosquittopp((Id+to_string(rand())).c_str()) {     _exit=false;     int rc=connect(mqtt_host.c_str(),mqtt_port,65535);     printf("MqttDevice connect %d\n",rc);     _threadMosquitto = thread(&MqttDevice::mosq_loop,this); } MqttDevice::~MqttDevice() {     _exit=true;     _threadMosquitto.join();     printf("MqttDevice delete\n"); }

Полный текст MqttDevice.cpp:

#define mqtt_port 1883  MqttDevice::MqttDevice(string Id,string mqtt_host,string startTag) : startTag{startTag},mqtt_host{mqtt_host}, mosquittopp((Id+to_string(rand())).c_str()) {     _exit=false;     int rc=connect(mqtt_host.c_str(),mqtt_port,65535);     printf("MqttDevice connect %d\n",rc);     _threadMosquitto = thread(&MqttDevice::mosq_loop,this); } MqttDevice::~MqttDevice() {     _exit=true;     _threadMosquitto.join();     printf("MqttDevice delete\n"); } void MqttDevice::addSubscribed(string topic, volatile bool* value) {     lwBool.push_back({topic,false,value,0});     if(subscribe(0,topic.c_str(),0)==MOSQ_ERR_SUCCESS) printf("[Mosquitto] Subscribing to topic %s\n",topic.c_str()); } void MqttDevice::addSubscribed(string topic, volatile float* value) {     lwFloat.push_back({topic,0.0f,value,1});     if(subscribe(0,topic.c_str(),0)==MOSQ_ERR_SUCCESS) printf("[Mosquitto] Subscribing to topic %s\n",topic.c_str()); } void MqttDevice::add(string topic,volatile bool* value) {     lrBool.push_back({topic,false,value,0}); } void MqttDevice::add(string topic,volatile float* value, int precision) {     lrFloat.push_back({topic,0.0f,value,precision}); } void MqttDevice::add(string topic,volatile double* value, int precision) {     lrDouble.push_back({topic,0.0,value,precision}); } void MqttDevice::on_connect(int rc) {     int err=0;         printf("[Mosquitto] Connect. rc=%d\n",rc);     if(rc==0) {         for(auto& v:lwBool) {             err=subscribe(0,v.topic.c_str(),0);             if(err==MOSQ_ERR_SUCCESS) printf("[Mosquitto] Subscribing to topic %s\n",v.topic.c_str());             else {reconnect();return;}             nsleep(10);         }         for(auto& v:lwFloat) {             err=subscribe(0,v.topic.c_str(),0);             if(err==MOSQ_ERR_SUCCESS) printf("[Mosquitto] Subscribing to topic %s\n",v.topic.c_str());             else {reconnect();return;}             nsleep(10);         }     }     else         printf("[Mosquitto] Connection failed. Aborting subscribing.\n"); } void MqttDevice::on_message(const struct mosquitto_message *message) {     string tag=message->topic;     string value=string((const char*)message->payload,message->payloadlen);     setTag(tag,value); } void MqttDevice::mosq_loop() {     int rc;     lib_init();     printf("[Mosquitto] Start loop...\n");     while(!_exit) {         uint64_t now = timeSinceEpochMillisec();           rc=loop();         if(rc) {             printf("[Mosquitto] Disconnected. Trying to reconnect...\n");             nsleep(500);             reconnect();             nsleep(500);             continue;         }         while(messages.size()>0) {             rc=publish(NULL,get<0>(messages.front()).c_str(),strlen(get<1>(messages.front()).c_str()),(const uint8_t*)get<1>(messages.front()).c_str(),0,true);             if(rc) {                 printf("[Mosquitto] Error send message\n");                 break;             }             else                 messages.pop_front();         }         if(now-tBool>300) {             tBool=now;             for(auto &v:lrBool) {                 if(v.current==NULL) continue;                 if(v.old!=*v.current || now-v.tFlag>3000) {                     v.tFlag=now;                     v.old=*v.current;                     message(v.topic,*v.current);                 }             }             for(auto &v:lwBool) {                 if(v.old!=*v.current || now-v.tFlag>3000) {                     v.tFlag=now;                     v.old=*v.current;                     message(v.topic,*v.current);                 }             }         }         if(now-tFloat>500) {             tFloat=now;             for(auto &v:lrFloat) {                 float r=round(*v.current*pow(10,v.precision))/pow(10,v.precision);                 if(v.old!=r || now-v.tFlag>5000) {                     v.tFlag=now;                     v.old=r;                     message(v.topic,r);                 }             }             for(auto &v:lrDouble) {                 double r=round(*v.current*pow(10,v.precision))/pow(10,v.precision);                 if(v.old!=r || now-v.tFlag>5000) {                     v.tFlag=now;                     v.old=r;                     message(v.topic,r);                 }             }             for(auto &v:lwFloat) {                 float r=round(*v.current*pow(10,v.precision))/pow(10,v.precision);                 if(v.old!=r || now-v.tFlag>5000) {                     v.tFlag=now;                     v.old=r;                     message(v.topic,r);                 }             }         }     }     lib_cleanup();     printf("[Mosquitto] Exit from mosquitto loop.\n"); } void MqttDevice::message(string name, volatile float value) {     messages.push_back(make_tuple(name,to_string(value))); } void MqttDevice::message(string name, volatile double value) {     messages.push_back(make_tuple(name,to_string(value))); } void MqttDevice::message(string name, volatile bool value) {     messages.push_back(make_tuple(name,(value?"true":"false"))); } void MqttDevice::message(string name, string value) {     messages.push_back(make_tuple(name,value)); } void MqttDevice::setTag(string topic,string value) {     try {         for(auto& v:lwBool)             if(v.topic==topic) {                 *(v.current)=(value=="true");                 v.old=*(v.current);                 return;             }         for(auto& v:lwFloat)             if(v.topic==topic) {                 *(v.current)=(float)atof(value.c_str());                 v.old=*(v.current);                 return;             }     }     catch(exception& ex) { } }

Для передачи более сложных объектов автоматизации, мы можем реализовать функции:

void MqttDevice::add(ValveD* valve) {     add(startTag+"/"+valve->name+"/SQL",valve->b_t1);     add(startTag+"/"+valve->name+"/SQH",valve->b_t2);     add(startTag+"/"+valve->name+"/out",valve->b_out);     addSubscribed(startTag+"/"+valve->name+"/auto",valve->b_mode);     addSubscribed(startTag+"/"+valve->name+"/man",valve->b_control); }

В данном примере передается указатель на объект отсечного клапана с двумя концевиками крайних положений, выходного сигнала, а также управления клапаном. Из управления есть сигнал в каком состоянии клапан auto. Если ручной режим auto=false, то управление с сигнала man. Если автоматический режим auto=true, то управление с внутреннего состояния, заданного программно по алгоритму.

Из прошлой статьи возьмем пример опроса модуля ввода/вывода и к нему добавим отправку полученных данных на брокер:

volatile bool b1=false,b2=false,b3=false; ModbusLine line1(modbus_new_rtu("/dev/ttyS2", 115200, 'N', 8, 1)); line1.addMDS_DIO_16BD(1,&b1,&b2,&b3);    MqttDevice mqtt("line","localhost"); mqtt.addSubscribed("device/b1",&b1); mqtt.add("device/b2",&b2); mqtt.add("device/b3",&b3);

При подключении к брокеру мы получили 3 переменных. При изменении переменной b1 через брокер в программе получаем изменение в значении переменной.

Мы добились нужного нам результата, а добавление новых переменных не занимает много кода.

В следующей статье мы опробуем обмен между контроллерами и устройствами через RS-485 по Modbus аналогичным образом. Это будет полезно при межконтроллерном обмене и общении с сенсорными панелями.