Support for using TAE events

This adds the necessary support to easily read "Super TAE" events.

DAQ/DAQUtils/CMakeLists.txt

@@ -14,7 +14,9 @@ DAQ/DAQUtils
 #]=======================================================================]
 
 gaudi_add_header_only_library(DAQUtilsLib
-    LINK Gaudi::GaudiAlgLib
+    LINK
+        Gaudi::GaudiAlgLib
+        LHCb::LHCbAlgsLib
 )
 
 gaudi_add_module(DAQUtils

DAQ/DAQUtils/include/Event/TAEUtils.h

+/*****************************************************************************\
+* (c) Copyright 2024 CERN for the benefit of the LHCb Collaboration           *
+*                                                                             *
+* This software is distributed under the terms of the GNU General Public      *
+* Licence version 3 (GPL Version 3), copied verbatim in the file "COPYING".   *
+*                                                                             *
+* In applying this licence, 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.                                       *
+\*****************************************************************************/
+#pragma once
+
+#include "LHCbAlgs/MergingTransformer.h" // Gaudi::Functional::vector_of_const_
+
+#include "Gaudi/Accumulators/Histogram.h"
+
+#include <algorithm>
+#include <map>
+#include <vector>
+
+namespace LHCb::TAE {
+  const auto defaultAxis = Gaudi::Accumulators::Axis<double>(
+      9, -4.5, 4.5, "TAE offset", {"Prev4", "Prev3", "Prev2", "Prev1", "Central", "Next1", "Next2", "Next3", "Next4"} );
+
+  /** @brief Helper class to combine ODIN and data objects for TAE.
+   *
+   *  The main method `arrangeTAE` takes three parameters:
+   *  - a vector of const ODIN objects
+   *  - a vector of const data objects
+   *  - `nSide`
+   *  The return value is a map with the TAE offset as key (Central = 0, Prev1 = -1, Next2 = 2, etc.)
+   *  and a pair (ODIN,DataType) as value.
+   *  The parameter `nSide` defines how many events to require on each side.
+   *  If the request cannot be satisfied (e.g. nSide is more than the half window),
+   *  an error will be logged and nothing will be returned.
+   *  If `nSide` is negative, the half window is automatically determined and all
+   *  TAE events are returned.
+   */
+
+  class Handler final {
+  public:
+    template <typename Owner>
+    Handler( Owner* owner )
+        : m_inconsistentSize{owner, "inconsistent number of ODIN and data objects"}
+        , m_noCentralEvent{owner, "no central event"}
+        , m_noFirstEvent{owner, "no first event"}
+        , m_notEnoughEvents{owner, "not enough events"}
+        , m_holeInBCIDs{owner, "hole in BCIDs"}
+        , m_missingData{owner, "data missing for some event(s)"} {}
+
+    template <typename T>
+    auto arrangeTAE( Gaudi::Functional::vector_of_const_<LHCb::ODIN const*> const& odins,
+                     Gaudi::Functional::vector_of_const_<T const*> const& data, int nSide = -1 ) const;
+
+  private:
+    mutable Gaudi::Accumulators::MsgCounter<MSG::ERROR> m_inconsistentSize;
+    mutable Gaudi::Accumulators::MsgCounter<MSG::ERROR> m_noCentralEvent;
+    mutable Gaudi::Accumulators::MsgCounter<MSG::ERROR> m_noFirstEvent;
+    mutable Gaudi::Accumulators::MsgCounter<MSG::ERROR> m_notEnoughEvents;
+    mutable Gaudi::Accumulators::MsgCounter<MSG::ERROR> m_holeInBCIDs;
+    mutable Gaudi::Accumulators::MsgCounter<MSG::ERROR> m_missingData;
+    mutable Gaudi::Accumulators::Counter<>              m_goodTAE;
+  };
+
+  template <typename T>
+  auto Handler::arrangeTAE( Gaudi::Functional::vector_of_const_<LHCb::ODIN const*> const& odins,
+                            Gaudi::Functional::vector_of_const_<T const*> const& data, int nSide ) const {
+    // std::map<int, std::tuple<LHCb::ODIN const &, T const &>> result;
+    std::map<int, std::pair<LHCb::ODIN const&, T const&>> result;
+
+    std::vector<unsigned int> indices;
+    indices.reserve( odins.size() );
+
+    if ( odins.size() != data.size() ) {
+      // inconsistent number of ODIN and data objects
+      ++m_inconsistentSize;
+      return result;
+    }
+
+    // remove missing events (can happen if the TAE half window was reduced but the python options were not adapted)
+    for ( auto const& [i, p] : LHCb::range::enumerate( odins ) ) {
+      if ( p ) indices.push_back( i );
+    }
+
+    // Find the central event.
+    auto itCentral =
+        std::find_if( indices.begin(), indices.end(), [&]( auto i ) { return odins[i]->timeAlignmentEventCentral(); } );
+    if ( itCentral == indices.end() ) {
+      // No central event -- this is a problem
+      ++m_noCentralEvent;
+      return result;
+    }
+    auto const* central = odins[*itCentral];
+
+    // Do we have enough events (with consecutive BCIDs) on either side of the central?
+    if ( nSide < 0 ) {
+      // We're requested to get the complete TAE, so find out the half window
+      auto itFirst = itCentral;
+      while ( itFirst >= indices.begin() && !odins[*itFirst]->timeAlignmentEventFirst() ) { --itFirst; }
+      if ( itFirst < indices.begin() ) {
+        // No first event
+        ++m_noFirstEvent;
+        return result;
+      }
+      nSide = itCentral - itFirst;
+    }
+
+    if ( itCentral - nSide < indices.begin() || itCentral + nSide >= indices.end() ) {
+      // Not enough events
+      ++m_notEnoughEvents;
+      return result;
+    }
+
+    // Do final checks and assemble the output map
+    for ( int j = -nSide; j <= nSide; ++j ) {
+      auto i    = *( itCentral + j );
+      auto dist = int( odins[i]->bunchId() ) - int( central->bunchId() ) +
+                  3564 * ( int( odins[i]->orbitNumber() ) - int( central->orbitNumber() ) );
+      if ( dist != j ) {
+        // There is a hole in the BCIDs...
+        ++m_holeInBCIDs;
+        return result;
+      }
+      if ( !data[i] ) {
+        // Some data is missing. This should never happen for a sane job control flow.
+        ++m_missingData;
+        return result;
+      }
+      result.emplace( j, std::make_pair( std::cref( *odins[i] ), std::cref( *data[i] ) ) );
+      // result.emplace(j, std::make_tuple(std::cref(*odins[i]), std::cref(*data[i])));
+    }
+    ++m_goodTAE;
+    return result;
+  }
+
+} // namespace LHCb::TAE
\ No newline at end of file

PyConf/python/PyConf/application.py

@@ -166,7 +166,7 @@ def mdf_writer(path,
         Connection=path)
 
 
-def online_writer(location):
+def online_writer(location, enable_tae=False):
     """Return an OnlineAlg writer which writes a single  TES `location`."""
     from PyConf.Algorithms import Online__OutputAlg, Online__RawEventToBanks
 
@@ -176,12 +176,21 @@ def online_writer(location):
         bank_types=["VP"], maker=make_raw_event_with_Online
     ).producer.inputs["RawData"].producer.RawGuard
 
+    tae_properties = {}
+    if enable_tae:
+        tae_properties = dict(
+            UseTAELeaves=True,  # Enable writing of TAE output
+            BankDirectory=
+            "OutputBanks",  # See the Hlt2TAE line Moore/Hlt/Hlt2Conf/python/Hlt2Conf/lines/hlt2calib/__init__.py
+        )
+
     raw_data = Online__RawEventToBanks(RawEvent=location).RawData
     return Online__OutputAlg(
         name="EventOutput_{hash}",
         RawData=raw_data,
-        # outputs={'RawGuard': force_location(location)},
-        RawGuard=guard,  # TODO this should also work
+        # outputs={'RawGuard': force_location(location)},  # TODO this should also work and we can get rid of the producer.inputs["RawData"].producer.RawGuard stuff above
+        RawGuard=guard,
+        **tae_properties,
     )
 
 
@@ -628,19 +637,54 @@ def make_raw_event_with_IOAlg(location,
 def make_raw_event_with_Online(location, bank_type=""):
     from PyConf.Algorithms import Online__InputAlg, Online__BanksToRawEvent
     # TODO what if we're called twice with a different location?
+    tae = location.removeprefix("/Event").replace("DAQ/RawEvent", "").replace(
+        "/", "")
+
+    def transform(RawData, RawGuard, **outputs):
+        # FIXME check outputs have the same prefix
+        return {
+            "BankDirectory": "Banks",
+            "RawData": RawData,
+            "RawGuard": RawGuard,
+            "ExtraOutputs": list(outputs.values())
+        }
+
+    if not location.startswith("/Event"):
+        location = "/Event/" + location
+
+    raw_data_loc = re.sub(r"/Event/(Prev|Next)[0-9]+", "/Event", location)
+    tae_locs = ({
+        f"Prev{i}": force_location(
+            raw_data_loc.replace("/Event", f"/Event/Prev{i}"))
+        for i in range(10)
+    } | {
+        f"Next{i}": force_location(
+            raw_data_loc.replace("/Event", f"/Event/Next{i}"))
+        for i in range(10)
+    })
+    tae_locs = (
+        {f"Prev{i}": force_location(f"Banks/Prev{i}")
+         for i in range(10)}
+        | {f"Next{i}": force_location(f"Banks/Next{i}")
+           for i in range(10)})
+
+    # print(location, raw_data_loc, tae_locs, flush=True)
+    # assert False
     alg = Online__InputAlg(
+        name="Online__InputAlg",
         outputs={
-            "RawData": force_location(location),
+            "RawData": force_location(raw_data_loc),
             "RawGuard": None,
             "DAQErrors": None,
-        },
+        } | tae_locs,
+        output_transform=transform,
         DeclareData=True,  # TODO what does this do?
         DeclareEvent=True,  # TODO what does this do?
         DeclareErrors=False,
     )
     return Online__BanksToRawEvent(
-        name="Online__BanksToRawEvent" + bank_type,
-        RawData=alg.RawData,
+        name="Online__BanksToRawEvent_" + bank_type + tae,
+        RawData=getattr(alg, tae or "RawData"),
         BankType=bank_type).RawEvent
 
 
@@ -709,17 +753,17 @@ def default_raw_event(bank_types=[],
 
 
 @configurable
-def default_raw_banks(bank_type, make_raw=default_raw_event, **kwargs):
-
-    import inspect
-    num_args = len(inspect.getfullargspec(make_raw).args)
-
-    if num_args == 0:
-        raw_event = make_raw()
-    else:
+def default_raw_banks(bank_type,
+                      make_raw=default_raw_event,
+                      tae_pos="",
+                      **kwargs):
+    if not tae_pos:
         raw_event = make_raw([bank_type])
+    else:
+        raw_event = make_raw([bank_type], stream=f"{tae_pos}/DAQ")
 
-    if 'name' not in kwargs: kwargs['name'] = f'UnpackRawEvent_{bank_type}'
+    if 'name' not in kwargs:
+        kwargs['name'] = f'UnpackRawEvent_{tae_pos}{bank_type}'
 
     def output_transform(RawBankLocations):
         return {"RawBankLocations": [RawBankLocations]}
@@ -733,7 +777,8 @@ def default_raw_banks(bank_type, make_raw=default_raw_event, **kwargs):
         BankTypes=[bank_type],
         output_transform=output_transform,
         outputs={
-            'RawBankLocations': force_location(f'/Event/RawBanks/{bank_type}')
+            'RawBankLocations':
+            force_location(f'/Event/RawBanks/{tae_pos}{bank_type}')
         },
         **kwargs,
     ).RawBankLocations

PyConf/python/PyConf/components.py

@@ -715,9 +715,15 @@ class Algorithm(object):
         instance._is_barrier = is_barrier or alg_properties.get(
             'hasOptionals', False)
 
-        if any(not isinstance(v, (type(None), force_location))
-               for v in outputs.values()):
-            raise ValueError('outputs values must be None or force_location')
+        bad_type = {
+            k: type(v)
+            for k, v in outputs.items()
+            if not isinstance(v, (type(None), force_location))
+        }
+        if bad_type:
+            raise ValueError(
+                f'values of outputs must be None or force_location, got {bad_type}'
+            )
 
         explicitly_set_outputs = set(kwargs).intersection(outputs)
         if explicitly_set_outputs: