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

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

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

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	str	The name of the source jet collection 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

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 : str
        The name of the source jet collection 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.
    '''
    source_jets: 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

    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 not (self.floats_to_copy or self.ints_to_copy):
            raise ValueError('You must provide at least one float or int to copy')

    def to_ca(self):
        ca = ComponentAccumulator()
        dr_str = f'deltaRTo{self.source_name}'
        dpt_str = f'deltaPtTo{self.source_name}'
        ca.addEventAlgo(
            CompFactory.JetMatcherAlg(
                f'{self.source_name}To{self.target_collection}CopyAlg',
                targetJet=self.target_collection,
                sourceJets=[self.source_jets],
                floatsToCopy={f:f for f in self.floats_to_copy},
                intsToCopy={i:i for i in self.ints_to_copy},
                dR=dr_str,
                dPt=dpt_str,
                ptPriorityWithDeltaR=self.pt_priority_with_delta_r,
                sourceMinimumPt=self.source_minimum_pt
            )
        )

        return ca

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
decorate_tracks	bool	Whether to decorate the tracks objects directly with the aux outputs.	True
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.
    decorate_tracks : bool
        Whether to decorate the tracks objects directly with the aux outputs.
    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
    decorate_tracks: bool = True
    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 = self.flags.BTagging.OutputFiles.Prefix + self.jet_collection.replace("Jets", "")
        elif self.target == "Jet":
            self.deco_alg = CompFactory.FlavorTagDiscriminants.JetTagDecoratorAlg
            self.container = self.jet_collection
        else:
            raise ValueError(f"Unknown target {self.target}")

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

    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,
                    decorateTracks=self.decorate_tracks,
                    variableRemapping=self.remap,
                )
            )
        )
        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

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'))

        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,
                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,
            )
        )
        return ca