File indexing completed on 2025-07-05 08:12:09
0001 import multiprocessing
0002 from pathlib import Path
0003 import sys
0004 import os
0005 import tempfile
0006 import shutil
0007 from typing import Dict
0008 import warnings
0009 import pytest_check as check
0010 from collections import namedtuple
0011
0012
0013 sys.path += [
0014 str(Path(__file__).parent.parent.parent.parent / "Examples/Scripts/Python/"),
0015 str(Path(__file__).parent),
0016 ]
0017
0018
0019 import helpers
0020 import helpers.hash_root
0021
0022 import pytest
0023
0024 import acts
0025 import acts.examples
0026 from acts.examples.odd import getOpenDataDetector
0027 from acts.examples.simulation import addParticleGun, EtaConfig, ParticleConfig
0028
0029 try:
0030 import ROOT
0031
0032 ROOT.gSystem.ResetSignals()
0033 except ImportError:
0034 pass
0035
0036 try:
0037 if acts.logging.getFailureThreshold() != acts.logging.WARNING:
0038 acts.logging.setFailureThreshold(acts.logging.WARNING)
0039 except RuntimeError:
0040
0041 errtype = (
0042 "negative"
0043 if acts.logging.getFailureThreshold() < acts.logging.WARNING
0044 else "positive"
0045 )
0046 warnings.warn(
0047 "Runtime log failure threshold could not be set. "
0048 "Compile-time value is probably set via CMake, i.e. "
0049 f"`ACTS_LOG_FAILURE_THRESHOLD={acts.logging.getFailureThreshold().name}` is set, "
0050 "or `ACTS_ENABLE_LOG_FAILURE_THRESHOLD=OFF`. "
0051 f"The pytest test-suite can produce false-{errtype} results in this configuration"
0052 )
0053
0054
0055 u = acts.UnitConstants
0056
0057
0058 class RootHashAssertionError(AssertionError):
0059 def __init__(
0060 self, file: Path, key: str, exp_hash: str, act_hash: str, *args, **kwargs
0061 ):
0062 super().__init__(f"{exp_hash} != {act_hash}", *args, **kwargs)
0063 self.file = file
0064 self.key = key
0065 self.exp_hash = exp_hash
0066 self.act_hash = act_hash
0067
0068
0069 hash_assertion_failures = []
0070
0071
0072 def _parse_hash_file(file: Path) -> Dict[str, str]:
0073 res = {}
0074 for line in file.open():
0075 if line.strip() == "" or line.strip().startswith("#"):
0076 continue
0077 key, h = line.strip().split(":", 1)
0078 res[key.strip()] = h.strip()
0079 return res
0080
0081
0082 @pytest.fixture(scope="session")
0083 def root_file_exp_hashes():
0084 path = Path(
0085 os.environ.get("ROOT_HASH_FILE", Path(__file__).parent / "root_file_hashes.txt")
0086 )
0087 return _parse_hash_file(path)
0088
0089
0090 @pytest.fixture(name="assert_root_hash")
0091 def assert_root_hash(request, root_file_exp_hashes):
0092 if not helpers.doHashChecks:
0093
0094 def fn(*args, **kwargs):
0095 pass
0096
0097 return fn
0098
0099 def fn(key: str, file: Path):
0100 """
0101 Assertion helper function to check the hashes of root files.
0102 Do NOT use this function directly by importing, rather use it as a pytest fixture
0103
0104 Arguments you need to provide:
0105 key: Explicit lookup key for the expected hash, should be unique per test function
0106 file: Root file to check the expected hash against
0107 """
0108 __tracebackhide__ = True
0109 gkey = f"{request.node.name}__{key}"
0110 act_hash = helpers.hash_root.hash_root_file(file)
0111 if not gkey in root_file_exp_hashes:
0112 warnings.warn(
0113 f'Hash lookup key "{key}" not found for test "{request.node.name}"'
0114 )
0115 check.equal(act_hash, "[MISSING]")
0116 exc = RootHashAssertionError(file, gkey, "[MISSING]", act_hash)
0117 hash_assertion_failures.append(exc)
0118
0119 else:
0120 refhash = root_file_exp_hashes[gkey]
0121 check.equal(act_hash, refhash)
0122 if act_hash != refhash:
0123 exc = RootHashAssertionError(file, gkey, refhash, act_hash)
0124 hash_assertion_failures.append(exc)
0125
0126 return fn
0127
0128
0129 def pytest_terminal_summary(terminalreporter, exitstatus, config):
0130 docs_url = "https://acts.readthedocs.io/en/latest/examples/python_bindings.html#root-file-hash-regression-checks"
0131 if len(hash_assertion_failures) > 0:
0132 terminalreporter.ensure_newline()
0133 terminalreporter.section(
0134 "RootHashAssertionErrors", sep="-", red=True, bold=True
0135 )
0136 terminalreporter.line(
0137 "The ROOT files produced by tests have changed since the last recorded reference."
0138 )
0139 terminalreporter.line(
0140 "This can be be expected if e.g. the underlying algorithm changed, or it can be a test failure symptom."
0141 )
0142 terminalreporter.line(
0143 "Please manually check the output files listed below and make sure that their content is correct."
0144 )
0145 terminalreporter.line(
0146 "If it is, you can update the test reference file Examples/Python/tests/root_file_hashes.txt with the new hashes below."
0147 )
0148 terminalreporter.line(f"See {docs_url} for more details")
0149 terminalreporter.line("")
0150
0151 for e in hash_assertion_failures:
0152 terminalreporter.line(f"{e.key}: {e.act_hash}")
0153
0154 if not helpers.doHashChecks:
0155 terminalreporter.section("Root file hash checks", sep="-", blue=True, bold=True)
0156 terminalreporter.line(
0157 "NOTE: Root file hash checks were skipped, enable with ROOT_HASH_CHECKS=on"
0158 )
0159 terminalreporter.line(f"See {docs_url} for more details")
0160
0161
0162 def kwargsConstructor(cls, *args, **kwargs):
0163 return cls(*args, **kwargs)
0164
0165
0166 def configKwConstructor(cls, *args, **kwargs):
0167 assert hasattr(cls, "Config")
0168 _kwargs = {}
0169 if "level" in kwargs:
0170 _kwargs["level"] = kwargs.pop("level")
0171 config = cls.Config()
0172 for k, v in kwargs.items():
0173 setattr(config, k, v)
0174 return cls(*args, config=config, **_kwargs)
0175
0176
0177 def configPosConstructor(cls, *args, **kwargs):
0178 assert hasattr(cls, "Config")
0179 _kwargs = {}
0180 if "level" in kwargs:
0181 _kwargs["level"] = kwargs.pop("level")
0182 config = cls.Config()
0183 for k, v in kwargs.items():
0184 setattr(config, k, v)
0185
0186 return cls(config, *args, **_kwargs)
0187
0188
0189 @pytest.fixture(params=[configPosConstructor, configKwConstructor, kwargsConstructor])
0190 def conf_const(request):
0191 return request.param
0192
0193
0194 @pytest.fixture
0195 def rng():
0196 return acts.examples.RandomNumbers(seed=42)
0197
0198
0199 @pytest.fixture
0200 def basic_prop_seq(rng):
0201 def _basic_prop_seq_factory(geo, s=None):
0202 if s is None:
0203 s = acts.examples.Sequencer(events=10, numThreads=1)
0204
0205 addParticleGun(
0206 s,
0207 ParticleConfig(num=10, pdg=acts.PdgParticle.eMuon, randomizeCharge=True),
0208 EtaConfig(-4.0, 4.0),
0209 rnd=rng,
0210 )
0211
0212 trkParamExtractor = acts.examples.ParticleTrackParamExtractor(
0213 level=acts.logging.WARNING,
0214 inputParticles="particles_generated",
0215 outputTrackParameters="params_particles_generated",
0216 )
0217 s.addAlgorithm(trkParamExtractor)
0218
0219 nav = acts.Navigator(trackingGeometry=geo)
0220 stepper = acts.StraightLineStepper()
0221
0222 prop = acts.examples.ConcretePropagator(acts.Propagator(stepper, nav))
0223
0224 alg = acts.examples.PropagationAlgorithm(
0225 level=acts.logging.WARNING,
0226 propagatorImpl=prop,
0227 sterileLogger=False,
0228 inputTrackParameters="params_particles_generated",
0229 outputSummaryCollection="propagation_summary",
0230 )
0231 s.addAlgorithm(alg)
0232
0233 return s, alg
0234
0235 return _basic_prop_seq_factory
0236
0237
0238 @pytest.fixture
0239 def trk_geo():
0240 detector = acts.examples.GenericDetector()
0241 trackingGeometry = detector.trackingGeometry()
0242 yield trackingGeometry
0243
0244
0245 DetectorConfig = namedtuple(
0246 "DetectorConfig",
0247 [
0248 "detector",
0249 "trackingGeometry",
0250 "decorators",
0251 "geometrySelection",
0252 "digiConfigFile",
0253 "name",
0254 ],
0255 )
0256
0257
0258 @pytest.fixture(params=["generic", pytest.param("odd", marks=pytest.mark.odd)])
0259 def detector_config(request):
0260 srcdir = Path(__file__).resolve().parent.parent.parent.parent
0261
0262 if request.param == "generic":
0263 detector = acts.examples.GenericDetector()
0264 trackingGeometry = detector.trackingGeometry()
0265 decorators = detector.contextDecorators()
0266 return DetectorConfig(
0267 detector,
0268 trackingGeometry,
0269 decorators,
0270 geometrySelection=(srcdir / "Examples/Configs/generic-seeding-config.json"),
0271 digiConfigFile=(
0272 srcdir / "Examples/Configs/generic-digi-smearing-config.json"
0273 ),
0274 name=request.param,
0275 )
0276 elif request.param == "odd":
0277 if not helpers.dd4hepEnabled:
0278 pytest.skip("DD4hep not set up")
0279
0280 matDeco = acts.IMaterialDecorator.fromFile(
0281 srcdir / "thirdparty/OpenDataDetector/data/odd-material-maps.root",
0282 level=acts.logging.INFO,
0283 )
0284 detector = getOpenDataDetector(matDeco)
0285 trackingGeometry = detector.trackingGeometry()
0286 decorators = detector.contextDecorators()
0287 return DetectorConfig(
0288 detector,
0289 trackingGeometry,
0290 decorators,
0291 digiConfigFile=(srcdir / "Examples/Configs/odd-digi-smearing-config.json"),
0292 geometrySelection=(srcdir / "Examples/Configs/odd-seeding-config.json"),
0293 name=request.param,
0294 )
0295 else:
0296 raise ValueError(f"Invalid detector {detector}")
0297
0298
0299 @pytest.fixture
0300 def ptcl_gun(rng):
0301 def _factory(s):
0302 evGen = acts.examples.EventGenerator(
0303 level=acts.logging.INFO,
0304 generators=[
0305 acts.examples.EventGenerator.Generator(
0306 multiplicity=acts.examples.FixedMultiplicityGenerator(n=2),
0307 vertex=acts.examples.GaussianVertexGenerator(
0308 stddev=acts.Vector4(0, 0, 0, 0), mean=acts.Vector4(0, 0, 0, 0)
0309 ),
0310 particles=acts.examples.ParametricParticleGenerator(
0311 p=(1 * u.GeV, 10 * u.GeV),
0312 eta=(-2, 2),
0313 phi=(0, 360 * u.degree),
0314 randomizeCharge=True,
0315 numParticles=2,
0316 ),
0317 )
0318 ],
0319 outputEvent="particle_gun_event",
0320 randomNumbers=rng,
0321 )
0322
0323 s.addReader(evGen)
0324
0325 hepmc3Converter = acts.examples.hepmc3.HepMC3InputConverter(
0326 level=acts.logging.INFO,
0327 inputEvent=evGen.config.outputEvent,
0328 outputParticles="particles_generated",
0329 outputVertices="vertices_input",
0330 )
0331 s.addAlgorithm(hepmc3Converter)
0332
0333 return evGen, hepmc3Converter
0334
0335 return _factory
0336
0337
0338 @pytest.fixture
0339 def fatras(ptcl_gun, trk_geo, rng):
0340 def _factory(s):
0341 evGen, h3conv = ptcl_gun(s)
0342
0343 field = acts.ConstantBField(acts.Vector3(0, 0, 2 * acts.UnitConstants.T))
0344 simAlg = acts.examples.FatrasSimulation(
0345 level=acts.logging.INFO,
0346 inputParticles=h3conv.config.outputParticles,
0347 outputParticles="particles_simulated",
0348 outputSimHits="simhits",
0349 randomNumbers=rng,
0350 trackingGeometry=trk_geo,
0351 magneticField=field,
0352 generateHitsOnSensitive=True,
0353 emScattering=False,
0354 emEnergyLossIonisation=False,
0355 emEnergyLossRadiation=False,
0356 emPhotonConversion=False,
0357 )
0358
0359 s.addAlgorithm(simAlg)
0360
0361
0362 digiCfg = acts.examples.DigitizationAlgorithm.Config(
0363 digitizationConfigs=acts.examples.readDigiConfigFromJson(
0364 str(
0365 Path(__file__).parent.parent.parent.parent
0366 / "Examples/Configs/generic-digi-smearing-config.json"
0367 )
0368 ),
0369 surfaceByIdentifier=trk_geo.geoIdSurfaceMap(),
0370 randomNumbers=rng,
0371 inputSimHits=simAlg.config.outputSimHits,
0372 )
0373 digiAlg = acts.examples.DigitizationAlgorithm(digiCfg, acts.logging.INFO)
0374
0375 s.addAlgorithm(digiAlg)
0376
0377 return evGen, simAlg, digiAlg
0378
0379 return _factory
0380
0381
0382 def _do_material_recording(d: Path):
0383 from material_recording import runMaterialRecording
0384
0385 s = acts.examples.Sequencer(events=2, numThreads=1)
0386
0387 with getOpenDataDetector() as detector:
0388 runMaterialRecording(detector, str(d), tracksPerEvent=100, s=s)
0389
0390 s.run()
0391
0392
0393 @pytest.fixture(scope="session")
0394 def material_recording_session():
0395 if not helpers.geant4Enabled:
0396 pytest.skip("Geantino recording requested, but Geant4 is not set up")
0397
0398 if not helpers.dd4hepEnabled:
0399 pytest.skip("DD4hep recording requested, but DD4hep is not set up")
0400
0401 with tempfile.TemporaryDirectory() as d:
0402
0403 spawn_context = multiprocessing.get_context("spawn")
0404 p = spawn_context.Process(target=_do_material_recording, args=(d,))
0405 p.start()
0406 p.join()
0407 if p.exitcode != 0:
0408 raise RuntimeError("Failure to exeecute material recording")
0409
0410 yield Path(d)
0411
0412
0413 @pytest.fixture
0414 def material_recording(material_recording_session: Path, tmp_path: Path):
0415 target = tmp_path / material_recording_session.name
0416 shutil.copytree(material_recording_session, target)
0417 yield target
