Skip to content

quast_decisiontree.nodes.algorithm_setup

quast_decisiontree.nodes.algorithm_setup

Setup nodes for classical and quantum solvers.

logger module-attribute 

logger = logging.getLogger('dt_logger')

BruteForceSetupNode

Bases: Node

Sets up the brute force solver.

Modifications at runtime: None

Source code in src/quast_decisiontree/nodes/algorithm_setup.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
class BruteForceSetupNode(Node):
    """Sets up the brute force solver.

    Modifications at runtime: None
    """

    _known_children = ["ClassicalAlgorithmExecuteNode"]

    def __init__(self, children: list):
        super().__init__(requires=[], creates="solver", children=children)

    def execute(self, problem_data: dict, path_info: dict) -> dict:
        problem_data["solver"] = BruteForce()
        return dict()

    def interpret_result(
        self,
        result: dict,
        problem_data: dict,
        next_node_info: dict | None = None,
        config: dict | None = None,
    ) -> dict:
        sol_vector, sol_val = result["raw"]
        result["solution_vector"] = sol_vector
        result["best_energy"] = sol_val
        result["best_bitstring"] = "".join([str(int(x)) for x in sol_vector[::-1]])
        result["best_qubo_value"] = sol_val

        result["solution_bitstring"] = result["best_bitstring"]
        result["solution_qubo_value"] = sol_val

        if config and config.get("saving_policy") == "discard_if_processed":
            result.setdefault("_no_save", set()).add("raw")
        return result

__init__ 

__init__(children)
Source code in src/quast_decisiontree/nodes/algorithm_setup.py 
33
34
def __init__(self, children: list):
    super().__init__(requires=[], creates="solver", children=children)

execute 

execute(problem_data, path_info)
Source code in src/quast_decisiontree/nodes/algorithm_setup.py 
36
37
38
def execute(self, problem_data: dict, path_info: dict) -> dict:
    problem_data["solver"] = BruteForce()
    return dict()

interpret_result 

interpret_result(
    result, problem_data, next_node_info=None, config=None
)
Source code in src/quast_decisiontree/nodes/algorithm_setup.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def interpret_result(
    self,
    result: dict,
    problem_data: dict,
    next_node_info: dict | None = None,
    config: dict | None = None,
) -> dict:
    sol_vector, sol_val = result["raw"]
    result["solution_vector"] = sol_vector
    result["best_energy"] = sol_val
    result["best_bitstring"] = "".join([str(int(x)) for x in sol_vector[::-1]])
    result["best_qubo_value"] = sol_val

    result["solution_bitstring"] = result["best_bitstring"]
    result["solution_qubo_value"] = sol_val

    if config and config.get("saving_policy") == "discard_if_processed":
        result.setdefault("_no_save", set()).add("raw")
    return result

TabuSetupNode

Bases: Node

Sets up a Tabu Sampler Solver.

Modifications at runtime: - num_reads: - {int} : the number of samples to generate from the solver

Source code in src/quast_decisiontree/nodes/algorithm_setup.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class TabuSetupNode(Node):
    """Sets up a Tabu Sampler Solver.

    Modifications at runtime:
    - num_reads:
        - {int} : the number of samples to generate from the solver
    """

    _known_children = ["ClassicalAlgorithmExecuteNode"]
    _path_keys = dict(num_reads=PathKey(int))

    def __init__(self, children: list) -> None:
        super().__init__(creates="solver", children=children, requires=[])
        self.query = IntQuery(
            question="How many samples should the Tabu Solver generate?",
            default=10,
            name="num_reads",
        )

    def execute(self, problem_data: dict, path_info: dict) -> dict:
        num_reads = path_info.get("num_reads")
        if num_reads is None:
            num_reads = self.query.input()
            path_info["num_reads"] = num_reads
        problem_data["solver"] = TabuSolver(num_reads=num_reads)
        return dict(solver=problem_data["solver"])

    def interpret_result(
        self,
        result: dict,
        problem_data: dict,
        next_node_info: dict | None = None,
        config: dict | None = None,
    ) -> dict:
        tabu_result = result["raw"]
        result["solution_vector"] = list(tabu_result.first.sample.values())
        result["best_energy"] = tabu_result.first.energy
        result["best_bitstring"] = "".join([str(int(x)) for x in result["solution_vector"][::-1]])
        result["best_qubo_value"] = result["best_energy"]

        result["solution_bitstring"] = result["best_bitstring"]
        result["solution_qubo_value"] = result["best_qubo_value"]

        result["eigenstate"] = next_node_info["solver"].build_fake_eigenstate(tabu_result)
        result["eigenvalue"] = next_node_info["solver"].build_fake_eigenvalue(tabu_result)

        return result

query instance-attribute 

query = IntQuery(
    question="How many samples should the Tabu Solver generate?",
    default=10,
    name="num_reads",
)

__init__ 

__init__(children)
Source code in src/quast_decisiontree/nodes/algorithm_setup.py 
72
73
74
75
76
77
78
def __init__(self, children: list) -> None:
    super().__init__(creates="solver", children=children, requires=[])
    self.query = IntQuery(
        question="How many samples should the Tabu Solver generate?",
        default=10,
        name="num_reads",
    )

execute 

execute(problem_data, path_info)
Source code in src/quast_decisiontree/nodes/algorithm_setup.py 
80
81
82
83
84
85
86
def execute(self, problem_data: dict, path_info: dict) -> dict:
    num_reads = path_info.get("num_reads")
    if num_reads is None:
        num_reads = self.query.input()
        path_info["num_reads"] = num_reads
    problem_data["solver"] = TabuSolver(num_reads=num_reads)
    return dict(solver=problem_data["solver"])

interpret_result 

interpret_result(
    result, problem_data, next_node_info=None, config=None
)
Source code in src/quast_decisiontree/nodes/algorithm_setup.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def interpret_result(
    self,
    result: dict,
    problem_data: dict,
    next_node_info: dict | None = None,
    config: dict | None = None,
) -> dict:
    tabu_result = result["raw"]
    result["solution_vector"] = list(tabu_result.first.sample.values())
    result["best_energy"] = tabu_result.first.energy
    result["best_bitstring"] = "".join([str(int(x)) for x in result["solution_vector"][::-1]])
    result["best_qubo_value"] = result["best_energy"]

    result["solution_bitstring"] = result["best_bitstring"]
    result["solution_qubo_value"] = result["best_qubo_value"]

    result["eigenstate"] = next_node_info["solver"].build_fake_eigenstate(tabu_result)
    result["eigenvalue"] = next_node_info["solver"].build_fake_eigenvalue(tabu_result)

    return result

LRQAOASetupNode

Bases: Node

Sets up a Linear Ramp QAOA solver (non-variational).

Expects in problem_data
  • cost_operator OR qubo_matrix
  • num_qubits (optional, derived from qubo_matrix)
  • reps: int (number of QAOA layers)
  • delta_gamma: float
  • delta_beta: float
  • backend: Backend instance (optional, defaults to Qrisp simulator)
  • mixer: callable (optional, defaults to RX mixer)
  • shots: int (optional, defaults to 128)
Creates
  • hybrid_algorithm: LRQAOA instance
  • hybrid_inputs: dict with cost_operator and num_qubits
  • ising_offset: float

Modifications at runtime: None

Source code in src/quast_decisiontree/nodes/algorithm_setup.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
class LRQAOASetupNode(Node):
    """Sets up a Linear Ramp QAOA solver (non-variational).

    Expects in problem_data:
        - cost_operator OR qubo_matrix
        - num_qubits (optional, derived from qubo_matrix)
        - reps: int (number of QAOA layers)
        - delta_gamma: float
        - delta_beta: float
        - backend: Backend instance (optional, defaults to Qrisp simulator)
        - mixer: callable (optional, defaults to RX mixer)
        - shots: int (optional, defaults to 128)

    Creates:
        - hybrid_algorithm: LRQAOA instance
        - hybrid_inputs: dict with cost_operator and num_qubits
        - ising_offset: float

    Modifications at runtime: None
    """

    _known_children = ["HybridAlgorithmExecuteNode"]

    def __init__(self, children: list) -> None:
        super().__init__(
            requires=[
                "reps",
                "delta_gamma",
                "delta_beta",
                ["cost_operator", "qubo_matrix"],
                ["optional:", "backend", "shots", "num_qubits", "mixer"],
            ],
            creates=["hybrid_algorithm", "hybrid_inputs", "ising_offset"],
            children=children,
        )

    def execute(self, problem_data: dict, path_info: dict) -> dict:
        if "num_qubits" not in problem_data:
            if "qubo_matrix" in problem_data:
                problem_data["num_qubits"] = len(problem_data["qubo_matrix"])
            else:
                raise ValueError(
                    "Cannot determine num_qubits: neither num_qubits nor "
                    "qubo_matrix found in problem_data."
                )

        if "cost_operator" not in problem_data:
            if "qubo_matrix" not in problem_data:
                raise ValueError("Either cost_operator or qubo_matrix must be in problem_data.")
            qubo_matrix = np.asarray(problem_data["qubo_matrix"], dtype=float)
            cost_operator, ising_offset = qubo_to_qrisp_cost_operator(qubo_matrix)
            problem_data["cost_operator"] = cost_operator
            problem_data["ising_offset"] = ising_offset
        else:
            if "ising_offset" not in problem_data:
                problem_data["ising_offset"] = 0.0

        mixer = problem_data.get("mixer")
        shots = problem_data.get("shots", 128)

        problem_data["hybrid_algorithm"] = LRQAOA(
            reps=problem_data["reps"],
            delta_gamma=problem_data["delta_gamma"],
            delta_beta=problem_data["delta_beta"],
            mixer=mixer,
            shots=shots,
        )
        problem_data["hybrid_inputs"] = {
            "cost_operator": problem_data["cost_operator"],
            "num_qubits": problem_data["num_qubits"],
        }

        return dict()

    def interpret_result(
        self,
        result: dict,
        problem_data: dict,
        next_node_info: dict | None = None,
        config: dict | None = None,
    ) -> dict:
        counts = result["raw"]
        cl_cost_function = problem_data.get("cl_cost_function")

        if cl_cost_function is not None:
            best_bitstring = None
            best_energy = float("inf")
            total_shots = sum(counts.values())
            expectation = 0.0

            for bitstring, count in counts.items():
                energy = cl_cost_function(bitstring)
                expectation += energy * count / total_shots
                if energy < best_energy:
                    best_energy = energy
                    best_bitstring = bitstring

            result["eigenvalue"] = expectation
            result["best_energy"] = best_energy
            result["best_bitstring"] = best_bitstring
            result["eigenstate"] = counts
        else:
            best_bitstring = max(counts, key=counts.get)
            result["best_bitstring"] = best_bitstring
            result["eigenstate"] = counts
            logger.warning(
                "No cl_cost_function in problem_data; cannot compute energies. "
                "Reporting most frequent bitstring."
            )

        result["solution_bitstring"] = result["best_bitstring"]

        if config and config.get("saving_policy") == "discard_if_processed":
            result.setdefault("_no_save", set()).add("raw")

        return result

__init__ 

__init__(children)
Source code in src/quast_decisiontree/nodes/algorithm_setup.py 
133
134
135
136
137
138
139
140
141
142
143
144
def __init__(self, children: list) -> None:
    super().__init__(
        requires=[
            "reps",
            "delta_gamma",
            "delta_beta",
            ["cost_operator", "qubo_matrix"],
            ["optional:", "backend", "shots", "num_qubits", "mixer"],
        ],
        creates=["hybrid_algorithm", "hybrid_inputs", "ising_offset"],
        children=children,
    )

execute 

execute(problem_data, path_info)
Source code in src/quast_decisiontree/nodes/algorithm_setup.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
def execute(self, problem_data: dict, path_info: dict) -> dict:
    if "num_qubits" not in problem_data:
        if "qubo_matrix" in problem_data:
            problem_data["num_qubits"] = len(problem_data["qubo_matrix"])
        else:
            raise ValueError(
                "Cannot determine num_qubits: neither num_qubits nor "
                "qubo_matrix found in problem_data."
            )

    if "cost_operator" not in problem_data:
        if "qubo_matrix" not in problem_data:
            raise ValueError("Either cost_operator or qubo_matrix must be in problem_data.")
        qubo_matrix = np.asarray(problem_data["qubo_matrix"], dtype=float)
        cost_operator, ising_offset = qubo_to_qrisp_cost_operator(qubo_matrix)
        problem_data["cost_operator"] = cost_operator
        problem_data["ising_offset"] = ising_offset
    else:
        if "ising_offset" not in problem_data:
            problem_data["ising_offset"] = 0.0

    mixer = problem_data.get("mixer")
    shots = problem_data.get("shots", 128)

    problem_data["hybrid_algorithm"] = LRQAOA(
        reps=problem_data["reps"],
        delta_gamma=problem_data["delta_gamma"],
        delta_beta=problem_data["delta_beta"],
        mixer=mixer,
        shots=shots,
    )
    problem_data["hybrid_inputs"] = {
        "cost_operator": problem_data["cost_operator"],
        "num_qubits": problem_data["num_qubits"],
    }

    return dict()

interpret_result 

interpret_result(
    result, problem_data, next_node_info=None, config=None
)
Source code in src/quast_decisiontree/nodes/algorithm_setup.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
def interpret_result(
    self,
    result: dict,
    problem_data: dict,
    next_node_info: dict | None = None,
    config: dict | None = None,
) -> dict:
    counts = result["raw"]
    cl_cost_function = problem_data.get("cl_cost_function")

    if cl_cost_function is not None:
        best_bitstring = None
        best_energy = float("inf")
        total_shots = sum(counts.values())
        expectation = 0.0

        for bitstring, count in counts.items():
            energy = cl_cost_function(bitstring)
            expectation += energy * count / total_shots
            if energy < best_energy:
                best_energy = energy
                best_bitstring = bitstring

        result["eigenvalue"] = expectation
        result["best_energy"] = best_energy
        result["best_bitstring"] = best_bitstring
        result["eigenstate"] = counts
    else:
        best_bitstring = max(counts, key=counts.get)
        result["best_bitstring"] = best_bitstring
        result["eigenstate"] = counts
        logger.warning(
            "No cl_cost_function in problem_data; cannot compute energies. "
            "Reporting most frequent bitstring."
        )

    result["solution_bitstring"] = result["best_bitstring"]

    if config and config.get("saving_policy") == "discard_if_processed":
        result.setdefault("_no_save", set()).add("raw")

    return result