dev: finishing orbitsox and orbitreset via zmq

Author: lietava

Detectors/CTP/workflowScalers/CMakeLists.txt

@@ -13,7 +13,8 @@ o2_add_library(CTPWorkflowScalers
                SOURCES src/RunManager.cxx
                PUBLIC_LINK_LIBRARIES O2::Framework
                                      O2::DataFormatsCTP
-                                    AliceO2::BookkeepingApi)
+                                     AliceO2::BookkeepingApi
+                                     AliceO2::InfoLogger)
 o2_target_root_dictionary(CTPWorkflowScalers HEADERS
   include/CTPWorkflowScalers/ctpCCDBManager.h)
 o2_add_executable(

Detectors/CTP/workflowScalers/include/CTPWorkflowScalers/RunManager.h

@@ -18,6 +18,9 @@
 #include "DataFormatsCTP/Configuration.h"
 #include "BookkeepingApi/BkpClientFactory.h"
 #include "BookkeepingApi/BkpClient.h"
+#include <InfoLogger/InfoLogger.hxx>
+
+
 using namespace o2::bkp::api;
 namespace o2
 {
@@ -71,6 +74,7 @@ class CTPRunManager : public ctpCCDBManager
   int mEOX = 0; // redundancy check
   int mNew = 1; // 1 - no CCDB: used for QC
   int mQCWritePeriod = 3; // Time in 10secs between two writes to QCCD
+  AliceO2::InfoLogger::InfoLogger mInfoLogger;
   ClassDefNV(CTPRunManager, 7);
 };
 } // namespace ctp

Detectors/CTP/workflowScalers/src/RunManager.cxx

@@ -18,6 +18,7 @@
 #include <regex>
 #include "CommonUtils/StringUtils.h"
 #include <fairlogger/Logger.h>
+
 using namespace o2::ctp;
 ///
 /// Active run to keep cfg and saclers of active runs
@@ -232,10 +233,53 @@ int CTPRunManager::processMessage(std::string& topic, const std::string& message
     return 0;
   }
   if (topic.find("soxorbit") != std::string::npos) {
-    return 0;
+    std::vector<std::string> tokens = o2::utils::Str::tokenize(message, ' ');
+    int ret = 0;
+    if(tokens.size() == 3) {
+      long timestamp = std::stol(tokens[0]);
+      uint32_t runnumber = std::stoul(tokens[1]);
+      uint32_t orbit = std::stoul(tokens[2]);
+      ret = saveSoxOrbit(timestamp,runnumber,orbit);
+      LOG(info) << "soxorbit: " << timestamp << " "  << runnumber << " " << orbit;
+      std::string logmessage;
+      if(ret) {
+        logmessage = "Failed to update CCDB with SOX orbit.";
+      } else {
+        logmessage = "CCDB updated with SOX orbit.";
+      }
+      int rlog = mInfoLogger.log(logmessage.c_str());
+      if(rlog) {
+        LOG(warn) << "Writing to infologger failed.";
+      }
+    } else {
+      LOG(error) << "Topic soxorbit dize !=3: " << message << " token size:" << tokens.size();
+      ret = 1;
+    }
+    return ret;
   }
   if (topic.find("orbitreset") != std::string::npos) {
-    return 0;
+    std::vector<std::string> tokens = o2::utils::Str::tokenize(message, ' ');
+    int ret = 0;
+    if(tokens.size()  == 1) {
+      long timestamp = std::stol(tokens[0]);
+      ret = saveOrbitReset(timestamp);
+      LOG(info) << "orbitreset: " << timestamp;
+      std::string logmessage;
+      if(ret) {
+        logmessage = "Failed to update CCDB with orbitreset.";
+        return 1;
+      } else {
+        logmessage = "CCDB updated with orbitreset.";
+      }
+      int rlog = mInfoLogger.log(logmessage.c_str());
+      if(rlog) {
+        LOG(warn) << "Writing to infologger failed.";
+      }
+    } else {
+      LOG(error) << "Topic orbit reset != 2: " << message << " token size:" << tokens.size();
+      ret = 1;
+    }
+    return ret;
   }
   static int nerror = 0;
   if (topic.find("sox") != std::string::npos) {

Detectors/CTP/workflowScalers/src/ctpCCDBManager.cxx

@@ -122,7 +122,7 @@ int ctpCCDBManager::saveSoxOrbit(uint32_t runNumber, uint32_t soxOrbit, long tim
   vect.push_back(timestamp);
   vect.push_back((uint64_t)runNumber);
   vect.push_back((uint64_t)soxOrbit);
-  long tmin = timestamp;
+  long tmin = timestamp/1000;
   long tmax = tmin + 381928219;
   o2::ccdb::CcdbApi api;
   map<string, string> metadata; // can be empty
@@ -149,9 +149,10 @@ int ctpCCDBManager::saveOrbitReset(long timeStamp)
   if (timeStamp == 0) {
     auto now = std::chrono::system_clock::now();
     timeStamp = std::chrono::duration_cast<std::chrono::microseconds>(now.time_since_epoch()).count();
+    LOG(warn) << "Received timestamp = 0 , using current time:" << timeStamp;
   }
   vect.push_back(timeStamp);
-  long tmin = timeStamp;
+  long tmin = timeStamp/1000;
   long tmax = tmin + 381928219;
   o2::ccdb::CcdbApi api;
   map<string, string> metadata; // can be empty