feat: added integration for PPVM simulator backend

pyproject.toml

@@ -57,6 +57,9 @@ msd-reprod = [
 clifft = [
     "clifft; python_version >= '3.12'"
 ]
+ppvm = [
+    "ppvm"
+]
 
 [tool.maturin]
 manifest-path = "crates/bloqade-lanes-bytecode-python/Cargo.toml"
@@ -187,3 +190,6 @@ testpaths = "python/tests/"
 markers = [
     "msd_numerical_regression: optional long-running, environment-specific MSD numerical regression",
 ]
+
+[tool.uv.sources]
+ppvm = { git = "https://github.com/QuEraComputing/ppvm.git", subdirectory = "ppvm-python" }

python/bloqade/gemini/__init__.py

@@ -18,6 +18,7 @@
     PhysicalResult as PhysicalResult,
     PhysicalSimulator as PhysicalSimulator,
     PhysicalSimulatorTask as PhysicalSimulatorTask,
+    PPVMSimulatorBackend as PPVMSimulatorBackend,
     Result as Result,
     SimulatorResult as SimulatorResult,
     TsimSimulatorBackend as TsimSimulatorBackend,

python/bloqade/gemini/device/__init__.py

@@ -22,5 +22,6 @@
     AbstractSimulatorBackend as AbstractSimulatorBackend,
     BackendSample as BackendSample,
     CliffTSimulatorBackend as CliffTSimulatorBackend,
+    PPVMSimulatorBackend as PPVMSimulatorBackend,
     TsimSimulatorBackend as TsimSimulatorBackend,
 )

python/bloqade/gemini/device/_task_runtime.py

@@ -281,17 +281,37 @@ def fidelity_bounds(self) -> tuple[float, float]:
         return min_fidelity, max_fidelity
 
     @staticmethod
-    def _normalize_matrix(payload: Any, *, name: str, shots: int) -> list[list[bool]]:
+    def _normalize_matrix(
+        payload: Any,
+        *,
+        name: str,
+        shots: int,
+        loss_replace: Any = None,
+        loss: Any = None,
+    ) -> list[list[bool]]:
+        """
+        Checks that the shape of the returned measurements is valid, and
+        adds configuration options for converting values of `loss` (representing atom loss values) to `loss_replace`.
+
+        By default, loss and loss_replace are no-ops (converts None to None).
+        """
         try:
-            array = np.asarray(payload, dtype=bool)
+            array = np.asarray(payload, dtype=object)
+
+            # NOTE: this is to model a lack of loss-resolved readout; atom loss is indistinguishable from measuring the |1> state
+            array[array == loss] = loss_replace
+            array = array.astype(bool)
         except (TypeError, ValueError) as exc:
             raise ValueError(f"Backend returned invalid {name} samples") from exc
+
         if array.ndim != 2:
             raise ValueError(f"Backend {name} samples must be a two-dimensional array")
+
         if array.shape[0] != shots:
             raise ValueError(
                 f"Backend returned {array.shape[0]} {name} rows for {shots} shots"
             )
+
         return array.tolist()
 
     def run(

python/bloqade/gemini/device/physical_simulator.py

@@ -4,6 +4,7 @@
 from functools import cache, cached_property
 from typing import (
     TYPE_CHECKING,
+    Any,
     Generic,
     TypeVar,
 )
@@ -190,6 +191,27 @@ def noiseless_physical_squin_kernel(self) -> ir.Method[[], RetType]:
             arch_spec=self.physical_arch_spec,
         ).emit(self.physical_move_kernel)
 
+    @staticmethod
+    def _normalize_matrix(
+        payload: Any,
+        *,
+        name: str,
+        shots: int,
+        loss_replace: Any = True,
+        loss: Any = None,
+    ) -> list[list[bool]]:
+        """
+        Overrides _normalize_matrix to convert the None values in the measurement output to True (to model a lack of loss-resolved readout
+        on Gemini)
+        """
+        return _SimulatorTaskBase._normalize_matrix(
+            payload=payload,
+            name=name,
+            shots=shots,
+            loss_replace=loss_replace,
+            loss=loss,
+        )
+
 
 @dataclass
 class GeminiPhysicalSimulator:

python/bloqade/gemini/device/simulator.py

@@ -4,6 +4,7 @@
 from functools import cached_property
 from typing import (
     TYPE_CHECKING,
+    Any,
     Generic,
     TypeVar,
 )
@@ -78,6 +79,27 @@ def noiseless_physical_squin_kernel(self) -> ir.Method[[], RetType]:
             add_noise=False,
         ).emit(self.physical_move_kernel)
 
+    @staticmethod
+    def _normalize_matrix(
+        payload: Any,
+        *,
+        name: str,
+        shots: int,
+        loss_replace: Any = True,
+        loss: Any = None,
+    ) -> list[list[bool]]:
+        """
+        Overrides _normalize_matrix to convert the None values in the measurement output to True (to model a lack of loss-resolved readout
+        on Gemini)
+        """
+        return _SimulatorTaskBase._normalize_matrix(
+            payload=payload,
+            name=name,
+            shots=shots,
+            loss_replace=loss_replace,
+            loss=loss,
+        )
+
 
 @dataclass
 class GeminiLogicalSimulator:

python/bloqade/gemini/device/simulator_backend.py

@@ -13,7 +13,7 @@
 from kirin.rewrite.abc import RewriteResult, RewriteRule
 
 if TYPE_CHECKING:
-    from stim import DetectorErrorModel
+    from stim import Circuit as StimCircuit, DetectorErrorModel
     from tsim import Circuit as TsimCircuit  # type: ignore[reportMissingImports]
 
 
@@ -191,6 +191,43 @@ def _clifft_compatible_stim_text(circuit: Any) -> str:
     )
 
 
+_PPVM_CORRELATED_LOSS_EVENT = re.compile(
+    r"^(?P<instruction>\s*I_ERROR)"
+    r"\[correlated_loss:(?P<event_id>\d+)\]"
+    r"(?P<arguments>\([^)\n]*\))"
+    r"(?P<targets>\s+[^#\n]+)"
+    r"(?P<comment>#.*)?$"
+)
+
+
+def _ppvm_compatible_stim_text(circuit: str | StimCircuit) -> str:
+    """Convert paired Bloqade correlated-loss tags to PPVM syntax."""
+
+    converted_lines: list[str] = []
+    for line in str(circuit).splitlines():
+        match = _PPVM_CORRELATED_LOSS_EVENT.match(line)
+        if match is None:
+            converted_lines.append(line)
+            continue
+
+        targets = match.group("targets").split()
+        if len(targets) != 2:
+            event_id = match.group("event_id")
+            raise ValueError(
+                "PPVM requires each Bloqade correlated-loss event to target "
+                f"exactly two qubits, but correlated_loss:{event_id} targets "
+                f"{len(targets)}."
+            )
+
+        converted_lines.append(
+            f"{match.group('instruction')}[correlated_loss]"
+            f"{match.group('arguments')}{match.group('targets')}"
+            f"{match.group('comment') or ''}"
+        )
+
+    return "\n".join(converted_lines)
+
+
 def _clifft() -> Any:
     try:
         import clifft  # type: ignore[reportMissingImports]
@@ -204,6 +241,18 @@ def _clifft() -> Any:
     return clifft
 
 
+def _ppvm() -> Any:
+    try:
+        import ppvm  # type: ignore[reportMissingImports]
+    except ImportError as exc:
+        raise ImportError(
+            "PPVM simulation requires the optional `ppvm` dependency. "
+            "Install it with `bloqade-lanes[ppvm,sim]`."
+        ) from exc
+
+    return ppvm
+
+
 def _clifft_tsim_import_error(exc: ImportError) -> ImportError:
     return ImportError(
         "CliffT simulation also requires `bloqade-lanes[sim]`: Tsim performs "
@@ -212,6 +261,14 @@ def _clifft_tsim_import_error(exc: ImportError) -> ImportError:
     )
 
 
+def _ppvm_tsim_import_error(exc: ImportError) -> ImportError:
+    return ImportError(
+        "PPVM simulation also requires `bloqade-lanes[sim]`: Tsim performs "
+        "physical SQuIn conversion and detector error model generation even "
+        "when CliffT is selected for sampling."
+    )
+
+
 @dataclass
 class CliffTSimulatorBackend(AbstractSimulatorBackend):
     """Backend using Tsim for conversion/DEM generation and CliffT for sampling.
@@ -289,6 +346,89 @@ def _pyqrack_tsim_import_error(exc: ImportError) -> ImportError:
     )
 
 
+@dataclass
+class PPVMSimulatorBackend(AbstractSimulatorBackend):
+    """
+    Backend using Tsim for conversion/DEM generation and PPVM for sampling.
+
+    ``seed`` is a construction-time root seed. Each seeded sampling request
+    receives the next derived child seed from this backend's private stream.
+    """
+
+    seed: int | None = None
+    _tsim_backend: TsimSimulatorBackend = field(
+        default_factory=TsimSimulatorBackend, repr=False
+    )
+    _programs: WeakKeyDictionary[ir.Method, Any] = field(
+        default_factory=WeakKeyDictionary, init=False, repr=False
+    )
+    _rng_state: np.random.Generator | None = field(init=False, repr=False)
+    _rng_lock: LockType = field(default_factory=Lock, init=False, repr=False)
+
+    def __post_init__(self):
+        _validate_seed(self.seed)
+        self._rng_state = (
+            np.random.default_rng(self.seed) if self.seed is not None else None
+        )
+
+    def _tsim_circuit(self, physical_squin_kernel: ir.Method) -> TsimCircuit:
+        try:
+            return self._tsim_backend._tsim_circuit(physical_squin_kernel)
+        except ImportError as exc:
+            raise _ppvm_tsim_import_error(exc) from exc
+
+    def _ppvm_program(self, physical_squin_kernel: ir.Method) -> Any:
+        try:
+            return self._programs[physical_squin_kernel]
+        except KeyError:
+            pass
+
+        ppvm = _ppvm()
+        tsim_circuit = self._tsim_circuit(physical_squin_kernel)
+        stim_text = _ppvm_compatible_stim_text(tsim_circuit.stim_circuit)
+
+        program = ppvm.StimProgram.parse(
+            stim_text
+        )  # parses and validates compatibility
+        self._programs[physical_squin_kernel] = program
+        return program
+
+    def sample(
+        self,
+        physical_squin_kernel: ir.Method,
+        *,
+        shots: int,
+    ) -> BackendSample:
+        effective_seed = _next_child_seed(self._rng_state, self._rng_lock)
+
+        program = self._ppvm_program(physical_squin_kernel)
+
+        ppvm = _ppvm()
+        samples = ppvm.sample_stim(
+            program,
+            n_qubits=program.num_qubits,
+            num_shots=int(shots),
+            seed=effective_seed,
+        )
+
+        raw_measurements = np.asarray(samples)
+
+        measurements = np.empty(raw_measurements.shape, dtype=object)
+        measurements[raw_measurements == ppvm.MeasurementResult.ZERO] = False
+        measurements[raw_measurements == ppvm.MeasurementResult.ONE] = True
+        measurements[raw_measurements == ppvm.MeasurementResult.LOST] = None
+
+        return BackendSample(measurements=measurements)
+
+    def _detector_error_model(
+        self, physical_squin_kernel: ir.Method
+    ) -> DetectorErrorModel:
+        try:
+            return self._tsim_backend._detector_error_model(physical_squin_kernel)
+        except ImportError as exc:
+            raise _ppvm_tsim_import_error(exc) from exc
+
+
 class _RemovePyQrackAnnotations(RewriteRule):
     def rewrite_Statement(self, node: ir.Statement) -> RewriteResult:
         from bloqade.decoders.dialects.annotate.stmts import (

python/tests/gemini/test_physical_simulator.py

@@ -125,6 +125,28 @@ def test_logical_and_physical_tasks_share_non_dataclass_runtime():
     assert issubclass(PhysicalSimulatorTask, _SimulatorTaskBase)
 
 
+@pytest.mark.parametrize(
+    "task_type", [GeminiLogicalSimulatorTask, PhysicalSimulatorTask]
+)
+def test_simulator_tasks_normalize_lost_measurements_as_true(task_type):
+    assert task_type._normalize_matrix(
+        [[False, None], [None, True]], name="measurement", shots=2
+    ) == [[False, True], [True, True]]
+
+
+@pytest.mark.parametrize(
+    "task_type", [GeminiLogicalSimulatorTask, PhysicalSimulatorTask]
+)
+def test_simulator_tasks_forward_loss_normalization_options(task_type):
+    assert task_type._normalize_matrix(
+        [[True, "lost"]],
+        name="measurement",
+        shots=1,
+        loss_replace=False,
+        loss="lost",
+    ) == [[True, False]]
+
+
 @pytest.mark.parametrize("simulator_type", [GeminiLogicalSimulator, PhysicalSimulator])
 @pytest.mark.parametrize(
     "method",