Skip to content

quast_decisiontree.nodes.qrisp_qaoa_setup

quast_decisiontree.nodes.qrisp_qaoa_setup

Setup node for QrispQAOA — assembles HybridAlgorithm + inputs for execution.

logger module-attribute 

logger = logging.getLogger('dt_logger')

QrispQAOASetupNode

Bases: Node

Sets up a QrispQAOA instance and prepares inputs for HybridAlgorithmExecuteNode.

Assembles the hybrid algorithm with its hyperparameters and gathers the required inputs (cost_operator, mixer, cl_cost_function, num_qubits) from problem_data.

If cost_operator/cl_cost_function are not already in problem_data, they are derived from qubo_matrix.

Modifications at runtime: None (all parameters come from prior nodes).

Source code in src/quast_decisiontree/nodes/qrisp_qaoa_setup.py 
 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
 59
 60
 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
108
109
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
class QrispQAOASetupNode(Node):
    """Sets up a QrispQAOA instance and prepares inputs for HybridAlgorithmExecuteNode.

    Assembles the hybrid algorithm with its hyperparameters and gathers the
    required inputs (cost_operator, mixer, cl_cost_function, num_qubits) from
    problem_data.

    If cost_operator/cl_cost_function are not already in problem_data, they
    are derived from qubo_matrix.

    Modifications at runtime: None (all parameters come from prior nodes).
    """

    _known_children = ["HybridAlgorithmExecuteNode"]

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

    def execute(self, problem_data: dict, path_info: dict) -> dict:
        backend = problem_data["backend"]
        if isinstance(backend, str):
            backends = self.request_info("backends")
            problem_data["backend"] = backends[backend]

        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:
            cost_operator = problem_data["cost_operator"]
            if "ising_offset" not in problem_data:
                problem_data["ising_offset"] = 0.0

        if "cl_cost_function" not in problem_data:
            if "qubo_matrix" not in problem_data:
                raise ValueError("Either cl_cost_function or qubo_matrix must be in problem_data.")
            problem_data["cl_cost_function"] = qubo_to_cl_cost_function(
                np.asarray(problem_data["qubo_matrix"], dtype=float)
            )

        if "mixer" not in problem_data:
            problem_data["mixer"] = RX_mixer
            logger.info("No mixer specified — defaulting to RX_mixer.")

        hyperparams = {
            "reps": problem_data["reps"],
            "optimizer": problem_data["optimizer"],
            "max_iter": problem_data.get("max_iter", 50),
            "init_type": problem_data.get("init_type", "random"),
            "init_params": problem_data.get("init_params"),
        }

        problem_data["hybrid_algorithm"] = QrispQAOA(**hyperparams)

        problem_data["hybrid_inputs"] = {
            "cost_operator": problem_data["cost_operator"],
            "mixer": problem_data["mixer"],
            "cl_cost_function": problem_data["cl_cost_function"],
            "num_qubits": problem_data["num_qubits"],
        }

        return dict(
            num_qubits=problem_data["num_qubits"],
            ising_offset=problem_data["ising_offset"],
        )

    def interpret_result(
        self,
        result: dict,
        problem_data: dict,
        next_node_info: dict | None = None,
        config: dict | None = None,
    ) -> dict:
        """Interpret QAOAResult into standard result dict."""
        qaoa_result: QAOAResult = result["raw"]

        result["eigenstate"] = qaoa_result.counts
        result["best_bitstring"] = qaoa_result.best_bitstring
        result["best_energy"] = qaoa_result.best_cost

        ising_offset = (next_node_info or {}).get("ising_offset", 0.0)
        result["qubo_offset"] = ising_offset
        result["best_qubo_value"] = qaoa_result.best_cost
        result["solution_bitstring"] = qaoa_result.best_bitstring
        result["solution_qubo_value"] = qaoa_result.best_cost

        if qaoa_result.cost_history:
            result["cost_history"] = qaoa_result.cost_history
            result["num_evals"] = qaoa_result.num_evals

        if qaoa_result.optimal_params is not None:
            result["optimal_params"] = qaoa_result.optimal_params

        if qaoa_result.counts and problem_data.get("cl_cost_function"):
            cl_cost = problem_data["cl_cost_function"]
            result["eigenvalue"] = cl_cost(qaoa_result.counts)
            result["qubo_eigenvalue"] = result["eigenvalue"]
        else:
            result["eigenvalue"] = qaoa_result.best_cost

        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/qrisp_qaoa_setup.py 
41
42
43
44
45
46
47
48
49
50
51
52
def __init__(self, children: list) -> None:
    super().__init__(
        requires=[
            "optimizer",
            "reps",
            "backend",
            ["cost_operator", "qubo_matrix"],
            ["optional:", "mixer", "num_qubits", "max_iter", "init_params", "init_type"],
        ],
        creates=["hybrid_algorithm", "hybrid_inputs", "ising_offset"],
        children=children,
    )

execute 

execute(problem_data, path_info)
Source code in src/quast_decisiontree/nodes/qrisp_qaoa_setup.py 
 54
 55
 56
 57
 58
 59
 60
 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
108
109
110
111
112
def execute(self, problem_data: dict, path_info: dict) -> dict:
    backend = problem_data["backend"]
    if isinstance(backend, str):
        backends = self.request_info("backends")
        problem_data["backend"] = backends[backend]

    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:
        cost_operator = problem_data["cost_operator"]
        if "ising_offset" not in problem_data:
            problem_data["ising_offset"] = 0.0

    if "cl_cost_function" not in problem_data:
        if "qubo_matrix" not in problem_data:
            raise ValueError("Either cl_cost_function or qubo_matrix must be in problem_data.")
        problem_data["cl_cost_function"] = qubo_to_cl_cost_function(
            np.asarray(problem_data["qubo_matrix"], dtype=float)
        )

    if "mixer" not in problem_data:
        problem_data["mixer"] = RX_mixer
        logger.info("No mixer specified — defaulting to RX_mixer.")

    hyperparams = {
        "reps": problem_data["reps"],
        "optimizer": problem_data["optimizer"],
        "max_iter": problem_data.get("max_iter", 50),
        "init_type": problem_data.get("init_type", "random"),
        "init_params": problem_data.get("init_params"),
    }

    problem_data["hybrid_algorithm"] = QrispQAOA(**hyperparams)

    problem_data["hybrid_inputs"] = {
        "cost_operator": problem_data["cost_operator"],
        "mixer": problem_data["mixer"],
        "cl_cost_function": problem_data["cl_cost_function"],
        "num_qubits": problem_data["num_qubits"],
    }

    return dict(
        num_qubits=problem_data["num_qubits"],
        ising_offset=problem_data["ising_offset"],
    )

interpret_result 

interpret_result(
    result, problem_data, next_node_info=None, config=None
)

Interpret QAOAResult into standard result dict.

Source code in src/quast_decisiontree/nodes/qrisp_qaoa_setup.py 
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
def interpret_result(
    self,
    result: dict,
    problem_data: dict,
    next_node_info: dict | None = None,
    config: dict | None = None,
) -> dict:
    """Interpret QAOAResult into standard result dict."""
    qaoa_result: QAOAResult = result["raw"]

    result["eigenstate"] = qaoa_result.counts
    result["best_bitstring"] = qaoa_result.best_bitstring
    result["best_energy"] = qaoa_result.best_cost

    ising_offset = (next_node_info or {}).get("ising_offset", 0.0)
    result["qubo_offset"] = ising_offset
    result["best_qubo_value"] = qaoa_result.best_cost
    result["solution_bitstring"] = qaoa_result.best_bitstring
    result["solution_qubo_value"] = qaoa_result.best_cost

    if qaoa_result.cost_history:
        result["cost_history"] = qaoa_result.cost_history
        result["num_evals"] = qaoa_result.num_evals

    if qaoa_result.optimal_params is not None:
        result["optimal_params"] = qaoa_result.optimal_params

    if qaoa_result.counts and problem_data.get("cl_cost_function"):
        cl_cost = problem_data["cl_cost_function"]
        result["eigenvalue"] = cl_cost(qaoa_result.counts)
        result["qubo_eigenvalue"] = result["eigenvalue"]
    else:
        result["eigenvalue"] = qaoa_result.best_cost

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

    return result