Warning, file /snippets/Tracking/svt_rate_per_source/test_svt_rate.py was not indexed
or was modified since last indexation (in which case cross-reference links may be missing, inaccurate or erroneous).
0001
0002 """Deterministic physics/IO checks; no network, ROOT bindings, or pytest needed.
0003
0004 Run: python test_svt_rate.py. Generated outputs are retained in ./tmp by default;
0005 set SVT_TEST_OUTPUT to choose another validation output directory.
0006 """
0007
0008 import csv
0009 import json
0010 import os
0011 import sys
0012 import tempfile
0013 import unittest
0014 from pathlib import Path
0015 from types import SimpleNamespace
0016 from unittest.mock import patch
0017
0018 import awkward as ak
0019 import numpy as np
0020
0021 import plot_hit_rate_per_source as rate
0022 from svt_sources import RELATIONS, assign_sources, read_branch, resolve_status
0023
0024
0025 class Branch:
0026 def __init__(self, data):
0027 self.data = data
0028
0029 def array(self, library, entry_start, entry_stop):
0030 return self.data[entry_start:entry_stop]
0031
0032
0033 class RootFile:
0034 def __init__(self, tree):
0035 self.tree = tree
0036
0037 def __enter__(self):
0038 return self
0039
0040 def __exit__(self, *args):
0041 pass
0042
0043 def __getitem__(self, name):
0044 assert name == "events"
0045 return self.tree
0046
0047
0048 class Tree(dict):
0049 num_entries = 2
0050
0051
0052 def fixture():
0053 """Two equal-duration frames, including one empty frame in normalization."""
0054 tree = Tree()
0055
0056 def add(name, fields):
0057 tree[name] = Branch(ak.zip({name + "." + field: ak.Array([values, []])
0058 for field, values in fields.items()}))
0059
0060 add("MCParticles", {"generatorStatus": [1, 0, 2001, 3001, 9999],
0061 "parents_begin": [0, 0, 1, 1, 1], "parents_end": [0, 1, 1, 1, 1]})
0062 add("_MCParticles_parents", {"index": [0]})
0063 for config in RELATIONS:
0064 if config["rec"] == "SiBarrelVertexRecHits":
0065 coordinates = {"position.x": [35.] * 8, "position.y": [0.] * 8, "position.z": [0.] * 8}
0066 rec_raw = [0, 1, 2, 3, 4, 5, 6, -1]
0067 raw, sim, weights, particles = [0, 1, 2, 2, 3, 4], [0, 1, 2, 3, -1, 4], [.5, 1., .2, .8, 1., 1.], [0, 1, 0, 2, 4]
0068 else:
0069 endcap = config["rec"] == "SiEndcapTrackerRecHits"
0070 coordinates = {"position.x": [10. if endcap else 270.], "position.y": [0.], "position.z": [250. if endcap else 0.]}
0071 rec_raw, raw, sim, weights, particles = [0], [0], [0], [1.], [2 if endcap else 3]
0072 add(config["rec"], coordinates)
0073 add(config["rec_raw"], {"index": rec_raw})
0074 add(config["association"], {"weight": weights})
0075 add(config["association_raw"], {"index": raw})
0076 add(config["association_sim"], {"index": sim})
0077 add(config["sim_particle"], {"index": particles})
0078 return tree
0079
0080
0081 def args(chunk_size=1):
0082 return SimpleNamespace(tile=False, entry_start=0, entry_stop=None,
0083 chunk_size=chunk_size, frame_us=2.)
0084
0085
0086 class RateTests(unittest.TestCase):
0087 def test_status_tracing_and_invalid_ancestry(self):
0088 self.assertEqual(resolve_status(1, [1, 0], [0, 0], [0, 1], [0]), (1, True, False))
0089 self.assertEqual(resolve_status(0, [0], [0], [1], [0]), (0, True, True))
0090 self.assertEqual(resolve_status(-1, [1], [0], [0], []), (None, False, False))
0091 self.assertEqual(resolve_status(0, [0], [0], [2], [0]), (0, True, True))
0092
0093 def test_weight_choice_secondary_unmatched_invalid_and_other(self):
0094 tree = fixture()
0095 config = RELATIONS[1]
0096 data = {key: read_branch(tree, value, 0, 1)[0] for key, value in config.items()}
0097 labels, diagnostic = assign_sources(data["rec_raw"], data["association"]["weight"],
0098 data["association_raw"], data["association_sim"],
0099 data["sim_particle"], read_branch(tree, "MCParticles", 0, 1)[0],
0100 read_branch(tree, "_MCParticles_parents", 0, 1)[0])
0101 self.assertEqual(labels.tolist(), ["DIS", "DIS", "SR", "invalid_relation", "other", "unmatched", "unmatched", "invalid_relation"])
0102 self.assertEqual(diagnostic["secondary_traced_count"], 1)
0103 self.assertEqual(diagnostic["ambiguous_source_count"], 1)
0104 self.assertEqual(diagnostic["invalid_relation_count"], 2)
0105 bad_weights = ak.Array([float("nan")] * 6)
0106 labels, diagnostic = assign_sources(data["rec_raw"], bad_weights, data["association_raw"], data["association_sim"], data["sim_particle"], read_branch(tree, "MCParticles", 0, 1)[0], read_branch(tree, "_MCParticles_parents", 0, 1)[0])
0107 self.assertEqual(diagnostic["invalid_weight_count"], 5)
0108
0109 def test_chunks_conservation_zero_frames_and_fractional_rate(self):
0110 for tile in (False, True):
0111 results = []
0112 for size in (1, 2):
0113 options = args(size)
0114 options.tile = tile
0115 with patch.object(rate.uproot, "open", side_effect=lambda *a, **k: RootFile(fixture())):
0116 maps, inputs, diagnostics, events, elapsed = rate.process_files(["synthetic.root"], options)
0117 self.assertEqual(events, 2)
0118 self.assertEqual(inputs[0]["processed_events"], 2)
0119 self.assertEqual(maps["L0"]["counts"]["ALL"].sum(), 8)
0120 np.testing.assert_array_equal(maps["L3"]["counts"]["other background"], maps["L3"]["counts"]["Bremstrahlung"])
0121 results.append(maps)
0122 for name in results[0]:
0123 for source in results[0][name]["counts"]:
0124 np.testing.assert_array_equal(results[0][name]["counts"][source], results[1][name]["counts"][source])
0125
0126 rows = rate.summarize(results[0], 1000 / (2 * 3000))
0127 dis = next(row for row in rows if row["lname"] == "L0" and row["source"] == "DIS")
0128 self.assertAlmostEqual(dis["rsu_or_tile_max_rate_khz"], 1 / 3)
0129
0130 def test_map_edge_coverage_and_outside_counter(self):
0131 maps = rate.make_surface_maps()
0132 hits = ak.Array([{"position.x": 35., "position.y": 0., "position.z": 195.},
0133 {"position.x": 35., "position.y": 0., "position.z": 205.}])
0134 rate.fill_event(maps, "SiBarrelVertexRecHits", hits, np.array(["DIS", "DIS"]))
0135 self.assertEqual(maps["L0"]["counts"]["ALL"].sum(), 1)
0136 self.assertEqual(maps["L0"]["counters"]["ALL"]["outside_map_hits"], 1)
0137 self.assertEqual(maps["L0"]["xedges"][-1], 200.)
0138
0139 def test_missing_fields_and_relation_length(self):
0140 with self.assertRaises(ValueError):
0141 read_branch(fixture(), "MCParticles", 0, 1, ["missing"])
0142 with self.assertRaises(ValueError):
0143 rate.fill_event(rate.make_surface_maps(), "SiBarrelVertexRecHits", ak.Array([]), np.array(["DIS"]))
0144
0145 def test_cli_outputs_and_input_list(self):
0146 parent = Path(os.environ.get("SVT_TEST_OUTPUT", str(rate.HERE / "tmp")))
0147 parent.mkdir(parents=True, exist_ok=True)
0148 output = Path(tempfile.mkdtemp(prefix="svt_rate_validation_", dir=str(parent)))
0149 listing = output / "inputs.txt"
0150 listing.write_text("# test fixture\n\nsynthetic.root:events\n")
0151 inputs = rate.build_inputs(SimpleNamespace(file_list=listing, input_files=None, max_files=None))
0152 self.assertEqual(inputs, [str(output / "synthetic.root")])
0153 command = ["plot_hit_rate_per_source.py", "--file-list", str(listing), "--output-dir", str(output), "--label", "smoke", "--chunk-size", "1"]
0154 with patch.object(sys, "argv", command), patch.object(rate.uproot, "open", side_effect=lambda *a, **k: RootFile(fixture())):
0155 rate.main()
0156 result = output / "smoke"
0157 record = json.loads((result / "smoke_run.json").read_text())
0158 self.assertEqual(record["processed_events"], 2)
0159 self.assertEqual(record["rate_scale_khz_per_count"], 250.)
0160 with (result / "smoke_summary.csv").open() as stream:
0161 rows = list(csv.DictReader(line for line in stream if not line.startswith("#")))
0162 all_l0 = next(row for row in rows if row["lname"] == "L0" and row["source"] == "ALL")
0163 self.assertEqual(float(all_l0["scaled_total_rate_khz"]), 2000.)
0164 with np.load(result / "smoke_maps.npz") as archive:
0165 index = next(row for row in record["maps"] if row["lname"] == "L0" and row["source"] == "ALL")
0166 self.assertEqual(archive[index["key"]].sum(), 8)
0167 for filename in record["pdfs"]:
0168 self.assertTrue((result / filename).read_bytes().startswith(b"%PDF"))
0169 with patch.object(sys, "argv", command), self.assertRaises(FileExistsError):
0170 rate.main()
0171 print(f"Validation outputs retained at {output}")
0172
0173
0174 if __name__ == "__main__":
0175 unittest.main()
