| """ |
| 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() |
|
|
