microsoft/qdk
Publicmirrored from https://github.com/microsoft/qdkAvailable
source/pip/qsharp/interop/cirq/_result.py
320lines · modecode
| 1 | # Copyright (c) Microsoft Corporation. |
| 2 | # Licensed under the MIT License. |
| 3 | |
| 4 | """Result types and conversion utilities for the Cirq–NeutralAtomDevice integration.""" |
| 5 | |
| 6 | from __future__ import annotations |
| 7 | |
| 8 | import ast |
| 9 | import re |
| 10 | from typing import Any, Dict, List, Optional, Sequence |
| 11 | |
| 12 | import cirq |
| 13 | import numpy as np |
| 14 | |
| 15 | |
| 16 | # --------------------------------------------------------------------------- |
| 17 | # Result type |
| 18 | # --------------------------------------------------------------------------- |
| 19 | |
| 20 | |
| 21 | class NeutralAtomCirqResult(cirq.ResultDict): |
| 22 | """A ``cirq.ResultDict`` that also carries raw (loss-inclusive) shot data. |
| 23 | |
| 24 | The inherited ``measurements`` field contains only *accepted* shots - those |
| 25 | where every measured qubit returned a clean ``{0, 1}`` outcome. Shots in |
| 26 | which one or more qubits were lost during the simulation are excluded from |
| 27 | ``measurements`` but are preserved in ``raw_shots``. |
| 28 | |
| 29 | Attributes: |
| 30 | raw_shots: The full list of simulation results, one entry per shot, |
| 31 | in the native simulator output format (tuple, list, or scalar). |
| 32 | This includes shots that contain qubit-loss markers. |
| 33 | |
| 34 | Methods: |
| 35 | raw_measurements(): Return the full per-shot data (including loss markers) |
| 36 | in the same ``{key: 2D-array (shots x bits)}`` format as |
| 37 | ``measurements``, but with Unicode string dtype so that non-binary |
| 38 | markers are preserved. |
| 39 | """ |
| 40 | |
| 41 | __slots__ = ("raw_shots", "_measurement_dict_data", "_raw_measurements_cache") |
| 42 | |
| 43 | def __init__( |
| 44 | self, |
| 45 | *, |
| 46 | params: cirq.ParamResolver, |
| 47 | measurements: Dict[str, np.ndarray], |
| 48 | raw_shots: List[Any], |
| 49 | measurement_dict: Dict[str, Sequence[int]], |
| 50 | ) -> None: |
| 51 | super().__init__(params=params, measurements=measurements) |
| 52 | self.raw_shots = raw_shots |
| 53 | self._measurement_dict_data = measurement_dict |
| 54 | self._raw_measurements_cache: Optional[Dict[str, Any]] = None |
| 55 | |
| 56 | def raw_measurements(self) -> Dict[str, Any]: |
| 57 | """Return unfiltered per-shot measurement symbols including loss markers. |
| 58 | |
| 59 | The structure mirrors ``measurements``: ``{key: 2D array (shots x bits)}``, |
| 60 | but the array dtype is ``"<U1"`` (single Unicode character) so that |
| 61 | non-binary markers (e.g. ``"-"`` for lost qubits) are preserved. |
| 62 | |
| 63 | The returned arrays should **not** be fed into Cirq tooling that |
| 64 | assumes ``{0, 1}`` integer bit data. |
| 65 | """ |
| 66 | if self._raw_measurements_cache is not None: |
| 67 | return self._raw_measurements_cache |
| 68 | |
| 69 | measurement_dict = self._measurement_dict_data or {"m": []} |
| 70 | measurement_keys = list(measurement_dict.keys()) |
| 71 | key_lengths = [len(measurement_dict[k]) for k in measurement_keys] |
| 72 | |
| 73 | rows_by_key: Dict[str, List[List[str]]] = {k: [] for k in measurement_keys} |
| 74 | |
| 75 | for shot in self.raw_shots: |
| 76 | bitstring = _qir_display_to_bitstring(shot) |
| 77 | registers = _split_registers(bitstring, key_lengths) |
| 78 | |
| 79 | if len(registers) == len(measurement_keys): |
| 80 | parts = registers |
| 81 | else: |
| 82 | flattened = "".join(registers) |
| 83 | parts = _split_registers(flattened, key_lengths) |
| 84 | |
| 85 | for key_index, key in enumerate(measurement_keys): |
| 86 | width = key_lengths[key_index] |
| 87 | if width == 0: |
| 88 | rows_by_key[key].append([]) |
| 89 | continue |
| 90 | |
| 91 | bits = parts[key_index] if key_index < len(parts) else "" |
| 92 | chars = list(str(bits).strip()) |
| 93 | if len(chars) < width: |
| 94 | chars = chars + [""] * (width - len(chars)) |
| 95 | elif len(chars) > width: |
| 96 | chars = chars[:width] |
| 97 | rows_by_key[key].append(chars) |
| 98 | |
| 99 | try: |
| 100 | raw_meas: Dict[str, Any] = { |
| 101 | k: np.asarray(v, dtype="<U1") if v else np.zeros((0, 0), dtype="<U1") |
| 102 | for k, v in rows_by_key.items() |
| 103 | } |
| 104 | except Exception: |
| 105 | raw_meas = rows_by_key # type: ignore[assignment] |
| 106 | |
| 107 | self._raw_measurements_cache = raw_meas |
| 108 | return raw_meas |
| 109 | |
| 110 | |
| 111 | # --------------------------------------------------------------------------- |
| 112 | # Circuit introspection |
| 113 | # --------------------------------------------------------------------------- |
| 114 | |
| 115 | |
| 116 | def measurement_dict(circuit: cirq.Circuit) -> Dict[str, List[int]]: |
| 117 | """Extract ``{measurement_key: [global_qubit_indices]}`` from a Cirq circuit. |
| 118 | |
| 119 | Qubit indices are determined by ``sorted(circuit.all_qubits())``, matching |
| 120 | the ordering that Cirq's ``to_qasm()`` uses when it numbers the qubits. |
| 121 | |
| 122 | Args: |
| 123 | circuit: The Cirq circuit to introspect. |
| 124 | |
| 125 | Returns: |
| 126 | An ordered dict mapping each measurement key to the list of global qubit |
| 127 | indices that key covers, in the order they are measured. |
| 128 | """ |
| 129 | ordered_qubits = sorted(circuit.all_qubits()) |
| 130 | index_by_qubit = {q: i for i, q in enumerate(ordered_qubits)} |
| 131 | |
| 132 | keys_in_order: List[str] = [] |
| 133 | key_to_qubits: Dict[str, List[int]] = {} |
| 134 | |
| 135 | for op in circuit.all_operations(): |
| 136 | if isinstance(op.gate, cirq.MeasurementGate): |
| 137 | key = op.gate.key |
| 138 | if key not in key_to_qubits: |
| 139 | keys_in_order.append(key) |
| 140 | key_to_qubits[key] = [] |
| 141 | key_to_qubits[key].extend(index_by_qubit[q] for q in op.qubits) |
| 142 | |
| 143 | return {k: key_to_qubits[k] for k in keys_in_order} |
| 144 | |
| 145 | |
| 146 | # --------------------------------------------------------------------------- |
| 147 | # Bit-string parsing utilities |
| 148 | # --------------------------------------------------------------------------- |
| 149 | |
| 150 | |
| 151 | def _qir_display_to_bitstring(obj: Any) -> str: |
| 152 | """Convert a raw QIR simulation result value to a flat bitstring. |
| 153 | |
| 154 | Handles the various formats the NeutralAtomDevice simulator may emit: |
| 155 | - ``qsharp.Result`` enum values (``Result.One`` -> ``"1"``, ``Result.Zero`` -> ``"0"``) |
| 156 | - ``tuple`` - multiple classical registers, joined with spaces |
| 157 | - ``list`` - single register bits, each element processed recursively |
| 158 | - ``str`` - already a representation, parsed with ``ast.literal_eval`` if needed |
| 159 | - other - converted to string with ``str()`` |
| 160 | """ |
| 161 | # Handle qsharp.Result enum values produced by the local simulator. |
| 162 | try: |
| 163 | from qsharp import Result as _Result |
| 164 | |
| 165 | if obj == _Result.One: |
| 166 | return "1" |
| 167 | if obj == _Result.Zero: |
| 168 | return "0" |
| 169 | if obj == _Result.Loss: |
| 170 | return "-" |
| 171 | except ImportError: |
| 172 | pass |
| 173 | |
| 174 | if isinstance(obj, str) and not re.match(r"[\d\s\-]+$", obj): |
| 175 | try: |
| 176 | obj = ast.literal_eval(obj) |
| 177 | except Exception: |
| 178 | return str(obj) |
| 179 | |
| 180 | if isinstance(obj, tuple): |
| 181 | return " ".join(_qir_display_to_bitstring(t) for t in obj) |
| 182 | if isinstance(obj, list): |
| 183 | # Recurse per element so Result.One/Zero inside lists are handled correctly. |
| 184 | return "".join(_qir_display_to_bitstring(bit) for bit in obj) |
| 185 | return str(obj) |
| 186 | |
| 187 | |
| 188 | def _split_registers(bitstring: str, key_lengths: List[int]) -> List[str]: |
| 189 | """Split a flat or space-delimited bitstring into per-register chunks. |
| 190 | |
| 191 | Args: |
| 192 | bitstring: The raw bitstring, possibly containing spaces between registers. |
| 193 | key_lengths: The expected width of each register, in order. |
| 194 | |
| 195 | Returns: |
| 196 | A list of register strings, one per key. |
| 197 | """ |
| 198 | raw = str(bitstring).strip() |
| 199 | |
| 200 | if " " in raw: |
| 201 | return raw.split(" ") |
| 202 | |
| 203 | if not key_lengths: |
| 204 | return [raw] |
| 205 | |
| 206 | total_len = sum(key_lengths) |
| 207 | if total_len == len(raw): |
| 208 | regs: List[str] = [] |
| 209 | start = 0 |
| 210 | for length in key_lengths: |
| 211 | regs.append(raw[start : start + length]) |
| 212 | start += length |
| 213 | return regs |
| 214 | |
| 215 | return [raw] |
| 216 | |
| 217 | |
| 218 | # --------------------------------------------------------------------------- |
| 219 | # Loss-filtering shot conversion |
| 220 | # --------------------------------------------------------------------------- |
| 221 | |
| 222 | |
| 223 | def _shots_to_rows( |
| 224 | shots: Sequence[Any], |
| 225 | measurement_dict_data: Optional[Dict[str, Sequence[int]]] = None, |
| 226 | ) -> Dict[str, List[List[int]]]: |
| 227 | """Convert raw simulation shots to ``{key: [[bit_per_shot]]}`` filtering loss. |
| 228 | |
| 229 | Shots where any qubit returned a non-binary value (loss marker) are silently |
| 230 | dropped. Only ``{0, 1}`` shots contribute to the returned arrays. |
| 231 | |
| 232 | Args: |
| 233 | shots: Raw simulation output, one entry per shot. |
| 234 | measurement_dict_data: ``{key: [qubit_indices]}`` - the measurement |
| 235 | register layout. Defaults to a single key ``"m"`` with no qubits. |
| 236 | |
| 237 | Returns: |
| 238 | ``{key: list_of_rows}`` where each row is a list of 0/1 integers. |
| 239 | """ |
| 240 | if measurement_dict_data is None: |
| 241 | measurement_dict_data = {"m": []} |
| 242 | |
| 243 | measurement_keys = list(measurement_dict_data.keys()) |
| 244 | key_lengths = [len(measurement_dict_data[k]) for k in measurement_keys] |
| 245 | |
| 246 | shots_by_key: Dict[str, List[List[int]]] = {k: [] for k in measurement_keys} |
| 247 | |
| 248 | for shot in shots: |
| 249 | bitstring = _qir_display_to_bitstring(shot) |
| 250 | registers = _split_registers(bitstring, key_lengths) |
| 251 | |
| 252 | if len(registers) == len(measurement_keys): |
| 253 | parts = registers |
| 254 | else: |
| 255 | flattened = "".join(registers) |
| 256 | parts = _split_registers(flattened, key_lengths) |
| 257 | |
| 258 | per_key_rows: Dict[str, List[int]] = {} |
| 259 | is_valid_shot = True |
| 260 | |
| 261 | for key, bits in zip(measurement_keys, parts): |
| 262 | bit_chars = list(str(bits).strip()) |
| 263 | if not all(ch in "01" for ch in bit_chars): |
| 264 | is_valid_shot = False |
| 265 | break |
| 266 | per_key_rows[key] = [1 if ch == "1" else 0 for ch in bit_chars] |
| 267 | |
| 268 | if not is_valid_shot: |
| 269 | continue |
| 270 | |
| 271 | for key in measurement_keys: |
| 272 | shots_by_key[key].append(per_key_rows.get(key, [])) |
| 273 | |
| 274 | return shots_by_key |
| 275 | |
| 276 | |
| 277 | # --------------------------------------------------------------------------- |
| 278 | # Result construction |
| 279 | # --------------------------------------------------------------------------- |
| 280 | |
| 281 | |
| 282 | def to_cirq_result( |
| 283 | raw_shots: List[Any], |
| 284 | meas_dict: Dict[str, List[int]], |
| 285 | param_resolver: Optional[cirq.ParamResolverOrSimilarType] = None, |
| 286 | ) -> NeutralAtomCirqResult: |
| 287 | """Build a :class:`NeutralAtomCirqResult` from raw simulation output. |
| 288 | |
| 289 | Args: |
| 290 | raw_shots: The raw per-shot results from ``NeutralAtomDevice.simulate()``. |
| 291 | meas_dict: ``{key: [qubit_indices]}`` as returned by :func:`measurement_dict`. |
| 292 | param_resolver: Cirq parameter resolver for the circuit. Defaults to the |
| 293 | empty resolver. |
| 294 | |
| 295 | Returns: |
| 296 | A ``NeutralAtomCirqResult`` whose ``measurements`` field contains only |
| 297 | loss-free shots, and whose ``raw_shots`` / ``raw_measurements()`` retain |
| 298 | all shots including those with loss markers. |
| 299 | """ |
| 300 | if param_resolver is None: |
| 301 | param_resolver = cirq.ParamResolver({}) |
| 302 | |
| 303 | normalized = meas_dict or {"m": []} |
| 304 | shots_by_key = _shots_to_rows(raw_shots, normalized) |
| 305 | measurement_keys = list(normalized.keys()) |
| 306 | |
| 307 | measurements: Dict[str, np.ndarray] = {} |
| 308 | for key in measurement_keys: |
| 309 | rows = shots_by_key.get(key, []) |
| 310 | if not rows: |
| 311 | measurements[key] = np.zeros((0, 0), dtype=np.int8) |
| 312 | else: |
| 313 | measurements[key] = np.asarray(rows, dtype=np.int8) |
| 314 | |
| 315 | return NeutralAtomCirqResult( |
| 316 | params=param_resolver, |
| 317 | measurements=measurements, |
| 318 | raw_shots=raw_shots, |
| 319 | measurement_dict=normalized, |
| 320 | ) |
| 321 | |