Validation of timestamps and validation of values for SHDR and JSON pipeline conversions.

CMakeLists.txt

@@ -2,7 +2,7 @@
 set(AGENT_VERSION_MAJOR 2)
 set(AGENT_VERSION_MINOR 5)
 set(AGENT_VERSION_PATCH 0)
-set(AGENT_VERSION_BUILD 4)
+set(AGENT_VERSION_BUILD 5)
 set(AGENT_VERSION_RC "")
 
 # This minimum version is to support Visual Studio 2019 and C++ feature checking and FetchContent

README.md

@@ -672,6 +672,10 @@ Configuration Parameters
 
     *Default*: `false`
 
+* `CorrectTimestamps` - Verify time is always progressing forward for each data item and correct if not.
+
+    *Default*: `false`
+
 * `Validation` - Turns on validation of model components and observations
 
     *Default*: `false`

agent_lib/CMakeLists.txt

@@ -189,6 +189,7 @@ set(AGENT_SOURCES
         "${SOURCE_DIR}/pipeline/topic_mapper.hpp"
         "${SOURCE_DIR}/pipeline/transform.hpp"
         "${SOURCE_DIR}/pipeline/upcase_value.hpp"
+        "${SOURCE_DIR}/pipeline/correct_timestamp.hpp"       
         "${SOURCE_DIR}/pipeline/validator.hpp"       
 
 # src/pipeline SOURCE_FILES_ONLY

src/mtconnect/agent.hpp

@@ -213,6 +213,10 @@ namespace mtconnect {
     /// @brief Get the MTConnect schema version the agent is supporting
     /// @return The MTConnect schema version as a string
     const auto &getSchemaVersion() const { return m_schemaVersion; }
+
+    /// @brief Get the validation state of the agent
+    /// @returns the validation state of the agent
+    bool isValidating() const { return m_validation; }
 
     /// @brief Get the integer schema version based on configuration.
     /// @returns the schema version as an integer [major * 100 + minor] as a 32bit integer.
@@ -589,6 +593,7 @@ namespace mtconnect {
       }
     }
     int32_t getSchemaVersion() const override { return m_agent->getIntSchemaVersion(); }
+    bool isValidating() const override { return m_agent->isValidating(); }
     void deliverObservation(observation::ObservationPtr obs) override
     {
       m_agent->receiveObservation(obs);

src/mtconnect/configuration/agent_config.cpp

@@ -830,7 +830,8 @@ namespace mtconnect::configuration {
                 {configuration::TlsClientCAs, ""s},
                 {configuration::SuppressIPAddress, false},
                 {configuration::AllowPutFrom, ""s},
-                {configuration::Validation, false}});
+                {configuration::Validation, false},
+                {configuration::CorrectTimestamps, false}});
 
     m_workerThreadCount = *GetOption<int>(options, configuration::WorkerThreads);
     m_monitorFiles = *GetOption<bool>(options, configuration::MonitorConfigFiles);

src/mtconnect/configuration/config_options.hpp

@@ -77,6 +77,7 @@ namespace mtconnect {
     DECLARE_CONFIGURATION(EnableSourceDeviceModels);
     DECLARE_CONFIGURATION(WorkerThreads);
     DECLARE_CONFIGURATION(Validation);
+    DECLARE_CONFIGURATION(CorrectTimestamps);
     ///@}
 
     /// @name MQTT Configuration

src/mtconnect/entity/entity.hpp

@@ -95,7 +95,12 @@ namespace mtconnect {
       /// @param props entity properties
       Entity(const std::string &name, const Properties &props) : m_name(name), m_properties(props)
       {}
-      Entity(const Entity &entity) = default;
+      Entity(const Entity &entity)
+      : m_name(entity.m_name), m_properties(entity.m_properties), m_order(entity.m_order)
+      {
+        if (entity.m_attributes)
+          setAttributes(*entity.m_attributes);
+      }
       virtual ~Entity() {}
 
       /// @brief Get a shared pointer
@@ -348,10 +353,21 @@ namespace mtconnect {
       auto erase(Properties::iterator &it) { return m_properties.erase(it); }
       /// @brief tells the entity which properties are attributes for XML generation
       /// @param[in] a the attributes
-      void setAttributes(AttributeSet a) { m_attributes = a; }
+      void setAttributes(AttributeSet a)
+      {
+        std::unique_ptr<AttributeSet> set(new AttributeSet(a));
+        m_attributes.swap(set);
+      }
       /// @brief get the attributes for XML generation
       /// @return attribute set
-      const auto &getAttributes() const { return m_attributes; }
+      const auto &getAttributes() const
+      {
+        static AttributeSet empty;
+        if (m_attributes)
+          return *m_attributes;
+        else
+          return empty;
+      }
 
       /// @brief compare two entities for equality
       /// @param other the other entity
@@ -436,7 +452,8 @@ namespace mtconnect {
       QName m_name;
       Properties m_properties;
       OrderMapPtr m_order;
-      AttributeSet m_attributes;
+      std::unique_ptr<AttributeSet> m_attributes;
+      std::unique_ptr<ErrorList> m_errors;
     };
 
     /// @brief variant visitor to compare two entity parameter values for equality

src/mtconnect/pipeline/correct_timestamp.hpp

@@ -0,0 +1,86 @@
+//
+// Copyright Copyright 2009-2024, AMT – The Association For Manufacturing Technology (“AMT”)
+// All rights reserved.
+//
+//    Licensed under the Apache License, Version 2.0 (the "License");
+//    you may not use this file except in compliance with the License.
+//    You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+//    Unless required by applicable law or agreed to in writing, software
+//    distributed under the License is distributed on an "AS IS" BASIS,
+//    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//    See the License for the specific language governing permissions and
+//    limitations under the License.
+//
+
+#pragma once
+
+#include "mtconnect/config.hpp"
+#include "transform.hpp"
+
+namespace mtconnect::pipeline {
+  /// @brief Filter duplicates
+  class AGENT_LIB_API CorrectTimestamp : public Transform
+  {
+  protected:
+    struct State : TransformState
+    {
+      std::unordered_map<std::string, Timestamp> m_timestamps;
+    };
+
+  public:
+    CorrectTimestamp(const CorrectTimestamp &) = default;
+    /// @brief Create a duplicate filter with shared state from the context
+    /// @param context the context
+    CorrectTimestamp(PipelineContextPtr context)
+      : Transform("ValidateTimestamp"),
+        m_context(context),
+        m_state(context->getSharedState<State>(m_name))
+    {
+      m_guard = TypeGuard<observation::Observation>(RUN);
+    }
+    ~CorrectTimestamp() override = default;
+
+    /// @brief check if the entity is a duplicate
+    /// @param[in] entity the entity to check
+    /// @return the result of the transform if not a duplicate or an empty entity
+    entity::EntityPtr operator()(entity::EntityPtr &&entity) override
+    {
+      using namespace observation;
+
+      auto obs = std::dynamic_pointer_cast<Observation>(entity);
+      if (obs->isOrphan())
+        return entity::EntityPtr();
+
+      auto di = obs->getDataItem();
+      auto &id = di->getId();
+      auto ts = obs->getTimestamp();
+
+      std::lock_guard<TransformState> guard(*m_state);
+
+      auto last = m_state->m_timestamps.find(id);
+      if (last != m_state->m_timestamps.end())
+      {
+        if (ts < last->second)
+        {
+          LOG(debug) << "Observation for data item " << id << " has timestamp " << format(ts)
+                     << " thats is before " << format(last->second);
+
+          // Set the timestamp to now.
+          ts = std::chrono::system_clock::now();
+          obs->setTimestamp(ts);
+        }
+      }
+
+      m_state->m_timestamps.emplace(id, ts);
+
+      return obs;
+    }
+
+  protected:
+    PipelineContextPtr m_context;
+    std::shared_ptr<State> m_state;
+  };
+}  // namespace mtconnect::pipeline

src/mtconnect/pipeline/duplicate_filter.hpp

@@ -25,12 +25,6 @@ namespace mtconnect::pipeline {
   class AGENT_LIB_API DuplicateFilter : public Transform
   {
   public:
-    /// @brief Shared states to check for duplicates
-    struct State : TransformState
-    {
-      std::unordered_map<std::string, entity::Value> m_values;
-    };
-
     DuplicateFilter(const DuplicateFilter &) = default;
     /// @brief Create a duplicate filter with shared state from the context
     /// @param context the context

src/mtconnect/pipeline/json_mapper.cpp

@@ -134,14 +134,19 @@ namespace mtconnect::pipeline {
           {
             LOG(warning) << "Error while parsing json: " << e->what();
           }
+
+          props.clear();
+          props["VALUE"] = "UNAVAILABLE"s;
+          if (m_pipelineContext->m_contract->isValidating())
+            props["quality"] = "INVALID"s;
+
+          obs = observation::Observation::make(dataItem, props, *m_timestamp, errors);
         }
-        else
-        {
-          if (m_source)
-            dataItem->setDataSource(*m_source);
-          m_entities.push_back(obs);
-          m_forward(std::move(obs));
-        }
+
+        if (m_source)
+          dataItem->setDataSource(*m_source);
+        m_entities.push_back(obs);
+        m_forward(std::move(obs));
       }
     }
 

src/mtconnect/pipeline/json_mapper.hpp

@@ -24,8 +24,6 @@
 #include "mtconnect/entity/entity.hpp"
 #include "mtconnect/observation/observation.hpp"
 #include "mtconnect/pipeline/timestamp_extractor.hpp"
-#include "shdr_tokenizer.hpp"
-#include "timestamp_extractor.hpp"
 #include "topic_mapper.hpp"
 #include "transform.hpp"
 

src/mtconnect/pipeline/pipeline_contract.hpp

@@ -76,6 +76,9 @@ namespace mtconnect {
       /// @brief get the current schema version as an integer
       /// @returns the schema version as an integer [major * 100 + minor] as a 32bit integer.
       virtual int32_t getSchemaVersion() const = 0;
+      /// @brief `true` if validation is turned on for the agent.
+      /// @returns the validation state for the pipeline
+      virtual bool isValidating() const = 0;
       /// @brief iterate through all the data items calling `fun` for each
       /// @param[in] fun The function or lambda to call
       virtual void eachDataItem(EachDataItem fun) = 0;

src/mtconnect/pipeline/shdr_token_mapper.cpp

@@ -137,7 +137,7 @@ namespace mtconnect {
                                         const entity::Requirements &reqs,
                                         TokenList::const_iterator &token,
                                         const TokenList::const_iterator &end, ErrorList &errors,
-                                        int32_t schemaVersion)
+                                        int32_t schemaVersion, bool validation)
     {
       NAMED_SCOPE("zipProperties");
       Properties props;
@@ -168,6 +168,10 @@ namespace mtconnect {
         {
           LOG(warning) << "Cannot convert value for data item id '" << dataItem->getId()
                        << "': " << *token << " - " << e.what();
+          if (schemaVersion >= SCHEMA_VERSION(2, 5) && validation)
+          {
+            props.insert_or_assign("quality", "INVALID"s);
+          }
         }
       }
 
@@ -276,7 +280,8 @@ namespace mtconnect {
       if (reqs != nullptr)
       {
         auto obs = zipProperties(dataItem, timestamp, *reqs, token, end, errors,
-                                 m_contract->getSchemaVersion());
+                                 m_contract->getSchemaVersion(),
+                                 m_contract->isValidating());
         if (dataItem->getConstantValue())
           return nullptr;
         if (obs && source)

src/mtconnect/pipeline/validator.hpp

@@ -39,7 +39,7 @@ namespace mtconnect::pipeline {
     Validator(PipelineContextPtr context)
       : Transform("Validator"), m_contract(context->m_contract.get())
     {
-      m_guard = TypeGuard<observation::Event>(RUN) || TypeGuard<observation::Observation>(SKIP);
+      m_guard = TypeGuard<observation::Observation>(RUN) || TypeGuard<entity::Entity>(SKIP);
     }
 
     /// @brief validate the Event
@@ -49,14 +49,14 @@ namespace mtconnect::pipeline {
     {
       using namespace observation;
       using namespace mtconnect::validation::observations;
-      auto evt = std::dynamic_pointer_cast<Event>(entity);
+      auto obs = std::dynamic_pointer_cast<Observation>(entity);
 
-      auto di = evt->getDataItem();
-      if (evt->isUnavailable() || di->isDataSet())
+      auto di = obs->getDataItem();
+      if (obs->isUnavailable() || di->isDataSet())
       {
-        evt->setProperty("quality", std::string("VALID"));
+        obs->setProperty("quality", std::string("VALID"));
       }
-      else
+      else if (auto evt = std::dynamic_pointer_cast<observation::Event>(obs))
       {
         auto &value = evt->getValue<std::string>();
 
@@ -114,8 +114,15 @@ namespace mtconnect::pipeline {
           evt->setProperty("quality", std::string("UNVERIFIABLE"));
         }
       }
+      else if (auto spl = std::dynamic_pointer_cast<observation::Sample>(obs))
+      {
+      }
+      else
+      {
+        obs->setProperty("quality", std::string("VALID"));
+      }
 
-      return next(std::move(evt));
+      return next(std::move(obs));
     }
 
   protected:

src/mtconnect/source/adapter/adapter_pipeline.cpp

@@ -28,6 +28,7 @@
 #include "mtconnect/pipeline/timestamp_extractor.hpp"
 #include "mtconnect/pipeline/topic_mapper.hpp"
 #include "mtconnect/pipeline/upcase_value.hpp"
+#include "mtconnect/pipeline/correct_timestamp.hpp"
 #include "mtconnect/pipeline/validator.hpp"
 #include "mtconnect/source/adapter/adapter.hpp"
 
@@ -140,6 +141,9 @@ namespace mtconnect {
       if (IsOptionSet(m_options, configuration::ConversionRequired))
         next = next->bind(make_shared<ConvertSample>());
 
+      if (IsOptionSet(m_options, configuration::CorrectTimestamps))
+        next = next->bind(make_shared<CorrectTimestamp>(m_context));
+
       // Validate Values
       if (IsOptionSet(m_options, configuration::Validation))
         next = next->bind(make_shared<Validator>(m_context));