eis/eqpalg/algs/exp_base.cpp

#include <eqpalg/algs/exp_base.h>
#include <eqpalg/exp_macro/get_macro_replaced_exp.h>
#include <eqpalg/table_struct/t_sample_mag.h>
#include <eqpalg/utility/build_alarm_info.h>
#include <float.h>
#include <memory>
#include <mix_cc/ihyper_db/utility.h>
#include <mix_cc/sql.h>
#include <mix_cc/sql/database/db2_t.h>
#include <mix_cc/type/mix_time.h>
#include <shm/TaskData.h>
#include <unordered_map>
#include <vector>
extern ProcessType glob_process_type;
ExpBase::ExpBase(const string &name, const mix_cc::json &rule_json,
                 const string &ruleId, size_t exp_type)
    : AlgBase(name, rule_json, ruleId), exp_type_(exp_type) {
  logger_.reset(
      new LOG("ExpBase-" + std::to_string(exp_type) + ":" + rule_name_,
              AUTO_CATCH_PID));
}
ExpBase::~ExpBase() {
}

int ExpBase::init() {

  int ret = 0;
  try {
    ret += AlgBase::init(); /*1.tag点；2.执行周期*/
    this->con_monitor_.setThreshold(100);
    // 重新载入数据源配置信息
    ret += this->reload_config_data_source(); /*3.数据源*/
    // 在载入数据源信息完成后，载入表达式配置之前，必须刷新变量，把变量信息初始化到mm_vars内
    if (glob_process_type == ProcessType::kMon ||
        glob_process_type == ProcessType::kTask) {
      ret += expr_engine_->firstFill(data_source_, now_time_, query_time_range_); /*4.数据项*/
    }

    // 必须在刷新变量后，才可以初始化表达式
    ret += this->reload_config_exp_act(); /*5.动作表达式*/
    if (feedback_mode_) {
      ret += this->reload_config_exp_feedback(); /*6.反馈表达式*/
      fb_fsm_.configure(keep_mode_, time_out_);
    }

    doInitExtend();
  } catch (const std::exception &e) {
    logger_->Debug() << e.what() << ".location:" << BOOST_CURRENT_LOCATION
                     << endl;
  }
  if (ret == 0) {
    this->exp_is_wrong_ = false;
  }
  return ret;
}

AlarmInfo ExpBase::exec_mon() {
  AlarmInfo out_alarm{};
  /*----规则配置参数检查----start*/
  if (this->exp_is_wrong_) {
    if (!exp_wrong_is_alarmed_) {
      std::string msg = rule_name_ + "配置参数加载异常，请检规则配置！";
      mix_cc::time_range_t time_range(chrono::system_clock::now() - 20s,
                                      chrono::system_clock::now());
      if (!this->error_code_list_.empty()) {
        for (auto item : this->error_code_list_) {
          msg +=
              ErrorCode::ErrorLocationDescription.at(int(item.error_location));
          msg += ErrorCode::ErrorTypeDescription.at(int(item.error_type));
          msg += "。";
        }
      }
      msg += this->error_message_str_;

      auto alarm_task = utility::build_alarm_info(
          MsgLevel::ERROR, rule_id_, rule_name_, "EXPACT", msg, time_range);

      if (alarm_task.alarmed) {
        last_alarm_time_ -= minutes(5);
        alarm_poster_.alarm(alarm_task, &last_alarm_time_);
      }
      exp_wrong_is_alarmed_ = true;
    }
    return out_alarm;
  }
  /*----规则配置参数检查----end*/

  /*----定时载入置信区间---错峰调用--start*/
  if (now_time_ - last_load_time_ >
      (minutes(30) + save_interval_ms_ + rule_state_update_interval_ms_)) {
    this->reload_ci_dist();
  }
  /*----定时载入置信区间-----end*/

  try {
    this->refresh_now_time();
    // 根据数据来源种类，决定最后的执行过程
    switch (data_source_) {
    // 如果是共享内存，只需要执行当前周期的数据和表达式
    // 并判断是否报警
    case DataSource::MEMORY: {
      try {
        expr_engine_->refreshFromMemory(now_time_, query_time_range_);
      } catch (const std::exception &e) {
        logger_->Error() << rule_id_ << ":refreshFromMemory()异常！"
                         << e.what() << ",location:" << BOOST_CURRENT_LOCATION
                         << endl;
        return out_alarm;
      }
      if (this->refresh_counts_ < 11) {
        /*首次载入，刷mm_vars 11次 刷新mm_vars[pv1_10]*/
        this->refresh_counts_++;
      } else {
        out_alarm = mon_proc();
      }
    } break;
    // 如果是ihyperDB
    case DataSource::IHDB:
      // 需要先重新载入ihyperDB的缓存数据
      refresh_ihd_cache();
      // 然后刷以此把缓存中的数据取出
      for (auto i = 0; i < queried_data_.rows(); i++) {
        expr_engine_->refreshFromIhdRow(i, queried_data_, queried_time_, now_time_, query_time_range_);
        if (!out_alarm.alarmed) {
          auto tmp = mon_proc();
          if (tmp.alarmed) {
            out_alarm = (tmp);
          }
        }
      }
      break;
    default:
      break;
    }
  } catch (const std::exception &e) {
    gb_logger_->log_exception(e);
  }
  return out_alarm;
}

std::vector<AlarmInfo> ExpBase::exec_task(mix_cc::time_range_t time_range) {

  std::vector<AlarmInfo> out_alarms;
  /*统计类，需要单独处理数据*/
  if (this->exp_type_ == ExpType::Bound ||
      this->exp_type_ == ExpType::CondBound ||
      this->exp_type_ == ExpType::BoundHoldTime) {
    this->refresh_counts_ = 0;
    try {
      /**
       * 1.获取样本id，数据查询时间范围，查数据
       * 2.计算参数
       * 3.存数据至 db2 T_SAMPLE_STAT表
       * 4.存样本管理 db2 T_SAMPLE_MAG
       */
      logger_->Debug()
          << "stime:"
          << mix_cc::mix_time_t(time_range.get_left()).to_formatted_time()
          << ",endtime:"
          << mix_cc::mix_time_t(time_range.get_right()).to_formatted_time()
          << endl;
      std::string sample_id = this->get_id(time_range);
      stat_collector_.ensureInitialized();

      logger_->Debug() << "reset_data()------1" << endl;
      if ((time_range.get_right() - time_range.get_left()) <
          this->query_interval_time_) {
        this->query_time_range_ = time_range;
        task_mon_pro(); /*累计数据*/
      } else {
        for (auto t = time_range.get_left(); t < time_range.get_right();
             t += query_interval_time_) {
          this->query_time_range_.set_left(t);
          if (time_range.get_right() - t > query_interval_time_) {
            this->query_time_range_.set_right(t + query_interval_time_);
          } else {
            this->query_time_range_.set_right(time_range.get_right());
          }
          task_mon_pro(); /*累计数据*/
        }
      }
      logger_->Debug() << "reset_data()------1" << endl;
      stat_collector_.resetData(); /*参数重置*/
      logger_->Debug() << "reset_data()------2" << endl;
      logger_->Debug() << "task_seq:" << task_seq << std::endl;
      auto &data_record = TaskShm::TaskRecordPtr.get()
                              ->
                              operator[](exp_type_ * 1000 + task_seq)
                              .data_record;
      logger_->Debug() << "dataSize:" << data_record.size() << std::endl;
      for (size_t j = 0; j < data_record.size(); j++) {
        double dataJ = data_record[j];
        stat_collector_.distAdd(dataJ);
      }
      logger_->Debug() << "reset_data()------3" << endl;
      auto store_res = stat_collector_.taskStoreDb2(sample_id);
      logger_->Debug() << "reset_data()------4" << endl;
      this->sample_result_ = stat_collector_.getSampleStatStr();
      this->update_t_sample_mag(store_res);
      this->alarm_poster_.zmqp_send(912, this->sample_result_);
      TaskShm::TaskRecordPtr.get()->erase(exp_type_ * 1000 + task_seq);
      logger_->Debug() << "|" << rule_id_ << "|" << rule_name_ << "|"
                       << "统计完成！info:" << sample_result_ << std::endl;
    } catch (const std::exception &e) {
      logger_->Error() << "ExpBase::exec_task:" << e.what()
                       << ",location:" << BOOST_CURRENT_LOCATION << endl;
      return out_alarms;
    }
    return out_alarms;
  }

  /*其他算法*/
  try {
    // 刷一遍 mmvar 防止p1_3数据错误带来的问题
    now_time_ = time_range.get_left() - query_interval_time_;
    if (this->delay_time_ > this->ihd_min_time_particles_ &&
        (this->delay_time_.count() % this->ihd_min_time_particles_.count() ==
         0)) {
      refresh_ihd_cache(this->delay_time_);
    } else {
      refresh_ihd_cache();
    }
    for (auto i = 0; i < queried_data_.rows(); i++) {
      expr_engine_->refreshFromIhdRow(i, queried_data_, queried_time_, now_time_, query_time_range_);
    }
    // 对每个ihdb 查询周期的数据，进行处理
    for (auto now_time = time_range.get_left();
         now_time <= time_range.get_right(); now_time += query_interval_time_) {
      this->now_time_ = now_time;
      if (this->delay_time_ > this->ihd_min_time_particles_ &&
          (this->delay_time_.count() % this->ihd_min_time_particles_.count() ==
           0)) {
        refresh_ihd_cache(this->delay_time_);
      } else {
        refresh_ihd_cache();
      }
      for (auto i = 0; i < queried_data_.rows(); i++) {
        expr_engine_->refreshFromIhdRow(i, queried_data_, queried_time_, now_time_, query_time_range_);
        auto tmp = mon_proc();
        if (tmp.alarmed) {
          out_alarms.push_back(tmp);
        }
      }
    }
    gb_logger_->log_info("本次测试报警数量：" +
                         std::to_string(out_alarms.size()));
    auto msg =
        rule_name_ + "本次测试报警数量：" + std::to_string(out_alarms.size());
    auto alarm_task = utility::build_alarm_info(
        MsgLevel::INFO, rule_id_, rule_name_, "EXPACT", msg, time_range);
    out_alarms.clear();
    out_alarms.push_back(alarm_task);
  } catch (const std::exception &e) {
    gb_logger_->log_exception(e);
  }

  return out_alarms;
}

AlarmInfo ExpBase::mon_proc() {
  expr_engine_->autoResetFunVars();
  try {
    return doMonProc();
  } catch (const std::exception &e) {
    gb_logger_->log_error("rule_name:" + rule_name_ + ",error:" + e.what());
    return AlarmInfo{};
  }
}
bool ExpBase::get_cycled_cron() {
  auto now = system_clock::now();
  if ((now - last_run_start_time_) > std::chrono::minutes(1)) {
    last_run_start_time_ = now;
  } else {
    return false;
  }
  return true;
}

mix_cc::json ExpBase::exec_cron() {
  if (this->get_cycled_cron()) {
    if (this->exp_type_ == ExpType::Bound ||
        this->exp_type_ == ExpType::CondBound ||
        this->exp_type_ == ExpType::BoundHoldTime) {
      int stat_size = this->cron_proc();
      logger_->Debug() << this->rule_name_ << ",cron 统计数据量：" << stat_size
                       << std::endl;
    }
  }
  return {};
}

int ExpBase::cron_proc() {
  int size_data = 0;
  this->refresh_now_time();
  if (exp_type_ == ExpType::Bound || exp_type_ == ExpType::CondBound ||
      exp_type_ == ExpType::BoundHoldTime) {
    /*只保存有上下限的,2-监控变量-上下限；4-动作反馈-上下限；5-监控变量-上下限-持续*/
    this->rule_stat_.stat_values.clear();
    SingletonTemp<EqpStat>::GetInstance().get_stat_values(this->rule_id_,
                                                          this->rule_stat_);
    if (!this->rule_stat_.stat_values.empty()) {
      size_data = stat_collector_.processCron(
          this->rule_stat_.stat_values, now_time_,
          CronUpdateDelay, last_load_time_);
    }
    this->rule_stat_.stat_values.clear();
    this->rule_stat_.stat_values.shrink_to_fit();
  }
  return size_data;
}

int ExpBase::reload_config_exp_feedback() {
  int res = 0;
  //  获取feedback信息
  if (rule_json_.at("function").at("action_start").contains("param")) {
    //是否保持
    keep_mode_ = std::stoi(rule_json_.at("function")
                               .at("action_start")
                               .at("param")
                               .at("hold")
                               .at("value")
                               .get<std::string>());
    logger_->Debug() << "keep:" << keep_mode_ << endl;
    //超时时间
    time_out_ = milliseconds(std::stoi(rule_json_.at("function")
                                           .at("action_end")
                                           .at("param")
                                           .at("timeout")
                                           .at("value")
                                           .get<std::string>()));
    logger_->Debug() << "init timeout:" << time_out_.count() << endl;
    // 如果超时时间小于3分钟，则默认为超时时间为3分钟
    if (time_out_ < minutes(3) && time_out_ != milliseconds(-32768)) {
      time_out_ = minutes(3);
    }
    logger_->Debug() << "timeout:" << time_out_.count() << endl;

    auto tmp_exp = string(rule_json_.at("function")
                              .at("action_end")
                              .at("value")
                              .get<std::string>());
    exp_str_ = get_macro_replaced_exp(tmp_exp);

    if (exp_str_ != "") {
      int reg_ret = expr_engine_->registerExpression("feedback", exp_str_);
      if (reg_ret != 0) {
        this->error_code_list_.push_back(
            {ErrorType::CalError, ErrorLocation::FBExp});
        return -1;
      }
    }

  } else {
    time_out_ = milliseconds(600000);
  }
  // 获取监控变量信息
  auto tmp_exp =
      rule_json_.at("function").at("result").at("value").get<std::string>();
  exp_str_ = get_macro_replaced_exp(tmp_exp);

  if (exp_str_ != "") {
    int reg_ret = expr_engine_->registerExpression("result", exp_str_);
    if (reg_ret != 0) {
      this->error_code_list_.push_back(
          {ErrorType::CalError, ErrorLocation::ResultExp});
      res += reg_ret;
    }
  }

  if (exp_str_.find("time", 0) != string::npos &&
      exp_type_ == ExpType::CondBound) {
    fb_fsm_.setTimeMode(true);
    rule_stat_.unit = "ms";
  } else {
    fb_fsm_.setTimeMode(false);
  }
  return res;
}

// 重新载入数据源配置
int ExpBase::reload_config_data_source() {
  int res = 0;
  if (!rule_json_.contains("datasource")) {
    data_source_ = 0;
    logger_->Debug()
        << "默认数据源为iHyerDB！ data source[0:iHyerDB，1:memory]："
        << data_source_ << endl;
    return 0;
  }
  try {
    data_source_ =
        std::stoi(rule_json_.at("datasource").at("value").get<std::string>());
    logger_->Debug() << "data source[0:iHyerDB，1:memory]:" << data_source_
                     << endl;
  } catch (const std::exception &e) {
    gb_logger_->log_error(std::string("ExpBase::reload_config_data_source()") +
                          e.what());
    this->error_code_list_.push_back(
        {ErrorType::Empty, ErrorLocation::DataSource});
    return -1;
  }

  return res;
}

int ExpBase::reload_config_exp_act() {
  // 根据key，对不同版本的算法都进行取值
  int res = 0;
  if (rule_json_.at("function").contains("action_start")) {
    auto tmp_exp = rule_json_.at("function")
                       .at("action_start")
                       .at("value")
                       .get<std::string>();
    exp_str_ = get_macro_replaced_exp(tmp_exp);
    feedback_mode_ = true;

  } else if (rule_json_.at("function").contains("result")) {
    if (rule_json_.at("function").contains("filter_exp")) {
      auto tmp_exp = rule_json_.at("function")
                         .at("filter_exp")
                         .at("value")
                         .get<std::string>();
      exp_str_ = get_macro_replaced_exp(tmp_exp);
      feedback_mode_ = false;
      if (exp_str_ != "") {
        int reg_ret = expr_engine_->registerExpression("feedback", exp_str_);
        if (reg_ret != 0) {
          this->error_code_list_.push_back(
              {ErrorType::CalError, ErrorLocation::FBExp});
          return -1;
        }
      } else {
        logger_->Error() << "filter_exp exp_feedback_:"
                         << "为空"
                         << ",location:" << BOOST_CURRENT_LOCATION << endl;
        this->error_code_list_.push_back(
            {ErrorType::CalError, ErrorLocation::FBExp});
        return -1;
      }
    }
    auto tmp_exp =
        rule_json_.at("function").at("result").at("value").get<std::string>();
    exp_str_ = get_macro_replaced_exp(tmp_exp);

    feedback_mode_ = false;
  }

  if (exp_str_ != "") {
    int reg_ret = expr_engine_->registerExpression("act", exp_str_);
    if (reg_ret != 0) {
      this->error_code_list_.push_back(
          {ErrorType::CalError, ErrorLocation::ActExp});
      return -1;
    }
  }
  if (rule_json_.at("output").contains("error")) {
    try {
      error_str_ =
          rule_json_.at("output").at("error").at("value").get<std::string>();
    } catch (const std::exception &e) {
      logger_->Error() << "output,error出错：" << e.what()
                       << ",location:" << BOOST_CURRENT_LOCATION << endl;
      error_message_str_ += "报警内容格式异常！";
      return -1;
    }
  }
  return 0;
}

int ExpBase::reload_config_up_down() {
  try {
    if (rule_json_.at("function").at("result").contains("param")) {
      if (rule_json_.at("function").at("result").at("param").contains("unit")) {
        unit_ = rule_json_.at("function")
                    .at("result")
                    .at("param")
                    .at("unit")
                    .at("value")
                    .get<std::string>();
      }
      rule_stat_.unit = unit_;
      limit_down_ = std::stod(rule_json_.at("function")
                                  .at("result")
                                  .at("param")
                                  .at("limit_down")
                                  .at("value")
                                  .get<std::string>());
      limit_up_ = std::stod(rule_json_.at("function")
                                .at("result")
                                .at("param")
                                .at("limit_up")
                                .at("value")
                                .get<std::string>());
      this->rule_stat_.current_value = (limit_down_ + limit_up_) / 2;
      if (-32768 == (int)limit_down_ && (int)limit_up_ != (int)limit_down_) {
        this->rule_stat_.current_value = 0;
        this->detect_mode_ = DetectMode::OnlyRight;
      } else if ((-32768 == (int)limit_up_ || 32768 == (int)limit_up_ ||
                  32767 == (int)limit_up_) &&
                 (int)limit_up_ != (int)limit_down_) {
        this->detect_mode_ = DetectMode::OnlyLeft;
        this->rule_stat_.current_value = limit_down_ + 1;
      } else if (-32768 == (int)limit_up_ &&
                 (int)limit_up_ == (int)limit_down_) {
        this->detect_mode_ = DetectMode::ErrorMode;
      }
      bound_checker_.setLimits(limit_down_, limit_up_);
      ///共享内存参数
      logger_->Info() << rule_name_
                      << ",detect_mode_[0-双侧，1-仅left；2-仅right；3-错误]:"
                      << static_cast<int>(bound_checker_.detectMode()) << ",limit_down:" << limit_down_ << ","
                      << "limit_up:" << limit_up_ << std::endl;
      this->rule_stat_.limit_down = limit_down_;
      this->rule_stat_.limit_up = limit_up_;
    }
    if (rule_json_.contains("self_learning")) {
      this->dist_mode_ = rule_json_.at("self_learning").at("mode").get<int>();
      this->is_learning_ =
          rule_json_.at("self_learning").at("is_learning").get<bool>();
      logger_->Debug() << "is_learning_:" << is_learning_ << std::endl;
    }
  } catch (const std::exception &e) {
    logger_->Error() << e.what() << ",location:" << BOOST_CURRENT_LOCATION
                     << endl;
    return -1;
  }

  return 0;
}
int ExpBase::reload_config_up_down_hold_time() {
  try {
    if (rule_json_.at("function")
            .at("result")
            .at("param")
            .contains("hold_time")) {
      hold_time_ = milliseconds(std::stoi(rule_json_.at("function")
                                              .at("result")
                                              .at("param")
                                              .at("hold_time")
                                              .at("value")
                                              .get<std::string>()));
      logger_->Info() << rule_name_ << "hold_time:" << hold_time_.count()
                      << std::endl;
    }
  } catch (const std::exception &e) {
    logger_->Error() << e.what() << ",location:" << BOOST_CURRENT_LOCATION
                     << endl;
    return -1;
  }
  return 0;
}

void ExpBase::set_last_alarm_time(TimePoint time_point) {
  this->refresh_counts_ = 0;
  if (this->is_usable_) {
    expr_engine_->firstFill(data_source_, now_time_, query_time_range_);
  }
  AlgBase::set_last_alarm_time(time_point);
}

void ExpBase::set_usable(bool usable) {
  this->refresh_counts_ = 0;
  if (this->is_usable_) {
    expr_engine_->firstFill(data_source_, now_time_, query_time_range_);
  }
  AlgBase::set_usable(usable);
}

int ExpBase::reload_ci_dist() {
  if (this->exp_type_ != ExpType::Bound &&
      this->exp_type_ != ExpType::CondBound &&
      this->exp_type_ != ExpType::BoundHoldTime) {
    return 0;
  }
  double newDown, newUp;
  if (stat_collector_.reloadCiDist(newDown, newUp, now_time_, last_load_time_)) {
    this->limit_down_ = newDown;
    this->limit_up_ = newUp;
    bound_checker_.setLimits(newDown, newUp);
    this->rule_stat_.limit_down = limit_down_;
    this->rule_stat_.limit_up = limit_up_;
    this->rule_stat_.current_value = (limit_down_ + limit_up_) / 2;
    logger_->Info()
        << "更新置信区间，[" << newDown << ","
        << newUp << "]"
        << ",type[0-手动设置的区间;1-在线更新的区间；2-离线分析的区间]:"
        << this->dist_mode_ << endl;
  }
  return 0;
}

void ExpBase::query_ihd_data() {
  vector<string> ihd_tags = m_tags;
  auto queried_batch_maybe = mix_cc::ihd::make_query_batch_maybe(
      ihd_tags, query_time_range_, this->ihd_min_time_particles_);
  if (queried_batch_maybe.is_nothing()) {
    queried_time_.clear();
    queried_data_ = typename decltype(queried_batch_maybe)::type::Mat2d{};
    gb_logger_->log_error("Tag点没有查到相应的信息");
    return;
  }
  auto queried_bath = queried_batch_maybe.unsafe_get_just();
  auto result_maybe = mix_cc::ihd::read_data_with_time_maybe(&queried_bath);
  if (result_maybe.is_nothing()) {
    queried_time_.clear();
    queried_data_ = typename decltype(queried_batch_maybe)::type::Mat2d{};
    gb_logger_->log_error("Tag点没有查到相应的数据");
    return;
  }
  std::tie(queried_time_, queried_data_) = result_maybe.unsafe_get_just();
}

void ExpBase::task_mon_pro() {
  /*1.查数据
   *2.执行
   */
  query_ihd_data(); /*查询m_tags+运行前提条件(如果有，放置在最后一列)*/
  logger_->Debug() << rule_name_ << ",ihd size:" << queried_data_.size()
                   << endl;
  for (auto i = 0; i < queried_data_.rows(); i++) {
    expr_engine_->refreshFromIhdRow(i, queried_data_, queried_time_, now_time_, query_time_range_);
    if (this->refresh_counts_ < 3) {
      this->refresh_counts_++;
    } else {
      auto res = task_base_proc();
      if (res.is_valid && this->task_prr(i)) {
        TaskShm::TaskRecordPtr.get()
            ->
            operator[](exp_type_ * 1000 + task_seq)
            .data_record.push_back(res.value);
        stat_collector_.runningStatAdd(res.value);
      }
    }
  }
}
string ExpBase::get_id(mix_cc::time_range_t time_range) {
  this->sample_id_ =
      this->rule_id_.substr(3, 21) + "_" +
      std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(
                         time_range.get_left().time_since_epoch())
                         .count()) +
      "_" +
      std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(
                         time_range.get_right().time_since_epoch())
                         .count());
  return this->sample_id_;
}

TaskReturnType ExpBase::task_base_proc() {
  TaskReturnType task_return_data;
  double result_value = 0;
  try {
    // 获得是否满足前提条件表达式
    result_value = expr_engine_->evaluate("act");
    bool act_triggered = static_cast<bool>(result_value);
    // 检测是否是表达式-反馈模式
    if (feedback_mode_) {
      // 使用 FbStateMachine 替代旧的手动状态管理
      auto [fbState, needFunReset] = fb_fsm_.update(
          act_triggered, now_time_, expr_engine_->vars(), m_tags.size());

      if (needFunReset) {
        expr_engine_->markFunVarsNeedReset();
      }

      if (fbState == FbState::Started || fbState == FbState::NotHold) {
        return task_return_data;
      }

      if (fbState == FbState::InProgress) {
        bool fbCond = expr_engine_->evaluateBool("feedback");
        bool done =
            fb_fsm_.checkFeedback(fbCond, now_time_, expr_engine_->vars());
        if (done) {
          result_value = expr_engine_->evaluate("result");
          expr_engine_->printVars();
          // 并且数据合法
          if (!std::isnan(result_value)) {
            task_return_data.is_valid = true;
            task_return_data.value = result_value;
            return task_return_data;
          }
        }
        return task_return_data;
      }

      if (fbState == FbState::Timeout) {
        return task_return_data;
      }

      // Idle or Done (terminal states that auto-reset) — nothing to do
      return task_return_data;
    }
    // 表达式-样本，无需反馈
    else if (!std::isnan(result_value)) {
      try {
        if (expr_engine_->evaluateBool("feedback")) {
          task_return_data.is_valid = true;
          task_return_data.value = result_value;
          logger_->Debug() << "实时值：" << result_value << endl;
          expr_engine_->printVars();
        }
      } catch (...) {
        task_return_data.is_valid = true;
        task_return_data.value = result_value;
      }

      return task_return_data;
    }

  } catch (const std::exception &e) {
    this->logger_->Error() << "ExpBase::task_base_proc():" << e.what()
                           << ",location:" << BOOST_CURRENT_LOCATION << endl;
    return task_return_data;
  }
  return task_return_data;
}

int ExpBase::update_t_sample_mag(bool vlid) {
  T_SAMPLE_MAG tsm;
  auto update_ret = exec<db2_t, size_t>(
      update(tsm)
          .set(tsm.result() = this->sample_result_,
               tsm.type() = this->sample_type_, tsm.usable() = 0,
               tsm.starttime() =
                   mix_cc::mix_time_t(this->task_time_range_.get_left()),
               tsm.endtime() =
                   mix_cc::mix_time_t(this->task_time_range_.get_right()),
               tsm.verify() = (vlid ? 1 : 3))
          .where(tsm.sampleid() = this->sample_id_,
                 tsm.ruleId() = this->rule_id_));
  if (!update_ret.is_nothing()) {
    auto res = update_ret.unsafe_get_just();
    if (res == 0) {
      auto inset_ret = exec<db2_t, size_t>(insert_into(tsm).set(
          tsm.ruleId() = this->rule_id_, tsm.result() = this->sample_result_,
          tsm.type() = this->sample_type_, tsm.sampleid() = this->sample_id_,
          tsm.usable() = 0,
          tsm.starttime() =
              mix_cc::mix_time_t(this->task_time_range_.get_left()),
          tsm.endtime() =
              mix_cc::mix_time_t(this->task_time_range_.get_right()),
          tsm.verify() = (vlid ? 1 : 3)));
      if (inset_ret.is_nothing()) {
        logger_->Error() << "T_SAMPLE_MAG,插入数据失败" << std::endl;
        return -2;
      }
    }
  } else {
    gb_logger_->log_error("T_SAMPLE_MAG 表更新失败:" + rule_name_);
    return -1;
  }

  gb_logger_->log_info(
      "T_SAMPLE_MAG更新:" + rule_name_ +
      ",vlid[true-计算成功；false-计算失败]:" + std::to_string(vlid)) +
      ",结果:" + this->sample_result_;
  return 0;
}

bool ExpBase::detect_up_down(const double &value) {
  switch (this->detect_mode_) {
  case DetectMode::Default:
    return value < this->limit_down_ || value > limit_up_;
    break;
  case DetectMode::OnlyLeft:
    return value < limit_down_;
    break;
  case DetectMode::OnlyRight:
    return value > limit_up_;
    break;
  default:
    return false;
    break;
  }
  return false;
}

bool ExpBase::task_prr(int row) {
  if (this->prr_ == 1) {
    expr_engine_->refreshFromIhdRow(row, queried_data_, queried_time_, now_time_, query_time_range_);
    bool prr_result = expr_engine_->evaluateBool("pre_result");
    return prr_result;
  }
  return true;
}

void ExpBase::reset_dev_data() {
  if (glob_process_type == ProcessType::kMon) {
    if (exp_type_ == ExpType::Bound || exp_type_ == ExpType::CondBound ||
        exp_type_ == ExpType::BoundHoldTime) {
      stat_collector_.reset(rule_id_);
      this->reload_config_up_down(); /*7.上下限*/
      this->reload_ci_dist();
      this->last_load_time_ = chrono::system_clock::now();
      SingletonTemp<EqpStat>::GetInstance().update_static(this->rule_id_,
                                                          false);
    }
  }
}

void ExpBase::save_rule_norm_data() {
  if (feedback_mode_ && fb_fsm_.currentState() == FbState::Done ||
      ((exp_type_ == ExpType::Bound || exp_type_ == ExpType::BoundHoldTime) &&
       filter_flag_ == true)) {
    int64_t default_lt =
        mix_cc::mix_time_t(now_time_).to_milliseconds() - 1000 * 5;
    int64_t stime = mm_vars["stime"];
    int64_t lt = std::min(default_lt, stime);
    data_info_.update(lt, mix_cc::mix_time_t(now_time_).to_milliseconds());
  } else {
    if (con_monitor_.getCurrentState()) {
      AlgBase::save_rule_norm_data();
    } else {
      /*查不到数据，无法保存*/
      data_info_.update(mix_cc::mix_time_t(now_time_).to_milliseconds(),
                        mix_cc::mix_time_t(now_time_).to_milliseconds());
    }
  }
}

bool ExpBase::get_prr() {
  if (this->prr_ == 1) {
    // 变量刷新已由 exec_mon() 中的 refreshFromMemory 完成，此处只需求值
    bool prr_result = expr_engine_->evaluateBool("pre_result");
    this->now_prr_ = prr_result;

    if (!this->now_prr_) {
      fb_fsm_.forceReset();
    }
    return prr_result;
  }
  now_prr_ = true;
  return true;
}