Adopt standard user I/O spec helpers

Author: knopers8

Framework/include/QualityControl/Aggregator.h

@@ -51,10 +51,6 @@ struct AggregatorSpec;
 class Aggregator
 {
  public:
-  /// \brief Number of bytes in data description used for hashing of AggregatorRunner names. See HashDataDescription.h for details
-  static constexpr size_t descriptionHashLength = 4;
-  static_assert(descriptionHashLength <= o2::header::DataDescription::size);
-
   /// Constructor
   /**
    * \brief Aggregator constructor
@@ -84,7 +80,6 @@ class Aggregator
   void endOfActivity(const core::Activity& activity);
 
   static AggregatorConfig extractConfig(const core::CommonSpec&, const AggregatorSpec&);
-  static framework::OutputSpec createOutputSpec(const std::string& detector, const std::string& aggregatorName);
 
  private:
   /**

Framework/include/QualityControl/AggregatorRunner.h

@@ -79,10 +79,6 @@ struct AggregatorSource;
 /// \author Barthélémy von Haller
 class AggregatorRunner : public framework::Task
 {
-  /// \brief Number of bytes in data description used for hashing of AggregatorRunner names. See HashDataDescription.h for details
-  static constexpr size_t descriptionHashLength = 4;
-  static_assert(descriptionHashLength <= o2::header::DataDescription::size);
-
  public:
   /// Constructor
   /**
@@ -111,7 +107,6 @@ class AggregatorRunner : public framework::Task
   static framework::DataProcessorLabel getLabel() { return { "qc-aggregator" }; }
   static std::string createAggregatorRunnerIdString() { return "qc-aggregator"; };
   static std::string createAggregatorRunnerName();
-  static header::DataDescription createAggregatorRunnerDataDescription(const std::string& aggregatorName);
 
   /// \brief Compute the detector name to be used in the infologger for this runner.
   /// Compute the detector name to be used in the infologger for this runner.

Framework/include/QualityControl/Check.h

@@ -45,10 +45,6 @@ struct CheckSpec;
 /// \author Rafal Pacholek
 class Check
 {
-  /// \brief Number of bytes in data description used for hashing of Check descrition names. See HashDataDescription.h for details
-  static constexpr size_t descriptionHashLength = 4;
-  static_assert(descriptionHashLength <= o2::header::DataDescription::size);
-
  public:
   /// Constructor
   /**
@@ -78,21 +74,12 @@ class Check
   void startOfActivity(const core::Activity& activity);
   void endOfActivity(const core::Activity& activity);
 
-  // TODO: Unique Input string
-  static o2::header::DataDescription createCheckDataDescription(const std::string& checkName);
-
-  /// \brief creates DataOrigin for Check task in form CDET
-  ///
-  /// \param detector Name of the detector to be used. If longer than 3B it will be truncated
-  static o2::header::DataOrigin createCheckDataOrigin(const std::string& detector);
-
   UpdatePolicyType getUpdatePolicyType() const;
   std::vector<std::string> getObjectsNames() const;
   bool getAllObjectsOption() const;
 
   // todo: probably make CheckFactory
   static CheckConfig extractConfig(const core::CommonSpec&, const CheckSpec&);
-  static framework::OutputSpec createOutputSpec(const std::string& detector, const std::string& checkName);
 
  private:
   void beautify(std::map<std::string, std::shared_ptr<core::MonitorObject>>& moMap, const core::Quality& quality);

Framework/include/QualityControl/PostProcessingDevice.h

@@ -36,10 +36,6 @@ struct PostProcessingRunnerConfig;
 /// \author Piotr Konopka
 class PostProcessingDevice : public framework::Task
 {
-  /// \brief Number of bytes in data description used for hashing DataDescription. See HashDataDescription.h for details
-  static constexpr size_t descriptionHashLength = 4;
-  static_assert(descriptionHashLength <= o2::header::DataDescription::size);
-
  public:
   /// \brief Constructor
   ///
@@ -61,10 +57,6 @@ class PostProcessingDevice : public framework::Task
   /// \brief Data Processor Label to identify all Task Runners
   static framework::DataProcessorLabel getLabel() { return { "qc-pp-task-runner" }; }
   static std::string createPostProcessingDeviceName(const std::string& taskName, const std::string& detectorName);
-  /// \brief Unified DataOrigin for Post-processing tasks
-  static header::DataOrigin createPostProcessingDataOrigin(const std::string& detectorCode);
-  /// \brief Unified DataDescription naming scheme for all Post-processing tasks
-  static header::DataDescription createPostProcessingDataDescription(const std::string& taskName);
 
  private:
   /// \brief Callback for CallbackService::Id::Start (DPL) a.k.a. RUN transition (FairMQ)

Framework/include/QualityControl/TaskRunner.h

@@ -77,10 +77,6 @@ class ObjectsManager;
 /// \author Barthelemy von Haller
 class TaskRunner : public framework::Task
 {
-  /// \brief Number of bytes in data description used for hashing of Task names. See HashDataDescription.h for details
-  static constexpr size_t taskDescriptionHashLength = 4;
-  static_assert(taskDescriptionHashLength <= o2::header::DataDescription::size);
-
  public:
   /// \brief Constructor
   ///
@@ -109,10 +105,6 @@ class TaskRunner : public framework::Task
   static framework::DataProcessorLabel getTaskRunnerLabel() { return { "qc-task" }; }
   /// \brief ID string for all TaskRunner devices
   static std::string createTaskRunnerIdString();
-  /// \brief Unified DataOrigin for Quality Control tasks
-  static header::DataOrigin createTaskDataOrigin(const std::string& detectorCode, bool movingWindows = false);
-  /// \brief Unified DataDescription naming scheme for all tasks
-  static header::DataDescription createTaskDataDescription(const std::string& taskName);
   /// \brief Unified DataDescription naming scheme for all timers
   static header::DataDescription createTimerDataDescription(const std::string& taskName);
 

Framework/src/Aggregator.cxx

@@ -26,7 +26,7 @@
 #include "QualityControl/Activity.h"
 #include <Common/Exceptions.h>
 #include "QualityControl/CommonSpec.h"
-#include "QualityControl/DataHeaderHelpers.h"
+#include "QualityControl/UserInputOutput.h"
 
 #include <utility>
 #include <algorithm>
@@ -253,33 +253,11 @@ AggregatorConfig Aggregator::extractConfig(const core::CommonSpec& commonSpec, c
     std::move(objectNames),
     checkAllObjects,
     std::move(inputs),
-    createOutputSpec(aggregatorSpec.detectorName, aggregatorSpec.aggregatorName),
+    createUserOutputSpec(DataSourceType::Aggregator, aggregatorSpec.detectorName, aggregatorSpec.aggregatorName),
     sources
   };
 }
 
-o2::header::DataOrigin createAggregatorDataOrigin(const std::string& detector)
-{
-  using Origin = o2::header::DataOrigin;
-  Origin header;
-  header.runtimeInit(std::string{ "A" }.append(detector.substr(0, Origin::size - 1)).c_str());
-  return header;
-}
-
-o2::header::DataDescription createAggregatorDataDescription(const std::string& aggregatorName)
-{
-  if (aggregatorName.empty()) {
-    BOOST_THROW_EXCEPTION(FatalException() << AliceO2::Common::errinfo_details("Empty aggregatorName for aggregator's data description"));
-  }
-
-  return quality_control::core::createDataDescription(aggregatorName, Aggregator::descriptionHashLength);
-}
-
-framework::OutputSpec Aggregator::createOutputSpec(const std::string& detector, const std::string& aggregatorName)
-{
-  return { createAggregatorDataOrigin(detector), createAggregatorDataDescription(aggregatorName), 0, framework::Lifetime::Sporadic };
-}
-
 void Aggregator::startOfActivity(const core::Activity& activity)
 {
   if (mAggregatorInterface) {

Framework/src/AggregatorRunner.cxx

@@ -96,14 +96,6 @@ void AggregatorRunner::prepareOutputs()
   }
 }
 
-header::DataDescription AggregatorRunner::createAggregatorRunnerDataDescription(const std::string& aggregatorName)
-{
-  if (aggregatorName.empty()) {
-    BOOST_THROW_EXCEPTION(FatalException() << errinfo_details("Empty taskName for task's data description"));
-  }
-  return quality_control::core::createDataDescription(aggregatorName, AggregatorRunner::descriptionHashLength);
-}
-
 std::string AggregatorRunner::createAggregatorRunnerName()
 {
   return AggregatorRunner::createAggregatorRunnerIdString(); // there is only one thus we can just take the idString

Framework/src/Check.cxx

@@ -29,7 +29,7 @@
 #include "QualityControl/RootClassFactory.h"
 #include "QualityControl/QcInfoLogger.h"
 #include "QualityControl/Quality.h"
-#include "QualityControl/DataHeaderHelpers.h"
+#include "QualityControl/UserInputOutput.h"
 #include "QualityControl/ObjectMetadataHelpers.h"
 
 #include <QualityControl/AggregatorRunner.h>
@@ -44,24 +44,6 @@ using namespace std;
 namespace o2::quality_control::checker
 {
 
-/// Static functions
-o2::header::DataDescription Check::createCheckDataDescription(const std::string& checkName)
-{
-  if (checkName.empty()) {
-    BOOST_THROW_EXCEPTION(FatalException() << errinfo_details("Empty checkName for check's data description"));
-  }
-
-  return quality_control::core::createDataDescription(checkName, Check::descriptionHashLength);
-}
-
-o2::header::DataOrigin Check::createCheckDataOrigin(const std::string& detector)
-{
-  using Origin = o2::header::DataOrigin;
-  Origin header;
-  header.runtimeInit(std::string{ "C" }.append(detector.substr(0, Origin::size - 1)).c_str());
-  return header;
-}
-
 /// Members
 Check::Check(CheckConfig config)
   : mCheckConfig(std::move(config))
@@ -281,15 +263,10 @@ CheckConfig Check::extractConfig(const CommonSpec& commonSpec, const CheckSpec&
     checkAllObjects,
     allowBeautify,
     std::move(inputs),
-    createOutputSpec(checkSpec.detectorName, checkSpec.checkName),
+    createUserOutputSpec(DataSourceType::Check, checkSpec.detectorName, checkSpec.checkName),
   };
 }
 
-framework::OutputSpec Check::createOutputSpec(const std::string& detector, const std::string& checkName)
-{
-  return { createCheckDataOrigin(detector), createCheckDataDescription(checkName), 0, framework::Lifetime::Sporadic };
-}
-
 void Check::startOfActivity(const core::Activity& activity)
 {
   if (mCheckInterface) {

Framework/src/InfrastructureGenerator.cxx

@@ -337,7 +337,7 @@ framework::WorkflowSpec InfrastructureGenerator::generateLocalBatchInfrastructur
     auto taskConfig = TaskRunnerFactory::extractConfig(infrastructureSpec.common, taskSpec, 0, 1);
     workflow.emplace_back(TaskRunnerFactory::create(taskConfig));
 
-    fileSinkInputs.emplace_back(taskSpec.taskName, TaskRunner::createTaskDataOrigin(taskSpec.detectorName), TaskRunner::createTaskDataDescription(taskSpec.taskName), Lifetime::Sporadic);
+    fileSinkInputs.emplace_back(createUserInputSpec(DataSourceType::Task, taskSpec.detectorName, taskSpec.taskName));
   }
 
   if (!fileSinkInputs.empty()) {
@@ -377,9 +377,8 @@ framework::WorkflowSpec InfrastructureGenerator::generateRemoteBatchInfrastructu
     // We create an OutputSpec for moving windows for this task only if they are expected.
     if (!taskConfig.movingWindows.empty()) {
       fileSourceOutputs.push_back(
-        { RootFileSource::outputBinding(taskSpec.detectorName, taskSpec.taskName, true),
-          TaskRunner::createTaskDataOrigin(taskSpec.detectorName, true),
-          TaskRunner::createTaskDataDescription(taskSpec.taskName), 0, Lifetime::Sporadic });
+        createUserOutputSpec(DataSourceType::TaskMovingWindow, taskSpec.detectorName, taskSpec.taskName, 0,
+                             RootFileSource::outputBinding(taskSpec.detectorName, taskSpec.taskName, true)));
     }
   }
   if (!fileSourceOutputs.empty()) {
@@ -530,7 +529,7 @@ void InfrastructureGenerator::generateLocalTaskLocalProxy(framework::WorkflowSpe
   std::string remotePort = std::to_string(taskSpec.remotePort);
   std::string proxyName = taskSpec.detectorName + "-" + taskName + "-proxy";
   std::string channelName = taskSpec.detectorName + "-" + taskName + "-proxy";
-  InputSpec proxyInput{ channelName, TaskRunner::createTaskDataOrigin(taskSpec.detectorName, false), TaskRunner::createTaskDataDescription(taskName), static_cast<SubSpec>(id), Lifetime::Sporadic };
+  InputSpec proxyInput = createUserInputSpec(DataSourceType::Task, taskSpec.detectorName, taskName, static_cast<SubSpec>(id), channelName);
   std::string channelConfig = "name=" + channelName + ",type=pub,method=connect,address=tcp://" +
                               taskSpec.remoteMachine + ":" + remotePort + ",rateLogging=60,transport=zeromq,sndBufSize=4";
 
@@ -554,7 +553,7 @@ void InfrastructureGenerator::generateLocalTaskRemoteProxy(framework::WorkflowSp
   Outputs proxyOutputs;
   for (size_t id = 1; id <= numberOfLocalMachines; id++) {
     proxyOutputs.emplace_back(
-      OutputSpec{ { channelName }, TaskRunner::createTaskDataOrigin(taskSpec.detectorName, false), TaskRunner::createTaskDataDescription(taskName), static_cast<SubSpec>(id), Lifetime::Sporadic });
+      createUserOutputSpec(DataSourceType::Task, taskSpec.detectorName, taskName, static_cast<SubSpec>(id), { channelName }));
   }
 
   std::string channelConfig = "name=" + channelName + ",type=sub,method=bind,address=tcp://*:" + remotePort +
@@ -586,20 +585,14 @@ void InfrastructureGenerator::generateMergers(framework::WorkflowSpec& workflow,
   Inputs mergerInputs;
   for (size_t id = 1; id <= numberOfLocalMachines; id++) {
     mergerInputs.emplace_back(
-      InputSpec{ { taskName + std::to_string(id) },
-                 TaskRunner::createTaskDataOrigin(detectorName, false),
-                 TaskRunner::createTaskDataDescription(taskName),
-                 static_cast<SubSpec>(id),
-                 Lifetime::Sporadic });
+      createUserInputSpec(DataSourceType::Task, detectorName, taskName, static_cast<SubSpec>(id), taskName + std::to_string(id)));
   }
 
   MergerInfrastructureBuilder mergersBuilder;
   mergersBuilder.setInfrastructureName(taskName);
   mergersBuilder.setInputSpecs(mergerInputs);
-  mergersBuilder.setOutputSpec(
-    { { "main" }, TaskRunner::createTaskDataOrigin(detectorName, false), TaskRunner::createTaskDataDescription(taskName), 0, Lifetime::Sporadic });
-  mergersBuilder.setOutputSpecMovingWindow(
-    { { "main_mw" }, TaskRunner::createTaskDataOrigin(detectorName, true), TaskRunner::createTaskDataDescription(taskName), 0, Lifetime::Sporadic });
+  mergersBuilder.setOutputSpec(createUserOutputSpec(DataSourceType::Task, detectorName, taskName, 0, { "main" }));
+  mergersBuilder.setOutputSpecMovingWindow(createUserOutputSpec(DataSourceType::TaskMovingWindow, detectorName, taskName, 0, { "main_mw" }));
   MergerConfig mergerConfig;
   // if we are to change the mode to Full, disable reseting tasks after each cycle.
   mergerConfig.inputObjectTimespan = { (mergingMode.empty() || mergingMode == "delta") ? InputObjectsTimespan::LastDifference : InputObjectsTimespan::FullHistory };
@@ -630,23 +623,20 @@ void InfrastructureGenerator::generateCheckRunners(framework::WorkflowSpec& work
 
   // todo: avoid code repetition
   for (const auto& taskSpec : infrastructureSpec.tasks | std::views::filter(&TaskSpec::active)) {
-    InputSpec taskOutput{ taskSpec.taskName, TaskRunner::createTaskDataOrigin(taskSpec.detectorName), TaskRunner::createTaskDataDescription(taskSpec.taskName), Lifetime::Sporadic };
+    InputSpec taskOutput{ createUserInputSpec(DataSourceType::Task, taskSpec.detectorName, taskSpec.taskName) };
     tasksOutputMap.insert({ DataSpecUtils::label(taskOutput), taskOutput });
 
     bool movingWindowsEnabled = !taskSpec.movingWindows.empty();
     bool synchronousRemote = taskSpec.location == TaskLocationSpec::Local && (infrastructureSpec.workflowType == WorkflowType::Remote || infrastructureSpec.workflowType == WorkflowType::FullChain);
     bool asynchronousRemote = infrastructureSpec.workflowType == WorkflowType::RemoteBatch;
     if (movingWindowsEnabled && (synchronousRemote || asynchronousRemote)) {
-      InputSpec taskMovingWindowOutput{ taskSpec.taskName, TaskRunner::createTaskDataOrigin(taskSpec.detectorName, true), TaskRunner::createTaskDataDescription(taskSpec.taskName), Lifetime::Sporadic };
+      InputSpec taskMovingWindowOutput{ createUserInputSpec(DataSourceType::TaskMovingWindow, taskSpec.detectorName, taskSpec.taskName) };
       tasksOutputMap.insert({ DataSpecUtils::label(taskMovingWindowOutput), taskMovingWindowOutput });
     }
   }
 
   for (const auto& ppTaskSpec : infrastructureSpec.postProcessingTasks | std::views::filter(&PostProcessingTaskSpec::active)) {
-    InputSpec ppTaskOutput{ ppTaskSpec.taskName,
-                            PostProcessingDevice::createPostProcessingDataOrigin(ppTaskSpec.detectorName),
-                            PostProcessingDevice::createPostProcessingDataDescription(ppTaskSpec.taskName),
-                            Lifetime::Sporadic };
+    InputSpec ppTaskOutput{ createUserInputSpec(DataSourceType::PostProcessingTask, ppTaskSpec.detectorName, ppTaskSpec.taskName) };
     tasksOutputMap.insert({ DataSpecUtils::label(ppTaskOutput), ppTaskOutput });
   }
 
@@ -779,27 +769,18 @@ void InfrastructureGenerator::generatePostProcessing(WorkflowSpec& workflow, con
   }
 }
 
-template <typename Type>
-auto createSinkInput(const std::string& detectorName, const std::string& name) -> framework::InputSpec
-{
-  const auto outputSpec = Type::createOutputSpec(detectorName, name);
-  auto input = DataSpecUtils::matchingInput(outputSpec);
-  input.binding = name;
-  return input;
-}
-
 void InfrastructureGenerator::generateBookkeepingQualitySink(WorkflowSpec& workflow, const InfrastructureSpec& infrastructureSpec)
 {
   framework::Inputs sinkInputs{};
 
   for (const auto& checkSpec : infrastructureSpec.checks | std::views::filter(&CheckSpec::active) | std::views::filter(&CheckSpec::exportToBookkeeping)) {
     ILOG(Debug, Support) << "Adding input to BookkeepingSink from check " << checkSpec.checkName << " and detector: " << checkSpec.detectorName << ENDM;
-    sinkInputs.emplace_back(createSinkInput<Check>(checkSpec.detectorName, checkSpec.checkName));
+    sinkInputs.emplace_back(createUserInputSpec(DataSourceType::Check, checkSpec.detectorName, checkSpec.checkName));
   }
 
   for (const auto& aggregatorSpec : infrastructureSpec.aggregators | std::views::filter(&AggregatorSpec::active) | std::views::filter(&AggregatorSpec::exportToBookkeeping)) {
     ILOG(Debug, Support) << "Adding input to BookkeepingSink from aggregator " << aggregatorSpec.aggregatorName << " and detector: " << aggregatorSpec.detectorName << ENDM;
-    sinkInputs.emplace_back(createSinkInput<Aggregator>(aggregatorSpec.detectorName, aggregatorSpec.aggregatorName));
+    sinkInputs.emplace_back(createUserInputSpec(DataSourceType::Aggregator, aggregatorSpec.detectorName, aggregatorSpec.aggregatorName));
   }
 
   if (sinkInputs.empty()) {