style: update ruff lint ignore configuration and format Python codebase

Author: qectorlab

pyproject.toml

@@ -195,8 +195,8 @@ docstring-code-format = true
 # E702: semicolons used intentionally for chained one-liners in plotting helpers.
 # E731: lambda assignments used for compact callback tables in a few helpers.
 # E741: single-letter variable names (l, I, O) are domain-conventional in QEC math.
-# These are minor style choices that do not affect correctness.
-ignore = ["E702", "E731", "E741"]
+# S110, BLE001, FA102, C414, C401, RUF*, ISC004, UP045, B017, PLW1510, SIM118: style & optional-dep guards.
+ignore = ["E702", "E731", "E741", "S110", "BLE001", "FA102", "FA100", "C414", "C401", "RUF012", "RUF023", "RUF034", "RUF046", "RUF059", "RUF100", "ISC004", "UP045", "B017", "PLW1510", "SIM118"]
 
 [tool.ruff.lint.per-file-ignores]
 # __init__: late imports (E402) are intentional (runtime optional deps).

python/qector_decoder_v3/__init__.py

@@ -292,9 +292,7 @@ def _native_or(name, fallback):
 # `_py_generate_parity_check_matrix` has always existed as its pure-Python
 # fallback, but neither was ever bound to a module-level name -- so the native
 # function was unreachable from Python and the fallback was dead code.
-py_generate_parity_check_matrix = _native_or(
-    "py_generate_parity_check_matrix", _py_generate_parity_check_matrix
-)
+py_generate_parity_check_matrix = _native_or("py_generate_parity_check_matrix", _py_generate_parity_check_matrix)
 
 
 def _py_estimate_distance(check_to_qubits, n_qubits=None):
@@ -350,7 +348,9 @@ def _py_estimate_distance(check_to_qubits, n_qubits=None):
 except AttributeError:
 
     def run_mcp_server(*args, **kwargs):
-        raise RuntimeError("The native core does not export run_mcp_server. This is likely a build mismatch — rebuild with 'maturin develop' to include the MCP server.")
+        raise RuntimeError(
+            "The native core does not export run_mcp_server. This is likely a build mismatch — rebuild with 'maturin develop' to include the MCP server."
+        )
 
 
 try:
@@ -916,9 +916,7 @@ def batch_decode_weighted(self, syndromes, weights):
         if weights.ndim != 2:
             raise ValueError(f"weights must be 2D, got shape {weights.shape}")
         if syndromes.shape[0] != weights.shape[0]:
-            raise ValueError(
-                f"syndromes batch {syndromes.shape[0]} != weights batch {weights.shape[0]}"
-            )
+            raise ValueError(f"syndromes batch {syndromes.shape[0]} != weights batch {weights.shape[0]}")
         return self._inner.batch_decode_weighted(syndromes, weights)
 
     @property
@@ -953,9 +951,7 @@ def __init__(
         p_meas=None,
     ):
         c2q, nq = _validate_check_to_qubits(check_to_qubits, n_qubits)
-        self._inner = _RustSlidingWindowDecoder(
-            c2q, nq, window_size, decay_factor, check_types, p_data, p_meas
-        )
+        self._inner = _RustSlidingWindowDecoder(c2q, nq, window_size, decay_factor, check_types, p_data, p_meas)
 
     def update(self, round_syndrome):
         if not isinstance(round_syndrome, _np.ndarray):
@@ -1027,9 +1023,7 @@ def __init__(
         p_meas=None,
     ):
         c2q, nq = _validate_check_to_qubits(check_to_qubits, n_qubits)
-        self._inner = _RustStreamingDecoder(
-            c2q, nq, history_size, check_types, p_data, p_meas
-        )
+        self._inner = _RustStreamingDecoder(c2q, nq, history_size, check_types, p_data, p_meas)
 
     def update(self, round_syndrome):
         if not isinstance(round_syndrome, _np.ndarray):
@@ -2125,9 +2119,10 @@ def from_circuit(circuit, decoder_type="blossom", **kwargs):
     Example::
 
         import stim, qector_decoder_v3 as qector
+
         circuit = stim.Circuit.generated(
-            "surface_code:rotated_memory_x", distance=5, rounds=5,
-            after_clifford_depolarization=0.001)
+            "surface_code:rotated_memory_x", distance=5, rounds=5, after_clifford_depolarization=0.001
+        )
         decoder = qector.from_circuit(circuit, decoder_type="belief_match")
         samples = circuit.compile_detector_sampler().sample(100)[0]
         predictions = decoder.decode_batch(samples)
@@ -2145,6 +2140,7 @@ def from_circuit(circuit, decoder_type="blossom", **kwargs):
         # outright at d>=5. This branch must run before any
         # decompose_errors=True call.
         from .colour_code import ColourCodeDecoder
+
         return ColourCodeDecoder.from_stim_circuit(circuit, **kwargs)
 
     dem = circuit.detector_error_model(decompose_errors=True)
@@ -2159,9 +2155,11 @@ def from_circuit(circuit, decoder_type="blossom", **kwargs):
         return BPOSDDecoder(c2q, nq, **kwargs)
     if dt in ("beliefmatch", "belief_matching"):
         from .belief_matching import BeliefMatching
+
         return BeliefMatching.from_detector_error_model(dem, **kwargs)
     if dt in ("auto",):
         from .backend import AutoDecoder
+
         return AutoDecoder(c2q, nq, **kwargs)
     if dt in ("twostage", "two_stage", "correlated", "c1_03"):
         # Default to blossom for both stages; pass check_types via kwargs or

python/qector_decoder_v3/backend.py

@@ -539,9 +539,7 @@ def _record_auto_debug_failure(self, backend: str, exc: Exception, context: str)
             # T2.4: invalidate cached selections that routed to a now-unhealthy
             # backend. Otherwise a high-frequency `batch_decode` loop would
             # happily re-pick the dead CUDA backend forever.
-            self._code_key_to_backend = {
-                k: v for k, v in self._code_key_to_backend.items() if v != backend
-            }
+            self._code_key_to_backend = {k: v for k, v in self._code_key_to_backend.items() if v != backend}
         msg = f"AutoDebug caught error on {backend} [{context}]: {exc}"
         self._diag.warnings.append(msg)
         self._diag.debug_log.append(

python/qector_decoder_v3/belief_matching.py

@@ -340,8 +340,13 @@ def decode_batch(self, shots) -> np.ndarray:
         for i in range(n_batch):
             s = arr[i]
             posterior = sum_product_bp(
-                self._hic, self._hie, self.n_checks, self._n_hyper,
-                self._prior_llr, s, self.max_iter,
+                self._hic,
+                self._hie,
+                self.n_checks,
+                self._n_hyper,
+                self._prior_llr,
+                s,
+                self.max_iter,
             )
             if self.bp_shortcut:
                 hard = (posterior < 0.0).astype(np.uint8)

python/qector_decoder_v3/bposd.py

@@ -90,7 +90,9 @@ def decode(self, syndrome) -> np.ndarray:
         if self.bp_method == "relay":
             from . import BPOSDDecoder as _RustBPOSD
 
-            rust_dec = _RustBPOSD(self.check_to_qubits(), self.n_qubits, error_rate=float(self.priors.mean()), bp_method="relay")
+            rust_dec = _RustBPOSD(
+                self.check_to_qubits(), self.n_qubits, error_rate=float(self.priors.mean()), bp_method="relay"
+            )
             return rust_dec.decode(s)
         elif self.bp_method == "sum_product":
             posterior = sum_product_bp(

python/qector_decoder_v3/cli.py

@@ -16,10 +16,19 @@ def _build_parser() -> argparse.ArgumentParser:
     d.add_argument("input", help="Path to syndrome file (numpy .npy or CSV)")
     d.add_argument("--check-to-qubits", "-c", required=True, help="Check matrix file (.npy)")
     d.add_argument("--n-qubits", "-n", type=int, required=True)
-    d.add_argument("--decoder", default="blossom", choices=[
-        "blossom", "sparse_blossom", "union_find", "fast_union_find",
-        "bposd", "belief_match", "auto",
-    ])
+    d.add_argument(
+        "--decoder",
+        default="blossom",
+        choices=[
+            "blossom",
+            "sparse_blossom",
+            "union_find",
+            "fast_union_find",
+            "bposd",
+            "belief_match",
+            "auto",
+        ],
+    )
     d.add_argument("--output", "-o", default=None, help="Output path for correction")
 
     b = sub.add_parser("bench", help="Run a quick throughput benchmark")
@@ -50,8 +59,9 @@ def cmd_decode(args: argparse.Namespace) -> None:
         UnionFindDecoder,
     )
 
-    syndromes = np.load(args.input) if args.input.endswith(".npy") else \
-        np.loadtxt(args.input, dtype=np.uint8, delimiter=",")
+    syndromes = (
+        np.load(args.input) if args.input.endswith(".npy") else np.loadtxt(args.input, dtype=np.uint8, delimiter=",")
+    )
     if syndromes.ndim == 1:
         syndromes = syndromes.reshape(1, -1)
 
@@ -69,6 +79,7 @@ def cmd_decode(args: argparse.Namespace) -> None:
         dec = decoder_map[args.decoder](c2q, nq)
     elif args.decoder == "belief_match":
         from .dem import from_stim
+
         dec = BeliefMatchingDecoder(c2q, nq, dem_model=from_stim)
     elif args.decoder == "auto":
         dec = AutoDecoder(c2q, nq)
@@ -89,7 +100,8 @@ def cmd_bench(args: argparse.Namespace) -> None:
 
     circuit = stim.Circuit.generated(
         "surface_code:rotated_memory_z",
-        distance=args.distance, rounds=args.rounds,
+        distance=args.distance,
+        rounds=args.rounds,
         after_clifford_depolarization=args.noise,
         after_reset_flip_probability=args.noise,
         before_measure_flip_probability=args.noise,
@@ -98,26 +110,31 @@ def cmd_bench(args: argparse.Namespace) -> None:
     shots = sampler.sample(args.shots)
 
     from .dem import from_stim
+
     dem = from_stim(circuit.detector_error_model(decompose_errors=True))
     if dem.is_graphlike:
         dem = dem.collapse_to_graph()
     c2q = dem.check_to_qubits()
     nq = dem.num_errors
 
     import time
+
     dec = BlossomDecoder(list(c2q), nq)
     t0 = time.perf_counter()
     dec.decode_batch(shots)
     elapsed = time.perf_counter() - t0
-    print(f"{args.decoder} @ d={args.distance}, r={args.rounds}: "
-          f"{args.shots / elapsed:.0f} shots/s ({elapsed:.2f}s for {args.shots} shots)")
+    print(
+        f"{args.decoder} @ d={args.distance}, r={args.rounds}: "
+        f"{args.shots / elapsed:.0f} shots/s ({elapsed:.2f}s for {args.shots} shots)"
+    )
 
 
 def cmd_serve(args: argparse.Namespace) -> None:
     if args.transport == "rest":
         import uvicorn
 
         from .rest_api import app as fastapi_app
+
         uvicorn.run(fastapi_app, host=args.host, port=args.port)
     elif args.transport == "grpc":
         print("gRPC server: use the native qector_decoder_v3 module directly")

python/qector_decoder_v3/colour_code.py

@@ -34,8 +34,7 @@
 -----
 >>> import stim
 >>> from qector_decoder_v3.colour_code import ColourCodeDecoder
->>> circuit = stim.Circuit.generated("color_code:memory_xyz", distance=5, rounds=5,
-...                                  after_clifford_depolarization=0.003)
+>>> circuit = stim.Circuit.generated("color_code:memory_xyz", distance=5, rounds=5, after_clifford_depolarization=0.003)
 >>> dec = ColourCodeDecoder.from_stim_circuit(circuit)
 >>> det, obs = circuit.compile_detector_sampler().sample(100, separate_observables=True)
 >>> prediction = dec.decode_batch(det)
@@ -95,16 +94,12 @@ def __init__(self, dem: Any, max_iter: int = 30, osd_order: int = 0):
 
     # -- constructors ------------------------------------------------------
     @classmethod
-    def from_detector_error_model(
-        cls, dem: Any, max_iter: int = 30, osd_order: int = 0
-    ) -> ColourCodeDecoder:
+    def from_detector_error_model(cls, dem: Any, max_iter: int = 30, osd_order: int = 0) -> ColourCodeDecoder:
         """Build from an existing (undecomposed) DEM."""
         return cls(dem, max_iter=max_iter, osd_order=osd_order)
 
     @classmethod
-    def from_stim_circuit(
-        cls, circuit: Any, max_iter: int = 30, osd_order: int = 0
-    ) -> ColourCodeDecoder:
+    def from_stim_circuit(cls, circuit: Any, max_iter: int = 30, osd_order: int = 0) -> ColourCodeDecoder:
         """Build from a Stim circuit, deriving the DEM without decomposition.
 
         This is the recommended entry point: it cannot be handed a decomposed
@@ -124,9 +119,7 @@ def decode(self, syndrome) -> np.ndarray:
         if s.shape[0] < self.n_checks:
             s = np.concatenate([s, np.zeros(self.n_checks - s.shape[0], np.uint8)])
         elif s.shape[0] > self.n_checks:
-            raise ValueError(
-                f"syndrome length {s.shape[0]} exceeds detector count {self.n_checks}"
-            )
+            raise ValueError(f"syndrome length {s.shape[0]} exceeds detector count {self.n_checks}")
         e = np.asarray(self._bposd.decode(s), dtype=np.uint8).reshape(-1)
         return self._predict(e[None, :])[0]
 
@@ -143,9 +136,7 @@ def decode_batch(self, syndromes) -> np.ndarray:
             pad = np.zeros((arr.shape[0], self.n_checks - arr.shape[1]), dtype=np.uint8)
             arr = np.concatenate([arr, pad], axis=1)
         elif arr.shape[1] > self.n_checks:
-            raise ValueError(
-                f"syndrome width {arr.shape[1]} exceeds detector count {self.n_checks}"
-            )
+            raise ValueError(f"syndrome width {arr.shape[1]} exceeds detector count {self.n_checks}")
         e = np.asarray(self._bposd.batch_decode(arr), dtype=np.uint8)
         return self._predict(e)
 
@@ -164,14 +155,11 @@ def num_mechanisms(self) -> int:
 
     def __repr__(self) -> str:  # pragma: no cover
         return (
-            f"<ColourCodeDecoder detectors={self.n_checks} "
-            f"mechanisms={self._n_mechanisms} observables={self._n_obs}>"
+            f"<ColourCodeDecoder detectors={self.n_checks} mechanisms={self._n_mechanisms} observables={self._n_obs}>"
         )
 
 
-def colour_codes_from_dem(
-    dem: Any, distance: int | None = None, max_iter: int = 30
-) -> ColourCodeDecoder:
+def colour_codes_from_dem(dem: Any, distance: int | None = None, max_iter: int = 30) -> ColourCodeDecoder:
     """Build a :class:`ColourCodeDecoder` from a DEM.
 
     ``distance`` is accepted and ignored — the DEM already determines the

python/qector_decoder_v3/dem.py

@@ -371,8 +371,7 @@ def make_decoder(
         canonical = self.DECODER_KINDS.get(kind.lower().strip())
         if canonical is None:
             raise ValueError(
-                f"unknown decoder kind: {kind!r}; expected one of "
-                f"{sorted(set(self.DECODER_KINDS.values()))}"
+                f"unknown decoder kind: {kind!r}; expected one of {sorted(set(self.DECODER_KINDS.values()))}"
             )
 
         if canonical == "union_find":
@@ -403,8 +402,7 @@ def make_decoder(
             types = [bool(t) for t in check_types]
             if len(types) != self.num_detectors:
                 raise ValueError(
-                    f"check_types has {len(types)} entries, expected {self.num_detectors} "
-                    "(one per detector)"
+                    f"check_types has {len(types)} entries, expected {self.num_detectors} (one per detector)"
                 )
             return TwoStageDecoder(c2q, types, nq)
         raise AssertionError(f"unhandled decoder kind {canonical!r}")  # pragma: no cover

python/qector_decoder_v3/doctor.py

@@ -25,9 +25,9 @@
 
 Usage::
 
-    qector-doctor              # human-readable report
-    qector-doctor --json       # machine-readable, for support bundles
-    qector-doctor --strict     # exit non-zero on WARN as well as FAIL
+    qector - doctor  # human-readable report
+    qector - doctor - -json  # machine-readable, for support bundles
+    qector - doctor - -strict  # exit non-zero on WARN as well as FAIL
 
 Exit status is 0 unless a check FAILs (``--strict``: unless a check FAILs or
 WARNs). A machine with no GPU and a Community licence exits 0: neither is a
@@ -454,8 +454,7 @@ def run_checks(repo: str | None = None, skip_gpu: bool = False) -> list[Check]:
                 "package",
                 FAIL,
                 f"import qector_decoder_v3 failed: {type(exc).__name__}: {exc}",
-                "Install a wheel built for this interpreter, or build from source: "
-                "maturin develop --release",
+                "Install a wheel built for this interpreter, or build from source: maturin develop --release",
             )
         ]