OPCUAExchange.h

/*
 * Copyright (c) 2023 Pavel Vainerman.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as
 * published by the Free Software Foundation, version 2.1.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 * Lesser General Lesser Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */
// -----------------------------------------------------------------------------
#ifndef OPCUAExchange_H_
#define OPCUAExchange_H_
// -----------------------------------------------------------------------------
#include <vector>
#include <memory>
#include <deque>
#include <string>
#include <regex>
#include <optional>
#include "UniXML.h"
#include "ThreadCreator.h"
#include "PassiveTimer.h"
#include "Trigger.h"
#include "IONotifyController.h"
#include "UniSetObject.h"
#include "Mutex.h"
#include "MessageType.h"
#include "SMInterface.h"
#include "IOBase.h"
#include "SharedMemory.h"
#include "LogServer.h"
#include "DebugStream.h"
#include "LogAgregator.h"
#include "OPCUAClient.h"
#include "USingleProcess.h"
// -------------------------------------------------------------------------
#ifndef vmonit
#define vmonit( var ) vmon.add( #var, var )
#endif
// -------------------------------------------------------------------------
namespace uniset
{
    // ---------------------------------------------------------------------
    // ---------------------------------------------------------------------
    class OPCUAExchange:
        private USingleProcess,
        public UniSetObject
    {
        public:
            OPCUAExchange( uniset::ObjectId id, xmlNode* cnode,
                           uniset::ObjectId icID, const std::shared_ptr<SharedMemory>& shm = nullptr,
                           const std::string& _prefix = "opcua" );
            virtual ~OPCUAExchange();
 
            static std::shared_ptr<OPCUAExchange> init_opcuaexchange(int argc, const char* const* argv,
                    uniset::ObjectId icID, const std::shared_ptr<SharedMemory>& ic = nullptr,
                    const std::string& prefix = "opcua");
 
            static void help_print( int argc, const char* const* argv );
 
            virtual uniset::SimpleInfo* getInfo( const char* userparam = 0 ) override;
 
            static uint8_t firstBit( uint32_t mask );
 
            // offset = firstBit(mask)
            static uint32_t getBits( uint32_t value, uint32_t mask, uint8_t offset );
            // if mask = 0 return value
            static uint32_t setBits( uint32_t value, uint32_t set, uint32_t mask, uint8_t offset );
            // if mask=0 return set
            static uint32_t forceSetBits( uint32_t value, uint32_t set, uint32_t mask, uint8_t offset );
 
            using Tick = uint8_t;
 
            static const size_t numChannels = 2;
            struct ReadGroup
            {
                std::vector<std::vector<OPCUAClient::ResultVar>> results;
                std::vector<std::vector<UA_ReadValueId>> ids;
            };
            struct WriteGroup
            {
                std::vector<std::vector<UA_WriteValue>> ids;
            };

            struct OPCAttribute:
                public IOBase
            {
                // т.к. IOBase содержит rwmutex с запрещённым конструктором копирования
                // приходится здесь тоже объявлять разрешенными только операции "перемещения"
                OPCAttribute(const OPCAttribute& r ) = delete;
                OPCAttribute& operator=(const OPCAttribute& r) = delete;
                OPCAttribute(OPCAttribute&& r ) = default;
                OPCAttribute& operator=(OPCAttribute&& r) = default;
                OPCAttribute() = default;
 
                uniset::uniset_rwmutex vmut;
                int32_t val { 0 };
                Tick tick = { 0 }; // на каждом ли тике работать с этим аттрибутом
                uint32_t mask = { 0 };
                uint8_t offset = { 0 };
                OPCUAClient::VarType vtype = { OPCUAClient::VarType::Int32 };
 
                // with precision
                float as_float();
 
                // with precision/noprecision
                int32_t set( float val );
 
                std::string attrName = {""};
                struct RdValue
                {
                    std::shared_ptr<ReadGroup> gr;
                    size_t grIndex = {0};  // индекс в запросе (номер в запросе)
                    size_t grNumber = {0}; // Номер группы запроса в общем списке
                    int32_t get();
                    float getF();
                    bool statusOk();
                    UA_StatusCode status();
                    const UA_ReadValueId& ref();
                    // Subscription
                    uint32_t subscriptionId = {0U};
                    uint32_t monitoredItemId = {0U};
                    bool subscriptionState = {false};
                };
                RdValue rval[numChannels];
 
                struct WrValue
                {
                    std::shared_ptr<WriteGroup> gr;
                    size_t grIndex = {0};  // индекс в запросе (номер в запросе)
                    size_t grNumber = {0}; // Номер группы запроса в общем списке
                    bool set( int32_t val );
                    bool setF( float val );
                    bool statusOk();
                    UA_StatusCode status();
                    const UA_WriteValue& ref();
                    static void init( UA_WriteValue* wval, const std::string& nodeId, const std::string& type, int32_t defvalue );
                };
                WrValue wval[numChannels];
 
                friend std::ostream& operator<<(std::ostream& os, const OPCAttribute& inf );
                friend std::ostream& operator<<(std::ostream& os, const std::shared_ptr<OPCAttribute>& inf );
            };

            enum ExchangeMode
            {
                emNone = 0,         
                emWriteOnly = 1,    
                emReadOnly = 2,     
                emSkipSaveToSM = 3, 
                emSkipExchange = 4, 
                emLastNumber
            };

            enum StopMode
            {
                smNone = 0,         
                smFirstOnly = 1,    
                smAny = 2,          
                smLastNumber
            };
 
            typedef std::list<IOBase> ThresholdList;
            // т.к. пороговые датчики не связаны напрямую с обменом, создаём для них отдельный список
            // и отдельно его проверяем потом
            ThresholdList thrlist;
 
#ifndef DISABLE_REST_API
            // HTTP API
            virtual Poco::JSON::Object::Ptr httpHelp( const Poco::URI::QueryParameters& p ) override;
            virtual Poco::JSON::Object::Ptr httpRequest( const std::string& req, const Poco::URI::QueryParameters& p ) override;
#endif
 
        protected:
 
            enum Timers
            {
                tmUpdates
            };
 
            struct Channel;
            void channel1Thread();
            void channel2Thread();
            void channelThread( Channel* ch );
            bool prepare();
            void channelExchange( Tick tick, Channel* ch, bool writeOn );
            void updateFromChannel( Channel* ch );
            void updateToChannel( Channel* ch );
            void updateFromSM();
            void writeToSM();
            bool isUpdateSM( bool wrFunc ) const noexcept;
 
            virtual void sysCommand( const uniset::SystemMessage* sm ) override;
            virtual void askSensors( UniversalIO::UIOCommand cmd );
            virtual void sensorInfo( const uniset::SensorMessage* sm ) override;
            virtual void timerInfo( const uniset::TimerMessage* sm ) override;
            virtual bool activateObject() override;
            virtual bool deactivateObject() override;
 
#ifndef DISABLE_REST_API
            // Публичные HTTP-обработчики параметров
            Poco::JSON::Object::Ptr httpGetParam( const Poco::URI::QueryParameters& p );
            Poco::JSON::Object::Ptr httpSetParam( const Poco::URI::QueryParameters& p );
            Poco::JSON::Object::Ptr httpStatus();
 
            // Защитный флаг: запретить /setparam при false
            bool httpEnabledSetParams { true };
#endif
 
 
            // чтение файла конфигурации
            void readConfiguration();
            bool initIOItem( UniXML::iterator& it );
            bool readItem( const std::shared_ptr<UniXML>& xml, UniXML::iterator& it, xmlNode* sec );
            bool waitSM();
            bool tryConnect(Channel* ch);
            void initOutputs();
            void createSubscription(int nchannel);
 
            xmlNode* confnode = { nullptr }; 
            timeout_t polltime = { 100 };   
            timeout_t updatetime = { 100 };   
 
            typedef std::vector< std::shared_ptr<OPCAttribute> > IOList;
            IOList iolist;    
            size_t maxItem = { 0 };
            size_t maxReadItems = { 0 };
            size_t maxWriteItems = { 0 };
 
            struct Channel
            {
                size_t num;
                size_t idx;
                std::shared_ptr<OPCUAClient> client;
                uniset::Trigger trStatus;
                uniset::PassiveTimer ptTimeout;
                std::atomic_bool status = { false };
                std::string addr;
                std::string user;
                std::string pass;
                std::unordered_map<Tick, std::shared_ptr<ReadGroup>> readValues;
                std::unordered_map<Tick, std::shared_ptr<WriteGroup>> writeValues;
                uniset::ObjectId respond_s = { uniset::DefaultObjectId };
                IOController::IOStateList::iterator respond_it;
            };
            Channel channels[numChannels];
            uniset::Trigger noConnections;
            std::atomic_uint32_t currentChannel = { 0 };
 
            uniset::timeout_t reconnectPause = { 10000 };
            int filtersize = { 0 };
            float filterT = { 0.0 };
 
            std::string s_field;
            std::string s_fvalue;
            std::optional<std::regex> s_fvalue_re;
 
            std::shared_ptr<SMInterface> shm;
            std::string prop_prefix;
            const std::string argprefix;
 
            PassiveTimer ptHeartBeat;
            uniset::ObjectId sidHeartBeat;
            int maxHeartBeat = { 10 };
            IOController::IOStateList::iterator itHeartBeat;
 
            bool force = { false };                     
            bool force_out = { false };                 
            bool writeToAllChannels = { false };        
            timeout_t smReadyTimeout = { 15000 };       
            bool enableSubscription = {false};
            double publishingInterval = { 0.0 };
            double samplingInterval = { -1.0 };
            uint16_t timeoutIterate = {100};
            uint16_t stopOnError = {0U};                
            uint32_t connectCount = {0U};               
            std::atomic_bool subscription_ok = {false};
 
            std::atomic_bool activated = { false };
            std::atomic_bool cancelled = { false };
            std::atomic_bool readconf_ok = { false };
 
            int activateTimeout;
            uniset::ObjectId sidTestSMReady = { uniset::DefaultObjectId };
            uniset::ObjectId sidRespond = {uniset::DefaultObjectId };
            IOController::IOStateList::iterator itRespond;
 
            std::shared_ptr<LogAgregator> loga;
            std::shared_ptr<DebugStream> opclog;
            std::shared_ptr<LogServer> logserv;
            std::string logserv_host = {""};
            int logserv_port = {0};
 
            std::shared_ptr< ThreadCreator<OPCUAExchange> > thrChannel[numChannels];
 
            uniset::ObjectId sidExchangeMode = { uniset::DefaultObjectId }; 
            IOController::IOStateList::iterator itExchangeMode;
            long exchangeMode = { emNone }; 
 
            VMonitor vmon;
 
        private:
    };
    // --------------------------------------------------------------------------
} // end of namespace uniset
// -----------------------------------------------------------------------------
#endif // OPCUAExchange_H_
// -----------------------------------------------------------------------------