Import from Quazim0t0/neural-cd-preserve; repoint refs to NeuralVerified

GF256.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:95a3bd56efcb21b046ccb4b400a0965d720bd4828e883cb543c584f2e9dca77d
+size 564882

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Dean Byrne (Quazim0t0)
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,71 @@
+---
+license: mit
+library_name: pytorch
+tags:
+  - verified-units
+  - reed-solomon
+  - archival
+  - preservation
+  - self-healing
+---
+
+# neural-cd-preserve — self-healing optical-disc archives
+
+**Repositories:** [GitHub](https://github.com/quzi93/neural-cd-preserve) · [🤗 HuggingFace](https://huggingface.co/NeuralVerified/neural-cd-preserve)
+
+Scan an optical disc into a self-healing `.pt` archive that **detects and repairs
+bit-rot** and reconstructs the disc even from a damaged copy. Built on the same
+neural-verified Reed-Solomon core as
+[neural-storage](https://huggingface.co/NeuralVerified/neural-storage): each unique
+chunk is RS-split into shards, every shard carries its own SHA-256, so silent
+corruption is detected, marked as an erasure, and repaired from survivors (as long
+as ≥ *k* of *n* shards remain).
+
+> **Honest by design:** RS self-healing protects the *archive* against future
+> bit-rot. It cannot recover disc sectors that were already unreadable at scan
+> time — those are recorded, and re-reads are merged. No entropy is beaten.
+
+## Use
+
+```bash
+pip install torch
+python step_cd.py                                   # full self-healing demo
+
+python cli.py archive \\.\D: mydisc.pt --label MY_CD   # drive or .iso
+python cli.py verify  mydisc.pt
+python cli.py heal    mydisc.pt
+python cli.py restore mydisc.pt out.iso
+```
+
+The demo injects 149 silent bit-flips; `verify` detects all, `heal` repairs all,
+`restore` returns bit-exact; corrupting beyond RS capacity is flagged **LOST**,
+never silently wrong.
+
+Weights: `GF256.pt` (verified GF(256) LOG/EXP).
+
+**Create your own verified unit** (template: `cdpreserve/gf256.py`): write the
+exact golden finite function → enumerate the domain (decompose big/linear ones
+into bit/byte slices, see `cdpreserve/rs.py`) → `common.train` → `common.verify`
+must be bit-exact on 100% of inputs → compose.
+
+## Citation
+
+```bibtex
+@misc{byrne2026neuralcdpreserve,
+  title  = {neural-cd-preserve: Self-Healing Optical-Disc Archival with Verified Erasure Coding},
+  author = {Byrne, Dean (Quazim0t0)},
+  year   = {2026},
+  howpublished = {\url{https://huggingface.co/NeuralVerified/neural-cd-preserve}}
+}
+```
+
+**Dean Byrne (Quazim0t0)** · 2026
+
+
+---
+
+<!-- neuralverified-relocation-note -->
+> **Now hosted by [NeuralVerified](https://huggingface.co/NeuralVerified).**
+>
+> This repo was moved into the NeuralVerified organization to help organize my profile.
+> Originally published at [`Quazim0t0/neural-cd-preserve`](https://huggingface.co/Quazim0t0/neural-cd-preserve).

cdpreserve/archive.py

@@ -0,0 +1,120 @@
+"""
+Self-healing .pt disc archive.
+
+A disc image is chunked (fixed size), duplicate chunks deduped, and each unique
+chunk Reed-Solomon split into n = k+m shards (verified GF(256) core). Every shard
+carries its own SHA-256, so SILENT corruption (bit-rot flipping bytes) is
+DETECTED, marked as an erasure, and REPAIRED from the surviving shards -- as long
+as at least k of the n shards per chunk are still intact.
+
+The whole thing serializes to a single .pt (torch.save):
+  { meta, manifest:[chunk_hash...], chunks:{ hash: {shards, shard_hashes, L, orig} } }
+
+Honesty: RS adds redundancy (archive is ~(k+m)/k x the deduped image). It repairs
+up to m damaged shards per chunk; beyond that a chunk is flagged as lost, never
+silently wrong. No entropy is beaten.
+"""
+from __future__ import annotations
+import hashlib
+import time
+import torch
+from . import rs
+
+CHUNK = 4096
+
+
+def _h(b: bytes) -> str:
+    return hashlib.sha256(b).hexdigest()
+
+
+def build(data: bytes, label="disc", k=4, m=2, chunk=CHUNK, bad_sectors=None) -> dict:
+    chunks = {}
+    manifest = []
+    for off in range(0, len(data), chunk):
+        ch = data[off:off + chunk]
+        h = _h(ch)
+        manifest.append(h)
+        if h in chunks:
+            continue
+        shards, L, orig = rs.encode(ch, k, m)
+        shards = [bytes(s) for s in shards]
+        chunks[h] = {"shards": shards, "shard_hashes": [_h(s) for s in shards],
+                     "L": L, "orig": orig, "k": k, "m": m}
+    return {
+        "meta": {"label": label, "size": len(data), "chunk": chunk, "k": k, "m": m,
+                 "bad_sectors": list(bad_sectors or []), "created": time.time(),
+                 "format": "neural-cd-preserve/1"},
+        "manifest": manifest, "chunks": chunks,
+    }
+
+
+def save(arc: dict, path: str):
+    torch.save(arc, path)
+
+
+def load(path: str) -> dict:
+    return torch.load(path, map_location="cpu", weights_only=False)
+
+
+def _good_shards(rec: dict) -> dict:
+    """{index: bytes} for shards whose hash still matches (uncorrupted)."""
+    out = {}
+    for i, s in enumerate(rec["shards"]):
+        if _h(s) == rec["shard_hashes"][i]:
+            out[i] = s
+    return out
+
+
+def verify(arc: dict) -> dict:
+    """Report health without modifying. -> {chunks, corrupted_shards, repairable,
+    lost} where lost chunks have < k intact shards (unrecoverable)."""
+    corrupted = repairable = lost = 0
+    for h, rec in arc["chunks"].items():
+        good = _good_shards(rec)
+        nbad = len(rec["shards"]) - len(good)
+        corrupted += nbad
+        if nbad and len(good) >= rec["k"]:
+            repairable += 1
+        if len(good) < rec["k"]:
+            lost += 1
+    return {"chunks": len(arc["chunks"]), "corrupted_shards": corrupted,
+            "repairable_chunks": repairable, "lost_chunks": lost,
+            "healthy": corrupted == 0 and lost == 0}
+
+
+def heal(arc: dict) -> int:
+    """Detect corrupted shards and regenerate them from survivors. Returns the
+    number of shards repaired. Chunks with < k intact shards cannot be healed."""
+    repaired = 0
+    for h, rec in arc["chunks"].items():
+        good = _good_shards(rec)
+        if len(good) == len(rec["shards"]):
+            continue
+        if len(good) < rec["k"]:
+            continue                                    # unrecoverable, leave as-is
+        ch = rs.decode(good, rec["k"], rec["m"], rec["L"], rec["orig"])
+        if _h(ch) != h:
+            continue                                    # decoded wrong -> don't trust
+        fresh, _, _ = rs.encode(ch, rec["k"], rec["m"])
+        for i in range(len(rec["shards"])):
+            if i not in good:
+                rec["shards"][i] = bytes(fresh[i])
+                rec["shard_hashes"][i] = _h(rec["shards"][i])
+                repaired += 1
+    return repaired
+
+
+def restore(arc: dict) -> bytes:
+    """Reconstruct the disc image, RS-correcting any detected corruption."""
+    out = bytearray()
+    for h in arc["manifest"]:
+        rec = arc["chunks"][h]
+        good = _good_shards(rec)
+        if len(good) < rec["k"]:
+            raise IOError(f"chunk {h[:8]}: only {len(good)} intact shards, "
+                          f"need {rec['k']} -- unrecoverable")
+        ch = rs.decode(good, rec["k"], rec["m"], rec["L"], rec["orig"])
+        if _h(ch) != h:
+            raise IOError(f"chunk {h[:8]}: hash mismatch after decode")
+        out += ch
+    return bytes(out[:arc["meta"]["size"]])

cdpreserve/common.py

@@ -0,0 +1,56 @@
+"""Shared helpers (same as the neural-DDR project): bit<->int, MLP, verify, train."""
+from __future__ import annotations
+import torch
+import torch.nn as nn
+
+DEV = "cuda" if torch.cuda.is_available() else "cpu"
+
+
+def bits_of(v: int, n: int) -> list[int]:
+    return [(v >> k) & 1 for k in range(n)]
+
+
+def int_of(bits) -> int:
+    return sum((1 << k) for k, b in enumerate(bits) if b > 0)
+
+
+def pm(bits) -> torch.Tensor:
+    return torch.tensor([1.0 if b else -1.0 for b in bits], dtype=torch.float32)
+
+
+def mlp(inp: int, out: int, h: int = 256, layers: int = 2) -> nn.Sequential:
+    mods = [nn.Linear(inp, h), nn.GELU()]
+    for _ in range(layers - 1):
+        mods += [nn.Linear(h, h), nn.GELU()]
+    mods += [nn.Linear(h, out)]
+    return nn.Sequential(*mods)
+
+
+@torch.no_grad()
+def verify(net, X, Ybits) -> tuple[int, int]:
+    net = net.to("cpu")
+    pred = (net(X) > 0).int()
+    ok = (pred == Ybits.int()).all(dim=1).sum().item()
+    return ok, X.shape[0]
+
+
+def train(net, X, Ybits, steps=8000, lr=2e-3, tag="", report=2000):
+    net = net.to(DEV)
+    Xd, Yd = X.to(DEV), Ybits.to(DEV)
+    opt = torch.optim.Adam(net.parameters(), lr=lr)
+    sch = torch.optim.lr_scheduler.CosineAnnealingLR(opt, T_max=steps)
+    lossfn = nn.BCEWithLogitsLoss()
+    for e in range(steps):
+        opt.zero_grad()
+        loss = lossfn(net(Xd), Yd)
+        loss.backward(); opt.step(); sch.step()
+        if e % report == 0 or e == steps - 1:
+            ok, tot = verify(net, X, Ybits); net.to(DEV)
+            print(f"    [{tag}] epoch {e:5d}  loss {loss.item():.2e}  verified {ok}/{tot}")
+            if ok == tot:
+                print(f"    [{tag}] -> N/N"); break
+    return net.to("cpu")
+
+
+def run8(net, v: int) -> int:
+    return int_of((net(pm(bits_of(v, 8)).unsqueeze(0))[0] > 0).int().tolist())

cdpreserve/gf256.py

@@ -0,0 +1,80 @@
+"""
+Step 1 keystone: verified GF(2^8) multiply -- the arithmetic core of Reed-Solomon
+erasure coding (Step 2).
+
+A direct 16-bit -> 8-bit neural multiply is hard to drive to N/N. So we decompose
+it the way real RS implementations do -- via log/exp tables:
+
+    a * b = EXP[ (LOG[a] + LOG[b]) mod 255 ]      (and 0 if a==0 or b==0)
+
+LOG and EXP are 256-entry lookups -> trivially N/N verifiable neural units. We
+train both, compose them into a full multiply, and verify the COMPOSED multiply
+against the golden field over ALL 65,536 (a,b) pairs.
+
+Field: GF(2^8), reduction polynomial 0x11D, generator 2 (standard for RS).
+"""
+from __future__ import annotations
+import torch
+from .common import mlp, train, verify, pm, bits_of, int_of, run8
+
+POLY = 0x11D
+
+
+def build_tables():
+    exp = [0] * 512
+    log = [0] * 256
+    x = 1
+    for i in range(255):
+        exp[i] = x
+        log[x] = i
+        x <<= 1
+        if x & 0x100:
+            x ^= POLY
+    for i in range(255, 512):
+        exp[i] = exp[i - 255]
+    return exp, log
+
+
+EXP, LOG = build_tables()
+
+
+def gf_mul(a: int, b: int) -> int:
+    if a == 0 or b == 0:
+        return 0
+    return EXP[LOG[a] + LOG[b]]
+
+
+# ---- neural LOG (inputs 1..255) and EXP (indices 0..254) ----
+def log_domain():
+    xs = [pm(bits_of(a, 8)) for a in range(1, 256)]
+    ys = [[float(b) for b in bits_of(LOG[a], 8)] for a in range(1, 256)]
+    return torch.stack(xs), torch.tensor(ys)
+
+
+def exp_domain():
+    xs = [pm(bits_of(i, 8)) for i in range(255)]
+    ys = [[float(b) for b in bits_of(EXP[i], 8)] for i in range(255)]
+    return torch.stack(xs), torch.tensor(ys)
+
+
+def gf_mul_neural(a: int, b: int, net_log, net_exp) -> int:
+    if a == 0 or b == 0:
+        return 0
+    s = (run8(net_log, a) + run8(net_log, b)) % 255
+    return run8(net_exp, s)
+
+
+def train_units():
+    print("  training LOG (8->8):")
+    net_log = train(mlp(8, 8, 256, 2), *log_domain(), steps=8000, tag="log")
+    print("  training EXP (8->8):")
+    net_exp = train(mlp(8, 8, 256, 2), *exp_domain(), steps=8000, tag="exp")
+    return net_log, net_exp
+
+
+def verify_mul(net_log, net_exp) -> tuple[int, int]:
+    ok = 0
+    for a in range(256):
+        for b in range(256):
+            ok += (gf_mul_neural(a, b, net_log, net_exp) == gf_mul(a, b))
+    return ok, 256 * 256

cdpreserve/rs.py

@@ -0,0 +1,104 @@
+"""
+Step 2: Reed-Solomon erasure coding over the verified GF(256) field.
+
+Split data into k systematic shards + m parity shards (n = k+m). ANY k of the n
+shards reconstruct the original -- so up to m shards can be lost or corrupted and
+the data still survives. This is the honest sense of "small pieces that together
+rebuild the whole": redundancy, not shrinkage (n shards total n/k x the data).
+
+All coefficients come from a Cauchy matrix over GF(256) (every square submatrix
+invertible => any-k-of-n recovery). Every multiply is the GF(256) multiply that
+the neural LOG/EXP units were verified bit-exact against (65536/65536), so the
+fast table path here is the proven-equivalent of the neural coding core.
+"""
+from __future__ import annotations
+from .gf256 import EXP, LOG, gf_mul
+
+MUL = [[gf_mul(a, b) for b in range(256)] for a in range(256)]   # full product table
+
+
+def gf_inv(a: int) -> int:
+    return EXP[255 - LOG[a]] if a else 0
+
+
+def cauchy(k: int, m: int):
+    return [[gf_inv((k + i) ^ j) for j in range(k)] for i in range(m)]
+
+
+def _gen_row(r: int, k: int, C):
+    if r < k:
+        row = [0] * k
+        row[r] = 1
+        return row
+    return C[r - k]
+
+
+def encode(data: bytes, k: int, m: int):
+    """-> (shards[n], shard_len, orig_len). shards[:k] are systematic data."""
+    orig = len(data)
+    L = (orig + k - 1) // k
+    d = data + bytes(L * k - orig)
+    shards = [bytearray(d[j * L:(j + 1) * L]) for j in range(k)]
+    C = cauchy(k, m)
+    for i in range(m):
+        P = bytearray(L)
+        Ci = C[i]
+        for j in range(k):
+            MC = MUL[Ci[j]]
+            Dj = shards[j]
+            for b in range(L):
+                P[b] ^= MC[Dj[b]]
+        shards.append(P)
+    return shards, L, orig
+
+
+def _invert(M, k):
+    A = [list(M[i]) + [1 if j == i else 0 for j in range(k)] for i in range(k)]
+    for col in range(k):
+        piv = col
+        while A[piv][col] == 0:
+            piv += 1
+        A[col], A[piv] = A[piv], A[col]
+        inv = gf_inv(A[col][col])
+        A[col] = [MUL[inv][x] for x in A[col]]
+        for r in range(k):
+            if r != col and A[r][col]:
+                f = A[r][col]
+                A[r] = [A[r][x] ^ MUL[f][A[col][x]] for x in range(2 * k)]
+    return [A[i][k:] for i in range(k)]
+
+
+def decode(present: dict, k: int, m: int, L: int, orig: int) -> bytes:
+    """present: {shard_index: bytes}; needs >= k entries."""
+    idxs = sorted(present.keys())[:k]
+    C = cauchy(k, m)
+    Minv = _invert([_gen_row(r, k, C) for r in idxs], k)
+    data_shards = [bytearray(L) for _ in range(k)]
+    for j in range(k):
+        row = Minv[j]
+        Dj = data_shards[j]
+        for t, r in enumerate(idxs):
+            c = row[t]
+            if not c:
+                continue
+            MC = MUL[c]
+            src = present[r]
+            for b in range(L):
+                Dj[b] ^= MC[src[b]]
+    return b"".join(bytes(s) for s in data_shards)[:orig]
+
+
+def neural_parity_equiv(net_log, net_exp, k=4, m=2, L=16) -> bool:
+    """Compute one parity shard with the NEURAL gf_mul and confirm it matches the
+    golden coding -- i.e. the verified units drive the erasure code bit-exactly."""
+    import os
+    from .gf256 import gf_mul_neural
+    data = os.urandom(L * k)
+    shards, _, _ = encode(data, k, m)
+    C = cauchy(k, m)
+    Dsh = [data[j * L:(j + 1) * L] for j in range(k)]
+    P = bytearray(L)
+    for j in range(k):
+        for b in range(L):
+            P[b] ^= gf_mul_neural(C[0][j], Dsh[j][b], net_log, net_exp)
+    return bytes(P) == bytes(shards[k])         # neural parity == golden parity

cdpreserve/scan.py

@@ -0,0 +1,57 @@
+"""
+Disc scanner.
+
+Reads a source sector by sector into bytes. The source can be a raw optical
+drive (Windows: r"\\\\.\\D:", may need admin) or an .iso / image file (for
+testing). Unreadable sectors (scratches / disc rot) are recorded as BAD and
+zero-filled -- ddrescue-style -- rather than aborting the scan.
+
+Important: bad sectors are data you never read. RS self-healing (archive.py)
+protects the STORED copy against future corruption; it cannot invent sectors the
+drive could not read. To recover those, re-scan the disc and merge good sectors
+across attempts (`merge_scans`).
+"""
+from __future__ import annotations
+
+SECTOR = 2048                     # CD-ROM Mode-1 user data sector size
+
+
+def scan_image(path: str, sector: int = SECTOR):
+    """-> (data: bytes, bad_sectors: list[int]). Robust to unreadable sectors."""
+    data = bytearray()
+    bad = []
+    with open(path, "rb", buffering=0) as f:
+        idx = 0
+        while True:
+            try:
+                blk = f.read(sector)
+            except OSError:
+                bad.append(idx)
+                data += bytes(sector)
+                try:
+                    f.seek((idx + 1) * sector)
+                except OSError:
+                    break
+                idx += 1
+                continue
+            if not blk:
+                break
+            data += blk                                # keep true length (last
+            idx += 1                                   # sector may be partial)
+    return bytes(data), bad
+
+
+def merge_scans(scans, sector: int = SECTOR):
+    """Combine multiple (data, bad) scans of the same failing disc: a sector is
+    good if ANY scan read it. Returns (merged_data, still_bad_sectors)."""
+    n = max(len(d) // sector for d, _ in scans)
+    merged = bytearray(n * sector)
+    known = [False] * n
+    for data, bad in scans:
+        badset = set(bad)
+        for s in range(len(data) // sector):
+            if s not in badset and not known[s]:
+                merged[s * sector:(s + 1) * sector] = data[s * sector:(s + 1) * sector]
+                known[s] = True
+    still_bad = [s for s in range(n) if not known[s]]
+    return bytes(merged), still_bad

cli.py

@@ -0,0 +1,58 @@
+"""
+neural-cd-preserve CLI -- scan a disc into a self-healing .pt archive.
+
+  archive <source> <out.pt> [--label L --k K --m M --sector S]
+        source = optical drive (e.g. \\\\.\\D:) or an .iso/image file
+  info    <archive.pt>              show metadata
+  verify  <archive.pt>              check health (corruption / recoverability)
+  heal    <archive.pt> [--out O]    repair detected corruption (RS)
+  restore <archive.pt> <out.iso>    reconstruct the disc image (RS-corrected)
+"""
+import argparse
+from cdpreserve import scan, archive
+
+
+def main():
+    ap = argparse.ArgumentParser(prog="neural-cd-preserve")
+    sub = ap.add_subparsers(dest="cmd", required=True)
+    a = sub.add_parser("archive"); a.add_argument("source"); a.add_argument("out")
+    a.add_argument("--label", default="disc"); a.add_argument("--k", type=int, default=4)
+    a.add_argument("--m", type=int, default=2); a.add_argument("--sector", type=int, default=scan.SECTOR)
+    for name in ("info", "verify"):
+        q = sub.add_parser(name); q.add_argument("archive")
+    h = sub.add_parser("heal"); h.add_argument("archive"); h.add_argument("--out")
+    r = sub.add_parser("restore"); r.add_argument("archive"); r.add_argument("out")
+    args = ap.parse_args()
+
+    if args.cmd == "archive":
+        data, bad = scan.scan_image(args.source, args.sector)
+        arc = archive.build(data, label=args.label, k=args.k, m=args.m, bad_sectors=bad)
+        archive.save(arc, args.out)
+        red = args.k + args.m
+        print(f"archived '{args.label}': {len(data)//1024} KB, {len(arc['chunks'])} unique chunks, "
+              f"RS {args.k}+{args.m} (~{red/args.k:.2f}x). bad sectors at scan: {len(bad)}")
+        if bad:
+            print(f"  WARNING: {len(bad)} sectors were unreadable at scan time (zero-filled). "
+                  f"Re-scan and merge to recover them.")
+    elif args.cmd == "info":
+        meta = archive.load(args.archive)["meta"]
+        for k, v in meta.items():
+            print(f"  {k}: {v if k != 'bad_sectors' else f'{len(v)} sectors'}")
+    elif args.cmd == "verify":
+        h = archive.verify(archive.load(args.archive))
+        print(f"health: {'HEALTHY' if h['healthy'] else 'DAMAGED'}  "
+              f"chunks={h['chunks']} corrupted_shards={h['corrupted_shards']} "
+              f"repairable={h['repairable_chunks']} lost={h['lost_chunks']}")
+    elif args.cmd == "heal":
+        arc = archive.load(args.archive)
+        n = archive.heal(arc)
+        archive.save(arc, args.out or args.archive)
+        print(f"repaired {n} shard(s) -> {args.out or args.archive}")
+    elif args.cmd == "restore":
+        data = archive.restore(archive.load(args.archive))
+        open(args.out, "wb").write(data)
+        print(f"restored {len(data)//1024} KB -> {args.out}")
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1 @@
+torch

step_cd.py

@@ -0,0 +1,80 @@
+"""
+CD preservation demo: scan -> self-healing .pt archive -> corrupt -> detect ->
+heal -> restore bit-exact. Uses a synthetic disc image (no drive needed here).
+
+  (a) scan an image (bad-sector-tolerant)
+  (b) build .pt archive + restore bit-exact
+  (c) inject silent bit-rot (<= m shards/chunk) -> verify DETECTS it
+  (d) heal() repairs from survivors -> healthy again -> restore bit-exact
+  (e) honesty: corrupt > m shards in a chunk -> flagged LOST, never silently wrong
+"""
+import os
+import copy
+import random
+import tempfile
+from cdpreserve import scan, archive
+
+random.seed(0)
+
+print("=" * 62)
+print("neural-cd-preserve  --  self-healing .pt disc archive")
+print("=" * 62)
+
+# a synthetic 'disc': structured header + repeated table + random payload
+disc = (b"CD-IMAGE-V1" + bytes(2037)                       # a 'sector'
+        + (b"TABLE" * 400) * 3                               # repeated -> dedup
+        + os.urandom(400 * 1024))
+
+# (a) scan (via a temp file standing in for the drive)
+tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".iso")
+tmp.write(disc); tmp.close()
+data, bad = scan.scan_image(tmp.name)
+print(f"(a) scan: {len(data)//1024} KB, bad sectors: {len(bad)}  "
+      f"-> {'PASS' if data == disc else 'FAIL'}")
+
+# (b) build archive + restore
+arc = archive.build(data, label="DEMO_DISC", k=4, m=2)
+red = 6 / 4
+print(f"(b) archive: {len(arc['chunks'])} unique chunks, RS 4+2 (~{red:.2f}x); "
+      f"restore bit-exact: {'PASS' if archive.restore(arc) == data else 'FAIL'}")
+
+# (c) inject silent bit-rot: flip bytes in 1..2 shards per chunk
+def corrupt(rec, nshards):
+    idxs = random.sample(range(len(rec["shards"])), nshards)
+    for i in idxs:
+        s = bytearray(rec["shards"][i]); s[0] ^= 0xFF; s[-1] ^= 0xFF
+        rec["shards"][i] = bytes(s)                          # hash now mismatches
+
+damaged = 0
+for rec in arc["chunks"].values():
+    n = random.randint(1, 2)                                 # <= m, recoverable
+    corrupt(rec, n); damaged += n
+h1 = archive.verify(arc)
+print(f"(c) injected bit-rot in {damaged} shards; verify DETECTS: "
+      f"corrupted={h1['corrupted_shards']} repairable={h1['repairable_chunks']} "
+      f"lost={h1['lost_chunks']}  -> {'PASS' if h1['corrupted_shards'] == damaged and h1['lost_chunks'] == 0 else 'FAIL'}")
+
+# (d) heal + restore
+repaired = archive.heal(arc)
+h2 = archive.verify(arc)
+restore_ok = archive.restore(arc) == data
+print(f"(d) heal repaired {repaired} shards; now {'HEALTHY' if h2['healthy'] else 'DAMAGED'}; "
+      f"restore bit-exact: {'PASS' if restore_ok and h2['healthy'] else 'FAIL'}")
+
+# (e) honesty: beyond RS capacity -> LOST, not silently wrong
+arc2 = copy.deepcopy(arc)
+victim = next(iter(arc2["chunks"].values()))
+corrupt(victim, 3)                                          # > m=2 -> unrecoverable
+h3 = archive.verify(arc2)
+try:
+    archive.restore(arc2); raised = False
+except IOError:
+    raised = True
+print(f"(e) corrupt 3 shards (> m) in one chunk -> lost={h3['lost_chunks']}, "
+      f"restore refuses: {'PASS (honest failure)' if h3['lost_chunks'] == 1 and raised else 'FAIL'}")
+
+os.unlink(tmp.name)
+allpass = (archive.restore(arc) == data and h1['lost_chunks'] == 0 and h2['healthy']
+           and restore_ok and h3['lost_chunks'] == 1 and raised)
+print("=" * 62)
+print(f"OVERALL: {'ALL PASS' if allpass else 'some checks failed'}")