Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 4 additions & 5 deletions ACID/src/acid/__init__.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,9 @@
__all__ = [
"BBMidCycle",
"Connection",
"Device",
"Embedding",
"GroupRing",
"Monomial",
"Polynomial",
"Embedding",
"Device",
"Connection",
"BBMidCycle",
]

1 change: 0 additions & 1 deletion ACID/src/acid/analysis/__init__.py
Original file line number Diff line number Diff line change
@@ -1,2 +1 @@
"""Analysis helpers for reports and schedule introspection."""

32 changes: 16 additions & 16 deletions ACID/src/acid/analysis/gauge_fix_nkd.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@
import time
from dataclasses import dataclass
from itertools import product
from typing import Dict, List, Tuple, Optional

from acid.defects.defective_code import DefectiveCode
from acid.pauli import PauliString, StabiliserCode
Expand All @@ -15,15 +14,17 @@ class GaugeFixNKDResult:
combos_total: int
eval_count: int
timed_out: bool
best_choice: Optional[List[int]]
best_n: Optional[int]
best_k: Optional[int]
best_d: Optional[int]
best_choice: list[int] | None
best_n: int | None
best_k: int | None
best_d: int | None


def _basis_rows_to_Hx_Hz(rows: List[PauliString], n: int) -> Tuple[List[List[int]], List[List[int]]]:
Hx: List[List[int]] = []
Hz: List[List[int]] = []
def _basis_rows_to_Hx_Hz(
rows: list[PauliString], n: int
) -> tuple[list[list[int]], list[list[int]]]:
Hx: list[list[int]] = []
Hz: list[list[int]] = []
for p in rows:
if any(b & 1 for b in p.Z):
# Treat as Z row
Expand All @@ -36,7 +37,7 @@ def _basis_rows_to_Hx_Hz(rows: List[PauliString], n: int) -> Tuple[List[List[int

def _build_code_for_choice(
dcode: DefectiveCode,
choice_bits: List[int],
choice_bits: list[int],
) -> StabiliserCode:
"""
Build a CSS StabiliserCode by gauge-fixing:
Expand All @@ -48,7 +49,7 @@ def _build_code_for_choice(
# Untouched + products
base = dcode.midcycle_untouched_stabilisers()
prod = dcode.midcycle_product_stabilisers()
rows: List[PauliString] = []
rows: list[PauliString] = []
rows.extend(base.rows)
if prod is not None:
rows.extend(prod.rows)
Expand All @@ -62,7 +63,7 @@ def _build_code_for_choice(
rows.append(Gz[i])
# Convert to Hx/Hz
Hx_rows, Hz_rows = _basis_rows_to_Hx_Hz(rows, n)
labels: List[str] = [f"r{i}" for i in range(len(Hx_rows) + len(Hz_rows))]
labels: list[str] = [f"r{i}" for i in range(len(Hx_rows) + len(Hz_rows))]
return StabiliserCode(num_qubits=n, row_labels=labels, Hx=Hx_rows, Hz=Hz_rows)


Expand All @@ -86,10 +87,10 @@ def gauge_fixed_nkd(
g = min(len(Gx), len(Gz))
combos_total = 1 << g
start = time.monotonic()
best_d: Optional[int] = None
best_n: Optional[int] = None
best_k: Optional[int] = None
best_choice: Optional[List[int]] = None
best_d: int | None = None
best_n: int | None = None
best_k: int | None = None
best_choice: list[int] | None = None
eval_count = 0
for bits in product([0, 1], repeat=g):
# Check global timeout
Expand Down Expand Up @@ -134,4 +135,3 @@ def gauge_fixed_nkd(
best_k=best_k,
best_d=best_d,
)

58 changes: 32 additions & 26 deletions ACID/src/acid/analysis/report.py
Original file line number Diff line number Diff line change
@@ -1,26 +1,25 @@
from __future__ import annotations

from pathlib import Path
from typing import Dict, List, Set

from acid.defects.defective_code import DefectiveCode
from acid.analysis.schedule import analyze_layers
from acid.defects.defective_code import DefectiveCode
from acid.defects.syndrome_extraction_circuit import SyndromeExtractionCircuit


def write_schedule_report(out_path: Path,
dcode: DefectiveCode,
Ls: List[int],
*,
solve_time: float = 60.0) -> None:
def write_schedule_report(
out_path: Path, dcode: DefectiveCode, Ls: list[int], *, solve_time: float = 60.0
) -> None:
"""
Generic report writer for any DefectiveCode + schedule lengths.
Includes stats, per-layer measured/in-process/completed, and product completions.
"""
lines: List[str] = []
lines.append(f"# Schedule Report\n")
lines: list[str] = []
lines.append("# Schedule Report\n")
stats = dcode.stats()
lines.append(f"- Quasis: {stats.get('num_quasi')} nontrivial={stats.get('num_nontrivial')} rank={stats.get('rank')}\n")
lines.append(
f"- Quasis: {stats.get('num_quasi')} nontrivial={stats.get('num_nontrivial')} rank={stats.get('rank')}\n"
)
products = dcode.products_list()
prod_members = {p.label: set(p.members) for p in products}
interesting = set(dcode.anticommutation_graph().nodes())
Expand All @@ -29,42 +28,49 @@ def write_schedule_report(out_path: Path,
try:
circuit = dcode.schedule(L, solve_time=solve_time)
layers = circuit.layers
result = analyze_layers(prod_members, layers, interesting_labels=interesting)
result = analyze_layers(
prod_members, layers, interesting_labels=interesting
)
# Per-layer table
lines.append("\n| Layer | Measured | In-process | Completed |\n|------:|----------|------------|-----------|\n")
for t, rec in enumerate(result['per_layer']):
m = ','.join(rec['measured']) if rec['measured'] else '-'
ip = ','.join(rec['in_process']) if rec['in_process'] else '-'
cp = ','.join(rec['completed']) if rec['completed'] else '-'
lines.append(
"\n| Layer | Measured | In-process | Completed |\n|------:|----------|------------|-----------|\n"
)
for t, rec in enumerate(result["per_layer"]):
m = ",".join(rec["measured"]) if rec["measured"] else "-"
ip = ",".join(rec["in_process"]) if rec["in_process"] else "-"
cp = ",".join(rec["completed"]) if rec["completed"] else "-"
lines.append(f"| {t} | {m} | {ip} | {cp} |\n")
# Product completions
lines.append("\nCompletions:\n")
for p, compl in sorted(result['product_completions'].items()):
for p, compl in sorted(result["product_completions"].items()):
lines.append(f"- {p}: {sorted(compl)}\n")
except Exception as e:
lines.append(f"- Solve failed: {e}\n")
out_path.parent.mkdir(parents=True, exist_ok=True)
out_path.write_text("\n".join(lines))

def write_schedule_report_from_circuit(out_path: Path,
dcode: DefectiveCode,
circuit: SyndromeExtractionCircuit) -> None:

def write_schedule_report_from_circuit(
out_path: Path, dcode: DefectiveCode, circuit: SyndromeExtractionCircuit
) -> None:
products = dcode.products_list()
prod_members = {p.label: set(p.members) for p in products}
interesting = set(dcode.anticommutation_graph().nodes())
lines = []
layers = circuit.layers
result = analyze_layers(prod_members, layers, interesting_labels=interesting)
# Per-layer table
lines.append("\n| Layer | Measured | In-process | Completed |\n|------:|----------|------------|-----------|\n")
for t, rec in enumerate(result['per_layer']):
m = ','.join(rec['measured']) if rec['measured'] else '-'
ip = ','.join(rec['in_process']) if rec['in_process'] else '-'
cp = ','.join(rec['completed']) if rec['completed'] else '-'
lines.append(
"\n| Layer | Measured | In-process | Completed |\n|------:|----------|------------|-----------|\n"
)
for t, rec in enumerate(result["per_layer"]):
m = ",".join(rec["measured"]) if rec["measured"] else "-"
ip = ",".join(rec["in_process"]) if rec["in_process"] else "-"
cp = ",".join(rec["completed"]) if rec["completed"] else "-"
lines.append(f"| {t} | {m} | {ip} | {cp} |\n")
# Product completions
lines.append("\nCompletions:\n")
for p, compl in sorted(result['product_completions'].items()):
for p, compl in sorted(result["product_completions"].items()):
lines.append(f"- {p}: {sorted(compl)}\n")
out_path.parent.mkdir(parents=True, exist_ok=True)
out_path.write_text("\n".join(lines))
40 changes: 22 additions & 18 deletions ACID/src/acid/analysis/schedule.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,14 @@
from __future__ import annotations

from typing import Dict, List, Set

from acid.scheduling.types import SyndromeExtractionLayer


def analyze_layers(product_members: Dict[str, Set[str]],
layers: List[SyndromeExtractionLayer],
*,
interesting_labels: Set[str] | None = None) -> Dict:
def analyze_layers(
product_members: dict[str, set[str]],
layers: list[SyndromeExtractionLayer],
*,
interesting_labels: set[str] | None = None,
) -> dict:
"""
Generic schedule analysis usable for any code and any schedule:
- per-layer measured labels (optionally filtered to 'interesting_labels')
Expand All @@ -27,17 +27,19 @@ def analyze_layers(product_members: Dict[str, Set[str]],
interesting_labels = set()

# Progress since last completion per product
progress: Dict[str, Set[str]] = {p: set() for p in product_members}
completions: Dict[str, List[int]] = {p: [] for p in product_members}
progress: dict[str, set[str]] = {p: set() for p in product_members}
completions: dict[str, list[int]] = {p: [] for p in product_members}

per_layer = []
for t, layer in enumerate(layers):
# actual measured set
all_measured = sorted(stab.label for stab in layer.chosen.keys())
all_measured = sorted(stab.label for stab in layer.chosen)
mset = set(all_measured)
# display-only measured (optional filter)
if interesting_labels:
measured = sorted([lab for lab in all_measured if lab in interesting_labels])
measured = sorted(
[lab for lab in all_measured if lab in interesting_labels]
)
else:
measured = []

Expand All @@ -47,21 +49,23 @@ def analyze_layers(product_members: Dict[str, Set[str]],
for p, members in product_members.items():
if mset & members:
in_process.append(p)
progress[p] |= (mset & members)
progress[p] |= mset & members
# Check completions at end of layer
for p, members in product_members.items():
if progress[p] and progress[p] >= members:
completions[p].append(t)
completed.append(p)
progress[p].clear()

per_layer.append({
'measured': measured,
'in_process': sorted(in_process),
'completed': sorted(completed),
})
per_layer.append(
{
"measured": measured,
"in_process": sorted(in_process),
"completed": sorted(completed),
}
)

return {
'per_layer': per_layer,
'product_completions': completions,
"per_layer": per_layer,
"product_completions": completions,
}
37 changes: 23 additions & 14 deletions ACID/src/acid/base_code.py
Original file line number Diff line number Diff line change
@@ -1,42 +1,51 @@
from __future__ import annotations

from dataclasses import dataclass
from typing import List, Optional, Tuple

import networkx as nx


@dataclass(frozen=True)
class StabiliserShape:
pauli_type: str # 'X' or 'Z'
connectivity_subgraph: nx.Graph # nodes 0..k-1
sec_cycle_length: int # nominal SEC cycle length to use for schedules later
qubit_map: List[int] # len = k; template index -> code qubit id
pauli_type: str # 'X' or 'Z'
connectivity_subgraph: nx.Graph # nodes 0..k-1
sec_cycle_length: int # nominal SEC cycle length to use for schedules later
qubit_map: list[int] # len = k; template index -> code qubit id
label: str
# Preference inputs (all optional)
preferred_roots: Optional[List[int]] = None # list of preferred local roots; if None, root not considered for preference
preferred_edges: Optional[dict[tuple[int, int], Optional[List[int]]]] = None # edge -> optional list of required timesteps
preferred_roots: list[int] | None = (
None # list of preferred local roots; if None, root not considered for preference
)
preferred_edges: dict[tuple[int, int], list[int] | None] | None = (
None # edge -> optional list of required timesteps
)
# Solver hints (optional and soft)
schedule_hint: Optional[List[List[tuple[int, int]]]] = None # pre-defined schedule used as a hint only
layer_hint: Optional[int] = None # layer index at which to hint this stabiliser schedule
schedule_hint: list[list[tuple[int, int]]] | None = (
None # pre-defined schedule used as a hint only
)
layer_hint: int | None = (
None # layer index at which to hint this stabiliser schedule
)
# Accepted but unused; maintained for builder compatibility
redundant_edges: Optional[set[tuple[int, int]]] = None
redundant_edges: set[tuple[int, int]] | None = None


@dataclass
class BaseCode:
num_qubits: int
connectivity_graph: nx.Graph
shapes: List[StabiliserShape]
shapes: list[StabiliserShape]
# Optional: labelled connection classes for visualisation (undirected)
connection_classes: Optional[List[Tuple[int, int, str]]] = None
connection_classes: list[tuple[int, int, str]] | None = None

def validate_local_connectivity(self) -> None:
G = self.connectivity_graph
for sh in self.shapes:
S = sh.connectivity_subgraph
qmap = sh.qubit_map
for (u, v) in S.edges():
a = qmap[u]; b = qmap[v]
for u, v in S.edges():
a = qmap[u]
b = qmap[v]
if not G.has_edge(a, b):
raise RuntimeError(
f"Connectivity mismatch for {sh.label}: edge {(u, v)} -> {(a, b)} not in device"
Expand Down
1 change: 0 additions & 1 deletion ACID/src/acid/codes/bb/__init__.py
Original file line number Diff line number Diff line change
@@ -1,2 +1 @@
"""Bivariate bicycle (BB/bb) code builders."""

Loading