feat(sampler): fix sampling base conversion

qiskit_scaleway/backends/base_job.py

@@ -80,7 +80,7 @@ def submit(self, session_id: str) -> None:
 
         options = self._config.copy()
         shots = options.pop("shots")
-        memory = options.pop("memory")
+        memory = options.pop("memory", False)
 
         programs = map(lambda c: QuantumProgram.from_qiskit_circuit(c), self._circuits)
 

qiskit_scaleway/primitives/sampler.py

@@ -11,7 +11,26 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-from qiskit.primitives import BackendSamplerV2
+import numpy as np
+import re
+
+from numpy.typing import NDArray
+
+from qiskit.primitives.backend_sampler_v2 import (
+    BackendSamplerV2,
+    _MeasureInfo,
+    _samples_to_packed_array,
+    QiskitError,
+    ResultMemory,
+)
+
+from qiskit.primitives.containers import (
+    BitArray,
+    DataBin,
+    SamplerPubResult,
+)
+
+_NON_BINARY_CHARS = re.compile(r"[^01]")
 
 
 class Sampler(BackendSamplerV2):
@@ -37,3 +56,68 @@ def __init__(
             backend=backend,
             options=options,
         )
+
+    def _postprocess_pub(
+        self,
+        result_memory: list[ResultMemory],
+        shots: int,
+        shape: tuple[int, ...],
+        meas_info: list[_MeasureInfo],
+        max_num_bytes: int,
+        circuit_metadata: dict,
+        meas_level: int | None,
+    ) -> SamplerPubResult:
+        """Converts the memory data into a sampler pub result
+
+        For level 2 data, the memory data are stored in an array of bit arrays
+        with the shape of the pub. For level 1 data, the data are stored in a
+        complex numpy array.
+        """
+        if meas_level == 2 or meas_level is None:
+            arrays = {
+                item.creg_name: np.zeros(
+                    shape + (shots, item.num_bytes), dtype=np.uint8
+                )
+                for item in meas_info
+            }
+            memory_array = _memory_array(result_memory, max_num_bytes)
+
+            for samples, index in zip(memory_array, np.ndindex(*shape)):
+                for item in meas_info:
+                    ary = _samples_to_packed_array(samples, item.num_bits, item.start)
+                    arrays[item.creg_name][index] = ary
+
+            meas = {
+                item.creg_name: BitArray(arrays[item.creg_name], item.num_bits)
+                for item in meas_info
+            }
+        elif meas_level == 1:
+            raw = np.array(result_memory)
+            cplx = raw[..., 0] + 1j * raw[..., 1]
+            cplx = np.reshape(cplx, (*shape, *cplx.shape[1:]))
+            meas = {item.creg_name: cplx for item in meas_info}
+        else:
+            raise QiskitError(f"Unsupported meas_level: {meas_level}")
+        return SamplerPubResult(
+            DataBin(**meas, shape=shape),
+            metadata={"shots": shots, "circuit_metadata": circuit_metadata},
+        )
+
+
+def _memory_array(results: list[list[str]], num_bytes: int) -> NDArray[np.uint8]:
+    """Converts the memory data into an array in an unpacked way."""
+    lst = []
+    # Heuristic: check only the first result format
+    if len(results) > 0 and len(results[0]) > 0:
+        base = 16 if _NON_BINARY_CHARS.search(results[0][0]) else 2
+
+    for memory in results:
+        if num_bytes > 0:
+            data = b"".join(int(i, base).to_bytes(num_bytes, "big") for i in memory)
+            data = np.frombuffer(data, dtype=np.uint8).reshape(-1, num_bytes)
+        else:
+            # no measure in a circuit
+            data = np.zeros((len(memory), num_bytes), dtype=np.uint8)
+        lst.append(data)
+    ary = np.asarray(lst)
+    return np.unpackbits(ary, axis=-1, bitorder="big")

tests/test_aer_multiple_circuits.py

@@ -12,42 +12,15 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import os
-import numpy as np
 import random
 
 from qiskit import QuantumCircuit
 from qiskit_scaleway import ScalewayProvider
 
-
-def _random_qiskit_circuit(size: int) -> QuantumCircuit:
-    num_qubits = size
-    num_gate = size
-
-    qc = QuantumCircuit(num_qubits)
-
-    for _ in range(num_gate):
-        random_gate = np.random.choice(["unitary", "cx", "cy", "cz"])
-
-        if random_gate == "cx" or random_gate == "cy" or random_gate == "cz":
-            control_qubit = np.random.randint(0, num_qubits)
-            target_qubit = np.random.randint(0, num_qubits)
-
-            while target_qubit == control_qubit:
-                target_qubit = np.random.randint(0, num_qubits)
-
-            getattr(qc, random_gate)(control_qubit, target_qubit)
-        else:
-            for q in range(num_qubits):
-                random_gate = np.random.choice(["h", "x", "y", "z"])
-                getattr(qc, random_gate)(q)
-
-    qc.measure_all()
-
-    return qc
+from qio.utils.circuit import random_square_qiskit_circuit
 
 
 def test_aer_multiple_circuits():
-
     provider = ScalewayProvider(
         project_id=os.environ["QISKIT_SCALEWAY_PROJECT_ID"],
         secret_key=os.environ["QISKIT_SCALEWAY_SECRET_KEY"],
@@ -69,10 +42,10 @@ def test_aer_multiple_circuits():
     assert session_id is not None
 
     try:
-        qc1 = _random_qiskit_circuit(20)
-        qc2 = _random_qiskit_circuit(15)
-        qc3 = _random_qiskit_circuit(21)
-        qc4 = _random_qiskit_circuit(17)
+        qc1 = random_square_qiskit_circuit(20)
+        qc2 = random_square_qiskit_circuit(15)
+        qc3 = random_square_qiskit_circuit(21)
+        qc4 = random_square_qiskit_circuit(17)
 
         run_result = backend.run(
             [qc1, qc2, qc3, qc4],
@@ -126,7 +99,6 @@ def _simple_one_state_circuit():
 
 
 def test_aer_with_noise_model():
-
     provider = ScalewayProvider(
         project_id=os.environ["QISKIT_SCALEWAY_PROJECT_ID"],
         secret_key=os.environ["QISKIT_SCALEWAY_SECRET_KEY"],

tests/test_aqt_multiple_circuits.py

@@ -12,38 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import os
-import numpy as np
 import random
 
-from qiskit import QuantumCircuit
 from qiskit_scaleway import ScalewayProvider
 
-
-def _random_qiskit_circuit(size: int) -> QuantumCircuit:
-    num_qubits = size
-    num_gate = size
-
-    qc = QuantumCircuit(num_qubits)
-
-    for _ in range(num_gate):
-        random_gate = np.random.choice(["unitary", "cx", "cy", "cz"])
-
-        if random_gate == "cx" or random_gate == "cy" or random_gate == "cz":
-            control_qubit = np.random.randint(0, num_qubits)
-            target_qubit = np.random.randint(0, num_qubits)
-
-            while target_qubit == control_qubit:
-                target_qubit = np.random.randint(0, num_qubits)
-
-            getattr(qc, random_gate)(control_qubit, target_qubit)
-        else:
-            for q in range(num_qubits):
-                random_gate = np.random.choice(["h", "x", "y", "z"])
-                getattr(qc, random_gate)(q)
-
-    qc.measure_all()
-
-    return qc
+from qio.utils.circuit import random_square_qiskit_circuit
 
 
 def test_aqt_multiple_circuits():
@@ -66,10 +39,10 @@ def test_aqt_multiple_circuits():
     assert session_id is not None
 
     try:
-        qc1 = _random_qiskit_circuit(10)
-        qc2 = _random_qiskit_circuit(12)
-        qc3 = _random_qiskit_circuit(9)
-        qc4 = _random_qiskit_circuit(10)
+        qc1 = random_square_qiskit_circuit(10)
+        qc2 = random_square_qiskit_circuit(12)
+        qc3 = random_square_qiskit_circuit(9)
+        qc4 = random_square_qiskit_circuit(10)
 
         run_result = backend.run(
             [qc1, qc2, qc3, qc4],