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