update QuantumRandomAccessEncoding

Author: t-imamichi

qiskit_optimization/algorithms/qrao/encoding_commutation_verifier.py

@@ -1,6 +1,6 @@
 # This code is part of a Qiskit project.
 #
-# (C) Copyright IBM 2023.
+# (C) Copyright IBM 2023, 2024.
 #
 # This code is licensed under the Apache License, Version 2.0. You may
 # obtain a copy of this license in the LICENSE.txt file in the root directory
@@ -14,7 +14,8 @@
 
 from __future__ import annotations
 
-from qiskit.primitives import BaseEstimator
+from qiskit.passmanager import BasePassManager
+from qiskit.primitives import BaseEstimatorV1, BaseEstimatorV2
 
 from qiskit_optimization.exceptions import QiskitOptimizationError
 
@@ -24,14 +25,21 @@
 class EncodingCommutationVerifier:
     """Class for verifying that the relaxation commutes with the objective function."""
 
-    def __init__(self, encoding: QuantumRandomAccessEncoding, estimator: BaseEstimator):
+    def __init__(
+        self,
+        encoding: QuantumRandomAccessEncoding,
+        estimator: BaseEstimatorV1 | BaseEstimatorV2,
+        passmanager: BasePassManager | None = None,
+    ):
         """
         Args:
             encoding: The encoding to verify.
             estimator: The estimator to use for the verification.
+            passmanager: The pass manager to tranpile the circuits
         """
         self._encoding = encoding
         self._estimator = estimator
+        self._passmanager = passmanager
 
     def __len__(self) -> int:
         return 2**self._encoding.num_vars
@@ -48,6 +56,8 @@ def __getitem__(self, i: int) -> tuple[str, float, float]:
         str_dvars = f"{i:0{encoding.num_vars}b}"
         dvars = [int(b) for b in str_dvars]
         encoded_bitstr_qc = encoding.state_preparation_circuit(dvars)
+        if self._passmanager:
+            encoded_bitstr_qc = self._passmanager.run(encoded_bitstr_qc)
 
         # Evaluate the original objective function
         problem = encoding.problem
@@ -58,13 +68,24 @@ def __getitem__(self, i: int) -> tuple[str, float, float]:
         encoded_op = encoding.qubit_op
         offset = encoding.offset
 
-        job = self._estimator.run([encoded_bitstr_qc], [encoded_op])
-
-        try:
-            encoded_obj_val = job.result().values[0] + offset
-        except Exception as exc:
-            raise QiskitOptimizationError(
-                "The primitive job to verify commutation failed!"
-            ) from exc
-
-        return (str_dvars, obj_val, encoded_obj_val)
+        if isinstance(self._estimator, BaseEstimatorV1):
+            job = self._estimator.run([encoded_bitstr_qc], [encoded_op])
+
+            try:
+                encoded_obj_val = job.result().values[0] + offset
+            except Exception as exc:
+                raise QiskitOptimizationError(
+                    "The primitive job to verify commutation failed!"
+                ) from exc
+        else:  # BaseEstimatorV2
+            job = self._estimator.run([(encoded_bitstr_qc, encoded_op)])
+
+            try:
+                result = job.result()
+                encoded_obj_val = result[0].data.evs.item() + offset
+            except Exception as exc:
+                raise QiskitOptimizationError(
+                    "The primitive job to verify commutation failed!"
+                ) from exc
+
+        return str_dvars, obj_val, encoded_obj_val

test/algorithms/qrao/test_quantum_random_access_encoding.py

@@ -1,6 +1,6 @@
 # This code is part of a Qiskit project.
 #
-# (C) Copyright IBM 2023.
+# (C) Copyright IBM 2023, 2024.
 #
 # This code is licensed under the Apache License, Version 2.0. You may
 # obtain a copy of this license in the LICENSE.txt file in the root directory
@@ -13,22 +13,21 @@
 """Tests for QuantumRandomAccessEncoding"""
 import itertools
 import unittest
-from test.optimization_test_case import QiskitOptimizationTestCase
+from test import QiskitOptimizationTestCase
 
-from ddt import ddt, data, unpack
-import numpy as np
 import networkx as nx
-
+import numpy as np
+from ddt import data, ddt, unpack
 from qiskit.circuit import QuantumCircuit
-from qiskit.primitives import Estimator
+from qiskit.primitives import Estimator, StatevectorEstimator
 from qiskit.quantum_info import SparsePauliOp
 
 from qiskit_optimization.algorithms.qrao import (
     EncodingCommutationVerifier,
     QuantumRandomAccessEncoding,
 )
-from qiskit_optimization.problems import QuadraticProgram, QuadraticObjective
 from qiskit_optimization.applications import Maxcut
+from qiskit_optimization.problems import QuadraticObjective, QuadraticProgram
 
 
 class TestQuantumRandomAccessEncoding(QiskitOptimizationTestCase):
@@ -252,17 +251,20 @@ def test_qrac_unsupported_encoding(self):
 class TestEncodingCommutationVerifier(QiskitOptimizationTestCase):
     """Tests for EncodingCommutationVerifier."""
 
-    def check_problem_commutation(self, problem: QuadraticProgram, max_vars_per_qubit: int):
+    def check_problem_commutation(
+        self, problem: QuadraticProgram, max_vars_per_qubit: int, version: str
+    ):
         """Utility function to check that the problem commutes with its encoding"""
         encoding = QuantumRandomAccessEncoding(max_vars_per_qubit=max_vars_per_qubit)
         encoding.encode(problem)
-        estimator = Estimator()
+        estimator = Estimator() if version == "v1" else StatevectorEstimator()
         verifier = EncodingCommutationVerifier(encoding, estimator)
         self.assertEqual(len(verifier), 2**encoding.num_vars)
         for _, obj_val, encoded_obj_val in verifier:
             np.testing.assert_allclose(obj_val, encoded_obj_val, atol=1e-5)
 
-    def test_encoding_commutation_verifier(self):
+    @data("v1", "v2")
+    def test_encoding_commutation_verifier(self, version):
         """Test EncodingCommutationVerifier"""
         problem = QuadraticProgram()
         problem.binary_var("x")
@@ -272,11 +274,11 @@ def test_encoding_commutation_verifier(self):
 
         encoding = QuantumRandomAccessEncoding(max_vars_per_qubit=3)
         encoding.encode(problem)
-        self.check_problem_commutation(problem, 3)
+        self.check_problem_commutation(problem, 3, version)
 
-    @data(*itertools.product([1, 2, 3], ["minimize", "maximize"]))
+    @data(*itertools.product([1, 2, 3], ["minimize", "maximize"], ["v1", "v2"]))
     @unpack
-    def test_one_qubit_qrac(self, max_vars_per_qubit, task):
+    def test_one_qubit_qrac(self, max_vars_per_qubit, task, version):
         """Test commutation of single qubit QRAC with non-uniform weights, degree 1 terms"""
 
         problem = QuadraticProgram()
@@ -287,15 +289,17 @@ def test_one_qubit_qrac(self, max_vars_per_qubit, task):
             problem.minimize(linear=obj)
         else:
             problem.maximize(linear=obj)
-        self.check_problem_commutation(problem, max_vars_per_qubit)
+        self.check_problem_commutation(problem, max_vars_per_qubit, version)
 
     @data(
         *itertools.product(
-            [1, 2, 3], [QuadraticObjective.Sense.MINIMIZE, QuadraticObjective.Sense.MAXIMIZE]
+            [1, 2, 3],
+            [QuadraticObjective.Sense.MINIMIZE, QuadraticObjective.Sense.MAXIMIZE],
+            ["v1", "v2"],
         )
     )
     @unpack
-    def test_uniform_weights_degree_2(self, max_vars_per_qubit, task):
+    def test_uniform_weights_degree_2(self, max_vars_per_qubit, task, version):
         """Test problem commutation with degree 2 terms"""
         # Note that the variable embedding has some qubits with 1, 2, and 3 qubits
         elist = [(0, 1), (1, 2), (2, 3), (3, 4), (4, 5), (5, 0), (0, 3), (1, 4), (2, 4)]
@@ -306,25 +310,27 @@ def test_uniform_weights_degree_2(self, max_vars_per_qubit, task):
         maxcut = Maxcut(graph)
         problem = maxcut.to_quadratic_program()
         problem.objective.sense = task
-        self.check_problem_commutation(problem, max_vars_per_qubit)
+        self.check_problem_commutation(problem, max_vars_per_qubit, version)
 
-    @data(1, 2, 3)
-    def test_random_unweighted_maxcut(self, max_vars_per_qubit):
+    @data(*itertools.product([1, 2, 3], ["v1", "v2"]))
+    @unpack
+    def test_random_unweighted_maxcut(self, max_vars_per_qubit, version):
         """Test problem commutation with random unweighted MaxCut"""
         graph = nx.random_regular_graph(3, 8)
         maxcut = Maxcut(graph)
         problem = maxcut.to_quadratic_program()
-        self.check_problem_commutation(problem, max_vars_per_qubit)
+        self.check_problem_commutation(problem, max_vars_per_qubit, version)
 
-    @data(1, 2, 3)
-    def test_random_weighted_maxcut(self, max_vars_per_qubit):
+    @data(*itertools.product([1, 2, 3], ["v1", "v2"]))
+    @unpack
+    def test_random_weighted_maxcut(self, max_vars_per_qubit, version):
         """Test problem commutation with random weighted MaxCut"""
         graph = nx.random_regular_graph(3, 8)
         for w, v in graph.edges:
             graph[w][v]["weight"] = np.random.randint(1, 10)
         maxcut = Maxcut(graph)
         problem = maxcut.to_quadratic_program()
-        self.check_problem_commutation(problem, max_vars_per_qubit)
+        self.check_problem_commutation(problem, max_vars_per_qubit, version)
 
 
 if __name__ == "__main__":