<commit_before>/**
 * @file
 * @author <NAME> <<EMAIL>
 * @author <NAME> <<EMAIL>
 * @version 0.1
 *
 * @section DESCRIPTION
 *
 * This is an implementation of a standard Table-based QLearner
 */

#include "Observer/RealtimeObserver.h"
#include <stdlib.h>
#include <vector>
#include <deque>
#include <utility>
#include "Primitives/QLearner/QTable.h"
#include "Primitives/QLearner/State.h"
#include "Common/Utils.h"
namespace Observation {

using std::pair;
using std::vector;
using std::string;
using Primitives::State;
using Primitives::QTable;
using Utils::Log;

bool RealtimeObserver::Observe(Task* task, double duration) {
    duration_ = duration;
    return Observe(task);
}

bool RealtimeObserver::Observe(Task* task) {
  FILE *log_stream = stderr;
  double LEARNING_RATE = 0.2;
  double DISCOUNT_FACTOR = 0.9;

  if (sensors_.size() == 0) {
    Log(log_stream, DEBUG, (string("No sensors defined on observer.")).c_str());
    return false;
  }
  if (primitives_.size() == 0) {
    Log(log_stream, DEBUG,
    (string("No primitives defined on observer.").c_str()));
    return false;
  }

  is_observing_ =  true;

  vector<double> unified_frame;

  vector<ObservablePrimitive *> primitives;
  // Pair of timestamps and primitive pointers
  vector<pair<double, ObservablePrimitive *> > timed_out_primitives;

  // Setup observable primitives based on skills lis
  vector<QLearner *>::iterator p_iter;
  for (p_iter = primitives_.begin(); p_iter != primitives_.end(); ++p_iter) {
    ObservablePrimitive *op =
      new ObservablePrimitive((*p_iter)->get_name(), (*p_iter));
    primitives.push_back(op);
  }

  timespec time;
  clock_gettime(CLOCK_REALTIME, &time);
  double cur_time_ms = (time.tv_sec * 1000.) +
                       (static_cast<double>(time.tv_nsec) / 1000000.);


  int cur_frame = 0;
  double start_time = cur_time_ms;
  double end_time = start_time + duration_;
  double last_frame_time_ms = 0.;

  timeline_.clear();
  frames_.clear();

  while (is_observing_ && cur_time_ms < end_time) {
    unified_frame.clear();
    bool clear_hit_states = false;

    while (sampling_rate_ > cur_time_ms - last_frame_time_ms) {
      clock_gettime(CLOCK_REALTIME, &time);
      cur_time_ms = (time.tv_sec * 1000.) +
                           (static_cast<double>(time.tv_nsec) / 1000000.);
    }
    last_frame_time_ms = cur_time_ms;

    // Retrieve latest sensor d
    // char buf[1024];
    // snprintf(buf,1024,"Capturing Frame %d with %ld sensors...",cur_frame,
    //          sensors_.size());
    // Log(log_stream, DEBUG, buf);

    unsigned int sens_iter;
    for (sens_iter = 0; sens_iter < sensors_.size();
         ++sens_iter) {
      Sensor *sensor = sensors_[sens_iter];

      const double *values = sensor->GetValues();
      // snprintf(buf,1024,"... contains %d values ...",
      //    sensor->get_num_values());
      // Log(log_stream,DEBUG,buf);
      for (int i = 0; i < sensor->get_num_values(); ++i) {
        double rand_factor = static_cast<double>(rand() % 6) / 100. + 0.97;
        unified_frame.push_back(values[i] * rand_factor);
      }
    }

    // snprintf(buf,1024,"...inc frame has size %ld", unified_frame.size());
    // Log(log_stream,DEBUG,buf);

    frames_.push_back(unified_frame);  // Store incoming frame d
    vector<pair<double, string> > timeline_entry;
    timeline_entry.push_back(pair<double, string>(0.3, "Unknown"));
    timeline_.push_back(timeline_entry);  // Add default guess

    // Log(log_stream, DEBUG,
    //  string("Done capturing frame. Beginning primitive loop").c_str());

    vector<ObservablePrimitive *>::iterator p_iter;
    for (p_iter = primitives.begin(); p_iter != primitives.end();
         ++p_iter) {
      ObservablePrimitive *p = *p_iter;

      /*char buf[1024];
       snprintf(buf,1024,"Analyzing primitive %s with state count %ld",
        p->name.c_str(), p->q_learner->get_q_table()->get_states().size());
      Log(log_stream, DEBUG, buf); */


      // QTable *qtable = p->q_learner->get_q_table();


      // @TODO Check for timeout status- if yes, then continue
      //                                 if no, check if should be released

      // @TODO Make sure the beginning of p's hit_states starts within the
      //       window of eligibility for it occurring.
      //       (cut out states beginning earlier than (now - p->duration)

      // Get current state from QTable with descriptor unified_frame
      State input_frame(unified_frame);
      // Log(log_stream,DEBUG, input_frame.to_string().c_str());

      State *current_state = p->q_learner->get_q_table()->GetState(
        input_frame, true);

      /*char buf[1024];
      snprintf(buf,1024,"Current_state has descriptorsize %ld",
        current_state->get_state_vector().size());
      Log(log_stream,DEBUG,buf);*/

      if (!current_state) {
        Log(log_stream, ERROR, "QTable failed to add state");
      }

      // If p has current_state set, get reward going from old to new state
      // if non-zero reward, then add state to hit_states for p
      State *prev_state = p->current_state;

      if (prev_state == current_state) {
        //  Log(stderr,DEBUG,"Duplicate frame.");
        //    continue;
      }

      p->current_state = current_state;


      double transition_reward = 0.;
      if (prev_state != NULL) {
        transition_reward = prev_state->GetRewardValue(current_state,
                                                       true, "");
        char buf[1024];
        snprintf(buf, sizeof(buf), "...State transition %d has reward %g. "
                "Prev_state has %ld"
                " outbound connections. Descriptor size %ld",
                cur_frame, transition_reward,
                prev_state->get_reward().size(),
                prev_state->get_state_vector().size());
        /*Log(log_stream,DEBUG,string(
          prev_state->to_string() + " to " + current_state->to_string()
          ).c_str());*/
      } else {
        continue;
      }

      if (transition_reward > 0.) {
        p->hit_states.push_back(pair<double, State*>(
          cur_time_ms, current_state));
      } else {
        // Log(log_stream, ERROR,
        //    "...Zero reward transition on training data..?");
        prev_state->set_reward(current_state, "base", -1);
      }

      // If duration represented by hit_states is greater than 50% of the
      // anticipated duration of the primitive and number of frames in
      // hit_states between hit_states' start and now is less than
      // 75% * number of frames elapsed between hit_states' start and now
      //  --> add p to the timeout pile
      double hit_state_duration = 0.;
      double first_hit_timestamp = 0.;
      if (p->hit_states.size() > 1) {
        first_hit_timestamp = p->hit_states[0].first;
        hit_state_duration = p->hit_states[p->hit_states.size()-1].first
                             - first_hit_timestamp;<commit_msg>Removed busywait in favor of more system-friendly nanosleep call.<commit_after>/**
 * @file
 * @author <NAME> <<EMAIL>
 * @author <NAME> <<EMAIL>
 * @version 0.1
 *
 * @section DESCRIPTION
 *
 * This is an implementation of a standard Table-based QLearner
 */

#include "Observer/RealtimeObserver.h"
#include <stdlib.h>
#include <unistd.h>
#include <vector>
#include <deque>
#include <utility>
#include "Primitives/QLearner/QTable.h"
#include "Primitives/QLearner/State.h"
#include "Common/Utils.h"
namespace Observation {

using std::pair;
using std::vector;
using std::string;
using Primitives::State;
using Primitives::QTable;
using Utils::Log;

bool RealtimeObserver::Observe(Task* task, double duration) {
    duration_ = duration;
    return Observe(task);
}

bool RealtimeObserver::Observe(Task* task) {
  FILE *log_stream = stderr;
  double LEARNING_RATE = 0.2;
  double DISCOUNT_FACTOR = 0.9;

  if (sensors_.size() == 0) {
    Log(log_stream, DEBUG, (string("No sensors defined on observer.")).c_str());
    return false;
  }
  if (primitives_.size() == 0) {
    Log(log_stream, DEBUG,
    (string("No primitives defined on observer.").c_str()));
    return false;
  }

  is_observing_ =  true;

  vector<double> unified_frame;

  vector<ObservablePrimitive *> primitives;
  // Pair of timestamps and primitive pointers
  vector<pair<double, ObservablePrimitive *> > timed_out_primitives;

  // Setup observable primitives based on skills list
  vector<QLearner *>::iterator p_iter;
  for (p_iter = primitives_.begin(); p_iter != primitives_.end(); ++p_iter) {
    ObservablePrimitive *op =
      new ObservablePrimitive((*p_iter)->get_name(), (*p_iter));
    primitives.push_back(op);
  }

  struct timespec time;
  clock_gettime(CLOCK_REALTIME, &time);
  double cur_time_ms = (time.tv_sec * 1000.) +
                       (static_cast<double>(time.tv_nsec) / 1000000.);


  int cur_frame = 0;
  double start_time = cur_time_ms;
  double end_time = start_time + duration_;
  double last_frame_time_ms = 0.;

  timeline_.clear();
  frames_.clear();

  // Observe for pre-specified duration
  while (is_observing_ && cur_time_ms < end_time) {
    unified_frame.clear();
    bool clear_hit_states = false;


    // Wait for next sensor update
    double wait_time_ms = (sampling_rate_ - (cur_time_ms - last_frame_time_ms));
    if (wait_time_ms > 0) {
      struct timespec wait_ts;
      wait_ts.tv_sec = 0;
      wait_ts.tv_nsec = wait_time_ms * 1000000;
      nanosleep(&wait_ts, NULL);
    }

    /* Busywait
    while (sampling_rate_ > cur_time_ms - last_frame_time_ms) {
      sleep()
      clock_gettime(CLOCK_REALTIME, &time);
      cur_time_ms = (time.tv_sec * 1000.) +
                           (static_cast<double>(time.tv_nsec) / 1000000.);
    } */
    
    clock_gettime(CLOCK_REALTIME, &time);
    cur_time_ms = (time.tv_sec * 1000.) +
                         (static_cast<double>(time.tv_nsec) / 1000000.);
    last_frame_time_ms = cur_time_ms;

    #ifdef VERBOSE_MODE
    Retrieve latest sensor d
    char buf[1024];
    snprintf(buf, sizeof(buf), 
            "Capturing Frame %d with %ld sensors...", cur_frame,
             sensors_.size());
    Log(log_stream, DEBUG, buf);
    #endif
    
    unsigned int sens_iter;
    for (sens_iter = 0; sens_iter < sensors_.size();
         ++sens_iter) {
      Sensor *sensor = sensors_[sens_iter];

      const double *values = sensor->GetValues();

      #ifdef VERBOSE_MODE
        snprintf(buf, sizeof(buf), "... contains %d values ...",
           sensor->get_num_values());
        Log(log_stream, DEBUG, buf);
      #endif
      
      for (int i = 0; i < sensor->get_num_values(); ++i) {
        double rand_factor = static_cast<double>(rand() % 6) / 100. + 0.97;
        unified_frame.push_back(values[i] * rand_factor);
      }
    }

    frames_.push_back(unified_frame);  // Store incoming frame d
    vector<pair<double, string> > timeline_entry;
    timeline_entry.push_back(pair<double, string>(0.3, "Unknown"));
    timeline_.push_back(timeline_entry);  // Add default guess
    #ifdef VERBOSE_MODE
      Log(log_stream, DEBUG,
          string("Done capturing frame. Beginning primitive loop").c_str());
    #endif

    vector<ObservablePrimitive *>::iterator p_iter;
    for (p_iter = primitives.begin(); p_iter != primitives.end();
         ++p_iter) {
      ObservablePrimitive *p = *p_iter;

      QTable *qtable = p->q_learner->get_q_table();

      #ifdef VERBOSE_MODE
        char buf[1024];
         snprintf(buf, sizeof(buf), 
                  "Analyzing primitive %s with state count %ld",
                  p->name.c_str(), 
                  qtable->get_states().size());
        Log(log_stream, DEBUG, buf);
      #endif



      // @TODO Check for timeout status- if yes, then continue
      //                                 if no, check if should be released

      // @TODO Make sure the beginning of p's hit_states starts within the
      //       window of eligibility for it occurring.
      //       (cut out states beginning earlier than (now - p->duration)

      // Get current state from QTable with descriptor unified_frame
      State input_frame(unified_frame);

      State *current_state = qtable->GetState(
        input_frame, true);

      if (!current_state) {
        Log(log_stream, ERROR, "QTable failed to add state");
      }

      // If p has current_state set, get reward going from old to new state
      // if non-zero reward, then add state to hit_states for p
      State *prev_state = p->current_state;

      if (prev_state == current_state) {
        Log(stderr, DEBUG, "Duplicate frame received in RealtimeObserver.");
        continue;
      }

      p->current_state = current_state;


      double transition_reward = 0.;
      if (prev_state != NULL) {
        transition_reward = prev_state->GetRewardValue(current_state,
                                                       true, "");
        #ifdef VERBOSE_MODE
          char buf[1024];
          snprintf(buf, sizeof(buf), "...State transition %d has reward %g. "
                  "Prev_state has %ld"
                  " outbound connections. Descriptor size %ld",
                  cur_frame, transition_reward,
                  prev_state->get_reward().size(),
                  prev_state->get_state_vector().size());
          Log(log_stream, DEBUG, string(
            prev_state->to_string() + " to " + 
            current_state->to_string()).c_str());
        #endif
      } else {
        continue;
      }

      if (transition_reward > 0.) {
        p->hit_states.push_back(pair<double, State*>(
          cur_time_ms, current_state));
      } else {
        // Log(log_stream, ERROR,
        //    "...Zero reward transition on training data..?");
        prev_state->set_reward(current_state, "base", -1);
      }

      // If duration represented by hit_states is greater than 50% of the
      // anticipated duration of the primitive and number of frames in
      // hit_states between hit_states' start and now is less than
      // 75% * number of frames elapsed between hit_states' start and now
      //  --> add p to the timeout pile
      double hit_state_duration = 0.;
      double first_hit_timestamp = 0.;
      if (p->hit_states.size() > 1) {
        first_hit_timestamp = p->hit_states[0].first;
        hit_state_duration = p->hit_states[p->hit_states.size()-1].first
                             - first_hit_timestamp;
{'fraction_non_alphanumeric': 0.1243220692532332, 'fraction_numerical': 0.012376581838409123, 'mean_word_length': 2.583208769307424, 'pattern_counts': {'":': 0, '<': 57, '<?xml version=': 0, '>': 107, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6021096', 'n_tokens_mistral': 4704, 'n_tokens_neox': 4386, 'n_words': 1092}