CA Configuration

You can configure the running of Component Accumulator based code using "blocks". Each block is a Python class that holds configuration keys as constructor arguments, and implements a to_ca() function that returns a ComponentAccumulator object. Blocks are defined in the FTagDumper/python/blocks/ directory, and inherit from the BaseBlock class.

Blocks are configured inside an optional "ca_blocks": [] list in the top-level JSON configuration. Each block is a dict with a "block" key that specifies the block name, and any other keys that are passed to the block constructor.

This page contains automatically generated documentation for the different CA blocks that can be configured with the dumpster.

BTagJetLinker dataclass

Bases: FTagDumper.python.blocks.BaseBlock.BaseBlock

Matches a jet collection to a b-tagging collection.

See BTagToJetLinkerAlg.h

Parameters:

Name	Type	Description	Default
jet_collection	str	The name of the jet collection to match. If not provided, it will be taken from the dumper config.	None
old_link	str	The name of the old link to match. If not provided, it will be set to {jet_collection}Jets.btaggingLink.	None
new_link	str	The name of the new link to match. If not provided, it will be set to BTagging_{jet_collection}.jetLink.	None

Source code in FTagDumper/python/blocks/BTagJetLinker.py

@dataclass
class BTagJetLinker(BaseBlock):
    """Matches a jet collection to a b-tagging collection.

    See [BTagToJetLinkerAlg.h](https://gitlab.cern.ch/npond/athena/-/blob/master/PhysicsAnalysis/JetTagging/FlavorTagDiscriminants/FlavorTagDiscriminants/BTagToJetLinkerAlg.h)

    Parameters
    ----------
    jet_collection : str
        The name of the jet collection to match. If not provided, it will be taken from the dumper config.
    old_link : str
        The name of the old link to match. If not provided, it will be set to `{jet_collection}Jets.btaggingLink`.
    new_link : str
        The name of the new link to match. If not provided, it will be set to `BTagging_{jet_collection}.jetLink`.
    """

    jet_collection: str = None
    old_link: str = None
    new_link: str = None

    def __post_init__(self):
        if self.jet_collection is None:
            self.jet_collection = self.dumper_config["jet_collection"]
            self.jet_collection = self.jet_collection.replace('Jets','')
        if self.old_link is None:
            self.old_link = f'{self.jet_collection}Jets.btaggingLink'
        if self.new_link is None:
            self.new_link = f'BTagging_{self.jet_collection}.jetLink'


    def to_ca(self):
        ca = ComponentAccumulator()
        ca.addEventAlgo(
            CompFactory.BTagToJetLinkerAlg(
                f'jetToBTagFor{self.jet_collection}',
                newLink=self.new_link,
                oldLink=self.old_link
            )
        )

        return ca

EventSelector dataclass

Bases: FTagDumper.python.blocks.BaseBlock.BaseBlock

Performs event cleaning, pileup reweighting, object calibration, overlap removal, and a subsequent event selection. This is based on the TopCPToolkit, see makeRecoConfiguration in TtbarCPalgoConfigBlocksAnalysis.py

Note that this block can and should drop events from being being processed in the dumper algorithm, but processing can only be stopped when using setMainAlgSeqToSeqAND from dumper.py

Parameters:

Name	Type	Description	Default
selections	dict[str, list[str]]	A list of selection cuts to apply, as documented in the TopCPToolkit. These selections should only use basic cuts that rely on electron, muon, jet or MET information. For example, we could add the following to our json config to do an e+mu ttbar selection: "selections": [ "SUM_EL_N_MU_N 27000 >= 1", "EL_N 27000 == 1", "MU_N 27000 == 1", "JET_N 20000 >= 1", "JET_N 20000 >= 2", "EL_N 28000 == 1", "MU_N 28000 == 1", "JET_N 20000 == 2" ]	required

Source code in FTagDumper/python/blocks/EventSelector.py

@dataclass
class EventSelector(BaseBlock):
    '''Performs event cleaning, pileup reweighting, object calibration,
    overlap removal, and a subsequent event selection.
    This is based on the TopCPToolkit, see `makeRecoConfiguration` in
    [TtbarCPalgoConfigBlocksAnalysis.py](https://gitlab.cern.ch/atlasphys-top/reco/TopCPToolkit/-/blob/main/source/TopCPToolkit/python/TtbarCPalgoConfigBlocksAnalysis.py)

    Note that this block can and should drop events from being being processed
    in the dumper algorithm, but processing can only be stopped when using
    `setMainAlgSeqToSeqAND` from
    [dumper.py](https://gitlab.cern.ch/atlas-flavor-tagging-tools/training-dataset-dumper/-/blob/main/FTagDumper/python/dumper.py)

    Parameters
    ----------
    selections : dict[str, list[str]]
        A list of selection cuts to apply, as documented in the
        [TopCPToolkit](https://topcptoolkit.docs.cern.ch/settings/eventselection/).
        These selections should only use basic cuts that rely on electron, muon, jet or
        MET information. For example, we could add the following to our json config to
        do an e+mu ttbar selection:

        "selections": [
            "SUM_EL_N_MU_N 27000 >= 1",
            "EL_N 27000 == 1",
            "MU_N 27000 == 1",
            "JET_N 20000 >= 1",
            "JET_N 20000 >= 2",
            "EL_N 28000 == 1",
            "MU_N 28000 == 1",
            "JET_N 20000 == 2"
        ]
    '''

    selections: list[str]

    def to_ca(self) -> ComponentAccumulator:
        config = ConfigSequence()

        factory = ConfigFactory()
        factory.addAlgConfigBlock('EventCounting', EventCountingBlock)
        factory.addAlgConfigBlock(
            'CountedEventSelection', configure_counted_event_selection
        )

        cutflow_output_path = "cutflow_counts.json"
        config += factory.makeConfig(
            'EventCounting',
            count_id='total',
            counts_output_path=cutflow_output_path
        )

        config += factory.makeConfig('CommonServices')
        config.setOptionValue('.runSystematics', False)

        config += factory.makeConfig(
            'EventCounting',
            count_id='common services',
            counts_output_path=cutflow_output_path
        )

        # event cleaning
        if not self.flags.Input.isMC:
            config += factory.makeConfig('EventCleaning')
            config.setOptionValue('.runEventCleaning', True)

        config += factory.makeConfig(
            'EventCounting',
            count_id='event cleaning',
            counts_output_path=cutflow_output_path
        )

        # pileup reweighting
        config += factory.makeConfig('PileupReweighting')

        config += factory.makeConfig(
            'EventCounting',
            count_id='pileup reweighting',
            counts_output_path=cutflow_output_path
        )

        # electrons
        config += factory.makeConfig(
            'Electrons', containerName='AnaElectrons'
        )
        config.setOptionValue('.crackVeto', True)

        config += factory.makeConfig(
            'Electrons.WorkingPoint',
            containerName='AnaElectrons',
            selectionName='loose'
        )
        config.setOptionValue('.identificationWP', "TightLH")
        config.setOptionValue('.isolationWP', "Tight_VarRad")

        config += factory.makeConfig(
            'Electrons.WorkingPoint',
            containerName='AnaElectrons',
            selectionName='tight'
        )
        config.setOptionValue('.identificationWP', "TightLH")
        config.setOptionValue('.isolationWP', "Tight_VarRad")

        config += factory.makeConfig(
            'Electrons.PtEtaSelection',
            containerName='AnaElectrons'
        )
        config.setOptionValue('.selectionDecoration', 'selectPtEta')
        config.setOptionValue('.minPt', 27e3)

        # muons
        config += factory.makeConfig('Muons', containerName='AnaMuons')

        config += factory.makeConfig(
            'Muons.WorkingPoint',
            containerName='AnaMuons',
            selectionName='loose'
        )
        config.setOptionValue('.quality', "Medium")
        config.setOptionValue('.isolation', "NonIso")

        config += factory.makeConfig(
            'Muons.WorkingPoint',
            containerName='AnaMuons',
            selectionName='tight'
        )
        config.setOptionValue('.quality', "Medium")
        config.setOptionValue('.isolation', "Tight_VarRad")

        config += factory.makeConfig(
            'Muons.PtEtaSelection', containerName='AnaMuons'
        )
        config.setOptionValue('.selectionDecoration', 'selectPtEta')
        config.setOptionValue('.minPt', 27e3)
        config.setOptionValue('.maxEta', 2.5)

        # jets
        config += factory.makeConfig(
            'Jets', containerName='AnaJets', jetCollection='AntiKt4EMPFlowJets'
        )
        config.setOptionValue('.runGhostMuonAssociation', True)
        config.setOptionValue('.runNNJvtUpdate', True)

        config += factory.makeConfig('Jets.JVT', containerName='AnaJets')

        config += factory.makeConfig(
            'Jets.PtEtaSelection', containerName='AnaJets'
        )
        config.setOptionValue('.selectionDecoration', 'selectPtEta')
        config.setOptionValue('.minPt', 20e3)
        config.setOptionValue('.maxEta', 2.5)

        # missing ET
        config += factory.makeConfig('MissingET', containerName='AnaMET')
        config.setOptionValue('.electrons', 'AnaElectrons.tight')
        config.setOptionValue('.muons', 'AnaMuons.tight')
        config.setOptionValue('.jets', 'AnaJets')
        config.setOptionValue('.photons', '')
        config.setOptionValue('.taus', '')

        # overlap removal
        config += factory.makeConfig('OverlapRemoval')
        config.setOptionValue('.electrons', 'AnaElectrons.tight')
        config.setOptionValue('.muons', 'AnaMuons.tight')
        config.setOptionValue('.photons', '')
        config.setOptionValue('.jets', 'AnaJets.baselineJvt')
        config.setOptionValue('.taus', '')
        config.setOptionValue('.inputLabel', 'preselectOR')
        config.setOptionValue('.outputLabel', 'passesOR')

        # event selection
        if (
            len(self.selections) > 0

            # don't select anythings when running a test, we most likely have
            # an empty output and the test will fail
            and not os.environ.get("DUMPSTER_CI_TEST") == "1"
        ):
            config += factory.makeConfig(
                'CountedEventSelection',
                selection_id='selection',
                selections=self.selections,
                electrons='AnaElectrons.tight',
                muons='AnaMuons.tight',
                jets='AnaJets.baselineJvt',
                met='AnaMET',
                cutflow_counts_output_path=cutflow_output_path,
                count_total_before_selection=False
            )

        config_accumulator = ConfigAccumulator(
            algSeq=None,
            autoconfigFromFlags=self.flags,
            noSysSuffix=True,
            noSystematics=True
        )
        assert config_accumulator.CA is not None

        config.fullConfigure(config_accumulator)
        return config_accumulator.CA

ExampleDecorator dataclass

Bases: FTagDumper.python.blocks.BaseBlock.BaseBlock

Example starting point to add info to objects

Parameters:

Name	Type	Description	Default
decorator_value	float	The value to decorate to a jet	None
target_collection	str	The name of the target jet collection to match to. If not provided, it will be taken from the dumper config.	None
decorator_name	str	Name of decoration to add	'exampleDecorator'

Source code in FTagDumper/python/blocks/ExampleDecorator.py

@dataclass
class ExampleDecorator(BaseBlock):
    '''Example starting point to add info to objects

    Parameters
    ----------
    decorator_value : float, optional
        The value to decorate to a jet
    target_collection : str, optional
        The name of the target jet collection to match to. If not
        provided, it will be taken from the dumper config.
    decorator_name : str
        Name of decoration to add
    '''
    decorator_value: str = None
    target_collection: str = None
    decorator_name: str = "exampleDecorator"

    def __post_init__(self):
        if self.target_collection is None:
            self.target_collection = self.dumper_config["jet_collection"]

    def to_ca(self):
        ca = ComponentAccumulator()
        opts = {}
        if self.decorator_value is not None:
            opts['decoratorValue'] = self.decorator_value
        target = f'{self.target_collection}.{self.decorator_name}'
        ca.addEventAlgo(
            CompFactory.ExampleDecoratorAlg(
                f'{self.target_collection}{self.decorator_name}Decorator',
                exampleDecorator=target,
                **opts,
            )
        )

        return ca

FixedConeAssociation dataclass

Bases: FTagDumper.python.blocks.BaseBlock.BaseBlock

Run a fixed cone track association. The fixed cone association is used for the training of the nntc model.

To run the nntc, you can use the following configuration: EMPFlow_fixedcone.json. The size of the cone is parametrized by the fixedConeRadius parameter.

Parameters:

Name	Type	Description	Default
radius	float	The radius of the fixed cone.	0.5

Source code in FTagDumper/python/blocks/FixedConeAssociation.py

@dataclass
class FixedConeAssociation(BaseBlock):
    """
    Run a fixed cone track association.
    The fixed cone association is used for the training of the nntc model.

    To run the nntc, you can use the following configuration: EMPFlow_fixedcone.json.
    The size of the cone is parametrized by the fixedConeRadius parameter.

    Parameters
    ----------
    radius : float
        The radius of the fixed cone.
    """
    radius: float = 0.5

    def to_ca(self):    
        ca = ComponentAccumulator()
        an = 'FixedCone_Tracks'
        temp_jets='AntiKt4EMPFlowJets'
        fs_tracks='InDetTrackParticles'
        bc='BTagging_AntiKt4EMPFlow'
        trackOnJetDecorator = f'{temp_jets}.{an}ForBTagging'

        ca.merge(JetParticleFixedConeAssociationAlgCfg(
            self.flags,
            fixedConeRadius=self.radius,
            JetCollection=temp_jets,
            InputParticleCollection=fs_tracks,
            OutputParticleDecoration=trackOnJetDecorator.split('.')[-1]
        ))
        Copier = CompFactory.FlavorTagDiscriminants.BTagTrackLinkCopyAlg
        copier = Copier(
            'TrackCopier',
            jetTracks=trackOnJetDecorator,
            btagTracks=f'{bc}.{an}',
            jetLinkName=f'{bc}.jetLink'
        )
        ca.addEventAlgo(copier)
        return ca

FlowSelector dataclass

Bases: FTagDumper.python.blocks.BaseBlock.BaseBlock

Matches a jet collection to a b-tagging collection.

Parameters:

Name	Type	Description	Default
jet_collection	str	The name of the jet collection to match. If not provided, it will be taken from the dumper config.	None

Source code in FTagDumper/python/blocks/FlowSelector.py

@dataclass
class FlowSelector(BaseBlock):
    """Matches a jet collection to a b-tagging collection.

    Parameters
    ----------
    jet_collection : str
        The name of the jet collection to match. If not provided, it will be taken from the dumper config.
    """

    jet_collection: str = None

    def __post_init__(self):
        if self.jet_collection is None:
            self.jet_collection = self.dumper_config["jet_collection"]


    def to_ca(self):
        ca = ComponentAccumulator()
        ca.addEventAlgo(
            CompFactory.FlowSelectorAlg(
                name="NeutralFlowSelectorAlg",
                Constituents=f'{self.jet_collection}.constituentLinks',
                OutConstituentsNeutral=f'{self.jet_collection}.neutralConstituentLinks',
                OutConstituentsCharged=f'{self.jet_collection}.chargedConstituentLinks'
            )
        )

        return ca

FoldHashDecorator dataclass

Bases: FTagDumper.python.blocks.BaseBlock.BaseBlock

Decorate jets with a hash for kfold training and evaluation.

See FoldDecoratorAlg.cxx and FoldDecoratorConfig.py

Parameters:

Name	Type	Description	Default
jet_collection	str	Name of the jet collection to decorate. Use the dumper's jet_collection if not provided.	None
prefix	str	Prefix to add to the hash.	''

Source code in FTagDumper/python/blocks/FoldHashDecorator.py

@dataclass
class FoldHashDecorator(BaseBlock):
    """Decorate jets with a hash for kfold training and evaluation.

    See [FoldDecoratorAlg.cxx](https://gitlab.cern.ch/atlas/athena/-/blob/main/PhysicsAnalysis/JetTagging/FlavorTagDiscriminants/src/FoldDecoratorAlg.cxx)
    and [FoldDecoratorConfig.py](https://gitlab.cern.ch/atlas/athena/-/blob/main/PhysicsAnalysis/JetTagging/FlavorTagDiscriminants/python/FoldDecoratorConfig.py)

    Parameters
    ----------
    jet_collection : str, optional
        Name of the jet collection to decorate. Use the dumper's jet_collection
        if not provided.
    prefix : str, optional
        Prefix to add to the hash.
    """
    jet_collection: str = None
    prefix: str = ''

    def __post_init__(self):
        if self.jet_collection is None:
            self.jet_collection = self.dumper_config['jet_collection']

    def to_ca(self):
        ca = ComponentAccumulator()
        ca.merge(FoldDecoratorCfg(self.flags, jetCollection=self.jet_collection, prefix=self.prefix))
        return ca

GNNAuxTaskMapper dataclass

Bases: FTagDumper.python.blocks.BaseBlock.BaseBlock

Schedule GNNAuxTaskDecoratorAlg. This alg reads aux task decos from the btagging object and copies them over to the tracks. For technical reasons (AFT-686) we can't decorate the track directly.

Note that track_aux_tasks must be provided and the destination variable name must be different from the source variable name, which is used to decorate the btagging object, otherwise the scheduler gets confused about the types.

Parameters:

Name	Type	Description	Default
btagging_container	str | None	BTagging container name to read the aux-task outputs from.	'BTagging_AntiKt4EMPFlow'
track_container	str	Track container name to decorate with the aux-task outputs.	'InDetTrackParticles'
track_links	str	TrackLinks name associated with the BTagging object.	'GN2v01_TrackLinks'
track_aux_tasks	dict(str, str)	Map between aux-task decorations for jets and decorations for tracks.	required

Source code in FTagDumper/python/blocks/GNNAuxTaskMapper.py

@dataclass
class GNNAuxTaskMapper(BaseBlock):
    """
    Schedule GNNAuxTaskDecoratorAlg. This alg reads aux task decos
    from the btagging object and copies them over to the tracks.
    For technical reasons (AFT-686) we can't decorate the track directly.

    Note that track_aux_tasks must be provided and the destination
    variable name must be different from the source variable name,
    which is used to decorate the btagging object, otherwise the 
    scheduler gets confused about the types.

    Parameters
    ----------
    btagging_container : str | None
        BTagging container name to read the aux-task outputs from.
    track_container : str
        Track container name to decorate with the aux-task outputs.
    track_links : str
        TrackLinks name associated with the BTagging object.
    track_aux_tasks : dict(str, str)
        Map between aux-task decorations for jets and decorations for tracks.
    """
    track_aux_tasks: Mapping[str, str]
    btagging_container: str = "BTagging_AntiKt4EMPFlow"
    track_container: str = "InDetTrackParticles"
    track_links: str = "GN2v01_TrackLinks"

    def __post_init__(self):
        self.deco_alg = CompFactory.FlavorTagDiscriminants.GNNAuxTaskDecoratorAlg
        self.name = 'GNNAuxTaskDecoratorAlg_' + '_'.join(self.track_aux_tasks.keys())

    def to_ca(self):
        ca = ComponentAccumulator()
        ca.addEventAlgo(
            self.deco_alg(
                name=self.name,
                btagging_container=self.btagging_container,
                track_container=self.track_container,
                track_links=self.track_links,
                track_aux_tasks=self.track_aux_tasks
            )
        )
        return ca

GeneratorWeights dataclass

Bases: FTagDumper.python.blocks.BaseBlock.BaseBlock

Add truth weights to EventInfo

Source code in FTagDumper/python/blocks/GeneratorWeights.py

@dataclass
class GeneratorWeights(BaseBlock):
    """
    Add truth weights to EventInfo
    """
    def to_ca(self):
        return self.PMGTruthWeightAlgCfg(self.flags)

    def PMGTruthWeightAlgCfg(self, flags):
        ca = ComponentAccumulator()
        # we need both the systematics service and the metadata service to
        # make tis tool work.
        ca.addService(
            CompFactory.CP.SystematicsSvc(
                name="SystematicsSvc",
                sigmaRecommended=1,
                systematicsRegex='.*',
            )
        )
        ca.merge(MetaDataSvcCfg(flags))
        ca.addEventAlgo(
            CompFactory.CP.PMGTruthWeightAlg(
                name="PMGTruthWeightAlg",
                truthWeightTool=CompFactory.PMGTools.PMGTruthWeightTool(
                    name="PMGTruthWeightTool"
                ),
                decoration = 'generatorWeight_%SYS%',
            )
        )
        return ca

JetMatcher dataclass

Bases: FTagDumper.python.blocks.BaseBlock.BaseBlock

Matches jets from a single source collection to jets in a target collection. For more details, see JetMatcherAlg.cxx

Parameters:

Name	Type	Description	Default
source_jets	list[str]	The name of the source collections to match.	required
source_name	str	The name of a singular jet in the source collection. Used for generating the variables deltaRTo{source_name} and deltaPtTo{source_name}. If not provided, it will be the same as the source collection.	None
target_collection	str	The name of the target jet collection to match to. If not provided, it will be taken from the dumper config.	None
floats_to_copy	list[str]	List of float variables to copy from the source jets to the target jets	None
ints_to_copy	list[str]	List of int variables to copy from the source jets to the target jets	None
pt_priority_with_delta_r	float	The priority of the pt variable when matching jets based on deltaR. Disabled by default.	-1
particle_link_name	str	Name of the vector of links to matched particles.	None

Source code in FTagDumper/python/blocks/JetMatcher.py

@dataclass
class JetMatcher(BaseBlock):
    '''Matches jets from a single source collection to jets in a target collection. For more
    details, see
    [JetMatcherAlg.cxx](https://gitlab.cern.ch/atlas-flavor-tagging-tools/training-dataset-dumper/-/blob/main/FTagDumper/src/JetMatcherAlg.cxx)

    Parameters
    ----------
    source_jets : list[str]
        The name of the source collections to match.
    source_name : str, optional
        The name of a singular jet in the source collection. Used for generating the variables
        `deltaRTo{source_name}` and `deltaPtTo{source_name}`. If not provided, it will be
        the same as the source collection.
    target_collection : str, optional
        The name of the target jet collection to match to. If not provided, it will be taken 
        from the dumper config.
    floats_to_copy : list[str], optional
        List of float variables to copy from the source jets to the target jets
    ints_to_copy : list[str], optional
        List of int variables to copy from the source jets to the target jets
    pt_priority_with_delta_r : float, optional
        The priority of the pt variable when matching jets based on deltaR.
        Disabled by default.
    particle_link_name : str, optional
        Name of the vector of links to matched particles.
    '''
    source_jets: list[str]
    source_name: str = None
    target_collection: str = None
    floats_to_copy: list[str] = None
    ints_to_copy: list[str] = None
    pt_priority_with_delta_r: float = -1
    source_minimum_pt: float = 0.0
    particle_link_name: str = None

    def __post_init__(self):
        if self.source_name is None:
            self.source_name = self.source_jets
        if self.target_collection is None:
            self.target_collection = self.dumper_config["jet_collection"]
        if self.floats_to_copy is None:
            self.floats_to_copy = []
        if self.ints_to_copy is None:
            self.ints_to_copy = []
        if self.particle_link_name is None:
            self.particle_link_name = ""

    def to_ca(self):
        ca = ComponentAccumulator()
        dr_str = f'deltaRTo{self.source_name}'
        dpt_str = f'deltaPtTo{self.source_name}'
        to_suffix = f'From{self.source_name}'
        match_str = f'matchedTo{self.source_name}'
        def to(f):
            return f + to_suffix
        ca.addEventAlgo(
            CompFactory.JetMatcherAlg(
                f'{self.source_name}To{self.target_collection}CopyAlg',
                targetJet=self.target_collection,
                sourceJets=self.source_jets,
                floatsToCopy={f: to(f) for f in self.floats_to_copy},
                intsToCopy={i: to(i) for i in self.ints_to_copy},
                dR=dr_str,
                dPt=dpt_str,
                match=match_str,
                ptPriorityWithDeltaR=self.pt_priority_with_delta_r,
                sourceMinimumPt=self.source_minimum_pt,
                particleLink=self.particle_link_name,
            )
        )

        return ca

JetReco dataclass

Bases: FTagDumper.python.blocks.BaseBlock.BaseBlock

Reconstructs a jet collection

Parameters:

Name	Type	Description	Default
Duplicate_EMTopojets	(bool,)	The jet collection to be reconstructed already exists in the AOD but links are broken. Need to duplicate to reconstruct jets with a new name	''

Source code in FTagDumper/python/blocks/JetReco.py

@dataclass
class JetReco(BaseBlock):
    '''Reconstructs a jet collection
    Parameters
    ----------
    Duplicate_EMTopojets : bool,
        The jet collection to be reconstructed already exists in the AOD but links are broken. 
        Need to duplicate to reconstruct jets with a new name
    '''
    Duplicate_EMTopojets: str = ''

    def __post_init__(self):
        if self.Duplicate_EMTopojets is None:
            self.Duplicate_EMTopojets = self.dumper_config['Duplicate_Jets']

    def to_ca(self):

        acc = ComponentAccumulator()




        if self.Duplicate_EMTopojets:
            AntiKt4EMTopov2 = AntiKt4EMTopo.clone(suffix="v2")

            acc.merge(JetRecCfg(self.flags, AntiKt4EMTopov2))

        else:
            acc.merge(JetRecCfg(self.flags, AntiKt4EMTopo))
        return acc

MultifoldTagger dataclass

Bases: FTagDumper.python.blocks.BaseBlock.BaseBlock

Run a multifold tagger.

Parameters:

Name	Type	Description	Default
nn_paths	list[str]	List of paths to the neural network files.	required
target	str	Whether to tag the BTagging object or the Jet.	'BTagging'
jet_collection	str | None	Name of the jet collection to decorate. If None, uses the jet collection from the dumper configuration.	None
remap	dict	Remap input and output variable names.	None
fold_hash_name	str	Name of the fold hash variable.	'jetFoldHash'
constituents	str	Name of the constituent container.	'InDetTrackParticles'

Source code in FTagDumper/python/blocks/MultifoldTagger.py

@dataclass
class MultifoldTagger(BaseBlock):
    """Run a multifold tagger.

    Parameters
    ----------
    nn_paths : list[str]
        List of paths to the neural network files.
    target : str
        Whether to tag the BTagging object or the Jet.
    jet_collection : str | None
        Name of the jet collection to decorate. If None, uses the jet collection
        from the dumper configuration.
    remap : dict
        Remap input and output variable names.
    fold_hash_name : str
        Name of the fold hash variable.
    constituents : str
        Name of the constituent container.
    """
    nn_paths: list[str]
    target: str = "BTagging"
    jet_collection: str = None
    remap: dict = None
    fold_hash_name: str = "jetFoldHash"
    constituents: str = "InDetTrackParticles"

    def __post_init__(self):

        if self.jet_collection is None:
            self.jet_collection = self.dumper_config["jet_collection"]
        if self.remap is None:
            self.remap = {}

        if self.target == "BTagging":
            self.deco_alg = CompFactory.FlavorTagDiscriminants.BTagDecoratorAlg
            self.container = 'BTagging_{}'.format(
                self.jet_collection.replace("Jets", ""))
            self.track_link_type = "TRACK_PARTICLE"
        elif self.target == "Jet":
            self.deco_alg = CompFactory.FlavorTagDiscriminants.JetTagDecoratorAlg
            self.container = self.jet_collection
            self.track_link_type = "IPARTICLE"
        else:
            raise ValueError(f"Unknown target {self.target}")

        self.name = '_'.join(Path(self.nn_paths[0]).parts[2:-1])

        if "BTagTrackToJetAssociator" in self.remap:
            self.name = f'{self.remap["BTagTrackToJetAssociator"]}_{self.name}' 

    def to_ca(self):
        ca = ComponentAccumulator()
        ca.addEventAlgo(
            self.deco_alg(
                name=f'{self.name}_Alg',
                container=self.container,
                constituentContainer=self.constituents,
                decorator=CompFactory.FlavorTagDiscriminants.MultifoldGNNTool(
                    name=f'{self.name}_Tool',
                    foldHashName=self.fold_hash_name,
                    nnFiles=self.nn_paths,
                    variableRemapping=self.remap,
                    trackLinkType=self.track_link_type
                )
            )
        )
        return ca

MuonCorrDecorator dataclass

Bases: FTagDumper.python.blocks.BaseBlock.BaseBlock

Do Muon Correction with BJetCalibrationTool .

Source code in FTagDumper/python/blocks/MuonCorrDecorator.py

@dataclass
class MuonCorrDecorator(BaseBlock):
    """Do Muon Correction with BJetCalibrationTool .

    """

    def to_ca(self):
        ca = ComponentAccumulator()
        ca.addService(CompFactory.CP.SystematicsSvc("SystematicsSvc"))
        ca.merge(BJetCalibrationAlgCfg(
            self.flags,
            jets = 'AntiKt10UFOCSSKSoftDropBeta100Zcut10Jets',
            jetsOut = 'AntiKt10UFOCSSKSoftDropBeta100Zcut10Jets_muonCorr',
            muons = 'Muons',
            doPtCorr = False,
        ))

        return ca

ShrinkingConeAssociation dataclass

Bases: FTagDumper.python.blocks.BaseBlock.BaseBlock

Associate tracks to jets using the shrinking cone algorithm.

More info: https://ftag.docs.cern.ch/algorithms/taggers/inputs/

Source code in FTagDumper/python/blocks/ShrinkingConeAssociation.py

@dataclass
class ShrinkingConeAssociation(BaseBlock):
    """
    Associate tracks to jets using the shrinking cone algorithm.

    More info: https://ftag.docs.cern.ch/algorithms/taggers/inputs/
    """
    jet_collection: str = None
    track_collection: str = "InDetTrackParticles"
    link_name: str = "DeltaRTracks"

    def __post_init__(self):
        if self.jet_collection is None:
            self.jet_collection = self.dumper_config["jet_collection"]

    def to_ca(self):
        ca = ComponentAccumulator()
        ca.merge(JetParticleAssociationAlgCfg(
            self.flags,
            JetCollection=self.jet_collection,
            InputParticleCollection=self.track_collection,
            OutputParticleDecoration=self.link_name,
        ))

        return ca

SoftElectronsDecorator dataclass

Bases: FTagDumper.python.blocks.BaseBlock.BaseBlock

Do electrons decoration and jet association.

See SoftElectronDecoratorAlg.h See SoftElectronTruthDecoratorAlg.h See FTagGhostElectronAssociationAlg.h

Parameters:

Name	Type	Description	Default
jet_collection	str	The name of the jet collection to which the electrons are associated. If not provided, it will be taken from the dumper config.	None
electrons_collection	str	The name of the electron collection to decorate. Default is "GhostFTagElectrons".	'GhostFTagElectrons'
decorate_vars	bool	Whether to decorate the variables. Default is False.	False
decorate_truth_vars	bool	Whether to decorate the truth variables. Default is False.	False
link_electrons	bool	Whether to link the electrons to the jet collection. Default is False.	False

Source code in FTagDumper/python/blocks/SoftElectronsDecorator.py

@dataclass
class SoftElectronsDecorator(BaseBlock):
    """Do electrons decoration and jet association.

    See [SoftElectronDecoratorAlg.h](https://gitlab.cern.ch/atlas/athena/-/blob/main/PhysicsAnalysis/JetTagging/FlavorTagDiscriminants/FlavorTagDiscriminants/SoftElectronDecoratorAlg.h)
    See [SoftElectronTruthDecoratorAlg.h](https://gitlab.cern.ch/atlas/athena/-/blob/main/PhysicsAnalysis/JetTagging/FlavorTagDiscriminants/FlavorTagDiscriminants/SoftElectronTruthDecoratorAlg.h)
    See [FTagGhostElectronAssociationAlg.h](https://gitlab.cern.ch/atlas/athena/-/blob/main/PhysicsAnalysis/JetTagging/FlavorTagDiscriminants/FlavorTagDiscriminants/FTagGhostElectronAssociationAlg.h)

    Parameters
    ----------
    jet_collection : str
        The name of the jet collection to which the electrons are associated. If not provided, it will be taken from the dumper config.
    electrons_collection : str
        The name of the electron collection to decorate. Default is "GhostFTagElectrons".
    decorate_vars : bool
        Whether to decorate the variables. Default is False.
    decorate_truth_vars : bool
        Whether to decorate the truth variables. Default is False.
    link_electrons : bool
        Whether to link the electrons to the jet collection. Default is False.
    """

    jet_collection: str = None
    electrons_collection: str = "GhostFTagElectrons"
    decorate_vars: bool = False
    decorate_truth_vars: bool = False
    link_electrons: bool = False


    def __post_init__(self):
        if self.jet_collection is None:
            self.jet_collection = self.dumper_config["jet_collection"]

    def to_ca(self):
        ca = ComponentAccumulator()
        if self.decorate_vars:
            ca.addEventAlgo(
                CompFactory.FlavorTagDiscriminants.SoftElectronDecoratorAlg(
                    'SoftElectronDecoratorAlg'
                )
            )
        if self.decorate_truth_vars:
            ca.addEventAlgo(
                CompFactory.FlavorTagDiscriminants.SoftElectronTruthDecoratorAlg(
                    'SoftElectronTruthDecoratorAlg'
                )
            )
        if self.link_electrons:
            ca.addEventAlgo(
                CompFactory.FlavorTagDiscriminants.FTagGhostElectronAssociationAlg(
                    'FTagGhostElectronAssociationAlg',
                    jetContainer='AntiKt4EMPFlowJets',
                    outElectrons="AntiKt4EMPFlowJets.GhostFTagElectrons",
                )
            )

        return ca

TrackFlowOverlapRemoval dataclass

Bases: FTagDumper.python.blocks.BaseBlock.BaseBlock

Matches a jet collection to a b-tagging collection.

Parameters:

Name	Type	Description	Default
jet_collection	str	The name of the jet collection to match. If not provided, it will be taken from the dumper config.	None

Source code in FTagDumper/python/blocks/TrackFlowOverlapRemoval.py

@dataclass
class TrackFlowOverlapRemoval(BaseBlock):
    """
    Matches a jet collection to a b-tagging collection.

    Parameters
    ----------
    jet_collection : str
        The name of the jet collection to match. If not provided, it will be taken from the dumper config.
    """

    jet_collection: str = None
    track_container: str = None

    def __post_init__(self):
        if self.jet_collection is None:
            self.jet_collection = self.dumper_config['jet_collection']



    def to_ca(self):
        ca = ComponentAccumulator()
        ca.addEventAlgo(
            CompFactory.TrackFlowOverlapRemovalAlg(
                name='nonConstituentAdder',
                Tracks=f'{self.jet_collection}.{self.track_container}',
                Constituents=f'{self.jet_collection}.constituentLinks',
                OutTracks=f'{self.jet_collection}.nonConstituentTracks'
            )
        )

        return ca

Trackless dataclass

Bases: FTagDumper.python.blocks.BaseBlock.BaseBlock

Source code in FTagDumper/python/blocks/Trackless.py

@dataclass
class Trackless(BaseBlock):

    def to_ca(self):
        ca = ComponentAccumulator()
        ca.addEventAlgo(CompFactory.HitELDecoratorAlg('HitElementLinkDecorator'))
        ca.addEventAlgo(CompFactory.HitdRMinDecoratorAlg('HitTrackMindR'))
        ca.addEventAlgo(CompFactory.HitTruthDecoratorAlg('HitTruthDecorator'))

        return ca

TruthLabelling dataclass

Bases: FTagDumper.python.blocks.BaseBlock.BaseBlock

Re-run the FTAG truth labelling.

Detailed FTAG truth variables, such as ftagTruthOriginLabel and ftagTruthVertexIndex are normally produced when running derivations. With this block you can reproduce these labels as part of your dumping job, for example if the labelling code has been updated and you don't want to run expensive derivations.

For more info see TruthParticleDecoratorAlg and TrackTruthDecoratorAlg in Athena, and this page in the FTAG docs.

Parameters:

Name	Type	Description	Default
suffix	str	Suffix to append to the truth labelling variables.	required

Examples:

{
    "ca_blocks": [{"block": "TruthLabelling", "suffix": "tdd"}]
}

Remember to add the new ftagTruthOriginLabelTDD etc variables to your output variables, or alternatively if you want to avoid the presence of new variable names in your h5 you can remap variable names as follows:

"tracks": [
    {
        ...
        "edm_names": {
            "ftagTruthOriginLabel": "ftagTruthOriginLabelTDD",
            "ftagTruthVertexIndex": "ftagTruthVertexIndexTDD"
        }
    }
]

Source code in FTagDumper/python/blocks/TruthLabelling.py

@dataclass
class TruthLabelling(BaseBlock):
    """
    Re-run the FTAG truth labelling.

    Detailed FTAG truth variables, such as `ftagTruthOriginLabel` and `ftagTruthVertexIndex`
    are normally produced when running derivations. With this block you can reproduce these
    labels as part of your dumping job, for example if the labelling code has been updated
    and you don't want to run expensive derivations.

    For more info see `TruthParticleDecoratorAlg` and `TrackTruthDecoratorAlg` in Athena,
    and [this page](https://ftag.docs.cern.ch/algorithms/labelling/track_labels/) in the 
    FTAG docs.

    Parameters
    ----------
    suffix : str
        Suffix to append to the truth labelling variables.

    Examples
    --------
    ```json
    {
        "ca_blocks": [{"block": "TruthLabelling", "suffix": "tdd"}]
    }
    ```

    Remember to add the new `ftagTruthOriginLabelTDD` etc variables to your output variables,
    or alternatively if you want to avoid the presence of new variable names in your h5 you 
    can remap variable names as follows:

    ```json
    "tracks": [
        {
            ...
            "edm_names": {
                "ftagTruthOriginLabel": "ftagTruthOriginLabelTDD",
                "ftagTruthVertexIndex": "ftagTruthVertexIndexTDD"
            }
        }
    ]
    ```
    """
    suffix: str

    def to_ca(self):
        ca = ComponentAccumulator()
        trackTruthOriginTool = CompFactory.InDet.InDetTrackTruthOriginTool(
            isFullPileUpTruth=False
        )
        ca.addEventAlgo(
            CompFactory.FlavorTagDiscriminants.TruthParticleDecoratorAlg(
                "TruthParticleDecoratorAlg",
                trackTruthOriginTool=trackTruthOriginTool,
                ftagTruthOriginLabel="ftagTruthOriginLabel" + self.suffix,
                ftagTruthVertexIndex="ftagTruthVertexIndex" + self.suffix,
                ftagTruthTypeLabel="ftagTruthTypeLabel" + self.suffix,
                ftagTruthSourceLabel="ftagTruthSourceLabel" + self.suffix,
                ftagTruthParentBarcode="ftagTruthParentBarcode" + self.suffix,
            )
        )
        ca.addEventAlgo(
            CompFactory.FlavorTagDiscriminants.TrackTruthDecoratorAlg(
                "TrackTruthDecoratorAlg",
                trackTruthOriginTool=trackTruthOriginTool,
                truthLeptonTool=CompFactory.TruthClassificationTool(
                    "TruthClassificationTool"),
                acc_ftagTruthVertexIndex="ftagTruthVertexIndex" + self.suffix,
                acc_ftagTruthTypeLabel="ftagTruthTypeLabel" + self.suffix,
                acc_ftagTruthSourceLabel="ftagTruthSourceLabel" + self.suffix,
                acc_ftagTruthParentBarcode="ftagTruthParentBarcode" + self.suffix,
                dec_ftagTruthOriginLabel="ftagTruthOriginLabel" + self.suffix,
                dec_ftagTruthVertexIndex="ftagTruthVertexIndex" + self.suffix,
                dec_ftagTruthTypeLabel="ftagTruthTypeLabel" + self.suffix,
                dec_ftagTruthSourceLabel="ftagTruthSourceLabel" + self.suffix,
                dec_ftagTruthBarcode="ftagTruthBarcode" + self.suffix,
                dec_ftagTruthParentBarcode="ftagTruthParentBarcode" + self.suffix,
                dec_ftagTruthMuonOriginLabel="ftagTruthMuonOriginLabel" + self.suffix,
            )
        )
        return ca

TruthTauMatcher dataclass

Bases: FTagDumper.python.blocks.BaseBlock.BaseBlock

Matches truth taus to reco. jets and copies variables from truth taus to the matched reco. jets. For more details, see TruthTauMatcher.cxx

Parameters:

Name	Type	Description	Default
truth_tau_collection	str	The name of the truth tau collection to use. By default "TruthTaus"	'TruthTaus'
reco_jet_collection	str	The name of the reco jet collection to which the truth taus are matched. By default the "jet_collection" from the config file used is taken.	None
floats_to_copy	list[str]	List of float variables to copy from the truth taus to the matched reco. jets. By default None	None
ints_to_copy	list[str]	List of integer variables to copy from the truth taus to the matched reco. jets. By default None	None
uints_to_copy	list[str]	List of unsigned integer variables to copy from the truth taus to the matched reco. jets. By default None	None
chars_to_copy	list[str]	List of character variables to copy from the truth taus to the matched reco. jets. By default None	None
particle_link_name	str	Name of the vector of links to matched particles.	None

Source code in FTagDumper/python/blocks/TruthTauMatcher.py

@dataclass
class TruthTauMatcher(BaseBlock):
    """Matches truth taus to reco. jets and copies variables from truth taus to the
    matched reco. jets. For more details, see
    [TruthTauMatcher.cxx](https://gitlab.cern.ch/atlas-flavor-tagging-tools/training-dataset-dumper/-/blob/main/FTagDumper/src/TruthTauMatcher.cxx)


    Parameters
    ----------
    truth_tau_collection : str
        The name of the truth tau collection to use. By default "TruthTaus"
    reco_jet_collection : str
        The name of the reco jet collection to which the truth taus are matched.
        By default the "jet_collection" from the config file used is taken.
    floats_to_copy : list[str], optional
        List of float variables to copy from the truth taus to the matched
        reco. jets. By default None
    ints_to_copy : list[str], optional
        List of integer variables to copy from the truth taus to the matched
        reco. jets. By default None
    uints_to_copy : list[str], optional
        List of unsigned integer variables to copy from the truth taus to the
        matched reco. jets. By default None
    chars_to_copy : list[str], optional
        List of character variables to copy from the truth taus to the matched
        reco. jets. By default None
    particle_link_name : str, optional
        Name of the vector of links to matched particles.
    """

    truth_tau_collection: str = "TruthTaus"
    reco_jet_collection: str = None
    floats_to_copy: list[str] = None
    ints_to_copy: list[str] = None
    uints_to_copy: list[str] = None
    chars_to_copy: list[str] = None
    max_delta_R: float = 0.3
    min_truth_tau_pT: float = 0.0
    particle_link_name: str = None

    def __post_init__(self):
        if self.truth_tau_collection is None or isinstance(
            self.truth_tau_collection, str
        ):
            self.truth_tau_collection = ["TruthTaus"]
            self.truth_tau_collection_name = self.truth_tau_collection[0]
        if self.reco_jet_collection is None:
            self.reco_jet_collection = self.dumper_config["jet_collection"]
        if self.floats_to_copy is None:
            self.floats_to_copy = []
        if self.ints_to_copy is None:
            self.ints_to_copy = []
        if self.uints_to_copy is None:
            self.uints_to_copy = []
        if self.chars_to_copy is None:
            self.chars_to_copy = []
        if self.particle_link_name is None:
            self.particle_link_name = ""

    def to_ca(self):
        ca = ComponentAccumulator()
        dr_str = f"deltaRTo{self.truth_tau_collection_name}"
        dpt_str = f"deltaPtTo{self.truth_tau_collection_name}"
        to_suffix = f"From{self.truth_tau_collection_name}"
        match_str = f"matchedTo{self.truth_tau_collection_name}"

        def to(f):
            return f + to_suffix

        ca.addEventAlgo(
            CompFactory.TruthTauMatcherAlg(
                f"{self.truth_tau_collection_name}To{self.reco_jet_collection}CopyAlg",
                recoJets=self.reco_jet_collection,
                truthTaus=self.truth_tau_collection,
                floatsToCopy={f: to(f) for f in self.floats_to_copy},
                intsToCopy={i: to(i) for i in self.ints_to_copy},
                uintsToCopy={i: to(i) for i in self.uints_to_copy},
                charsToCopy={i: to(i) for i in self.chars_to_copy},
                dR=dr_str,
                dPt=dpt_str,
                match=match_str,
                particleLink=self.particle_link_name,
                maxDeltaR=self.max_delta_R,
                minTruthTauPt=self.min_truth_tau_pT,
            )
        )

        return ca