create-grp-ecs.cxx

Go to the documentation of this file.

// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
// All rights not expressly granted are reserved.
//
// This software is distributed under the terms of the GNU General Public
// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
//
// In applying this license CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
 
#include <boost/program_options.hpp>
#include <ctime>
#include <chrono>
#include <regex>
#include <TSystem.h>
#include "DataFormatsParameters/GRPECSObject.h"
#include "DataFormatsCTP/Configuration.h"
#include "DetectorsCommonDataFormats/DetID.h"
#include "CCDB/CcdbApi.h"
#include "CCDB/BasicCCDBManager.h"
#include "CommonUtils/NameConf.h"
#include "CommonUtils/StringUtils.h"
 
using DetID = o2::detectors::DetID;
using CcdbApi = o2::ccdb::CcdbApi;
using GRPECSObject = o2::parameters::GRPECSObject;
namespace bpo = boost::program_options;
 
enum CCDBRefreshMode { NONE,
                       ASYNC,
                       SYNC };
 
int createGRPECSObject(const std::string& dataPeriod,
                       int run,
                       int runOrig, // in case of replay
                       int runTypeI,
                       int nHBPerTF,
                       const std::string& _detsReadout,
                       const std::string& _detsContinuousRO,
                       const std::string& _detsTrigger,
                       const std::string& flpList,
                       long tstart,
                       long tend,
                       long tstartCTP,
                       long tendCTP,
                       long marginAtSOR,
                       long marginAtEOR,
                       const std::string& ccdbServer = "",
                       std::string ccdbServerInp = "",
                       const std::string& metaDataStr = "",
                       CCDBRefreshMode refresh = CCDBRefreshMode::NONE)
{
  int retValGLO = 0;
  int retValRCT = 0;
  int retValGLOmd = 0;
  int retValCTP = 0;
  // substitute TRG by CTP
  std::regex regCTP(R"((^\s*|,\s*)(TRG)(\s*,|\s*$))");
  std::string detsReadout{std::regex_replace(_detsReadout, regCTP, "$1CTP$3")};
  std::string detsContinuousRO{std::regex_replace(_detsContinuousRO, regCTP, "$1CTP$3")};
  std::string detsTrigger{std::regex_replace(_detsTrigger, regCTP, "$1CTP$3")};
 
  auto detMask = DetID::getMask(detsReadout);
  if (detMask.count() == 0) {
    throw std::runtime_error("empty detectors list is provided");
  }
  if (runTypeI < 0 || runTypeI >= int(GRPECSObject::RunType::NRUNTYPES)) {
    LOGP(warning, "run type {} is not recognized, consider updating GRPECSObject.h", runTypeI);
  }
  auto runType = (GRPECSObject::RunType)runTypeI;
  auto detMaskCont = detMask & DetID::getMask(detsContinuousRO);
  auto detMaskTrig = detMask & DetID::getMask(detsTrigger);
  LOG(info) << tstart << " " << tend;
  if (tstart == 0) {
    tstart = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::high_resolution_clock::now().time_since_epoch()).count();
  }
  long tendVal = 0;
  if (tend < tstart) {
    tendVal = tstart + marginAtSOR;
  } else if (tendVal < tend) {
    tendVal = tend + marginAtEOR;
  }
  GRPECSObject grpecs;
  o2::ctp::CTPConfiguration* ctpConfig = nullptr;
  o2::ctp::CTPConfiguration ctpConfigNew;
  grpecs.setTimeStart(tstart);
  grpecs.setTimeEnd(tend);
  grpecs.setTimeStartCTP(tstartCTP);
  grpecs.setTimeEndCTP(tendCTP);
 
  if (runType == GRPECSObject::RunType::LASER && detMask[DetID::TPC] && detMaskCont[DetID::TPC]) {
    LOGP(important, "Overriding TPC readout mode to triggered for runType={}", o2::parameters::GRPECS::RunTypeNames[runTypeI]);
    detMaskCont.reset(DetID::TPC);
  }
  grpecs.setNHBFPerTF(nHBPerTF);
  grpecs.setDetsReadOut(detMask);
  grpecs.setDetsContinuousReadOut(detMaskCont);
  grpecs.setDetsTrigger(detMaskTrig);
  grpecs.setRun(run);
  grpecs.setRunType((GRPECSObject::RunType)runType);
  grpecs.setDataPeriod(dataPeriod);
  auto flpsVec = o2::utils::Str::tokenize(flpList, ',');
  for (const auto& s : flpsVec) {
    try {
      grpecs.addFLP((unsigned short)std::stoi(s));
    } catch (const std::exception& e) {
      LOG(alarm) << "could not convert string " << s << " to integer FLP ID, error : " << e.what();
    }
  }
  grpecs.print();
  std::map<std::string, std::string> metadata;
 
  auto toKeyValPairs = [&metadata](const std::string& str) {
    auto v0 = o2::utils::Str::tokenize(str, ';', true);
    for (auto& token : v0) {
      auto keyval = o2::utils::Str::tokenize(token, '=', false);
      if (keyval.size() != 2) {
        LOG(error) << "Illegal command-line key/value string: " << token;
        continue;
      }
      o2::utils::Str::trim(keyval[1]);
      LOGP(info, "Adding key {} with value {}", keyval[0], keyval[1]);
      metadata[keyval[0]] = keyval[1];
    }
  };
 
  if (ccdbServerInp.empty()) {
    ccdbServerInp = ccdbServer;
  }
  if (runOrig > 0 && runOrig != run && tend <= tstart && !ccdbServerInp.empty()) { // create CTP config
    try {
      auto& bcm = o2::ccdb::BasicCCDBManager::instance();
      bcm.setURL(ccdbServerInp);
      bcm.setFatalWhenNull(false);
      ctpConfig = bcm.getForRun<o2::ctp::CTPConfiguration>("CTP/Config/Config", runOrig);
      if (!ctpConfig) {
        throw std::runtime_error(fmt::format("Failed to access CTP/Config/Config for original run {}", runOrig));
      }
      std::string cfstr = ctpConfig->getConfigString(), srun{fmt::format("run {}", run)}, srunOrig{fmt::format("run {}", runOrig)};
      o2::utils::Str::replaceAll(cfstr, srunOrig, srun);
      ctpConfigNew.loadConfigurationRun3(cfstr);
      ctpConfigNew.setRunNumber(run);
    } catch (std::exception e) {
      LOGP(error, "Failed to create CTP/Config/Config from the original run {}, reason: {}", runOrig, e.what());
    }
  }
 
  toKeyValPairs(metaDataStr);
 
  if (!ccdbServer.empty()) {
    CcdbApi api;
 
    const std::string objPath{"GLO/Config/GRPECS"};
    api.init(ccdbServer);
    metadata["responsible"] = "ECS";
    metadata[o2::base::NameConf::CCDBRunTag.data()] = std::to_string(run);
    metadata["EOR"] = fmt::format("{}", tend);
    retValGLO = api.storeAsTFileAny(&grpecs, objPath, metadata, tstart, tendVal); // making it 1-year valid to be sure we have something
    if (retValGLO == 0) {
      LOGP(info, "Uploaded to {}/{} with validity {}:{} for SOR:{}/EOR:{}", ccdbServer, objPath, tstart, tendVal, tstart, tend);
    } else {
      LOGP(alarm, "Upload to {}/{} with validity {}:{} for SOR:{}/EOR:{} FAILED, returned with code {}", ccdbServer, objPath, tstart, tendVal, tstart, tend, retValGLO);
    }
    if ((runType == GRPECSObject::RunType::PHYSICS || runType == GRPECSObject::RunType::COSMICS) && tstart >= tend) { // also create the RCT/Info/RunInformation entry in case the run type is PHYSICS, to be finalized at EOR
      char tempChar{};
      std::map<std::string, std::string> mdRCT;
      mdRCT["SOR"] = std::to_string(tstart);
      mdRCT["EOR"] = std::to_string(tend);
      mdRCT["SOX"] = std::to_string(tstartCTP);
      mdRCT["EOX"] = std::to_string(tendCTP);
      long startValRCT = (long)run;
      long endValRCT = (long)(run + 1);
      retValRCT = api.storeAsBinaryFile(&tempChar, sizeof(tempChar), "tmp.dat", "char", "RCT/Info/RunInformation", mdRCT, startValRCT, endValRCT);
      if (retValRCT == 0) {
        LOGP(info, "Uploaded initial RCT object to {}/{} with validity {}:{}", ccdbServer, "RCT/Info/RunInformation", startValRCT, endValRCT);
      } else {
        LOGP(alarm, "Upload of initial RCT object to {}/{} with validity {}:{} FAILED, returned with code {}", ccdbServer, "RCT/Info/RunInformation", startValRCT, endValRCT, retValRCT);
      }
    }
    if (tend > tstart) {
      // override SOR version to the same limits
      metadata.erase("EOR");
      auto prevHeader = api.retrieveHeaders(objPath, metadata, tendVal + 1); // is there an object to override
      const auto itETag = prevHeader.find("ETag");
      if (itETag != prevHeader.end()) {
        std::string etag = itETag->second;
        etag.erase(remove(etag.begin(), etag.end(), '\"'), etag.end());
        LOGP(info, "Overriding run {} SOR-only version {}{}{}/{} validity to match complete SOR/EOR version validity", run, ccdbServer, ccdbServer.back() == '/' ? "" : "/", prevHeader["Valid-From"], etag);
        retValGLOmd = api.updateMetadata(objPath, {}, std::max(tstart, tendVal - 1), etag, tendVal);
        if (retValGLOmd != 0) {
          LOGP(alarm, "Overriding run {} SOR-only version {}{}{}/{} validity to match complete SOR/EOR version validity FAILED", run, ccdbServer, ccdbServer.back() == '/' ? "" : "/", prevHeader["Valid-From"], etag);
        }
      }
      if (runType == GRPECSObject::RunType::PHYSICS || runType == GRPECSObject::RunType::COSMICS) { // also storing the RCT/Info/RunInformation entry in case the run type is PHYSICS and if we are at the end of run
        char tempChar{};
        std::map<std::string, std::string> mdRCT;
        mdRCT["SOR"] = std::to_string(tstart);
        mdRCT["EOR"] = std::to_string(tend);
        mdRCT["SOX"] = std::to_string(tstartCTP);
        mdRCT["EOX"] = std::to_string(tendCTP);
        long startValRCT = (long)run;
        long endValRCT = (long)(run + 1);
        retValRCT = api.updateMetadata("RCT/Info/RunInformation", mdRCT, startValRCT);
        if (retValRCT == 0) {
          LOGP(info, "Updated RCT object to SOR:{}/EOR:{} SOX:{}/EOX:{}", tstart, tend, tstartCTP, tendCTP);
        } else {
          LOGP(alarm, "Update of RCT object to SOR:{}/EOR:{} SOX:{}/EOX:{} FAILED, returned with code {}", tstart, tend, tstartCTP, tendCTP, retValRCT);
        }
      }
    }
 
    if (ctpConfig && ctpConfigNew.getRunNumber() == run) { // create CTP config
      std::map<std::string, std::string> metadataCTP;
      metadataCTP["runNumber"] = fmt::format("{}", run);
      metadataCTP["comment"] = fmt::format("cloned from run {}", runOrig);
      retValCTP = api.storeAsTFileAny(&ctpConfigNew, "CTP/Config/Config", metadataCTP, tstart, tendVal);
      if (retValCTP == 0) {
        LOGP(info, "Uploaded to {}/{} with validity {}:{} for SOR:{}/EOR:{}, cloned from run {}", ccdbServer, "CTP/Config/Config", tstart, tendVal, tstart, tend, runOrig);
      } else {
        LOGP(alarm, "Upload to {}/{} with validity {}:{} for SOR:{}/EOR:{} (cloned from run {}) FAILED, returned with code {}", ccdbServer, "CTP/Config/Config", tstart, tendVal, tstart, tend, runOrig, retValCTP);
      }
    }
  } else { // write a local file
    auto fname = o2::base::NameConf::getGRPECSFileName();
    TFile grpF(fname.c_str(), "recreate");
    grpF.WriteObjectAny(&grpecs, grpecs.Class(), o2::base::NameConf::CCDBOBJECT.data());
    grpF.Close();
    LOGP(info, "Stored GRPECS to local file {}", fname);
    if (ctpConfig && ctpConfigNew.getRunNumber() == run) {
      std::string ctnpfname = fmt::format("CTPConfig_{}_from_{}.root", run, runOrig);
      TFile ctpF(ctnpfname.c_str(), "recreate");
      ctpF.WriteObjectAny(&ctpConfigNew, ctpConfigNew.Class(), o2::base::NameConf::CCDBOBJECT.data());
      ctpF.Close();
      LOGP(info, "Stored CTPConfig to local file {}", ctnpfname);
    }
  }
 
  if (refresh != CCDBRefreshMode::NONE && !ccdbServer.empty()) {
    auto cmd = fmt::format("curl -I -i -s \"{}{}latest/%5Cw%7B3%7D/.*/`date +%s000`/?prepare={}\"", ccdbServer, ccdbServer.back() == '/' ? "" : "/", refresh == CCDBRefreshMode::SYNC ? "sync" : "true");
    auto t0 = std::chrono::high_resolution_clock::now();
    auto res = gSystem->Exec(cmd.c_str());
    auto t1 = std::chrono::high_resolution_clock::now();
    LOGP(info, "Executed [{}] -> {} in {:.3f} s", cmd, res, std::chrono::duration_cast<std::chrono::milliseconds>(t1 - t0).count() / 1000.f);
  }
  if (retValGLO != 0 || retValRCT != 0 || retValGLOmd != 0 || retValCTP != 0) {
    return 4;
  }
  return 0;
}
 
int main(int argc, char** argv)
{
  bpo::variables_map vm;
  bpo::options_description opt_general(
    "Create GRP-ECS object and upload to the CCDB\n"
    "Usage:\n  " +
    std::string(argv[0]) +
    "");
  bpo::options_description opt_hidden("");
  bpo::options_description opt_all;
  bpo::positional_options_description opt_pos;
 
  try {
    auto add_option = opt_general.add_options();
    add_option("help,h", "Print this help message");
    add_option("period,p", bpo::value<std::string>(), "data taking period");
    add_option("run,r", bpo::value<int>(), "run number");
    add_option("run-type,t", bpo::value<int>()->default_value(int(GRPECSObject::RunType::NONE)), "run type");
    add_option("hbf-per-tf,n", bpo::value<int>()->default_value(128), "number of HBFs per TF");
    add_option("detectors,d", bpo::value<string>()->default_value("all"), "comma separated list of detectors");
    add_option("continuous,c", bpo::value<string>()->default_value("ITS,TPC,TOF,MFT,MCH,MID,ZDC,FT0,FV0,FDD,CTP"), "comma separated list of detectors in continuous readout mode");
    add_option("triggering,g", bpo::value<string>()->default_value("FT0,FV0"), "comma separated list of detectors providing a trigger");
    add_option("flps,f", bpo::value<string>()->default_value(""), "comma separated list of FLPs in the data taking");
    add_option("start-time,s", bpo::value<long>()->default_value(0), "ECS run start time in ms, now() if 0");
    add_option("end-time,e", bpo::value<long>()->default_value(0), "ECS run end time in ms, start-time+3days is used if 0");
    add_option("start-time-ctp", bpo::value<long>()->default_value(0), "run start CTP time in ms, same as ECS if not set or 0");
    add_option("end-time-ctp", bpo::value<long>()->default_value(0), "run end CTP time in ms, same as ECS if not set or 0");
    add_option("ccdb-server", bpo::value<std::string>()->default_value("http://alice-ccdb.cern.ch"), "CCDB server for upload, local file if empty");
    add_option("ccdb-server-input", bpo::value<std::string>()->default_value(""), "CCDB server for inputs (if needed, e.g. CTPConfig), dy default ccdb-server is used");
    add_option("meta-data,m", bpo::value<std::string>()->default_value("")->implicit_value(""), "metadata as key1=value1;key2=value2;..");
    add_option("refresh", bpo::value<string>()->default_value("")->implicit_value("async"), R"(refresh server cache after upload: "none" (or ""), "async" (non-blocking) and "sync" (blocking))");
    add_option("marginSOR", bpo::value<long>()->default_value(4 * o2::ccdb::CcdbObjectInfo::DAY), "validity at SOR");
    add_option("marginEOR", bpo::value<long>()->default_value(10 * o2::ccdb::CcdbObjectInfo::MINUTE), "validity margin to add after EOR");
    add_option("original-run,o", bpo::value<int>()->default_value(0), "if >0, use as the source run to create CTP/Config/Config object");
    opt_all.add(opt_general).add(opt_hidden);
    bpo::store(bpo::command_line_parser(argc, argv).options(opt_all).positional(opt_pos).run(), vm);
 
    if (vm.count("help")) {
      std::cout << opt_general << std::endl;
      exit(0);
    }
 
    bpo::notify(vm);
  } catch (bpo::error& e) {
    std::cerr << "ERROR: " << e.what() << std::endl
              << std::endl;
    std::cerr << opt_general << std::endl;
    exit(1);
  } catch (std::exception& e) {
    std::cerr << e.what() << ", application will now exit" << std::endl;
    exit(2);
  }
  if (vm.count("run") == 0) {
    std::cerr << "ERROR: "
              << "obligatory run number is missing" << std::endl;
    std::cerr << opt_general << std::endl;
    exit(3);
  }
  if (vm.count("period") == 0) {
    std::cerr << "ERROR: "
              << "obligatory data taking period name is missing" << std::endl;
    std::cerr << opt_general << std::endl;
    exit(3);
  }
  std::string refreshStr = vm["refresh"].as<string>();
  CCDBRefreshMode refresh = CCDBRefreshMode::NONE;
  if (!refreshStr.empty() && refreshStr != "none") {
    if (refreshStr == "async") {
      refresh = CCDBRefreshMode::ASYNC;
    } else if (refreshStr == "sync") {
      refresh = CCDBRefreshMode::SYNC;
    } else {
      LOGP(fatal, R"(Wrong CCDB refresh mode {}, supported are "none" (or ""), "async" and "sync")", refreshStr);
    }
  }
 
  int retVal = createGRPECSObject(
    vm["period"].as<std::string>(),
    vm["run"].as<int>(),
    vm["original-run"].as<int>(),
    vm["run-type"].as<int>(),
    vm["hbf-per-tf"].as<int>(),
    vm["detectors"].as<std::string>(),
    vm["continuous"].as<std::string>(),
    vm["triggering"].as<std::string>(),
    vm["flps"].as<std::string>(),
    vm["start-time"].as<long>(),
    vm["end-time"].as<long>(),
    vm["start-time-ctp"].as<long>(),
    vm["end-time-ctp"].as<long>(),
    vm["marginSOR"].as<long>(),
    vm["marginEOR"].as<long>(),
    vm["ccdb-server"].as<std::string>(),
    vm["ccdb-server-input"].as<std::string>(),
    vm["meta-data"].as<std::string>(),
    refresh);
 
  if (retVal != 0) {
    std::cerr << "ERROR: "
              << "Either GLO/Config/GRPECS or RCT/Info/RunInformation could not be stored correctly to CCDB" << std::endl;
    std::cerr << opt_general << std::endl;
    exit(retVal);
  }
}