diff --git a/DataFormats/Detectors/TPC/src/DCS.cxx b/DataFormats/Detectors/TPC/src/DCS.cxx index b56d07acd7c73..8a082a8d21901 100644 --- a/DataFormats/Detectors/TPC/src/DCS.cxx +++ b/DataFormats/Detectors/TPC/src/DCS.cxx @@ -354,6 +354,17 @@ void fillBuffer(std::pair, std::vector>& 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> 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); diff --git a/Detectors/TPC/dcs/include/TPCdcs/DCSProcessor.h b/Detectors/TPC/dcs/include/TPCdcs/DCSProcessor.h index e6ead9b0cb302..95d02f47b80cf 100644 --- a/Detectors/TPC/dcs/include/TPCdcs/DCSProcessor.h +++ b/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 diff --git a/Detectors/TPC/dcs/src/DCSProcessor.cxx b/Detectors/TPC/dcs/src/DCSProcessor.cxx index 4c9d196432687..6f892d94ed07a 100644 --- a/Detectors/TPC/dcs/src/DCSProcessor.cxx +++ b/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(); + } } } diff --git a/Detectors/TPC/dcs/src/DCSSpec.cxx b/Detectors/TPC/dcs/src/DCSSpec.cxx index ea4e3a29ff630..1c55940e69cf2 100644 --- a/Detectors/TPC/dcs/src/DCSSpec.cxx +++ b/Detectors/TPC/dcs/src/DCSSpec.cxx @@ -60,7 +60,7 @@ class DCSDevice : public o2::framework::Task void run(o2::framework::ProcessingContext& pc) final; template - 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 -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 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(pressureTime.back()) + std::max(genericMargin, tailMargin); + sendObject(output, mDCS.getPressure(), CDBType::CalPressure, mDCS.getPressureCCDBStartTime(), pressureEnd); + } } /// ===| create DCS processor |=================================================