Fix 11 ruff errors: unused vars, missing Optional import, f-string, variable name typo

Author: qectorlab

python/qector_decoder_v3/__init__.py

@@ -40,6 +40,7 @@ def _load_native_module():
 try:
     run_mcp_server = _native_module.run_mcp_server
 except AttributeError:
+
     def run_mcp_server(*args, **kwargs):
         raise RuntimeError("run_mcp_server requires the 'grpc' feature (maturin develop --features full)")
 
@@ -863,6 +864,7 @@ def __init__(self, node_feat_dim=None, edge_feat_dim=None, hidden_size=16, n_lay
         # Load PyTorch implementation if torch is installed.
         try:
             from .torch_predecoder import TorchGNNPredecoder
+
             self._torch_model = TorchGNNPredecoder(nfd, efd, hidden_size, n_layers)
             self._torch_model.double()  # Cast to float64 to match Rust f64.
             self.use_torch = True
@@ -873,6 +875,7 @@ def _sync_to_rust(self):
         if self.use_torch:
             import tempfile
             import os
+
             with tempfile.NamedTemporaryFile(suffix=".safetensors", delete=False) as tmp:
                 tmp_path = tmp.name
             try:
@@ -886,6 +889,7 @@ def _sync_to_torch(self):
         if self.use_torch:
             import tempfile
             import os
+
             with tempfile.NamedTemporaryFile(suffix=".safetensors", delete=False) as tmp:
                 tmp_path = tmp.name
             try:
@@ -936,6 +940,7 @@ def forward(self, graph):
         """Predict adjusted edge weights for a DetectorGraph."""
         if self.use_torch:
             import torch
+
             self._torch_model.eval()
             with torch.no_grad():
                 node_feats = torch.tensor(graph.node_features, dtype=torch.float64)
@@ -954,17 +959,18 @@ def train(self, graphs, targets, n_epochs):
                 raise ImportError("PyTorch is required to train GNNPredecoder")
             import torch
             import torch.optim as optim
+
             optimizer = optim.SGD(self._torch_model.parameters(), lr=self.learning_rate, weight_decay=self.l2_lambda)
             self._torch_model.train()
-            
+
             for epoch in range(n_epochs):
                 for graph, target in zip(graphs, targets):
                     node_feats = torch.tensor(graph.node_features, dtype=torch.float64)
                     edge_feats = torch.tensor(graph.edge_features, dtype=torch.float64)
                     edge_src = torch.tensor(graph.edge_src, dtype=torch.long)
                     edge_dst = torch.tensor(graph.edge_dst, dtype=torch.long)
                     target_t = torch.tensor(target, dtype=torch.float64)
-                    
+
                     optimizer.zero_grad()
                     preds = self._torch_model(node_feats, edge_feats, edge_src, edge_dst)
                     loss = F.mse_loss(preds, target_t)
@@ -979,35 +985,35 @@ def predict_with_node_probs(self, graph):
         """Predict edge weights and node error probabilities."""
         if self.use_torch:
             import torch
+
             self._torch_model.eval()
             with torch.no_grad():
                 node_feats = torch.tensor(graph.node_features, dtype=torch.float64)
                 edge_feats = torch.tensor(graph.edge_features, dtype=torch.float64)
                 edge_src = torch.tensor(graph.edge_src, dtype=torch.long)
                 edge_dst = torch.tensor(graph.edge_dst, dtype=torch.long)
-                
+
                 weights = self._torch_model(node_feats, edge_feats, edge_src, edge_dst)
-                
+
                 N = node_feats.size(0)
                 node_probs = torch.zeros(N, dtype=torch.float64)
                 node_counts = torch.zeros(N, dtype=torch.float64)
-                
+
                 node_probs.scatter_add_(0, edge_src, weights)
                 node_probs.scatter_add_(0, edge_dst, weights)
-                
+
                 node_counts.scatter_add_(0, edge_src, torch.ones_like(edge_src, dtype=torch.float64))
                 node_counts.scatter_add_(0, edge_dst, torch.ones_like(edge_dst, dtype=torch.float64))
-                
+
                 mask = node_counts > 0
                 node_probs[mask] /= node_counts[mask]
                 node_probs = 1.0 / (1.0 + (-node_probs + 1.0).exp())
-                
+
                 return weights.cpu().numpy().tolist(), node_probs.cpu().numpy().tolist()
         else:
             return self._inner.predict_with_node_probs(graph._inner if isinstance(graph, DetectorGraph) else graph)
 
 
-
 class DetectorGraph:
     """Detector graph used by the GNN and hybrid decoder paths."""
 

python/qector_decoder_v3/belief_matching.py

@@ -211,8 +211,6 @@ def from_numpy_h(cls, H, error_rate: float = 0.05, max_iter: int = 20) -> "Belie
         hyper_ids = {}
         edge_ids = {}
         hyper_to_edges = {}
-        hyper_obs_dict = {}
-        edge_obs_dict = {}
         priors = {}
 
         for q in range(nQ):

python/qector_decoder_v3/bposd.py

@@ -22,7 +22,7 @@
 
 from __future__ import annotations
 
-from typing import Any, List, Tuple, cast
+from typing import Any, List, Optional, Tuple, cast
 
 import numpy as np
 
@@ -82,6 +82,7 @@ def decode(self, syndrome) -> np.ndarray:
 
         if self.max_latency_ms is not None:
             import time as _time
+
             t_start = _time.perf_counter()
             max_seconds = self.max_latency_ms / 1000.0
 
@@ -110,6 +111,7 @@ def decode(self, syndrome) -> np.ndarray:
 
         if self.max_latency_ms is not None and (_time.perf_counter() - t_start) > max_seconds:
             from . import UnionFindDecoder
+
             checks = [sorted(int(c) for c in np.nonzero(self.H[r])[0]) for r in range(self.n_checks)]
             uf = UnionFindDecoder(checks, self.n_qubits)
             return np.asarray(uf.decode(s), dtype=np.uint8).reshape(-1)
@@ -226,9 +228,7 @@ def _osd(self, s: np.ndarray, posterior: np.ndarray) -> np.ndarray:
 # ---------------------------------------------------------------------------
 # GF(2) ordered-statistics solve
 # ---------------------------------------------------------------------------
-def _gf2_osd_solve(
-    H: np.ndarray, s: np.ndarray, order: np.ndarray, hard: np.ndarray
-) -> Tuple[np.ndarray, np.ndarray]:
+def _gf2_osd_solve(H: np.ndarray, s: np.ndarray, order: np.ndarray, hard: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
     """OSD-0 solve. ``order`` lists columns least-reliable first; the first
     rank(H) independent of them form the basis, the rest (free) take their BP hard
     decision, and the basis is solved so ``H x == s (mod 2)``.

python/qector_decoder_v3/decode_mmap.py

@@ -12,12 +12,10 @@
 >>> checks, nq = generate_repetition_code_checks(distance=5)
 >>> n_shots = 10000
 >>> # Create a memory-mapped syndrome file
->>> syn = np.memmap("syndromes.bin", dtype=np.uint8, mode="w+",
-...                  shape=(n_shots, len(checks)))
+>>> syn = np.memmap("syndromes.bin", dtype=np.uint8, mode="w+", shape=(n_shots, len(checks)))
 >>> syn[:] = (np.random.random((n_shots, len(checks))) < 0.1).astype(np.uint8)
 >>> syn.flush()
->>> decode_mmap("syndromes.bin", "corrections.bin",
-...             checks, nq, batch_size=1024, n_shots=n_shots)
+>>> decode_mmap("syndromes.bin", "corrections.bin", checks, nq, batch_size=1024, n_shots=n_shots)
 """
 
 from __future__ import annotations
@@ -66,14 +64,14 @@ def decode_mmap(
         n_shots = file_size // bytes_per_shot
     if file_size < n_shots * bytes_per_shot:
         raise ValueError(
-            f"syndrome file too small: {file_size} bytes, "
-            f"need {n_shots * bytes_per_shot} for {n_shots} shots"
+            f"syndrome file too small: {file_size} bytes, need {n_shots * bytes_per_shot} for {n_shots} shots"
         )
 
     syndromes = _np.memmap(syndrome_path, dtype=dtype, mode="r", shape=(n_shots, n_checks))
     output = _np.memmap(output_path, dtype=_np.uint8, mode="w+", shape=(n_shots, n_qubits))
 
     from . import CPUBatchDecoder, UnionFindDecoder
+
     if decoder_type == "cpu_batch":
         decoder = CPUBatchDecoder(c2q, n_qubits)
         batch_fn = decoder.batch_decode

python/qector_decoder_v3/decoder_cache.py

@@ -19,7 +19,7 @@
 import hashlib as _hashlib
 import json as _json
 import os as _os
-from typing import Any, Dict, Optional, Tuple
+from typing import Optional, Tuple
 
 import numpy as _np
 
@@ -29,19 +29,24 @@
 def _normalize_decoder_name(name: str) -> str:
     """Normalize decoder name aliases to canonical form."""
     mapping = {
-        'uf': 'union_find', 'unionfind': 'union_find',
-        'fast_uf': 'fast_union_find', 'fastuf': 'fast_union_find', 'fastunionfind': 'fast_union_find',
-        'sparse': 'sparse_blossom',
-        'batch': 'cpu_batch', 'cpu': 'cpu_batch',
+        "uf": "union_find",
+        "unionfind": "union_find",
+        "fast_uf": "fast_union_find",
+        "fastuf": "fast_union_find",
+        "fastunionfind": "fast_union_find",
+        "sparse": "sparse_blossom",
+        "batch": "cpu_batch",
+        "cpu": "cpu_batch",
     }
-    return mapping.get(name.lower().replace('-', '_').replace(' ', '_'), name)
+    return mapping.get(name.lower().replace("-", "_").replace(" ", "_"), name)
 
 
 @_functools.lru_cache(maxsize=256)
 def _build_decoder(checks_tuple: Tuple[Tuple[int, ...]], n_qubits: int, decoder_type: str):
     """Internal LRU-cached decoder factory."""
     checks = [list(c) for c in checks_tuple]
     from . import BlossomDecoder, CPUBatchDecoder, FastUnionFindDecoder, SparseBlossomDecoder, UnionFindDecoder
+
     builders = {
         "union_find": lambda: UnionFindDecoder(checks, n_qubits),
         "fast_union_find": lambda: FastUnionFindDecoder(checks, n_qubits),
@@ -91,6 +96,7 @@ def get_decoder_pool(
         A :class:`DecoderPool` instance.
     """
     from .decoder_pool import DecoderPool
+
     return DecoderPool(
         [list(c) for c in checks_tuple],
         n_qubits=n_qubits,

python/qector_decoder_v3/decoder_pool.py

@@ -29,7 +29,7 @@
 import multiprocessing as _mp
 import os
 import platform as _platform
-from typing import Any, List, Optional
+from typing import Optional
 
 import numpy as np
 
@@ -45,6 +45,7 @@ def _worker_init(checks_tuple, n_qubits, decoder_type):
     global _WORKER_DECODER
     checks = [list(c) for c in checks_tuple]
     from . import BlossomDecoder, CPUBatchDecoder, FastUnionFindDecoder, SparseBlossomDecoder, UnionFindDecoder
+
     builders = {
         "union_find": lambda: UnionFindDecoder(checks, n_qubits),
         "fast_union_find": lambda: FastUnionFindDecoder(checks, n_qubits),
@@ -74,10 +75,16 @@ def _worker_decode(chunk_and_idx):
 
 
 _DECODER_BUILDERS = {
-    "union_find": lambda c2q, nq: __import__("qector_decoder_v3", fromlist=["UnionFindDecoder"]).UnionFindDecoder(c2q, nq),
-    "fast_union_find": lambda c2q, nq: __import__("qector_decoder_v3", fromlist=["FastUnionFindDecoder"]).FastUnionFindDecoder(c2q, nq),
+    "union_find": lambda c2q, nq: __import__("qector_decoder_v3", fromlist=["UnionFindDecoder"]).UnionFindDecoder(
+        c2q, nq
+    ),
+    "fast_union_find": lambda c2q, nq: __import__(
+        "qector_decoder_v3", fromlist=["FastUnionFindDecoder"]
+    ).FastUnionFindDecoder(c2q, nq),
     "blossom": lambda c2q, nq: __import__("qector_decoder_v3", fromlist=["BlossomDecoder"]).BlossomDecoder(c2q, nq),
-    "sparse_blossom": lambda c2q, nq: __import__("qector_decoder_v3", fromlist=["SparseBlossomDecoder"]).SparseBlossomDecoder(c2q, nq),
+    "sparse_blossom": lambda c2q, nq: __import__(
+        "qector_decoder_v3", fromlist=["SparseBlossomDecoder"]
+    ).SparseBlossomDecoder(c2q, nq),
     "cpu_batch": lambda c2q, nq: __import__("qector_decoder_v3", fromlist=["CPUBatchDecoder"]).CPUBatchDecoder(c2q, nq),
 }
 
@@ -150,7 +157,7 @@ def decode(self, syndromes) -> np.ndarray:
 
         nw = min(self._n_workers, n)
         chunk_size = (n + nw - 1) // nw
-        chunks = [(arr[i:i + chunk_size], i // chunk_size) for i in range(0, n, chunk_size)]
+        chunks = [(arr[i : i + chunk_size], i // chunk_size) for i in range(0, n, chunk_size)]
 
         results = [None] * len(chunks)
         for idx, result in self._pool.imap_unordered(_worker_decode, chunks):

python/qector_decoder_v3/predecoder.py

@@ -81,7 +81,7 @@ def __init__(self, check_to_qubits, n_qubits=None, backend: str = "blossom"):
         if backend in _backend_map:
             backend = _backend_map[backend]
         if backend not in _valid_backends:
-            raise ValueError(f"backend must be one of ['blossom', 'union_find', 'sparse_blossom']")
+            raise ValueError("backend must be one of ['blossom', 'union_find', 'sparse_blossom']")
         self.backend = backend
         residual: Union[BlossomDecoder, UnionFindDecoder, SparseBlossomDecoder]
         if backend == "blossom":