Skip to content

DPL: add initial set of helpers for analysis #1746

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ktf merged 1 commit into from
Feb 28, 2019
Merged

DPL: add initial set of helpers for analysis #1746

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Framework/Core/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ endif()

set(SRCS
src/AODReaderHelpers.cxx
src/AnalysisHelpers.cxx
src/BoostOptionsRetriever.cxx
src/ConfigParamsHelper.cxx
src/CompletionPolicy.cxx
Expand Down
44 changes: 44 additions & 0 deletions Framework/Core/include/Framework/AnalysisHelpers.h
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
// Copyright CERN and copyright holders of ALICE O2. This software is
// distributed under the terms of the GNU General Public License v3 (GPL
// Version 3), copied verbatim in the file "COPYING".
//
// See http://alice-o2.web.cern.ch/license for full licensing information.
//
// In applying this license 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.
#ifndef o2_framework_AnalysisHelpers_H_DEFINED
#define o2_framework_AnalysisHelpers_H_DEFINED

#include "Framework/AnalysisHelpers.h"

#include <ROOT/RDataFrame.hxx>
#include <string>

using namespace ROOT::RDF;

namespace o2
{
namespace framework
{
class TableConsumer;
}

namespace analysis
{

/// Do a single loop on all the entries of the @a input table
ROOT::RDataFrame doSingleLoopOn(std::unique_ptr<framework::TableConsumer>& input);

/// Do a double loop on all the entries with the same value for the \a grouping
/// of the @a input table, where the entries for the outer index are prefixed
/// with `<name>_` while the entries for the inner loop are prefixed with
/// `<name>bar_`.
ROOT::RDataFrame doSelfCombinationsWith(std::unique_ptr<framework::TableConsumer>& input,
std::string name = "p",
std::string grouping = "eventID");

} // namespace analysis
} // namespace o2

#endif // o2_framework_AnalysisHelpers_H_DEFINED
45 changes: 45 additions & 0 deletions Framework/Core/src/AnalysisHelpers.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
// Copyright CERN and copyright holders of ALICE O2. This software is
// distributed under the terms of the GNU General Public License v3 (GPL
// Version 3), copied verbatim in the file "COPYING".
//
// See http://alice-o2.web.cern.ch/license for full licensing information.
//
// In applying this license 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.
#include "Framework/AnalysisHelpers.h"
#include "Framework/RCombinedDS.h"
#include "Framework/TableBuilder.h"
#include "Framework/TableConsumer.h"

#include <ROOT/RDataFrame.hxx>
#include <ROOT/RArrowDS.hxx>

using namespace ROOT::RDF;

namespace o2
{
namespace analysis
{

ROOT::RDataFrame doSingleLoopOn(std::unique_ptr<framework::TableConsumer>& input)
{
auto flat = std::make_unique<RArrowDS>(input->asArrowTable(), std::vector<std::string>{});
ROOT::RDataFrame rdf(std::move(flat));
return rdf;
}

ROOT::RDataFrame doSelfCombinationsWith(std::unique_ptr<framework::TableConsumer>& input, std::string name, std::string grouping)
{
auto table = input->asArrowTable();
using Index = RCombinedDSBlockJoinIndex<int>;
auto left = std::make_unique<RArrowDS>(table, std::vector<std::string>{});
auto right = std::make_unique<RArrowDS>(table, std::vector<std::string>{});
auto combined = std::make_unique<RCombinedDS>(std::move(left), std::move(right), std::move(std::make_unique<Index>(grouping, true, BlockCombinationRule::StrictlyUpper)), name + "_", name + "bar_");

ROOT::RDataFrame rdf(std::move(combined));
return rdf;
}

} // namespace analysis
} // namespace o2
120 changes: 62 additions & 58 deletions Framework/TestWorkflows/src/o2D0Analysis.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,85 +8,89 @@
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
#include "Framework/runDataProcessing.h"
#include "Framework/RCombinedDS.h"
#include "Framework/TableBuilder.h"
#include "Framework/AnalysisHelpers.h"

#include <ROOT/RDataFrame.hxx>
#include <ROOT/RArrowDS.hxx>

using namespace ROOT::RDF;
using namespace o2::framework;

// A dummy workflow which creates a few of the tables proposed by Ruben,
// using ARROW
WorkflowSpec defineDataProcessing(ConfigContext const& specs)
{
// Workflow definition. A workflow can be one or more DataProcessors
// each implementing (part of) an analysis. Each DataProcessor has
// (at least) a name, some Inputs, some Outputs and they get arranged
// together accordingly.
WorkflowSpec workflow{
/// Minimal analysis example
// Multiple DataProcessor specs
// can be provided per workflow
DataProcessorSpec{
// The name of my analysis
"d0-analysis",
{
Inputs{
// Dangling inputs of type AOD will be automatically picked up
// by DPL and an extra reader device will be instanciated to
// read them.
InputSpec{ "DZeroFlagged", "AOD", "DZEROFLAGGED" },
// read them. In this particular case the signature
// AOD/DZEROFLAGGED is associated to Gianmichele's
// D0 candidates schema. The first string is just a label
// so that the algorithm can be in principle be reused for different
// kind of candidates.
InputSpec{ "candidates", "AOD", "DZEROFLAGGED" },
},
{},
// No outputs for the time being.
Outputs{},
AlgorithmSpec{
[](InitContext& setup) {
return [](ProcessingContext& ctx) {
auto s = ctx.inputs().get<TableConsumer>("DZeroFlagged");
/// From the handle, we construct the actual arrow table
/// which is then used as a source for the RDataFrame.
/// This is probably easy to change to a:
///
/// auto rdf = ctx.inputs().get<RDataSource>("xz");
auto table = s->asArrowTable();
using namespace ROOT::RDF;
// This is the actual per "message" loop, where a message could
// be the contents of a file or part of it.
// FIXME: Too much boilerplate.
adaptStateless([](InputRecord& inputs) {
auto input = inputs.get<TableConsumer>("candidates");

TFile f("result.root", "RECREATE");
// This does a single loop on all the candidates in the input message
// using a simple mask on the cand_type_ML column and does
// a simple 1D histogram of the filtered entries.
auto candidates = o2::analysis::doSingleLoopOn(input);

auto flatD0 = std::make_unique<RArrowDS>(table, std::vector<std::string>{});
ROOT::RDataFrame rdf1(std::move(flatD0));
/// A single loop to do an invariant mass plot, where we use the preselected
/// candidates
auto candFilter = [](int x) -> bool { return x & 0x1; };
auto h1 = rdf1.Filter(candFilter, { "cand_type_ML" }).Histo1D("inv_mass_ML");
h1->SetName("InvariantMass");
h1->Write();
auto h1 = candidates.Filter("(bool)(cand_type_ML & 0x1)").Histo1D("inv_mass_ML");

/// Double loops on all supported loop types.
using Index = RCombinedDSBlockJoinIndex<int>;
auto types = {
BlockCombinationRule::Anti,
BlockCombinationRule::Full,
BlockCombinationRule::Diagonal,
BlockCombinationRule::StrictlyUpper,
BlockCombinationRule::Upper,
};
// A few helpers
auto bothCandFilter = [](int x, int y) -> bool { return x & 0x1 && y & 0x1; };
auto delta = [](float x, float y) { return x - y; };
// A lambda function subtracting two quantities. This defines
// a function "delta" which can be invoked with
//
// delta(1,2)
//
// and will return 1 - 2.
auto delta = [](float x, float y) { return x - y; };

for (auto combinationType : types) {
auto d0 = std::make_unique<RArrowDS>(table, std::vector<std::string>{});
auto d0bar = std::make_unique<RArrowDS>(table, std::vector<std::string>{});
auto d0d0bar = std::make_unique<RCombinedDS>(std::move(d0), std::move(d0bar), std::move(std::make_unique<Index>("cand_evtID_ML", true, combinationType)), "d0_", "d0bar_");
// This does all the combinations for all the candidates which have
// the same value for cand_evtID_ML (the Event ID).
// d0_ is the prefix assigned to the outer variable of the double loop.
// d0bar_ is the prefix assigned to the inner variable of the double loop.
//
// The lines below will:
// * Filter the combinations according to some mask
// * Define a column delta_phi with the difference in phi between d0 and d0bar phi
// * Define a column delta_eta with the difference in phi between d0 and d0bar eta
// * Do two histograms with delta_phi, delta_eta
auto combinations = o2::analysis::doSelfCombinationsWith(input, "d0", "cand_evtID_ML");
auto deltas = combinations.Filter("d0_cand_type_ML & 0x1 && d0bar_cand_type_ML & 0x1")
.Define("delta_phi", delta, { "d0_phi_cand_ML", "d0bar_phi_cand_ML" })
.Define("delta_eta", delta, { "d0_eta_cand_ML", "d0bar_eta_cand_ML" });
auto h2 = deltas.Histo1D("delta_phi");
auto h3 = deltas.Histo1D("delta_eta");

ROOT::RDataFrame rdf2(std::move(d0d0bar));
auto combinatorics = rdf2.Filter(bothCandFilter, { "d0_cand_type_ML", "d0bar_cand_type_ML" })
.Define("delta_phi", delta, { "d0_phi_cand_ML", "d0bar_phi_cand_ML" })
.Define("delta_eta", delta, { "d0_eta_cand_ML", "d0bar_eta_cand_ML" });
auto h2 = combinatorics.Histo1D("delta_phi");
auto h3 = combinatorics.Histo1D("delta_eta");

std::string rule = RCombinedDSIndexHelpers::combinationRuleAsString(combinationType);
h2->SetName(("DeltaPhi/" + rule).c_str());
h2->Write();
h3->SetName(("DeltaEta/" + rule).c_str());
h3->Write();
}
};
} } }
// FIXME: For the moment we hardcode saving the histograms.
// In reality it should send the results as outputs to a downstream merger
// process which merges them as wished.
TFile f("result.root", "RECREATE");
h1->SetName("InvariantMass");
h1->Write();
h2->SetName("DeltaPhi");
h2->Write();
h3->SetName("DeltaEta");
h3->Write();
}) } }
};
return workflow;
}