oversampling_provider_subscription.c

/******************************************************************************
**
**   **************************** auto-generated ****************************
**     This code was generated by a tool: UaModeler
**     Runtime Version: 1.2.0, using ANSI C DemoProvider 1.4.0 template
**
**     This is a template file that was generated for your convenience.
**     This file will not be overwritten when generating code again.
**     ADD YOUR IMPLEMTATION HERE!
**   **************************** auto-generated ****************************

** Copyright (c) 2006-2014 Unified Automation GmbH All rights reserved.
** Web: http://www.unifiedautomation.com
**
** Software License Agreement ("SLA") Version 2.3
**
** Unless explicitly acquired and licensed from Licensor under another
** license, the contents of this file are subject to the Software License
** Agreement ("SLA") Version 2.3, or subsequent versions
** as allowed by the SLA, and You may not copy or use this file in either
** source code or executable form, except in compliance with the terms and
** conditions of the SLA.
**
** All software distributed under the SLA is provided strictly on an
** "AS IS" basis, WITHOUT WARRANTY OF ANY KIND, EITHER EXPRESS OR IMPLIED,
** AND LICENSOR HEREBY DISCLAIMS ALL SUCH WARRANTIES, INCLUDING WITHOUT
** LIMITATION, ANY WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
** PURPOSE, QUIET ENJOYMENT, OR NON-INFRINGEMENT. See the SLA for specific
** language governing rights and limitations under the SLA.
**
** Project: OPC Ansi C OverSamplingProvider for namespace http://www.unifiedautomation.com/OverSampling
**
** Description: OPC Unified Architecture Software Development Kit.
**
** The complete license agreement can be found here:
** http://unifiedautomation/License/SLA/2.3/
**
** Created: 04.02.2014
**
******************************************************************************/

#include <uaserver_config.h>

#include <opcua_statuscodes.h>
#include <opcua_attributes.h>

#include <uaserver_basenode.h>
#include <uaserver_utilities.h>
#include <uaserver_p_atomic.h>
#include <uaserver_monitoring.h>
#include <uaserver_monitoreditemmanager.h>

#include "uaprovider_oversampling_helper.h"
#include "uaprovider_oversampling_identifiers_1.h"

/* include headers for POSIX timer */
#include <signal.h>
#include <time.h>
#include <pthread.h>
#include <time.h>

/* define signal number to use for timer */
#define SIGTIMER SIGALRM

OPCUA_BEGIN_EXTERN_C

#if OPCUA_MULTITHREADED || OPCUA_USE_SYNCHRONISATION
static OpcUa_Mutex g_mtxList;
# define LOCK_LIST() OpcUa_Mutex_Lock(g_mtxList)
# define UNLOCK_LIST() OpcUa_Mutex_Unlock(g_mtxList)
#else
# define LOCK_LIST()
# define UNLOCK_LIST()
#endif

#define NUM_SAMPLINGRATES 7

static OpcUa_Int   g_aiSamplingRates[NUM_SAMPLINGRATES] = {1, 50, 100, 250, 500, 1000, 2500};
static timer_t g_hSamplingTimers[NUM_SAMPLINGRATES];
static UaServer_Vector g_vecSamplingLists[NUM_SAMPLINGRATES];

/* timer thread id */
static pthread_t g_timerthreadid;
static pthread_attr_t g_timerthreadattr;
static volatile int g_bExitTimerThread = 0;

/* simulate variables */
double g_Voltage = 0.0;
double g_Current = 0.0;

/* enable debug trace to see if the timers cannot be handled fast enough */
#define TIMER_OVERRUN_TRACE 0

static void UaProvider_OverSampling_SampleData(UaServer_Vector *pvecList)
{
    unsigned int i = 0;
    UaServer_MonitoredItemData *pMonitoredItemData = OpcUa_Null;
    OpcUa_Variable *pVariable = OpcUa_Null;
    OpcUa_DataValue *pValue = OpcUa_Null;
    unsigned int size;

    if (pvecList == &g_vecSamplingLists[0])
    {
        /* simulate value at 1Khz interval */
        g_Voltage += 0.1;
        g_Current += 0.17;
    }

    LOCK_LIST();
    size = UaServer_Vector_GetSize(pvecList);
    /* iterate over all monitored items of this sampling rate */
    for (i = 0; i < size; i++)
    {
        pMonitoredItemData = (UaServer_MonitoredItemData*)UaServer_Vector_Get(pvecList, i);

        /* the status of the items is always good, so status change will never trigger a data change */
        if (pMonitoredItemData->MonitoringMode == OpcUa_MonitoringMode_Disabled)
        {
            continue;
        }

        /* create new data value */
        pValue = (OpcUa_DataValue*)OpcUa_Alloc(sizeof(OpcUa_DataValue));
        if (pValue == 0) break;
        OpcUa_DataValue_Initialize(pValue);
        /* set timestamps */
        pValue->ServerTimestamp = OpcUa_DateTime_UtcNow();
        /* check attribute type */
        if (pMonitoredItemData->AttributeId == OpcUa_Attributes_Value)
        {
            pValue->SourceTimestamp = pValue->ServerTimestamp;
            switch (pMonitoredItemData->NodeId.Identifier.Numeric)
            {
            case OverSampling_Objects_S1KHz_Voltage:
                pValue->Value.Datatype = OpcUaType_Double;
                pValue->Value.Value.Double = g_Voltage;
                break;
            case OverSampling_Objects_S1KHz_Current:
                pValue->Value.Datatype = OpcUaType_Double;
                pValue->Value.Value.Double = g_Current;
                break;
            default:
                /* default fallback to node based values */
                UaServer_GetNode(&g_pOverSamplingProvider->AddressSpace, &pMonitoredItemData->NodeId, (OpcUa_BaseNode**)&pVariable);
                if (pVariable)
                {
                    OpcUa_Variant_Clone(OpcUa_Variable_GetValue(pVariable), &pValue->Value);
                }
                break;
            }
        }
        else
        {
            /* all other attributes are read from nodes */
            UaServer_GetNode(&g_pOverSamplingProvider->AddressSpace, &pMonitoredItemData->NodeId, (OpcUa_BaseNode**)&pVariable);
            if (pVariable)
            {
                UaServer_GetNonValueAttribute(pMonitoredItemData->AttributeId,
                                              (OpcUa_BaseNode*)pVariable,
                                              &pValue->Value,
                                              UaServer_Subscription_GetSession(pMonitoredItemData->pSubscription));
            }
        }

        /* push new value into SDK */
        UaServer_NewItemValue(pMonitoredItemData, pValue);
    }
    UNLOCK_LIST();
}

/* Timer thread which handles SIGTIMER using sigwaitinfo.
 * The configured signal SIGTIMER is blocked by this thread using sigprocmask,
 * so no signal handler is needed.
 * sigwaitinfo suspends the thread until the signal is pending and removes it from pending list.
 *
 * We are using this timer for 1kHz sampling of UA data.
 * [posix thread] */
void* timer_thread (void *arg)
{
    UaServer_Vector *pvecList = OpcUa_Null;
    sigset_t waitset;
    int sig;
    siginfo_t info;
    int index;

    OpcUa_ReferenceParameter(arg);

    /* block signals processed by the timer thread */
    sigemptyset( &waitset );
    sigaddset( &waitset, SIGTIMER );
    sigprocmask( SIG_BLOCK, &waitset, NULL );

    while (g_bExitTimerThread == 0)
    {
        sig = sigwaitinfo( &waitset, &info );
        switch(sig)
        {
        case SIGTIMER:
            index = info.si_value.sival_int;
            if (index < 0 || index >= NUM_SAMPLINGRATES) break;
            pvecList = &g_vecSamplingLists[index];
            if (info.si_code == SI_TIMER)
            {
#if TIMER_OVERRUN_TRACE
                timer_t timerid;
                timerid  = &g_hSamplingTimers[index];

                /* debug printf
                printf("timer expired: %p\n", pvecList);
                */

                if (timer_getoverrun(timerid) > 0)
                {
                    /* if you see this output your system is not able to process the timers fast enough.
                     * It make your sampling routine faster or use a slower sampling rate.
                     * You should remove this printf from productive code.
                     */
                    fprintf(stderr, "timer overrun\n");
                }
#endif /* TIMER_OVERRUN_TRACE */
                /* timer expired, sample data */
                UaProvider_OverSampling_SampleData(pvecList);
            }
            break;
        default:
            /* debug info, this should not happen */
            perror("sigwaitinfo");
            break;
        }
    }

    return 0;
}
OpcUa_Void UaProvider_OverSampling_Subscription_Initialize()
{
    int i;
    struct sigevent se;
    struct itimerspec interval;
    int ret;
    sigset_t waitset;

#if OPCUA_MULTITHREADED || OPCUA_USE_SYNCHRONISATION
    OpcUa_Mutex_Create(&g_mtxList);
#endif
    LOCK_LIST();

    /* block signals processed by the timer thread */
    sigemptyset( &waitset );
    sigaddset( &waitset, SIGTIMER );
    sigprocmask( SIG_BLOCK, &waitset, NULL );

    pthread_attr_init(&g_timerthreadattr);
    pthread_attr_setstacksize(&g_timerthreadattr, 1024*1024);

    /* start timer thread */
    pthread_create(&g_timerthreadid, &g_timerthreadattr, timer_thread, 0);

    /* init vectors and start sampling timers */
    for (i = 0; i < NUM_SAMPLINGRATES; i++)
    {
        UaServer_Vector_Initialize(&g_vecSamplingLists[i], 10, 10);

        /* create sampling timer */
        OpcUa_MemSet(&se, 0, sizeof(struct sigevent));
        se.sigev_notify = SIGEV_SIGNAL;            /* notifiy by own thread function */
        se.sigev_signo  = SIGTIMER;
        se.sigev_value.sival_int = i;              /* store sampling index */
        ret = timer_create(CLOCK_MONOTONIC, &se, &g_hSamplingTimers[i]);
        if (ret != 0)
        {
            perror("timer_create");
        }

        /* arm timer */
        interval.it_interval.tv_sec = g_aiSamplingRates[i] / 1000;
        interval.it_interval.tv_nsec = (g_aiSamplingRates[i] % 1000) * 1000000;
        interval.it_value.tv_sec = interval.it_interval.tv_sec;
        interval.it_value.tv_nsec = interval.it_interval.tv_nsec;
        ret = timer_settime(g_hSamplingTimers[i], 0, &interval, 0);
        if (ret != 0)
        {
            perror("timer_settime");
        }
    }
    UNLOCK_LIST();
}
OpcUa_Void UaProvider_OverSampling_Subscription_Cleanup()
{
    int i;
    void *retval = 0;

    /* wait for timer thread to terminate */
    g_bExitTimerThread = 1;
    printf("wait for timer thread to terminate\n");
    pthread_join(g_timerthreadid, &retval);

    LOCK_LIST();
    for (i = 0; i < NUM_SAMPLINGRATES; i++)
    {
        /* delete the timer associated with this thread */
        timer_delete(g_hSamplingTimers[i]);
        /* cleanup our sampling list */
        UaServer_Vector_Clear(&g_vecSamplingLists[i]);
    }
    UNLOCK_LIST();

#if OPCUA_MULTITHREADED || OPCUA_USE_SYNCHRONISATION
    OpcUa_Mutex_Delete(&g_mtxList);
#endif
}

OpcUa_Range* UaProvider_OverSampling_GetEURange(const OpcUa_NodeId *pNodeId)
{
    OpcUa_Range         *pRange = OpcUa_Null;
    OpcUa_BaseNode      *pNode = OpcUa_Null, *pProperty = OpcUa_Null;
    OpcUa_QualifiedName  qnEURange;

    OpcUa_QualifiedName_Initialize(&qnEURange);
    qnEURange.NamespaceIndex = 0;
    OpcUa_String_AttachReadOnly(&qnEURange.Name, OpcUa_BrowseName_EURange);

    UaServer_GetNode(&g_pOverSamplingProvider->AddressSpace, (OpcUa_NodeId*)pNodeId, &pNode);
    if (pNode)
    {
        unsigned int iReference;
        UaServer_Vector *pReferencesVector = OpcUa_BaseNode_GetReferences(pNode);

        for (iReference = 0; iReference < UaServer_Vector_GetSize(pReferencesVector); iReference++)
        {
            OpcUa_Reference *pReference = UaServer_Vector_Get(pReferencesVector, iReference);
            if (pReference->refType.NamespaceIndex == 0
                && pReference->refType.IdentifierType == OpcUa_IdentifierType_Numeric
                && pReference->refType.Identifier.Numeric == OpcUaId_HasProperty
                && pReference->targetType == TargetTypeNode
                && OpcUa_BaseNode_GetClass(pReference->target.pNode) == OpcUa_NodeClass_Variable)
            {
                if (OpcUa_QualifiedName_Compare(OpcUa_BaseNode_GetBrowseName(pReference->target.pNode), &qnEURange) == 0)
                {
                    pProperty = pReference->target.pNode;
                    break;
                }
            }
        }

        if (pProperty)
        {
            OpcUa_Variant *pValue = OpcUa_Variable_GetValue(pProperty);

            if (pValue->Datatype == OpcUaType_ExtensionObject
                && pValue->ArrayType == OpcUa_VariantArrayType_Scalar
                && pValue->Value.ExtensionObject != OpcUa_Null
                && pValue->Value.ExtensionObject->Encoding == OpcUa_ExtensionObjectEncoding_EncodeableObject
                && pValue->Value.ExtensionObject->Body.EncodeableObject.Type == &OpcUa_Range_EncodeableType)
            {
                pRange = (OpcUa_Range*)pValue->Value.ExtensionObject->Body.EncodeableObject.Object;
            }
        }
    }
    return pRange;
}

IFMETHODIMP(UaProvider_OverSampling_AddItem)(UaServer_MonitoredItem* a_pItem)
{
    OpcUa_StatusCode uStatus = OpcUa_Good;
    OpcUa_Int               i               = 0;
    OpcUa_Int               iDist           = 0;
    OpcUa_Int               iTmp            = 0;
    OpcUa_Int               iRevised        = 0; /* revised index */
    OpcUa_BaseNode          *pNode          = OpcUa_Null;
    UaServer_AddressSpace   *pAddressSpace  = &(g_pOverSamplingProvider->AddressSpace);
#if UASERVER_SUPPORT_AUTHORIZATION
    UaServer_UserCtx        *pUserContext   = UaServer_UserMgt_GetUserContext(a_pItem);
#endif

    iDist = UaServer_P_Abs(a_pItem->iSamplingInterval - g_aiSamplingRates[0]);

    /* find closest sampling rate */
    for (i = 1; i < NUM_SAMPLINGRATES; i++)
    {
        iTmp = UaServer_P_Abs(a_pItem->iSamplingInterval - g_aiSamplingRates[i]);

        if (iTmp < iDist)
        {
            iDist    = iTmp;
            iRevised = i;
        }
    }

    /* revise values */
    a_pItem->uProviderHandle = iRevised;
    a_pItem->iSamplingInterval = g_aiSamplingRates[iRevised];

    /* set data type */
    UaServer_GetNode(pAddressSpace, &a_pItem->NodeId, &pNode);

    if (pNode)
    {
        if (a_pItem->ItemType == DATA)
        {
            UaServer_MonitoredItemData* pMonitoredItemData = (UaServer_MonitoredItemData*)a_pItem;
            if (OpcUa_BaseNode_GetClass(pNode) == OpcUa_NodeClass_Variable
                && pMonitoredItemData->AttributeId == OpcUa_Attributes_Value)
            {
                OpcUa_DataVariable* pVariable = (OpcUa_DataVariable*)pNode;

#if UASERVER_SUPPORT_AUTHORIZATION
                /* check if user is allowed to read the value */
                if (!UaServer_UserMgt_IsReadable(OpcUa_BaseNode_GetINode(pNode), pUserContext))
                {
                    uStatus = OpcUa_BadNotReadable;
                }
#endif

                /* check index range */
                if (OpcUa_IsGood(uStatus) && pMonitoredItemData->NoOfIndexRanges > 0)
                {
                    uStatus = UaServer_VariantNumericRangePlausible(OpcUa_Variable_GetValue(pVariable),
                                                                    pMonitoredItemData->IndexRanges,
                                                                    pMonitoredItemData->NoOfIndexRanges);
                }

                /* check data change filter */
                if (OpcUa_IsGood(uStatus) && pMonitoredItemData->DataChangeFilter.DeadbandType != OpcUa_DeadbandType_None)
                {
                    switch (OpcUa_Variable_GetValue(pVariable)->Datatype)
                    {
                    case OpcUaType_SByte:
                    case OpcUaType_Byte:
                    case OpcUaType_Int16:
                    case OpcUaType_UInt16:
                    case OpcUaType_Int32:
                    case OpcUaType_UInt32:
                    case OpcUaType_Int64:
                    case OpcUaType_UInt64:
                    case OpcUaType_Float:
                    case OpcUaType_Double:
                        break;
                    default:
                        uStatus = OpcUa_BadFilterNotAllowed;
                        break;
                    }

                    if (OpcUa_IsGood(uStatus) && pMonitoredItemData->DataChangeFilter.DeadbandType == OpcUa_DeadbandType_Percent)
                    {
                        if (pMonitoredItemData->DataChangeFilter.DeadbandValue < 0.0
                            || pMonitoredItemData->DataChangeFilter.DeadbandValue > 100.0)
                        {
                            uStatus = OpcUa_BadDeadbandFilterInvalid;
                        }
                    }
                }

                /* get EURange property if DeadbandPercent */
                if (OpcUa_IsGood(uStatus) && pMonitoredItemData->DataChangeFilter.DeadbandType == OpcUa_DeadbandType_Percent)
                {
                    OpcUa_Range *pRange = UaProvider_OverSampling_GetEURange(&pMonitoredItemData->NodeId);
                    if (pRange)
                    {
                        pMonitoredItemData->pEURange = pRange;
                    }
                    else
                    {
                        uStatus = OpcUa_BadMonitoredItemFilterUnsupported;
                    }
                }

                if (OpcUa_IsGood(uStatus))
                {
                    /* add to list */
                    LOCK_LIST();
                    UaServer_Vector_Add(&g_vecSamplingLists[iRevised], pMonitoredItemData);
                    UNLOCK_LIST();
                }
            }
            else
            {
                OpcUa_Variant currentValue;
                OpcUa_Variant_Initialize(&currentValue);

#if UASERVER_SUPPORT_AUTHORIZATION
                /* check if user is allowed to read the attribute */
                if (!UaServer_UserMgt_IsAttributeReadable(OpcUa_BaseNode_GetINode(pNode), pUserContext))
                {
                    uStatus = OpcUa_BadNotReadable;
                }
#endif

                /* check data change filter */
                if (OpcUa_IsGood(uStatus) && pMonitoredItemData->DataChangeFilter.DeadbandType != OpcUa_DeadbandType_None)
                {
                    uStatus = OpcUa_BadFilterNotAllowed;
                }

                /* index range not available for non-value attribute */
                if (OpcUa_IsGood(uStatus) && pMonitoredItemData->NoOfIndexRanges > 0)
                {
                    uStatus = OpcUa_BadIndexRangeNoData;
                }

                /* get value of the attribute */
                if (OpcUa_IsGood(uStatus))
                {
                    uStatus = UaServer_GetNonValueAttribute(pMonitoredItemData->AttributeId,
                                                            pNode,
                                                            &currentValue,
                                                            UaServer_Subscription_GetSession(pMonitoredItemData->pSubscription));
                }

                if (OpcUa_IsGood(uStatus))
                {
                    /* add to list */
                    LOCK_LIST();
                    UaServer_Vector_Add(&g_vecSamplingLists[iRevised], pMonitoredItemData);
                    UNLOCK_LIST();
                }
                OpcUa_Variant_Clear(&currentValue);
            }
        }
    }
    else
    {
        uStatus = OpcUa_BadNodeIdUnknown;
    }

    return uStatus;
}

IFMETHODIMP(UaProvider_OverSampling_RemoveItem)(UaServer_MonitoredItem* a_pItem)
{
    OpcUa_StatusCode uStatus = OpcUa_Good;
    UaServer_Vector *pVec = &g_vecSamplingLists[a_pItem->uProviderHandle];
    unsigned int size;
    unsigned int i;
    int found = 0;

    if (a_pItem->ItemType == DATA)
    {
        LOCK_LIST();
        size = UaServer_Vector_GetSize(pVec);

        for (i = 0; i < size; i++)
        {
            if (UaServer_Vector_Get(pVec, i) == a_pItem)
            {
                found = 1;
                break;
            }
        }
        if (found)
        {
            for (; i < size - 1; i++)
            {
                UaServer_Vector_Set(pVec, i, UaServer_Vector_Get(pVec, i + 1));
            }
            UaServer_Vector_Resize(pVec, size - 1);
        }
        else
        {
            uStatus = OpcUa_Bad;
        }
        UNLOCK_LIST();
    }

    return uStatus;
}

IFMETHODIMP(UaProvider_OverSampling_Subscribe)(UaServer_ProviderSubscribeContext *a_pCtx)
{
    OpcUa_StatusCode uStatus = OpcUa_Good;

    /* send the callback */
    UaServer_Atomic_Increment(&a_pCtx->nOutstandingCbs);
    UaServer_SubscribeComplete(a_pCtx);

    return uStatus;
}


OPCUA_END_EXTERN_C