使用APD和Snakemake访问分析产出数据

课程目标

学习如何使用APD Python包将分析产出数据与分析脚本对接
学习如何将APD集成到Snakemake工作流管理器中
另可参考HSF课程中关于Snakemake的入门介绍

分析产出数据（Analysis Production Data，APD）是一个Python包，用于与分析产出数据库交互，以确定每个任务的文件位置。它默认存在于LHCb环境和lb-conda中，也可通过pip或conda在其他环境中获取。APD的主要文档可在此处查看。

APD基础

与分析产出课程类似，要访问数据库需先获取认证令牌，可通过执行apd-login并遵循指示完成操作。

之后，你可以通过以下基本示例访问特定任务的输出：

>>> from apd import AnalysisData
>>> datasets = AnalysisData("dpa", "starterkit")
>>> bu2jpsik_24c4_magdown = datasets(name="bu2jpsik_24c4_magdown")
>>> bu2jpsik_24c4_magdown
['root://eoslhcb.cern.ch//eos/lhcb/grid/prod/lhcb/anaprod/lhcb/LHCb/Collision24/DATA.ROOT/00254511/0000/00254511_00000001_1.data.root', ...]

这将创建bu2jpsik_24c4_magdown，它是该任务输出的所有物理文件名称（PFN）的列表。之后你可以使用偏好的Python版ROOT（如PyROOT、Uproot等）访问这些文件。

然而，典型分析会包含多个任务，每个任务对应不同条件（如数据采集周期、LHCb磁铁极性、数据类型等）。因此，获取每个任务与其输出的映射会更有用。要获取可用于筛选任务的所有条件列表，可执行：

>>> datasets.summary()
{'tags': {'config': {'lhcb', 'mc'}, 'polarity': {'magup', 'magdown'}, 'eventtype': {'27163002', '94000000'}, 'datatype': {'2016', '2024'}, 'version': {'v0r0p4434304', 'v1r2161'}, 'name': {'bu2jpsik_24c4_magup', '2016_magup_promptmc_d02kk', 'bu2jpsik_24c4_magdown', '2016_magdown_promptmc_d02kk'}, 'state': {'ready'}}, 'analysis': 'starterkit', 'working_group': 'dpa', 'Number_of_files': 62, 'Bytecount': 3794798402}

then a mapping can be made by simply doing

>>> outputs = {name: datasets(name=name) for name in datasets.summary()["tags"]["name"]}
>>> outputs
{'bu2jpsik_24c4_magup': ['root://eoslhcb.cern.ch//eos/lhcb/grid/prod/lhcb/anaprod/lhcb/LHCb/Collision24/DATA.ROOT/00254507/0000/00254507_00000003_1.data.root', ...],
'2016_magup_promptmc_d02kk': ['root://eoslhcb.cern.ch//eos/lhcb/grid/prod/lhcb/anaprod/lhcb/MC/2016/D02KK.ROOT/00166693/0000/00166693_00000001_1.d02kk.root', ...],
'bu2jpsik_24c4_magdown': ['root://eoslhcb.cern.ch//eos/lhcb/grid/prod/lhcb/anaprod/lhcb/LHCb/Collision24/DATA.ROOT/00254511/0000/00254511_00000001_1.data.root', ...],
'2016_magdown_promptmc_d02kk': ['root://eoslhcb.cern.ch//eos/lhcb/grid/prod/lhcb/anaprod/lhcb/MC/2016/D02KK.ROOT/00166695/0000/00166695_00000001_1.d02kk.root', ...]}

之后你可以直接迭代产出中的所有文件，或将字典保存为JSON文件供后续使用。如需更细致的映射，可使用极性、数据类型、事件类型或其他标签创建形如{event_type: {year: {polarity: {[PFNs]}}}}的结构，以定位产出中的部分任务。

多重匹配

如果用于筛选任务的标签对应多个任务，为避免歧义，系统会抛出异常。

例如，在上述示例中：

>>> bu2jpsik_24c4 = datasets(datatype="2024")
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/cvmfs/lhcb.cern.ch/lib/var/lib/LbEnv/3526/stable/linux-64/lib/python3.12/site-packages/apd/analysis_data.py", line 399, in __call__
    raise ValueError("Error loading data: " + error_txt)
ValueError: Error loading data: 1 problem(s) found
{'datatype': '2024'}: 2 samples for the same configuration found, this is ambiguous:
    {'config': 'lhcb', 'polarity': 'magdown', 'eventtype': '94000000', 'datatype': '2024', 'version': 'v1r2161', 'name': 'bu2jpsik_24c4_magdown', 'state': 'ready'}
    {'config': 'lhcb', 'polarity': 'magup', 'eventtype': '94000000', 'datatype': '2024', 'version': 'v1r2161', 'name': 'bu2jpsik_24c4_magup', 'state': 'ready'}

相反，你需要做

>>> bu2jpsik_24c4_magup = datasets(datatype="2024", polarity="magup")
>>> bu2jpsik_24c4_magdown = datasets(datatype="2024", polarity="magdown")

Snakemake集成

LHCb中的许多分析使用Snakemake管理工作流。APD提供了额外的措施，用于将分析产出数据集成到Snakemake工作流中。

首先，可将Snakemake的APD工具导入Snakefile：

from apd.snakemake import get_analysis_data

数据集的访问方式与上文类似：

dataset = get_analysis_data("dpa", "starterkit")

通过指定额外参数，Snakemake可自动监控和处理数据集。以下是使用该技术的简单规则示例：

rule example_rule:
    input:
        data=lambda w: dataset(datatype=w.datatype, eventtype=w.eventtype, polarity=w.polarity)
    output:
        temp("filename_{config}_{datatype}_{eventtype}_{polarity}.root")
    shell:
        # Some script to run accessing the files
        "process_data.py -o {output} {input}"

完整的Snakefile示例可在此处查看。