Coverage for tests / cqg_test_utils.py: 6%
108 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-05-06 08:35 +0000
1# This file is part of ctrl_bps.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This software is dual licensed under the GNU General Public License and also
10# under a 3-clause BSD license. Recipients may choose which of these licenses
11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt,
12# respectively. If you choose the GPL option then the following text applies
13# (but note that there is still no warranty even if you opt for BSD instead):
14#
15# This program is free software: you can redistribute it and/or modify
16# it under the terms of the GNU General Public License as published by
17# the Free Software Foundation, either version 3 of the License, or
18# (at your option) any later version.
19#
20# This program is distributed in the hope that it will be useful,
21# but WITHOUT ANY WARRANTY; without even the implied warranty of
22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
23# GNU General Public License for more details.
24#
25# You should have received a copy of the GNU General Public License
26# along with this program. If not, see <https://www.gnu.org/licenses/>.
27"""ClusteredQuantumGraph-related utilities to support ctrl_bps testing."""
29import uuid
31from qg_test_utils import make_test_helper
33from lsst.ctrl.bps import ClusteredQuantumGraph, QuantaCluster
36def check_cqg(cqg, truth=None):
37 """Check ClusteredQuantumGraph for correctness used by unit
38 tests.
40 Parameters
41 ----------
42 cqg : `lsst.ctrl.bps.ClusteredQuantumGraph`
43 ClusteredQuantumGraph to be checked for correctness.
44 truth : `dict` [`str`, `~typing.Any`], optional
45 Information describing what this cluster should look like.
46 """
47 cqg.validate()
49 # If given what should be there, check values.
50 if truth:
51 cqg_info = dump_cqg(cqg)
52 compare_cqg_dicts(truth, cqg_info)
55def replace_node_name(name, label, dims):
56 """Replace node id in cluster name because they
57 change every run and thus make testing difficult.
59 Parameters
60 ----------
61 name : `str`
62 Cluster name.
63 label : `str`
64 Cluster label.
65 dims : `dict` [`str`, `~typing.Any`]
66 Dimension names and values in order to make new name unique.
68 Returns
69 -------
70 name : `str`
71 New name of cluster.
72 """
73 try:
74 name_parts = name.split("_")
75 _ = uuid.UUID(name_parts[0])
76 if len(name_parts) == 1:
77 name = f"NODEONLY_{label}_{str(dims)}"
78 else:
79 name = f"NODENAME_{'_'.join(name_parts[1:])}"
80 except ValueError:
81 pass
82 return name
85def dump_cqg(cqg):
86 """Represent ClusteredQuantumGraph as dictionary for testing.
88 Parameters
89 ----------
90 cqg : `lsst.ctrl.bps.ClusteredQuantumGraph`
91 ClusteredQuantumGraph to be represented as a dictionary.
93 Returns
94 -------
95 info : `dict` [`str`, `~typing.Any`]
96 Dictionary represention of ClusteredQuantumGraph.
97 """
98 info = {"name": cqg.name, "nodes": {}}
100 orig_to_new = {}
101 for cluster in cqg.clusters():
102 dims = {}
103 for key, value in cluster.tags.items():
104 if key == "label":
105 info["tags_label"] = value
106 elif key != "node_number":
107 dims[key] = value
108 name = replace_node_name(cluster.name, cluster.label, dims)
109 orig_to_new[cluster.name] = name
110 info["nodes"][name] = {"label": cluster.label, "dims": dims, "counts": dict(cluster.quanta_counts)}
112 info["edges"] = []
113 for edge in cqg._cluster_graph.edges:
114 info["edges"].append((orig_to_new[edge[0]], orig_to_new[edge[1]]))
116 return info
119def compare_cqg_dicts(truth, cqg):
120 """Compare dicts representing two ClusteredQuantumGraphs.
122 Parameters
123 ----------
124 truth : `dict` [`str`, `~typing.Any`]
125 Representation of the expected ClusteredQuantumGraph.
126 cqg : `dict` [`str`, `~typing.Any`]
127 Representation of the calculated ClusteredQuantumGraph.
129 Raises
130 ------
131 AssertionError
132 Whenever discover discrepancy between dicts.
133 """
134 assert truth["name"] == cqg["name"], f"Mismatch name: truth={truth['name']}, cqg={cqg['name']}"
135 assert len(truth["nodes"]) == len(cqg["nodes"]), (
136 f"Mismatch number of nodes: truth={len(truth['nodes'])}, cqg={len(cqg['nodes'])}"
137 )
138 for tkey in truth["nodes"]:
139 assert tkey in cqg["nodes"], f"Could not find {tkey} in cqg"
140 tnode = truth["nodes"][tkey]
141 cnode = cqg["nodes"][tkey]
142 assert tnode["label"] == cnode["label"], (
143 f"Mismatch cluster label: truth={tnode['label']}, cqg={cnode['label']}"
144 )
145 assert tnode["dims"] == cnode["dims"], (
146 f"Mismatch cluster dims: truth={tnode['dims']}, cqg={cnode['dims']}"
147 )
148 assert tnode["counts"] == cnode["counts"], (
149 f"Mismatch cluster quanta counts: truth={tnode['counts']}, cqg={cnode['counts']}"
150 )
151 assert set(truth["edges"]) == set(cqg["edges"]), (
152 f"Mismatch edges: truth={truth['edges']}, cqg={cqg['edges']}"
153 )
156# T1(1,2) T1(1,4) T1(3,4) T5(1,2) T5(1,4) T5(3,4)
157# | | |
158# T2(1,2) T2(1,4) T2(3,4)
159# | | | | | |
160# | T2b(1,2) | T2b(1,4) | T2b(3,4)
161# | | |
162# T3(1,2) T3(1,4) T3(3,4)
163# | | |
164# T4(1,2) T4(1,4) T4(3,4)
165def make_test_clustered_quantum_graph(outdir):
166 """Make a ClusteredQuantumGraph for testing.
168 Parameters
169 ----------
170 outdir : `str`
171 Root used for the quantum graph filename stored
172 in the ClusteredQuantumGraph. The quantum graph is always saved to
173 this location.
175 Returns
176 -------
177 qgraph : `lsst.pipe.base.quantum_graph.PredictedQuantumGraph`
178 The fake QuantumGraph created for the test
179 ClusteredQuantumGraph returned separately.
180 cqg : `lsst.ctrl.bps.ClusteredQuantumGraph`
181 Clustered quantum graph.
182 """
183 with make_test_helper() as helper:
184 qgc = helper.make_quantum_graph_builder(output_run="run").finish(attach_datastore_records=False)
185 qg_filename = f"{outdir}/test_file.qg"
186 qgc.write(qg_filename)
187 qgraph = qgc.assemble()
188 cqg = ClusteredQuantumGraph("cqg1", qgraph, qg_filename)
190 # since random hash ids, create mapping for tests
191 test_lookup = {}
192 for task_label, quanta_for_task in qgraph.quanta_by_task.items():
193 for data_coordinate, quantum_id in quanta_for_task.items():
194 data_id = dict(data_coordinate.required)
195 key = f"{task_label}_{data_id['D1']}_{data_id['D2']}"
196 test_lookup[key] = (quantum_id, cqg.qxgraph.nodes[quantum_id])
198 def get_add_quantum_args(key: str) -> tuple[uuid.UUID, str]:
199 quantum_id, quantum_info = test_lookup[key]
200 return quantum_id, quantum_info["task_label"]
202 # T1 -> T2,T3 -> T4 Dim1 = 1, Dim2 = 2
203 qc1 = QuantaCluster.from_quantum_info(*test_lookup["T1_1_2"], template="T1_1_2")
204 qc23 = QuantaCluster.from_quantum_info(*test_lookup["T2_1_2"], template="T23_1_2")
205 qc23.add_quantum(*get_add_quantum_args("T3_1_2"))
206 qc23.label = "clusterT2T3" # update label so doesnt look like only T2
207 qc23.tags["label"] = "clusterT2T3"
208 cqg.add_cluster([qc23, qc1]) # reversed to check order is corrected in tests
209 cqg.add_dependency(qc1, qc23)
210 qc4 = QuantaCluster.from_quantum_info(*test_lookup["T4_1_2"], template="T4_1_2")
211 cqg.add_cluster(qc4)
212 cqg.add_dependency(qc23, qc4)
213 qc2b = QuantaCluster.from_quantum_info(*test_lookup["T2b_1_2"], template="T2b_1_2")
214 cqg.add_cluster(qc2b)
215 cqg.add_dependency(qc23, qc2b)
217 # T1 -> T2,T3 -> T4 Dim1 = 1, Dim2 = 4
218 qc1 = QuantaCluster.from_quantum_info(*test_lookup["T1_1_4"], template="T1_1_4")
219 qc23 = QuantaCluster.from_quantum_info(*test_lookup["T2_1_4"], template="T23_1_4")
220 qc23.add_quantum(*get_add_quantum_args("T3_1_4"))
221 qc23.label = "clusterT2T3" # update label so doesnt look like only T2
222 qc23.tags["label"] = "clusterT2T3"
223 cqg.add_cluster([qc23, qc1]) # reversed to check order is corrected in tests
224 cqg.add_dependency(qc1, qc23)
225 qc4 = QuantaCluster.from_quantum_info(*test_lookup["T4_1_4"], template="T4_1_4")
226 cqg.add_cluster(qc4)
227 cqg.add_dependency(qc23, qc4)
228 qc2b = QuantaCluster.from_quantum_info(*test_lookup["T2b_1_4"], template="T2b_1_4")
229 cqg.add_cluster(qc2b)
230 cqg.add_dependency(qc23, qc2b)
232 # T1 -> T2,T3 -> T4 Dim1 = 3, Dim2 = 4
233 qc1 = QuantaCluster.from_quantum_info(*test_lookup["T1_3_4"], template="T1_3_4")
234 qc23 = QuantaCluster.from_quantum_info(*test_lookup["T2_3_4"], template="T23_3_4")
235 qc23.add_quantum(*get_add_quantum_args("T3_3_4"))
236 qc23.label = "clusterT2T3" # update label so doesnt look like only T2
237 qc23.tags["label"] = "clusterT2T3"
238 cqg.add_cluster([qc23, qc1]) # reversed to check order is corrected in tests
239 cqg.add_dependency(qc1, qc23)
240 qc4 = QuantaCluster.from_quantum_info(*test_lookup["T4_3_4"], template="T4_3_4")
241 cqg.add_cluster(qc4)
242 cqg.add_dependency(qc23, qc4)
243 qc2b = QuantaCluster.from_quantum_info(*test_lookup["T2b_3_4"], template="T2b_3_4")
244 cqg.add_cluster(qc2b)
245 cqg.add_dependency(qc23, qc2b)
247 # Add remaining
248 cluster = QuantaCluster.from_quantum_info(*test_lookup["T5_1_2"], template="T5_1_2")
249 cqg.add_cluster(cluster)
250 cluster = QuantaCluster.from_quantum_info(*test_lookup["T5_1_4"], template="T5_1_4")
251 cqg.add_cluster(cluster)
252 cluster = QuantaCluster.from_quantum_info(*test_lookup["T5_3_4"], template="T5_3_4")
253 cqg.add_cluster(cluster)
255 return qgraph, cqg