<commit_before>/*
 * Copyright Copyright 2012, System Insights, Inc.
 *
 *    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.
 */

#include "agent.hpp"
#include "dlib/logger.h"
#include <sys/stat.h>
#include <fcntl.h>
#include <sstream>
#include <stdexcept>
#include <dlib/tokenizer.h>
#include <dlib/misc_api.h>
#include <dlib/array.h>
#include <dlib/dir_nav.h>
#include <dlib/config_reader.h>
#include <dlib/queue.h>


using namespace std;

static const string sUnavailable("UNAVAILABLE");
static const string sConditionUnavailable("UNAVAILABLE|||");

static const string sAvailable("AVAILABLE");
static dlib::logger sLogger("agent");

/* Agent public methods */
Agent::Agent(const string& configXmlPath, int aBufferSize, int aMaxAssets, int aCheckpointFreq)
  : mPutEnabled(false), mLogStreamData(false)
{
  try
  {
    // Load the configuration for the Agent
    mXmlParser = new XmlParser();
    mDevices = mXmlParser->parseFile(configXmlPath);  
    std::vector<Device *>::iterator device;
    std::set<std::string> uuids;
    for (device = mDevices.begin(); device != mDevices.end(); ++device) 
    {
      if (uuids.count((*device)->getUuid()) > 0)
          throw runtime_error("Duplicate UUID: " + (*device)->getUuid());
      
      uuids.insert((*device)->getUuid());
      (*device)->resolveReferences();
    }
  }
  catch (runtime_error & e)
  {
    sLogger << LFATAL << "Error loading xml configuration: " + configXmlPath;
    sLogger << LFATAL << "Error detail: " << e.what();
    cerr << e.what() << endl;
    throw e;
  }
  catch (exception &f)
  {
    sLogger << LFATAL << "Error loading xml configuration: " + configXmlPath;
    sLogger << LFATAL << "Error detail: " << f.what();
    cerr << f.what() << endl;
    throw f;    
  }
  
  // Grab data from configuration
  string time = getCurrentTime(GMT_UV_SEC);
  
  // Unique id number for agent instance
  mInstanceId = getCurrentTimeInSec();
  
  // Sequence number and sliding buffer for data
  mSequence = 1;
  mSlidingBufferSize = 1 << aBufferSize;
  mSlidingBuffer = new sliding_buffer_kernel_1<ComponentEventPtr>();
  mSlidingBuffer->set_size(aBufferSize);
  mCheckpointFreq = aCheckpointFreq;
  mCheckpointCount = (mSlidingBufferSize / aCheckpointFreq) + 1;
  
  // Asset sliding buffer
  mMaxAssets = aMaxAssets;
  
  // Create the checkpoints at a regular frequency
  mCheckpoints = new Checkpoint[mCheckpointCount];
  
  // Mutex used for synchronized access to sliding buffer and sequence number
  mSequenceLock = new dlib::mutex;
  mAssetLock = new dlib::mutex;
  
  // Add the devices to the device map and create availability and 
  // asset changed events if they don't exist
  std::vector<Device *>::iterator device;
  for (device = mDevices.begin(); device != mDevices.end(); ++device) 
  {
    mDeviceMap[(*device)->getName()] = *device;
    
    // Make sure we have two device level data items:
    // 1. Availability
    // 2. AssetChanged
    if ((*device)->getAvailability() == NULL)
    {
      // Create availability data item and add it to the device.
      std::map<string,string> attrs;
      attrs["type"] = "AVAILABILITY";
      attrs["id"] = (*device)->getId() + "_avail";
      attrs["category"] = "EVENT";
      
      DataItem *di = new DataItem(attrs);
      di->setComponent(*(*device));
      (*device)->addDataItem(*di);
      (*device)->addDeviceDataItem(*di);
      (*device)->mAvailabilityAdded = true;
    }
    
    int major, minor;
    char c;
    stringstream ss(XmlPrinter::getSchemaVersion());
    ss >> major >> c >> minor;
    if ((*device)->getAssetChanged() == NULL && (major >= 1 || (major == 1 && minor >= 2)))
    {
      // Create availability data item and add it to the device.
      std::map<string,string> attrs;
      attrs["type"] = "ASSET_CHANGED";
      attrs["id"] = (*device)->getId() + "_asset_chg";
      attrs["category"] = "EVENT";
      
      DataItem *di = new DataItem(attrs);
      di->setComponent(*(*device));
      (*device)->addDataItem(*di);
      (*device)->addDeviceDataItem(*di);
    }
  }
  
  // Reload the document for path resolution
  mXmlParser->loadDocument(XmlPrinter::printProbe(mInstanceId, mSlidingBufferSize, 
                                                  mMaxAssets,
                                                  mAssets.size(),
                                                  mSequence, mDevices));
   
  /* Initialize the id mapping for the devices and set all data items to UNAVAILABLE */
  for (device = mDevices.begin(); device != mDevices.end(); ++device) 
  {
    const std::map<string, DataItem*> &items = (*device)->getDeviceDataItems();
    std::map<string, DataItem *>::const_iterator item;

    for (item = items.begin(); item != items.end(); ++item)
    {
      // Check for single valued constrained data items.
      DataItem *d = item->second;
      const string *value = &sUnavailable;
      if (d->isCondition()) {
        value = &sConditionUnavailable;
      } else if (d->hasConstraints()) { 
        std::vector<std::string> &values = d->getConstrainedValues();
        if (values.size() == 1)
          value = &values[0];
      }
      
      addToBuffer(d, *value, time);
      if (mDataItemMap.count(d->getId()) == 0)
        mDataItemMap[d->getId()] = d;
      else {
        sLogger << LFATAL << "Duplicate DataItem id " << d->getId() <<
        " for device: " << (*device)->getName() << " and data item name: " <<
        d->getName();
        exit(1);
      }
    }
  }
}

Agent::~Agent()
{
  delete mSlidingBuffer;
  delete mSequenceLock;
  delete mAssetLock;
  delete mXmlParser;<commit_msg>Fixed formatting.<commit_after>/*
 * Copyright Copyright 2012, System Insights, Inc.
 *
 *    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.
 */

#include "agent.hpp"
#include "dlib/logger.h"
#include <sys/stat.h>
#include <fcntl.h>
#include <sstream>
#include <stdexcept>
#include <dlib/tokenizer.h>
#include <dlib/misc_api.h>
#include <dlib/array.h>
#include <dlib/dir_nav.h>
#include <dlib/config_reader.h>
#include <dlib/queue.h>


using namespace std;

static const string sUnavailable("UNAVAILABLE");
static const string sConditionUnavailable("UNAVAILABLE|||");

static const string sAvailable("AVAILABLE");
static dlib::logger sLogger("agent");

/* Agent public methods */
Agent::Agent(const string& configXmlPath, int aBufferSize, int aMaxAssets, int aCheckpointFreq)
  : mPutEnabled(false), mLogStreamData(false)
{
  try
  {
    // Load the configuration for the Agent
    mXmlParser = new XmlParser();
    mDevices = mXmlParser->parseFile(configXmlPath);  
    std::vector<Device *>::iterator device;
    std::set<std::string> uuids;
    for (device = mDevices.begin(); device != mDevices.end(); ++device) 
    {
      if (uuids.count((*device)->getUuid()) > 0)
          throw runtime_error("Duplicate UUID: " + (*device)->getUuid());
      
      uuids.insert((*device)->getUuid());
      (*device)->resolveReferences();
    }
  }
  catch (runtime_error & e)
  {
    sLogger << LFATAL << "Error loading xml configuration: " + configXmlPath;
    sLogger << LFATAL << "Error detail: " << e.what();
    cerr << e.what() << endl;
    throw e;
  }
  catch (exception &f)
  {
    sLogger << LFATAL << "Error loading xml configuration: " + configXmlPath;
    sLogger << LFATAL << "Error detail: " << f.what();
    cerr << f.what() << endl;
    throw f;    
  }
  
  // Grab data from configuration
  string time = getCurrentTime(GMT_UV_SEC);
  
  // Unique id number for agent instance
  mInstanceId = getCurrentTimeInSec();
  
  // Sequence number and sliding buffer for data
  mSequence = 1;
  mSlidingBufferSize = 1 << aBufferSize;
  mSlidingBuffer = new sliding_buffer_kernel_1<ComponentEventPtr>();
  mSlidingBuffer->set_size(aBufferSize);
  mCheckpointFreq = aCheckpointFreq;
  mCheckpointCount = (mSlidingBufferSize / aCheckpointFreq) + 1;
  
  // Asset sliding buffer
  mMaxAssets = aMaxAssets;
  
  // Create the checkpoints at a regular frequency
  mCheckpoints = new Checkpoint[mCheckpointCount];
  
  // Mutex used for synchronized access to sliding buffer and sequence number
  mSequenceLock = new dlib::mutex;
  mAssetLock = new dlib::mutex;
  
  // Add the devices to the device map and create availability and 
  // asset changed events if they don't exist
  std::vector<Device *>::iterator device;
  for (device = mDevices.begin(); device != mDevices.end(); ++device) 
  {
    mDeviceMap[(*device)->getName()] = *device;
    
    // Make sure we have two device level data items:
    // 1. Availability
    // 2. AssetChanged
    if ((*device)->getAvailability() == NULL)
    {
      // Create availability data item and add it to the device.
      std::map<string,string> attrs;
      attrs["type"] = "AVAILABILITY";
      attrs["id"] = (*device)->getId() + "_avail";
      attrs["category"] = "EVENT";
      
      DataItem *di = new DataItem(attrs);
      di->setComponent(*(*device));
      (*device)->addDataItem(*di);
      (*device)->addDeviceDataItem(*di);
      (*device)->mAvailabilityAdded = true;
    }
    
    int major, minor;
    char c;
    stringstream ss(XmlPrinter::getSchemaVersion());
    ss >> major >> c >> minor;
    if ((*device)->getAssetChanged() == NULL && (major >= 1 || (major == 1 && minor >= 2)))
    {
      // Create availability data item and add it to the device.
      std::map<string,string> attrs;
      attrs["type"] = "ASSET_CHANGED";
      attrs["id"] = (*device)->getId() + "_asset_chg";
      attrs["category"] = "EVENT";
      
      DataItem *di = new DataItem(attrs);
      di->setComponent(*(*device));
      (*device)->addDataItem(*di);
      (*device)->addDeviceDataItem(*di);
    }
  }
  
  // Reload the document for path resolution
  mXmlParser->loadDocument(XmlPrinter::printProbe(mInstanceId, mSlidingBufferSize, 
                                                  mMaxAssets,
                                                  mAssets.size(),
                                                  mSequence, mDevices));
   
  /* Initialize the id mapping for the devices and set all data items to UNAVAILABLE */
  for (device = mDevices.begin(); device != mDevices.end(); ++device) 
  {
    const std::map<string, DataItem*> &items = (*device)->getDeviceDataItems();
    std::map<string, DataItem *>::const_iterator item;

    for (item = items.begin(); item != items.end(); ++item)
    {
      // Check for single valued constrained data items.
      DataItem *d = item->second;
      const string *value = &sUnavailable;
      if (d->isCondition()) {
        value = &sConditionUnavailable;
      } else if (d->hasConstraints()) { 
        std::vector<std::string> &values = d->getConstrainedValues();
        if (values.size() == 1)
          value = &values[0];
      }
      
      addToBuffer(d, *value, time);
      if (mDataItemMap.count(d->getId()) == 0)
        mDataItemMap[d->getId()] = d;
      else {
        sLogger << LFATAL << "Duplicate DataItem id " << d->getId() <<
                  " for device: " << (*device)->getName() << " and data item name: " <<
                  d->getName();
        exit(1);
      }
    }
  }
}

Agent::~Agent()
{
  delete mSlidingBuffer;
  delete mSequenceLock;
  delete mAssetLock;
  delete mXmlParser;
{'fraction_non_alphanumeric': 0.13833251913369157, 'fraction_numerical': 0.0035824784237094938, 'mean_word_length': 2.888255777144666, 'pattern_counts': {'":': 0, '<': 129, '<?xml version=': 0, '>': 117, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 4}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6367499', 'n_tokens_mistral': 3761, 'n_tokens_neox': 3706, 'n_words': 1020}