File indexing completed on 2025-01-30 09:14:57
0001 import os
0002 import inspect
0003 from pathlib import Path
0004 import shutil
0005 import math
0006 import sys
0007 import tempfile
0008
0009 import pytest
0010
0011 from helpers import (
0012 dd4hepEnabled,
0013 hepmc3Enabled,
0014 geant4Enabled,
0015 edm4hepEnabled,
0016 podioEnabled,
0017 AssertCollectionExistsAlg,
0018 )
0019
0020 import acts
0021 from acts import UnitConstants as u
0022 from acts.examples import (
0023 ObjPropagationStepsWriter,
0024 TrackFinderNTupleWriter,
0025 SeedingPerformanceWriter,
0026 RootPropagationStepsWriter,
0027 RootParticleWriter,
0028 RootTrackParameterWriter,
0029 RootMaterialTrackWriter,
0030 RootMaterialWriter,
0031 RootSimHitWriter,
0032 RootTrackStatesWriter,
0033 RootTrackSummaryWriter,
0034 VertexNTupleWriter,
0035 RootMeasurementWriter,
0036 CsvParticleWriter,
0037 CsvSimHitWriter,
0038 CsvTrackWriter,
0039 CsvTrackingGeometryWriter,
0040 CsvMeasurementWriter,
0041 JsonMaterialWriter,
0042 JsonFormat,
0043 Sequencer,
0044 GenericDetector,
0045 )
0046 from acts.examples.odd import getOpenDataDetectorDirectory
0047
0048
0049 @pytest.mark.obj
0050 def test_obj_propagation_step_writer(tmp_path, trk_geo, conf_const, basic_prop_seq):
0051 with pytest.raises(TypeError):
0052 ObjPropagationStepsWriter()
0053
0054 obj = tmp_path / "obj"
0055 obj.mkdir()
0056
0057 s, alg = basic_prop_seq(trk_geo)
0058 w = conf_const(
0059 ObjPropagationStepsWriter,
0060 acts.logging.INFO,
0061 collection=alg.config.outputSummaryCollection,
0062 outputDir=str(obj),
0063 )
0064
0065 s.addWriter(w)
0066
0067 s.run()
0068
0069 assert len([f for f in obj.iterdir() if f.is_file()]) == s.config.events
0070 for f in obj.iterdir():
0071 assert f.stat().st_size > 1024
0072
0073
0074 @pytest.mark.csv
0075 def test_csv_particle_writer(tmp_path, conf_const, ptcl_gun):
0076 s = Sequencer(numThreads=1, events=10)
0077 evGen = ptcl_gun(s)
0078
0079 out = tmp_path / "csv"
0080
0081 out.mkdir()
0082
0083 s.addWriter(
0084 conf_const(
0085 CsvParticleWriter,
0086 acts.logging.INFO,
0087 inputParticles=evGen.config.outputParticles,
0088 outputStem="particle",
0089 outputDir=str(out),
0090 )
0091 )
0092
0093 s.run()
0094
0095 assert len([f for f in out.iterdir() if f.is_file()]) == s.config.events
0096 assert all(f.stat().st_size > 200 for f in out.iterdir())
0097
0098
0099 @pytest.mark.root
0100 def test_root_prop_step_writer(
0101 tmp_path, trk_geo, conf_const, basic_prop_seq, assert_root_hash
0102 ):
0103 with pytest.raises(TypeError):
0104 RootPropagationStepsWriter()
0105
0106 file = tmp_path / "prop_steps.root"
0107 assert not file.exists()
0108
0109 s, alg = basic_prop_seq(trk_geo)
0110 w = conf_const(
0111 RootPropagationStepsWriter,
0112 acts.logging.INFO,
0113 collection=alg.config.outputSummaryCollection,
0114 filePath=str(file),
0115 )
0116
0117 s.addWriter(w)
0118
0119 s.run()
0120
0121 assert file.exists()
0122 assert file.stat().st_size > 2**10 * 50
0123 assert_root_hash(file.name, file)
0124
0125
0126 @pytest.mark.root
0127 def test_root_particle_writer(tmp_path, conf_const, ptcl_gun, assert_root_hash):
0128 s = Sequencer(numThreads=1, events=10)
0129 evGen = ptcl_gun(s)
0130
0131 file = tmp_path / "particles.root"
0132
0133 assert not file.exists()
0134
0135 s.addWriter(
0136 conf_const(
0137 RootParticleWriter,
0138 acts.logging.INFO,
0139 inputParticles=evGen.config.outputParticles,
0140 filePath=str(file),
0141 )
0142 )
0143
0144 s.run()
0145
0146 assert file.exists()
0147 assert file.stat().st_size > 1024 * 10
0148 assert_root_hash(file.name, file)
0149
0150
0151 @pytest.mark.root
0152 def test_root_meas_writer(tmp_path, fatras, trk_geo, assert_root_hash):
0153 s = Sequencer(numThreads=1, events=10)
0154 evGen, simAlg, digiAlg = fatras(s)
0155
0156 out = tmp_path / "meas.root"
0157
0158 assert not out.exists()
0159
0160 config = RootMeasurementWriter.Config(
0161 inputMeasurements=digiAlg.config.outputMeasurements,
0162 inputClusters=digiAlg.config.outputClusters,
0163 inputSimHits=simAlg.config.outputSimHits,
0164 inputMeasurementSimHitsMap=digiAlg.config.outputMeasurementSimHitsMap,
0165 filePath=str(out),
0166 surfaceByIdentifier=trk_geo.geoIdSurfaceMap(),
0167 )
0168 s.addWriter(RootMeasurementWriter(level=acts.logging.INFO, config=config))
0169 s.run()
0170
0171 assert out.exists()
0172 assert out.stat().st_size > 40000
0173 assert_root_hash(out.name, out)
0174
0175
0176 @pytest.mark.root
0177 def test_root_simhits_writer(tmp_path, fatras, conf_const, assert_root_hash):
0178 s = Sequencer(numThreads=1, events=10)
0179 evGen, simAlg, digiAlg = fatras(s)
0180
0181 out = tmp_path / "meas.root"
0182
0183 assert not out.exists()
0184
0185 s.addWriter(
0186 conf_const(
0187 RootSimHitWriter,
0188 level=acts.logging.INFO,
0189 inputSimHits=simAlg.config.outputSimHits,
0190 filePath=str(out),
0191 )
0192 )
0193
0194 s.run()
0195 assert out.exists()
0196 assert out.stat().st_size > 2e4
0197 assert_root_hash(out.name, out)
0198
0199
0200 @pytest.mark.root
0201 def test_root_tracksummary_writer(tmp_path, fatras, conf_const):
0202 detector = GenericDetector()
0203 trackingGeometry = detector.trackingGeometry()
0204 field = acts.ConstantBField(acts.Vector3(0, 0, 2 * u.T))
0205 s = Sequencer(numThreads=1, events=10)
0206
0207 from truth_tracking_kalman import runTruthTrackingKalman
0208
0209
0210 runTruthTrackingKalman(
0211 trackingGeometry,
0212 field,
0213 digiConfigFile=Path(
0214 str(
0215 Path(__file__).parent.parent.parent.parent
0216 / "Examples/Algorithms/Digitization/share/default-smearing-config-generic.json"
0217 )
0218 ),
0219 outputDir=tmp_path,
0220 s=s,
0221 )
0222
0223
0224 s.addWriter(
0225 conf_const(
0226 RootTrackSummaryWriter,
0227 level=acts.logging.INFO,
0228 inputTracks="tracks",
0229 filePath=str(tmp_path / "track_summary_kf_no_truth.root"),
0230 )
0231 )
0232
0233 s.run()
0234 assert (tmp_path / "tracksummary_kf.root").exists()
0235 assert (tmp_path / "track_summary_kf_no_truth.root").exists()
0236
0237
0238 @pytest.mark.csv
0239 def test_csv_meas_writer(tmp_path, fatras, trk_geo, conf_const):
0240 s = Sequencer(numThreads=1, events=10)
0241 evGen, simAlg, digiAlg = fatras(s)
0242
0243 out = tmp_path / "csv"
0244 out.mkdir()
0245
0246 s.addWriter(
0247 conf_const(
0248 CsvMeasurementWriter,
0249 level=acts.logging.INFO,
0250 inputMeasurements=digiAlg.config.outputMeasurements,
0251 inputClusters=digiAlg.config.outputClusters,
0252 inputMeasurementSimHitsMap=digiAlg.config.outputMeasurementSimHitsMap,
0253 outputDir=str(out),
0254 )
0255 )
0256 s.run()
0257
0258 assert len([f for f in out.iterdir() if f.is_file()]) == s.config.events * 3
0259 assert all(f.stat().st_size > 10 for f in out.iterdir())
0260
0261
0262 @pytest.mark.csv
0263 def test_csv_simhits_writer(tmp_path, fatras, conf_const):
0264 s = Sequencer(numThreads=1, events=10)
0265 evGen, simAlg, digiAlg = fatras(s)
0266
0267 out = tmp_path / "csv"
0268 out.mkdir()
0269
0270 s.addWriter(
0271 conf_const(
0272 CsvSimHitWriter,
0273 level=acts.logging.INFO,
0274 inputSimHits=simAlg.config.outputSimHits,
0275 outputDir=str(out),
0276 outputStem="hits",
0277 )
0278 )
0279
0280 s.run()
0281 assert len([f for f in out.iterdir() if f.is_file()]) == s.config.events
0282 assert all(f.stat().st_size > 200 for f in out.iterdir())
0283
0284
0285 @pytest.mark.parametrize(
0286 "writer",
0287 [
0288 RootPropagationStepsWriter,
0289 RootParticleWriter,
0290 TrackFinderNTupleWriter,
0291 SeedingPerformanceWriter,
0292 RootTrackParameterWriter,
0293 RootMaterialTrackWriter,
0294 RootMeasurementWriter,
0295 RootMaterialWriter,
0296 RootSimHitWriter,
0297 RootTrackStatesWriter,
0298 RootTrackSummaryWriter,
0299 VertexNTupleWriter,
0300 SeedingPerformanceWriter,
0301 ],
0302 )
0303 @pytest.mark.root
0304 def test_root_writer_interface(writer, conf_const, tmp_path, trk_geo):
0305 assert hasattr(writer, "Config")
0306
0307 config = writer.Config
0308
0309 assert hasattr(config, "filePath")
0310 assert hasattr(config, "fileMode")
0311
0312 f = tmp_path / "target.root"
0313 assert not f.exists()
0314
0315 kw = {"level": acts.logging.INFO, "filePath": str(f)}
0316
0317 for k, _ in inspect.getmembers(config):
0318 if k.startswith("input"):
0319 kw[k] = "collection"
0320 if k == "surfaceByIdentifier":
0321 kw[k] = trk_geo.geoIdSurfaceMap()
0322
0323 assert conf_const(writer, **kw)
0324
0325 assert f.exists()
0326
0327
0328 @pytest.mark.parametrize(
0329 "writer",
0330 [
0331 CsvParticleWriter,
0332 CsvMeasurementWriter,
0333 CsvSimHitWriter,
0334 CsvTrackWriter,
0335 CsvTrackingGeometryWriter,
0336 ],
0337 )
0338 @pytest.mark.csv
0339 def test_csv_writer_interface(writer, conf_const, tmp_path, trk_geo):
0340 assert hasattr(writer, "Config")
0341
0342 config = writer.Config
0343
0344 assert hasattr(config, "outputDir")
0345
0346 kw = {"level": acts.logging.INFO, "outputDir": str(tmp_path)}
0347
0348 for k, _ in inspect.getmembers(config):
0349 if k.startswith("input"):
0350 kw[k] = "collection"
0351 if k == "trackingGeometry":
0352 kw[k] = trk_geo
0353 if k == "outputStem":
0354 kw[k] = "stem"
0355
0356 assert conf_const(writer, **kw)
0357
0358
0359 @pytest.mark.root
0360 @pytest.mark.odd
0361 @pytest.mark.skipif(not dd4hepEnabled, reason="DD4hep not set up")
0362 def test_root_material_writer(tmp_path, assert_root_hash):
0363 from acts.examples.dd4hep import DD4hepDetector
0364
0365 detector = DD4hepDetector(
0366 xmlFileNames=[str(getOpenDataDetectorDirectory() / "xml/OpenDataDetector.xml")]
0367 )
0368 trackingGeometry = detector.trackingGeometry()
0369
0370 out = tmp_path / "material.root"
0371
0372 assert not out.exists()
0373
0374 rmw = RootMaterialWriter(level=acts.logging.WARNING, filePath=str(out))
0375 assert out.exists()
0376 assert out.stat().st_size > 0 and out.stat().st_size < 500
0377 rmw.write(trackingGeometry)
0378
0379 assert out.stat().st_size > 1000
0380 assert_root_hash(out.name, out)
0381
0382
0383 @pytest.mark.json
0384 @pytest.mark.odd
0385 @pytest.mark.parametrize("fmt", [JsonFormat.Json, JsonFormat.Cbor])
0386 @pytest.mark.skipif(not dd4hepEnabled, reason="DD4hep not set up")
0387 def test_json_material_writer(tmp_path, fmt):
0388 from acts.examples.dd4hep import DD4hepDetector
0389
0390 detector = DD4hepDetector(
0391 xmlFileNames=[str(getOpenDataDetectorDirectory() / "xml/OpenDataDetector.xml")]
0392 )
0393 trackingGeometry = detector.trackingGeometry()
0394
0395 out = (tmp_path / "material").with_suffix("." + fmt.name.lower())
0396
0397 assert not out.exists()
0398
0399 jmw = JsonMaterialWriter(
0400 level=acts.logging.WARNING, fileName=str(out.with_suffix("")), writeFormat=fmt
0401 )
0402 assert not out.exists()
0403 jmw.write(trackingGeometry)
0404
0405 assert out.stat().st_size > 1000
0406
0407
0408 @pytest.mark.csv
0409 def test_csv_multitrajectory_writer(tmp_path):
0410 detector = GenericDetector()
0411 trackingGeometry = detector.trackingGeometry()
0412 field = acts.ConstantBField(acts.Vector3(0, 0, 2 * u.T))
0413
0414 from truth_tracking_kalman import runTruthTrackingKalman
0415
0416 s = Sequencer(numThreads=1, events=10)
0417 runTruthTrackingKalman(
0418 trackingGeometry,
0419 field,
0420 digiConfigFile=Path(
0421 str(
0422 Path(__file__).parent.parent.parent.parent
0423 / "Examples/Algorithms/Digitization/share/default-smearing-config-generic.json"
0424 )
0425 ),
0426 outputDir=tmp_path,
0427 s=s,
0428 )
0429
0430 csv_dir = tmp_path / "csv"
0431 csv_dir.mkdir()
0432 s.addWriter(
0433 CsvTrackWriter(
0434 level=acts.logging.INFO,
0435 inputTracks="tracks",
0436 inputMeasurementParticlesMap="measurement_particles_map",
0437 outputDir=str(csv_dir),
0438 )
0439 )
0440 s.run()
0441 assert len([f for f in csv_dir.iterdir() if f.is_file()]) == 10
0442 assert all(f.stat().st_size > 20 for f in csv_dir.iterdir())
0443
0444
0445 @pytest.fixture(scope="session")
0446 def hepmc_data_impl(tmp_path_factory):
0447 import subprocess
0448
0449 script = (
0450 Path(__file__).parent.parent.parent.parent
0451 / "Examples"
0452 / "Scripts"
0453 / "Python"
0454 / "event_recording.py"
0455 )
0456 assert script.exists()
0457
0458 with tempfile.TemporaryDirectory() as tmp_path:
0459 env = os.environ.copy()
0460 env["NEVENTS"] = "1"
0461 subprocess.check_call([sys.executable, str(script)], cwd=tmp_path, env=env)
0462
0463 outfile = Path(tmp_path) / "hepmc3/event000000000-events.hepmc3"
0464
0465
0466
0467
0468
0469 assert outfile.exists()
0470
0471 yield outfile
0472
0473
0474 @pytest.fixture
0475 def hepmc_data(hepmc_data_impl: Path, tmp_path):
0476 dest = tmp_path / hepmc_data_impl.name
0477 shutil.copy(hepmc_data_impl, dest)
0478
0479 return dest
0480
0481
0482 @pytest.mark.skipif(not hepmc3Enabled, reason="HepMC3 plugin not available")
0483 @pytest.mark.skipif(not dd4hepEnabled, reason="DD4hep not set up")
0484 @pytest.mark.skipif(not geant4Enabled, reason="Geant4 not set up")
0485 @pytest.mark.odd
0486 @pytest.mark.slow
0487 def test_hepmc3_histogram(hepmc_data, tmp_path):
0488 from acts.examples.hepmc3 import (
0489 HepMC3AsciiReader,
0490 HepMCProcessExtractor,
0491 )
0492
0493 s = Sequencer(numThreads=1)
0494
0495 s.addReader(
0496 HepMC3AsciiReader(
0497 level=acts.logging.INFO,
0498 inputDir=str(hepmc_data.parent),
0499 inputStem="events",
0500 outputEvents="hepmc-events",
0501 )
0502 )
0503
0504 s.addAlgorithm(
0505 HepMCProcessExtractor(
0506 level=acts.logging.INFO,
0507 inputEvents="hepmc-events",
0508 extractionProcess="Inelastic",
0509 )
0510 )
0511
0512
0513
0514
0515
0516
0517
0518
0519 alg = AssertCollectionExistsAlg(
0520 "hepmc-events", name="check_alg", level=acts.logging.INFO
0521 )
0522 s.addAlgorithm(alg)
0523
0524 s.run()
0525
0526
0527 @pytest.mark.edm4hep
0528 @pytest.mark.skipif(not edm4hepEnabled, reason="EDM4hep is not set up")
0529 def test_edm4hep_measurement_writer(tmp_path, fatras):
0530 from acts.examples.edm4hep import EDM4hepMeasurementWriter
0531
0532 s = Sequencer(numThreads=1, events=10)
0533 _, simAlg, digiAlg = fatras(s)
0534
0535 out = tmp_path / "measurements_edm4hep.root"
0536
0537 s.addWriter(
0538 EDM4hepMeasurementWriter(
0539 level=acts.logging.VERBOSE,
0540 inputMeasurements=digiAlg.config.outputMeasurements,
0541 inputClusters=digiAlg.config.outputClusters,
0542 outputPath=str(out),
0543 )
0544 )
0545
0546 s.run()
0547
0548 assert os.path.isfile(out)
0549 assert os.stat(out).st_size > 10
0550
0551
0552 @pytest.mark.edm4hep
0553 @pytest.mark.skipif(not edm4hepEnabled, reason="EDM4hep is not set up")
0554 def test_edm4hep_simhit_writer(tmp_path, fatras, conf_const):
0555 from acts.examples.edm4hep import EDM4hepSimHitWriter
0556
0557 s = Sequencer(numThreads=1, events=10)
0558 _, simAlg, _ = fatras(s)
0559
0560 out = tmp_path / "simhits_edm4hep.root"
0561
0562 s.addWriter(
0563 conf_const(
0564 EDM4hepSimHitWriter,
0565 level=acts.logging.INFO,
0566 inputSimHits=simAlg.config.outputSimHits,
0567 outputPath=str(out),
0568 )
0569 )
0570
0571 s.run()
0572
0573 assert os.path.isfile(out)
0574 assert os.stat(out).st_size > 200
0575
0576
0577 @pytest.mark.edm4hep
0578 @pytest.mark.skipif(not edm4hepEnabled, reason="EDM4hep is not set up")
0579 def test_edm4hep_particle_writer(tmp_path, conf_const, ptcl_gun):
0580 from acts.examples.edm4hep import EDM4hepParticleWriter
0581
0582 s = Sequencer(numThreads=1, events=10)
0583 evGen = ptcl_gun(s)
0584
0585 out = tmp_path / "particles_edm4hep.root"
0586
0587 out.mkdir()
0588
0589 s.addWriter(
0590 conf_const(
0591 EDM4hepParticleWriter,
0592 acts.logging.INFO,
0593 inputParticles=evGen.config.outputParticles,
0594 outputPath=str(out),
0595 )
0596 )
0597
0598 s.run()
0599
0600 assert os.path.isfile(out)
0601 assert os.stat(out).st_size > 200
0602
0603
0604 @pytest.mark.edm4hep
0605 @pytest.mark.skipif(not edm4hepEnabled, reason="EDM4hep is not set up")
0606 def test_edm4hep_multitrajectory_writer(tmp_path):
0607 from acts.examples.edm4hep import EDM4hepMultiTrajectoryWriter
0608
0609 detector = GenericDetector()
0610 trackingGeometry = detector.trackingGeometry()
0611 field = acts.ConstantBField(acts.Vector3(0, 0, 2 * u.T))
0612
0613 from truth_tracking_kalman import runTruthTrackingKalman
0614
0615 s = Sequencer(numThreads=1, events=10)
0616 runTruthTrackingKalman(
0617 trackingGeometry,
0618 field,
0619 digiConfigFile=Path(
0620 str(
0621 Path(__file__).parent.parent.parent.parent
0622 / "Examples/Algorithms/Digitization/share/default-smearing-config-generic.json"
0623 )
0624 ),
0625 outputDir=tmp_path,
0626 s=s,
0627 )
0628
0629 s.addAlgorithm(
0630 acts.examples.TracksToTrajectories(
0631 level=acts.logging.INFO,
0632 inputTracks="tracks",
0633 outputTrajectories="trajectories",
0634 )
0635 )
0636
0637 out = tmp_path / "trajectories_edm4hep.root"
0638
0639 s.addWriter(
0640 EDM4hepMultiTrajectoryWriter(
0641 level=acts.logging.VERBOSE,
0642 inputTrajectories="trajectories",
0643 inputMeasurementParticlesMap="measurement_particles_map",
0644 outputPath=str(out),
0645 )
0646 )
0647
0648 s.run()
0649
0650 assert os.path.isfile(out)
0651 assert os.stat(out).st_size > 200
0652
0653
0654 @pytest.mark.edm4hep
0655 @pytest.mark.skipif(not edm4hepEnabled, reason="EDM4hep is not set up")
0656 def test_edm4hep_tracks_writer(tmp_path):
0657 from acts.examples.edm4hep import EDM4hepTrackWriter
0658
0659 detector = GenericDetector()
0660 trackingGeometry = detector.trackingGeometry()
0661 field = acts.ConstantBField(acts.Vector3(0, 0, 2 * u.T))
0662
0663 from truth_tracking_kalman import runTruthTrackingKalman
0664
0665 s = Sequencer(numThreads=1, events=10)
0666 runTruthTrackingKalman(
0667 trackingGeometry,
0668 field,
0669 digiConfigFile=Path(
0670 str(
0671 Path(__file__).parent.parent.parent.parent
0672 / "Examples/Algorithms/Digitization/share/default-smearing-config-generic.json"
0673 )
0674 ),
0675 outputDir=tmp_path,
0676 s=s,
0677 )
0678
0679 out = tmp_path / "tracks_edm4hep.root"
0680
0681 s.addWriter(
0682 EDM4hepTrackWriter(
0683 level=acts.logging.VERBOSE,
0684 inputTracks="kf_tracks",
0685 outputPath=str(out),
0686 Bz=2 * u.T,
0687 )
0688 )
0689
0690 s.run()
0691
0692 assert os.path.isfile(out)
0693 assert os.stat(out).st_size > 200
0694
0695 if not podioEnabled:
0696 import warnings
0697
0698 warnings.warn(
0699 "edm4hep output checks were skipped, because podio was not on the python path"
0700 )
0701 return
0702
0703 from podio.root_io import Reader
0704 import cppyy
0705
0706 reader = Reader(str(out))
0707
0708 actual = []
0709
0710 for frame in reader.get("events"):
0711 tracks = frame.get("ActsTracks")
0712 for track in tracks:
0713 actual.append(
0714 (track.getChi2(), track.getNdf(), len(track.getTrackStates()))
0715 )
0716
0717 locs = []
0718
0719 perigee = None
0720 for ts in track.getTrackStates():
0721 if ts.location == cppyy.gbl.edm4hep.TrackState.AtIP:
0722 perigee = ts
0723 continue
0724 locs.append(ts.location)
0725
0726 rp = ts.referencePoint
0727 r = math.sqrt(rp.x**2 + rp.y**2)
0728 assert r > 25
0729
0730 assert locs[0] == cppyy.gbl.edm4hep.TrackState.AtLastHit
0731 assert locs[-1] == cppyy.gbl.edm4hep.TrackState.AtFirstHit
0732
0733 assert perigee is not None
0734 rp = perigee.referencePoint
0735 assert rp.x == 0.0
0736 assert rp.y == 0.0
0737 assert rp.z == 0.0
0738 assert abs(perigee.D0) < 1e0
0739 assert abs(perigee.Z0) < 1e1