File indexing completed on 2026-08-12 08:24:58
0001 """Test ePIC stack/workflow configuration."""
0002
0003 import yaml
0004
0005 from aid2e.utilities.epic_utils.epic_stack_config import (
0006 EpicLayerConfig,
0007 EpicJobDefinition,
0008 EpicStageDefinition,
0009 EpicBranchDefinition,
0010 EpicWorkflowDefinition,
0011 EpicWorkflowsConfiguration,
0012 )
0013 from aid2e.utilities.workflows.dag_executor import DAGExecutor
0014
0015 def _sample_epic_workflow_payload() -> dict:
0016 """Build ePIC workflow configuration."""
0017 return {
0018 "workflows": [
0019 {
0020 "name": "imaging_optimization",
0021 "description": "Optimize the number of AstroPix layers in the BIC",
0022 "stack_type": "epic",
0023 "branches": [
0024 {
0025 "name": "photon_phi_resolution",
0026 "description": "Calculate phi resolution for single photons",
0027 "stages": [
0028 {
0029 "name": "geo",
0030 "description": "Check for overlaps in modified geomtry",
0031 "jobs" : [
0032 {
0033 "name": "geo_job",
0034 "layers": [
0035 {
0036 "layer": "geo",
0037 "inputs": [
0038 "{{geometry_dir}}/install/share/epic/epic.xml"
0039 ],
0040 "outputs": [
0041 "{{execution_dir}}/geo.overlaps.txt"
0042 ]
0043 }
0044 ],
0045 "payload": {
0046 "evaluator_type": "stack",
0047 "stack_type": "epic",
0048 "job_id": "geo"
0049 }
0050 }
0051 ]
0052 },
0053 {
0054 "name": "sim",
0055 "description": "Simulate single photons",
0056 "jobs": [
0057 {
0058 "name": "sim_job",
0059 "layers": [
0060 {
0061 "layer": "sim",
0062 "inputs": [
0063 "inputs/central_photons.py"
0064 ],
0065 "outputs": [
0066 "{{execution_dir}}/central_photons.edm4hep.root"
0067 ]
0068 }
0069 ],
0070 "payload": {
0071 "evaluator_type": "stack",
0072 "stack_type": "epic",
0073 "job_id": "sim"
0074 }
0075 }
0076 ]
0077 },
0078 {
0079 "name": "rec_and_ana",
0080 "description": "Run reconstruction and analysis",
0081 "jobs": [
0082 {
0083 "name": "rec_ana_job",
0084 "layers": [
0085 {
0086 "layer": "rec",
0087 "inputs": [
0088 "{{outputs[sim:sim_job:sim]}}/central_photons.edm4hep.root"
0089 ],
0090 "outputs": [
0091 "{{execution_dir}}/central_photons.edm4eic.root"
0092 ],
0093 "arguments": [
0094 "-Pnthreads=8",
0095 "-Peicrecon:LogLevel=debug"
0096 ]
0097 },
0098 {
0099 "layer": "ana",
0100 "inputs": [
0101 "{{outputs[rec_and_ana:rec_ana_job:rec]}}/central_photons.edm4eic.root"
0102 ],
0103 "outputs": [
0104 "{{execution_dir}}/central_photons.hist.root"
0105 ],
0106 "arguments": [
0107 "-c phi",
0108 "-s 22"
0109 ],
0110 "command": "scripts/bic_angular_reso.py",
0111 "rule": "python {{command}} -i {{inputs}} -o {{outputs}} {{arguments}}"
0112 }
0113 ],
0114 "payload": {
0115 "evaluator_type": "stack",
0116 "stack_type": "epic",
0117 "job_id": "rec_ana"
0118 }
0119 }
0120 ]
0121 }
0122 ]
0123 }
0124 ]
0125 }
0126 ]
0127 }
0128
0129 def test_epic_workflows_validation(tmp_path):
0130 """Validate workflow model."""
0131 config_path = tmp_path / "epic_workflow.config"
0132 config_path.write_text(yaml.safe_dump(_sample_epic_workflow_payload()))
0133
0134 payload = None
0135 with open(config_path, 'r') as data:
0136 payload = yaml.safe_load(data)
0137 assert payload is not None
0138
0139 config = EpicWorkflowsConfiguration(**payload)
0140 assert 1 == len(config.workflows)
0141 assert 1 == len(config.workflows[0].branches)
0142 assert 3 == len(config.workflows[0].branches[0].stages)
0143 assert 1 == len(config.workflows[0].branches[0].stages[0].jobs)
0144 assert 1 == len(config.workflows[0].branches[0].stages[1].jobs)
0145 assert 1 == len(config.workflows[0].branches[0].stages[2].jobs)
0146 assert 1 == len(config.workflows[0].branches[0].stages[0].jobs[0].layers)
0147 assert 1 == len(config.workflows[0].branches[0].stages[1].jobs[0].layers)
0148 assert 2 == len(config.workflows[0].branches[0].stages[2].jobs[0].layers)
0149
0150
0151 assert isinstance(config, EpicWorkflowsConfiguration)
0152 assert isinstance(config.workflows[0], EpicWorkflowDefinition)
0153 assert isinstance(config.workflows[0].branches[0], EpicBranchDefinition)
0154 assert isinstance(config.workflows[0].branches[0].stages[0], EpicStageDefinition)
0155 assert isinstance(config.workflows[0].branches[0].stages[0].jobs[0], EpicJobDefinition)
0156 assert isinstance(config.workflows[0].branches[0].stages[0].jobs[0].layers[0], EpicLayerConfig)
0157
0158
0159 assert "epic" == config.workflows[0].stack_type
0160 assert "photon_phi_resolution" == config.workflows[0].branches[0].name
0161 assert "geo" == config.workflows[0].branches[0].stages[0].name
0162 assert "epic" == config.workflows[0].branches[0].stages[1].jobs[0].payload["stack_type"]
0163 assert "rec" == config.workflows[0].branches[0].stages[2].jobs[0].layers[0].layer
0164 assert "-s 22" == config.workflows[0].branches[0].stages[2].jobs[0].layers[1].arguments[1]
0165
0166 def test_epic_executor_from_config(tmp_path):
0167 """Create DAGExecutor from epic workflow."""
0168 config_path = tmp_path / "epic_workflow.config"
0169 config_path.write_text(yaml.safe_dump(_sample_epic_workflow_payload()))
0170
0171 payload = None
0172 with open(config_path, 'r') as data:
0173 payload = yaml.safe_load(data)
0174 assert payload is not None
0175
0176 config = payload["workflows"][0]
0177 workflow = EpicWorkflowDefinition(**config)
0178 executor = DAGExecutor(workflow, base_output_dir=tmp_path)
0179 assert executor is not None
0180 assert "photon_phi_resolution" == executor.workflow.branches[0].name
