Python/tests/test_writer.py

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