step_cd.py

"""
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'}")