extract filters/transformations into helpers; merge identical functions

Author: aalkin

Framework/Core/include/Framework/AnalysisSupportHelpers.h

@@ -39,11 +39,6 @@ struct AnalysisSupportHelpers {
                                          std::vector<InputSpec> const& requestedSpecials,
                                          std::vector<InputSpec>& requestedAODs,
                                          DataProcessorSpec& publisher);
-  static void addMissingOutputsToAnalysisCCDBFetcher(std::vector<OutputSpec> const& providedSpecials,
-                                                     std::vector<InputSpec> const& requestedSpecials,
-                                                     std::vector<InputSpec>& requestedAODs,
-                                                     std::vector<InputSpec>& requestedDYNs,
-                                                     DataProcessorSpec& publisher);
   static void addMissingOutputsToBuilder(std::vector<InputSpec> const& requestedSpecials,
                                          std::vector<InputSpec>& requestedAODs,
                                          std::vector<InputSpec>& requestedDYNs,

Framework/Core/include/Framework/DataSpecViews.h

@@ -35,6 +35,25 @@ static auto filter_matching(auto const& provided)
 {
   return std::views::filter([&provided](auto const& input) { return std::any_of(provided.begin(), provided.end(), [&input](auto const& output) { return DataSpecUtils::match(input, output); }); });
 }
+
+static auto filter_string_params_with(std::string match)
+{
+  return std::views::filter([match](auto const& param) {
+    return (param.type == VariantType::String) && (param.name.find(match) != std::string::npos);
+  });
+}
+
+static auto input_to_output_specs()
+{
+  return std::views::transform([](auto const& input){ return DataSpecUtils::asOutputSpec(input); });
+}
+
+static auto params_to_input_specs()
+{
+  return std::views::transform([](auto const& param) {
+    return DataSpecUtils::fromMetadataString(param.defaultValue.template get<std::string>());
+  });
+}
 } // namespace o2::framework::views
 //
 namespace o2::framework::sinks

Framework/Core/src/AnalysisSupportHelpers.cxx

@@ -128,9 +128,7 @@ void AnalysisSupportHelpers::addMissingOutputsToReader(std::vector<OutputSpec> c
 {
   requestedInputs |
     views::filter_not_matching(providedOutputs) | // filter the inputs that are already provided
-    std::views::transform([](auto const& req) {   // create outputspecs for unmatched inputs
-      return DataSpecUtils::asOutputSpec(req);
-    }) |
+    views::input_to_output_specs() |
     sinks::update_output_list{publisher.outputs}; // append them to the publisher outputs
 }
 
@@ -141,20 +139,14 @@ void AnalysisSupportHelpers::addMissingOutputsToSpawner(std::vector<OutputSpec>
 {
   requestedSpecials |
     views::filter_not_matching(providedSpecials) | // filter the inputs that are already provided
-    std::views::transform([](auto const& req) {    // create outputspecs for unmatched inputs
-      return DataSpecUtils::asOutputSpec(req);
-    }) |
+    views::input_to_output_specs() |
     sinks::append_to(publisher.outputs); // append them to the publisher outputs
 
   std::vector<InputSpec> additionalInputs;
   for (auto& input : requestedSpecials | views::filter_not_matching(providedSpecials)) {
     input.metadata |
-      std::views::filter([](auto const& param) { // filter config params that are strings starting with "input:"
-        return (param.type == VariantType::String) && (param.name.find("input:") != std::string::npos);
-      }) |
-      std::views::transform([](auto const& param) { // parse them into InputSpecs
-        return DataSpecUtils::fromMetadataString(param.defaultValue.template get<std::string>());
-      }) |
+      views::filter_string_params_with("input:") |
+      views::params_to_input_specs() |
       sinks::update_input_list{additionalInputs}; // store into a temporary
   }
   additionalInputs | sinks::update_input_list{requestedAODs};    // update requestedAODs
@@ -167,55 +159,14 @@ void AnalysisSupportHelpers::addMissingOutputsToBuilder(std::vector<InputSpec> c
                                                         DataProcessorSpec& publisher)
 {
   requestedSpecials |
-    std::views::transform([](auto const& req) { // create outputspecs for inputs
-      return DataSpecUtils::asOutputSpec(req);
-    }) |
+    views::input_to_output_specs() |
     sinks::append_to{publisher.outputs}; // append them to the publisher outputs
 
   std::vector<InputSpec> additionalInputs;
   for (auto const& input : requestedSpecials) {
     input.metadata |
-      std::views::filter([](auto const& param) { // filter config params that are strings starting with "input:"
-        return (param.type == VariantType::String) && (param.name.find("input:") != std::string::npos);
-      }) |
-      std::views::transform([](auto const& param) { // parse them into InputSpecs
-        return DataSpecUtils::fromMetadataString(param.defaultValue.template get<std::string>());
-      }) |
-      sinks::update_input_list{additionalInputs}; // store into a temporary
-  }
-
-  additionalInputs | sinks::update_input_list(publisher.inputs); // update publisher inputs
-  // FIXME: until we have a single list of pairs
-  additionalInputs |
-    views::partial_match_filter(AODOrigins) |
-    sinks::update_input_list{requestedAODs}; // update requestedAODs
-  additionalInputs |
-    views::partial_match_filter(header::DataOrigin{"DYN"}) |
-    sinks::update_input_list{requestedDYNs}; // update requestedDYNs
-}
-
-void AnalysisSupportHelpers::addMissingOutputsToAnalysisCCDBFetcher(
-  std::vector<OutputSpec> const& providedSpecials,
-  std::vector<InputSpec> const& requestedSpecials,
-  std::vector<InputSpec>& requestedAODs,
-  std::vector<InputSpec>& requestedDYNs,
-  DataProcessorSpec& publisher)
-{
-  requestedSpecials |
-    std::views::transform([](auto const& req) { // create outputspecs for inputs
-      return DataSpecUtils::asOutputSpec(req);
-    }) |
-    sinks::append_to{publisher.outputs}; // append them to the publisher outputs
-
-  std::vector<InputSpec> additionalInputs;
-  for (auto& input : requestedSpecials) {
-    input.metadata |
-      std::views::filter([](auto const& param) { // filter config params that are strings starting with "input:"
-        return (param.type == VariantType::String) && (param.name.find("input:") != std::string::npos);
-      }) |
-      std::views::transform([](auto const& param) { // parse them into InputSpecs
-        return DataSpecUtils::fromMetadataString(param.defaultValue.template get<std::string>());
-      }) |
+      views::filter_string_params_with("input:") |
+      views::params_to_input_specs() |
       sinks::update_input_list{additionalInputs}; // store into a temporary
   }
 

Framework/Core/src/ArrowSupport.cxx

@@ -658,7 +658,7 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
         // FIXME: it should be made more generic, so it does not need replacement...
         // FIXME how can I make the lookup depend on DYN tables as well??
         analysisCCDB->algorithm = PluginManager::loadAlgorithmFromPlugin("O2FrameworkCCDBSupport", "AnalysisCCDBFetcherPlugin", ctx);
-        AnalysisSupportHelpers::addMissingOutputsToAnalysisCCDBFetcher({}, ac.analysisCCDBInputs, ac.requestedAODs, ac.requestedDYNs, *analysisCCDB);
+        AnalysisSupportHelpers::addMissingOutputsToBuilder(ac.analysisCCDBInputs, ac.requestedAODs, ac.requestedDYNs, *analysisCCDB);
       }
 
       if (writer != workflow.end()) {

Framework/Core/src/WorkflowHelpers.cxx

@@ -397,7 +397,7 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
   ac.requestedTIMs | views::filter_not_matching(ac.providedTIMs) | sinks::append_to{ac.analysisCCDBInputs};
   DeploymentMode deploymentMode = DefaultsHelpers::deploymentMode();
   if (deploymentMode != DeploymentMode::OnlineDDS && deploymentMode != DeploymentMode::OnlineECS) {
-    AnalysisSupportHelpers::addMissingOutputsToAnalysisCCDBFetcher({}, ac.analysisCCDBInputs, ac.requestedAODs, ac.requestedTIMs, analysisCCDBBackend);
+    AnalysisSupportHelpers::addMissingOutputsToBuilder(ac.analysisCCDBInputs, ac.requestedAODs, ac.requestedTIMs, analysisCCDBBackend);
   }
 
   ac.requestedDYNs | views::filter_not_matching(ac.providedDYNs) | sinks::append_to{ac.spawnerInputs};