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 <list>
#include <string>
#include <regex>
#include <optional>
#include <chrono>
#include <mutex>
#include <unordered_map>
#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<opcua::ua::ReadValueId>> ids;
            };
            struct WriteGroup
            {
                std::vector<std::vector<opcua::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();
                    opcua::StatusCode status();
                    const opcua::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 );
                    opcua::StatusCode status();
                    const opcua::ua::WriteValue& ref();
                    static opcua::Variant initValue(const std::string& stype, 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
            };
 
            struct ErrorRecord
            {
                std::chrono::system_clock::time_point time;      // first occurrence
                std::chrono::system_clock::time_point lastSeen;  // last occurrence
                size_t channel { 0 };
                std::string operation;  // "read" | "write" | "connect"
                UA_StatusCode statusCode { UA_STATUSCODE_GOOD };
                std::string nodeid;     // опционально, для read/write
                size_t count { 1 };
            };
 
            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 UHttp::HttpRequestContext& ctx ) override;
            virtual Poco::JSON::Object::Ptr httpGetMyInfo( Poco::JSON::Object::Ptr root ) 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( Poco::Net::HTTPServerResponse& resp, const Poco::URI::QueryParameters& p );
            Poco::JSON::Object::Ptr httpStatus();
            Poco::JSON::Object::Ptr buildLogServerInfo();
 
            // Новые HTTP endpoints
            Poco::JSON::Object::Ptr httpGet( const Poco::URI::QueryParameters& p );
            Poco::JSON::Object::Ptr httpSensors( const Poco::URI::QueryParameters& p );
            Poco::JSON::Object::Ptr httpSensor( Poco::Net::HTTPServerResponse& resp, const Poco::URI::QueryParameters& p );
            Poco::JSON::Object::Ptr httpDiagnostics( const Poco::URI::QueryParameters& p );
            Poco::JSON::Object::Ptr httpTakeControl( Poco::Net::HTTPServerResponse& resp, const Poco::URI::QueryParameters& p );
            Poco::JSON::Object::Ptr httpReleaseControl( Poco::Net::HTTPServerResponse& resp, const Poco::URI::QueryParameters& p );
 
            // Вспомогательные методы
            std::string formatTime( const std::chrono::system_clock::time_point& tp ) const;
            Poco::JSON::Object::Ptr sensorToJson( const std::shared_ptr<OPCAttribute>& attr, bool detailed = false, const std::string& cachedName = "" ) const;
 
            // Защитный флаг: запретить /setparam при false
            bool httpEnabledSetParams { true };
#endif
 
            // Вспомогательные методы
            void addError( size_t channel, const std::string& operation, UA_StatusCode status, const std::string& nodeid = "" );
 
            // Флаги управления режимом через HTTP (используются и вне REST API)
            bool httpControlAllow { false };              
            std::atomic_bool httpControlActive { false }; 
            // чтение файла конфигурации
            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();
 
            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;
                std::atomic_bool needSubscription = { false };  // Требуется создать подписку
            };
            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};                
            std::atomic<uint32_t> connectCount = {0U};  
            std::atomic_bool subscription_ok = {false};
            std::atomic_bool newSessionActivated = {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;
            std::atomic<long> exchangeMode = { emNone }; 
            std::atomic<long> sensorExchangeMode = { emNone }; 
            VMonitor vmon;
 
            struct ErrorKey
            {
                size_t channel;
                std::string operation;
                UA_StatusCode statusCode;
                std::string nodeid;
 
                bool operator==( const ErrorKey& other ) const noexcept
                {
                    return channel == other.channel &&
                           statusCode == other.statusCode &&
                           operation == other.operation &&
                           nodeid == other.nodeid;
                }
            };
 
            struct ErrorKeyHash
            {
                size_t operator()( const ErrorKey& k ) const noexcept
                {
                    size_t h1 = std::hash<size_t> {}(k.channel);
                    size_t h2 = std::hash<UA_StatusCode> {}(k.statusCode);
                    size_t h3 = std::hash<std::string> {}(k.operation);
                    size_t h4 = std::hash<std::string> {}(k.nodeid);
                    return ((h1 ^ (h2 << 1)) ^ (h3 << 2)) ^ (h4 << 3);
                }
            };
 
            // Диагностика: кольцевой буфер ошибок (с дедупликацией)
            mutable std::list<ErrorRecord> errorHistory;
            mutable std::unordered_map<ErrorKey, std::list<ErrorRecord>::iterator, ErrorKeyHash> errorHistoryIndex;
            size_t errorHistoryMax { 100 };  
            mutable std::mutex errorHistoryMutex;
 
            // Счётчики ошибок (не сбрасываются)
            std::atomic<uint64_t> totalReadErrors { 0 };
            std::atomic<uint64_t> totalWriteErrors { 0 };
            std::atomic<uint64_t> totalConnectionLosses { 0 };
 
            // Время запуска для uptime
            std::chrono::steady_clock::time_point startTime;
 
        private:
            void doCreateSubscription(int nchannel);
    };
    // --------------------------------------------------------------------------
} // end of namespace uniset
// -----------------------------------------------------------------------------
#endif // OPCUAExchange_H_
// -----------------------------------------------------------------------------