Pressure: Fix last smoothed pressure value

The rolling median in makeRobustPressure was computed with a one-sided
(past-only) window for the last query time(s) of each ~10-minute processing
slot, because no future data existed yet in the buffer. This caused the
last value(s) of each slot to lag behind any pressure trend, followed by a
visible jump when the next slot started with a symmetric window. Fixed by
withholding trailing query points until a 2x-timeInterval look-ahead margin
of future data is available, picking them up seamlessly in the next slot.

Also fixes:
- CCDB objects were uploaded even for slots with no new data for that
  quantity; now skipped when there's nothing to store.
- Pressure's CCDB end-validity now anchors on its own last data point
  plus a margin, instead of mLastCreationTime + generic extension

Author: matthias-kleiner

DataFormats/Detectors/TPC/src/DCS.cxx

@@ -354,6 +354,17 @@ void fillBuffer(std::pair<std::vector<float>, std::vector<TimeStampType>>& buffe
   buffer = std::move(buffTmp);
 }
 
+/// truncate all parallel vectors of a RollingStats to size n, keeping the leading n entries.
+/// Used to keep RobustPressure's members aligned with a `times` vector that got trimmed.
+void trimStats(o2::math_utils::RollingStats& stats, size_t n)
+{
+  stats.median.resize(n);
+  stats.std.resize(n);
+  stats.nPoints.resize(n);
+  stats.closestDistanceL.resize(n);
+  stats.closestDistanceR.resize(n);
+}
+
 void Pressure::makeRobustPressure(TimeStampType timeInterval, TimeStampType timeIntervalRef, TimeStampType tStart, TimeStampType tEnd, const int nthreads)
 {
   const auto surfaceAtmosPressurePair = surfaceAtmosPressure.getPairOfVector();
@@ -380,9 +391,9 @@ void Pressure::makeRobustPressure(TimeStampType timeInterval, TimeStampType time
 
   /// minimum number of points in the interval - otherwise use the n closest points
   const int minPoints = 4;
-  const auto cavernAtmosPressureStats = o2::math_utils::getRollingStatistics(mCavernAtmosPressure1Buff.second, mCavernAtmosPressure1Buff.first, times, timeInterval, nthreads, minPoints, minPoints);
-  const auto cavernAtmosPressure2Stats = o2::math_utils::getRollingStatistics(mCavernAtmosPressure2Buff.second, mCavernAtmosPressure2Buff.first, times, timeInterval, nthreads, minPoints, minPoints);
-  const auto surfaceAtmosPressureStats = o2::math_utils::getRollingStatistics(mSurfaceAtmosPressureBuff.second, mSurfaceAtmosPressureBuff.first, times, timeInterval, nthreads, minPoints, minPoints);
+  auto cavernAtmosPressureStats = o2::math_utils::getRollingStatistics(mCavernAtmosPressure1Buff.second, mCavernAtmosPressure1Buff.first, times, timeInterval, nthreads, minPoints, minPoints);
+  auto cavernAtmosPressure2Stats = o2::math_utils::getRollingStatistics(mCavernAtmosPressure2Buff.second, mCavernAtmosPressure2Buff.first, times, timeInterval, nthreads, minPoints, minPoints);
+  auto surfaceAtmosPressureStats = o2::math_utils::getRollingStatistics(mSurfaceAtmosPressureBuff.second, mSurfaceAtmosPressureBuff.first, times, timeInterval, nthreads, minPoints, minPoints);
 
   // subtract the moving median values from the different sensors if they are ok
   std::pair<std::vector<float>, std::vector<TimeStampType>> cavernAtmosPressure12;
@@ -421,9 +432,9 @@ void Pressure::makeRobustPressure(TimeStampType timeInterval, TimeStampType time
   fillBuffer(mPressure2SBuff, cavernAtmosPressure2S, tStartRef, minPointsRef);
 
   // get long term median of diffs - this is used for normalization of the pressure values -
-  const auto cavernAtmosPressure12Stats = o2::math_utils::getRollingStatistics(mPressure12Buff.second, mPressure12Buff.first, times, timeIntervalRef, nthreads, 3, minPointsRef);
-  const auto cavernAtmosPressure1SStats = o2::math_utils::getRollingStatistics(mPressure1SBuff.second, mPressure1SBuff.first, times, timeIntervalRef, nthreads, 3, minPointsRef);
-  const auto cavernAtmosPressure2SStats = o2::math_utils::getRollingStatistics(mPressure2SBuff.second, mPressure2SBuff.first, times, timeIntervalRef, nthreads, 3, minPointsRef);
+  auto cavernAtmosPressure12Stats = o2::math_utils::getRollingStatistics(mPressure12Buff.second, mPressure12Buff.first, times, timeIntervalRef, nthreads, 3, minPointsRef);
+  auto cavernAtmosPressure1SStats = o2::math_utils::getRollingStatistics(mPressure1SBuff.second, mPressure1SBuff.first, times, timeIntervalRef, nthreads, 3, minPointsRef);
+  auto cavernAtmosPressure2SStats = o2::math_utils::getRollingStatistics(mPressure2SBuff.second, mPressure2SBuff.first, times, timeIntervalRef, nthreads, 3, minPointsRef);
 
   // calculate diffs of median values
   const float maxDist = 20 * timeInterval;
@@ -518,6 +529,27 @@ void Pressure::makeRobustPressure(TimeStampType timeInterval, TimeStampType time
 
   fillBuffer(mRobPressureBuff, robustPressureTmp, tStartRef, minPointsRef);
 
+  // drop trailing query times that don't yet have a full look-ahead margin of data
+  // to their right in the buffer: the smoothing window is ±timeInterval, so without
+  // it those points would be smoothed with a partially or fully one-sided (past-only)
+  // window, biasing them low/high and causing a jump at the slot boundary.
+  const auto& robBuffTimes = mRobPressureBuff.second;
+  const TimeStampType lookaheadMargin = 2 * timeInterval;
+  while (times.size() > 1 && !robBuffTimes.empty() && times.back() + lookaheadMargin > robBuffTimes.back()) {
+    times.pop_back();
+  }
+  isOk.resize(times.size());
+
+  // the *Stats above were computed for the untrimmed query grid; truncate them to
+  // match so all vectors stored in RobustPressure stay parallel/same length as time.
+  // The dropped tail is simply recomputed (with a proper symmetric window) next slot.
+  trimStats(cavernAtmosPressureStats, times.size());
+  trimStats(cavernAtmosPressure2Stats, times.size());
+  trimStats(surfaceAtmosPressureStats, times.size());
+  trimStats(cavernAtmosPressure12Stats, times.size());
+  trimStats(cavernAtmosPressure1SStats, times.size());
+  trimStats(cavernAtmosPressure2SStats, times.size());
+
   RobustPressure& pOut = robustPressure;
   pOut.surfaceAtmosPressure = std::move(surfaceAtmosPressureStats);
   pOut.cavernAtmosPressure2 = std::move(cavernAtmosPressure2Stats);

Detectors/TPC/dcs/include/TPCdcs/DCSProcessor.h

@@ -102,6 +102,10 @@ class DCSProcessor
   const auto& getTimeGas() const { return mTimeGas; }
   const auto& getTimePressure() const { return mTimePressure; }
 
+  /// CCDB validity start for the pressure object: last output time of the previous slot
+  /// (0 on the first slot, falls back to mTimePressure.first in finalizePressure)
+  auto getPressureCCDBStartTime() const { return mPressureCCDBStartTime; }
+
   auto& getTemperature() { return mTemperature; }
   auto& getHighVoltage() { return mHighVoltage; }
   auto& getGas() { return mGas; }
@@ -121,6 +125,8 @@ class DCSProcessor
   dcs::TimeStampType mFitInterval{5 * 60 * 1000};                ///< fit interval (ms) e.g. for temparature data
   dcs::TimeStampType mPressureInterval{200 * 1000};              ///< interval (ms) for averaging pressure values
   dcs::TimeStampType mPressureIntervalRef{60 * 60 * 1000};       ///< interval (ms) for averaging pressure values for longer reference time interval
+  dcs::TimeStampType mLastPressureOutputEndTime{0};              ///< last time stamp in pOut.time from previous finalizePressure call
+  dcs::TimeStampType mPressureCCDBStartTime{0};                  ///< CCDB validity start for current pressure slot
   bool mWriteDebug{false};                                       ///< switch to dump debug tree
   bool mRoundToInterval{false};                                  ///< round to full fit interval e.g. full minute
   bool mHasData{false};                                          ///< if there are data to process

Detectors/TPC/dcs/src/DCSProcessor.cxx

@@ -178,7 +178,20 @@ void DCSProcessor::finalizePressure()
   mTimePressure = {mPressure.getMinTime(), mPressure.getMaxTime()};
   // if there is data perform the processing
   if (mTimePressure.last > 0) {
-    mPressure.makeRobustPressure(mPressureInterval, mPressureIntervalRef, mTimePressure.first, mTimePressure.last);
+    // capture start for CCDB validity before updating mLastPressureOutputEndTime
+    mPressureCCDBStartTime = (mLastPressureOutputEndTime > 0) ? mLastPressureOutputEndTime : mTimePressure.first;
+    // if the previous slot withheld trailing points (no full look-ahead margin yet),
+    // start a half-interval earlier so times[0] = mLastPressureOutputEndTime +
+    // timeInterval picks up exactly where the previous slot's kept data ended,
+    // regardless of how many trailing points it withheld
+    auto tStart = mTimePressure.first;
+    if (mLastPressureOutputEndTime > 0) {
+      tStart = std::min(tStart, mLastPressureOutputEndTime + mPressureInterval / 2);
+    }
+    mPressure.makeRobustPressure(mPressureInterval, mPressureIntervalRef, tStart, mTimePressure.last);
+    if (!mPressure.robustPressure.time.empty()) {
+      mLastPressureOutputEndTime = mPressure.robustPressure.time.back();
+    }
   }
 }
 

Detectors/TPC/dcs/src/DCSSpec.cxx

@@ -60,7 +60,7 @@ class DCSDevice : public o2::framework::Task
   void run(o2::framework::ProcessingContext& pc) final;
 
   template <typename T>
-  void sendObject(DataAllocator& output, T& obj, const CDBType calibType);
+  void sendObject(DataAllocator& output, T& obj, const CDBType calibType, uint64_t startTime, uint64_t endTimeOverride = 0);
 
   void updateCCDB(DataAllocator& output);
 
@@ -162,14 +162,17 @@ void DCSDevice::run(o2::framework::ProcessingContext& pc)
 }
 
 template <typename T>
-void DCSDevice::sendObject(DataAllocator& output, T& obj, const CDBType calibType)
+void DCSDevice::sendObject(DataAllocator& output, T& obj, const CDBType calibType, uint64_t startTime, uint64_t endTimeOverride)
 {
   LOGP(info, "Prepare CCDB for {}", CDBTypeMap.at(calibType));
 
   std::map<std::string, std::string> md = mCDBStorage.getMetaData();
   o2::ccdb::CcdbObjectInfo w;
-  // for online processing extend the validity range. Will be truncated with the adjustableEOV procedure
-  o2::calibration::Utils::prepareCCDBobjectInfo(obj, w, CDBTypeMap.at(calibType), md, mUpdateIntervalStart, mLastCreationTime + 2 * mCCDBupdateInterval * 1000);
+  // for online processing extend the validity range. Will be truncated with the adjustableEOV procedure.
+  // endTimeOverride==0 (the default) means "use the generic extension"; callers that need a
+  // different end-validity (currently only pressure - see updateCCDB()) pass their own value.
+  const uint64_t endTime = endTimeOverride > 0 ? endTimeOverride : mLastCreationTime + 2 * mCCDBupdateInterval * 1000;
+  o2::calibration::Utils::prepareCCDBobjectInfo(obj, w, CDBTypeMap.at(calibType), md, startTime, endTime);
   auto image = o2::ccdb::CcdbApi::createObjectImage(&obj, &w);
 
   LOGP(info, "Sending object {} / {} of size {} bytes, valid for {} : {} ", w.getPath(), w.getFileName(), image->size(), w.getStartValidityTimestamp(), w.getEndValidityTimestamp());
@@ -179,10 +182,25 @@ void DCSDevice::sendObject(DataAllocator& output, T& obj, const CDBType calibTyp
 
 void DCSDevice::updateCCDB(DataAllocator& output)
 {
-  sendObject(output, mDCS.getTemperature(), CDBType::CalTemperature);
-  sendObject(output, mDCS.getHighVoltage(), CDBType::CalHV);
-  sendObject(output, mDCS.getGas(), CDBType::CalGas);
-  sendObject(output, mDCS.getPressure(), CDBType::CalPressure);
+  // only store an object if it actually received new data this slot; otherwise
+  // we'd upload an empty object, for pressure additionally tagged with a stale
+  // start-validity time left over from a previous slot
+  if (mDCS.getTimeTemperature().last > 0) {
+    sendObject(output, mDCS.getTemperature(), CDBType::CalTemperature, mUpdateIntervalStart);
+  }
+  if (mDCS.getTimeHighVoltage().last > 0) {
+    sendObject(output, mDCS.getHighVoltage(), CDBType::CalHV, mUpdateIntervalStart);
+  }
+  if (mDCS.getTimeGas().last > 0) {
+    sendObject(output, mDCS.getGas(), CDBType::CalGas, mUpdateIntervalStart);
+  }
+  if (mDCS.getTimePressure().last > 0) {
+    const auto& pressureTime = mDCS.getPressure().robustPressure.time;
+    const uint64_t genericMargin = 2 * uint64_t(mCCDBupdateInterval) * 1000;
+    const uint64_t tailMargin = 2 * uint64_t(mDCS.getPressureInterval());
+    const uint64_t pressureEnd = pressureTime.empty() ? 0 : static_cast<uint64_t>(pressureTime.back()) + std::max(genericMargin, tailMargin);
+    sendObject(output, mDCS.getPressure(), CDBType::CalPressure, mDCS.getPressureCCDBStartTime(), pressureEnd);
+  }
 }
 
 /// ===| create DCS processor |=================================================