#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ M15 Abgeschlossenheit: Hat der Loesungsraum einer Instanz eine idiomatische Regel? Fuer jede loesbare Instanz werden gleichverteilt Loesungen gezogen (exakt aus dem BDD) und vier Operationen bitweise angewandt: UND (min), ODER (max), MEHRHEIT (3-stellig), MINDERHEIT = x^y^z (3-stellig) Gemessen wird, wie oft das Ergebnis wieder eine Loesung ist ("Treue"). Nur verschiedene Eingaben; triviale Faelle (Ergebnis = eine Eingabe) werden getrennt gezaehlt und aus der Treue herausgenommen. Erwartung aus der Theorie (Polymorphismen der Sprache): horn3 -> UND = 1, sat2 -> MEHRHEIT = 1, xor3 -> MINDERHEIT = 1 3sat -> keine Operation = 1 Basis: dieselben Operationen auf einer Zufallsmenge gleicher Groesse. lift = Treue / Basis (1 = nicht besser als Zufall) Abhaengigkeit: m13_durchlass.py (Generator, BDD) im selben Ordner. Seed wie M13 (seed*1000003 + n*7919 + i*104729 + Typ-Offset). Aufruf: nohup python3 m15_abschluss.py --jobs 16 > m15.log 2>&1 & """ import argparse, csv, random, signal, statistics, sys, time from multiprocessing import Pool import m13_durchlass as m13 OPS = ("und", "oder", "mehr", "minder") TYPEN = ["3sat", "horn3", "sat2", "xor3", "rel5", "rel4"] def pruefer(typ, cons, n): """Liefert Liste (pos, neg) je Klausel; x erfuellt Klausel gdw (x&pos) or (~x&neg).""" voll = (1 << n) - 1 kl = [] for c in m13.cnf(typ, cons): pos = neg = 0 for l in c: if l > 0: pos |= 1 << (l - 1) else: neg |= 1 << (-l - 1) kl.append((pos, neg)) def ist_loesung(x): nx = ~x & voll for pos, neg in kl: if not (x & pos) and not (nx & neg): return False return True return ist_loesung def ziehe(B, u, n, anzahl, r): """Gleichverteilt Loesungen aus dem BDD ziehen (natuerliche Ordnung: Ebene l = Variable l).""" memo = {0: 0, 1: 1} def c(v): if v in memo: return memo[v] l = B.lev[v]; lo, hi = B.lo[v], B.hi[v] w = c(lo) * (1 << (B.lev[lo] - l - 1)) + c(hi) * (1 << (B.lev[hi] - l - 1)) memo[v] = w; return w c(u) out = [] for _ in range(anzahl): x = 0; v = u; l = 0 while True: lv = B.lev[v] for j in range(l, lv): # uebersprungene Ebenen frei if r.random() < 0.5: x |= 1 << j if v == 1: break lo, hi = B.lo[v], B.hi[v] wl = c(lo) * (1 << (B.lev[lo] - lv - 1)); wh = c(hi) * (1 << (B.lev[hi] - lv - 1)) if r.random() * (wl + wh) < wh: x |= 1 << lv; v = hi else: v = lo l = lv + 1 out.append(x) return out def anwenden(op, a, b, cc): if op == "und": return a & b if op == "oder": return a | b if op == "mehr": return (a & b) | (a & cc) | (b & cc) return a ^ b ^ cc def treue(ziehen, istin, proben, r): """ziehen() liefert eine zufaellige Loesung. Rueckgabe je op: (treffer, nichttrivial, trivial).""" erg = {} for op in OPS: t = nt = tv = 0; versuche = 0 while nt < proben and versuche < proben * 20: versuche += 1 a, b, cc = ziehen(), ziehen(), ziehen() if a == b or (op in ("mehr", "minder") and (a == cc or b == cc)): continue y = anwenden(op, a, b, cc) if y in (a, b) or (op in ("mehr", "minder") and y == cc): tv += 1; continue nt += 1 if istin(y): t += 1 erg[op] = (t, nt, tv) return erg class Zeit(Exception): pass def _alarm(*_): raise Zeit() def aufgabe(a): typ, n, faktor, seed, i, proben, pool_max, basis_max, tmax = a signal.signal(signal.SIGALRM, _alarm); signal.alarm(tmax) t0 = time.time() z = dict(typ=typ, n=n, faktor=faktor, i=i, sat=0, log2S="", status="ok") for op in OPS: z[f"t_{op}"] = ""; z[f"b_{op}"] = ""; z[f"triv_{op}"] = "" try: cons = m13.erzeuge(typ, n, faktor, seed, i) B, u = m13.bdd_loesung(typ, cons, n, list(range(n))) if u != 0: S = B.zaehle(u) z.update(sat=1, log2S=round(S.bit_length() - 1 + (S / (1 << (S.bit_length() - 1)) - 1), 4)) if S >= 4: r = random.Random(seed * 3 + n * 5 + i * 7 + m13.OFFSET[typ]) istin = pruefer(typ, cons, n) pool = ziehe(B, u, n, pool_max, r) er = treue(lambda: r.choice(pool), istin, proben, r) # Basis: Zufallsmenge gleicher Groesse if S <= basis_max: menge = r.sample(range(1 << n), S) ms = set(menge) eb = treue(lambda: r.choice(menge), ms.__contains__, proben, r) else: eb = None for op in OPS: t, nt, tv = er[op] if nt: z[f"t_{op}"] = round(t / nt, 4) z[f"triv_{op}"] = round(tv / (nt + tv), 4) if eb: tb, ntb, _ = eb[op] if ntb: z[f"b_{op}"] = round(tb / ntb, 4) else: z[f"b_{op}"] = round(S / (1 << n), 6) # Naeherung: Dichte der Menge except Zeit: z["status"] = "zeit" finally: signal.alarm(0) z["sek"] = round(time.time() - t0, 3) return z def selbsttest(): fehler = 0 r = random.Random(5) # (1) Ziehen liefert nur Loesungen und ungefaehr gleichverteilt cons = m13.erzeuge("3sat", 10, 0.6, 3, 0) istin = pruefer("3sat", cons, 10) loes = [x for x in range(1 << 10) if istin(x)] B, u = m13.bdd_loesung("3sat", cons, 10, list(range(10))) if B.zaehle(u) != len(loes): fehler += 1 zug = ziehe(B, u, 10, 20000, r) if any(not istin(x) for x in zug): fehler += 1 haeuf = {} for x in zug: haeuf[x] = haeuf.get(x, 0) + 1 erw = 20000 / len(loes) if max(haeuf.values()) > 3 * erw + 10 or len(haeuf) < 0.9 * len(loes): fehler += 1 # (2) Polymorphismus-Kontrollen: Treue muss exakt 1 sein for typ, op in (("horn3", "und"), ("sat2", "mehr"), ("xor3", "minder")): for i in range(5): z = aufgabe((typ, 14, 0.6, 9, i, 300, 2000, 100000, 120)) if z["sat"] and z[f"t_{op}"] != "" and z[f"t_{op}"] != 1.0: fehler += 1 return fehler def med(xs): return statistics.median(xs) if xs else float("nan") def bericht(zs, args): out = []; p = out.append p("M15 Abgeschlossenheit — Treue = Anteil, mit dem op(Loesungen) wieder Loesung ist") p(f"seed={args.seed} inst={args.inst} proben={args.proben} pool={args.pool}") st = {} for z in zs: st[z["status"]] = st.get(z["status"], 0) + 1 p(f"Status: {st}") p("Erwartung: horn3 UND=1, sat2 MEHR=1, xor3 MINDER=1; 3sat keine =1") p("") for lo, hi in ((0.3, 0.5), (0.5, 0.7)): p(f"HAUPTBLOCK: Raumanteil log2S/n in [{lo},{hi}) — Treue Median (Basis Median) je n") for typ in args.typen: p(f" {typ}") for op in OPS: teile = [] for n in args.ns: r = [z for z in zs if z["typ"] == typ and z["n"] == n and z["status"] == "ok" and z["sat"] and z["log2S"] != "" and lo <= z["log2S"] / n < hi and z[f"t_{op}"] != ""] if len(r) >= 8: teile.append(f"n{n} {med([z[f't_{op}'] for z in r]):.3f} " f"({med([z[f'b_{op}'] for z in r]):.3f})") else: teile.append(f"n{n} -") p(f" {op:>6}: " + " ".join(teile)) p("") p("Beste Operation je Instanz (max Treue), alle Faktoren — Median je n:") for typ in args.typen: teile = [] for n in args.ns: r = [z for z in zs if z["typ"] == typ and z["n"] == n and z["status"] == "ok" and z["sat"] and all(z[f"t_{op}"] != "" for op in OPS)] if r: teile.append(f"n{n} {med([max(z[f't_{op}'] for op in OPS) for z in r]):.3f}/{len(r)}") p(f" {typ:>5}: " + " ".join(teile)) p("") p("Anteil Instanzen mit exakt geschlossener Operation (Treue = 1) je Typ:") for typ in args.typen: r = [z for z in zs if z["typ"] == typ and z["status"] == "ok" and z["sat"] and all(z[f"t_{op}"] != "" for op in OPS)] if r: p(f" {typ:>5}: " + " ".join(f"{op} {sum(1 for z in r if z[f't_{op}'] == 1.0)/len(r):.2f}" for op in OPS) + f" N {len(r)}") return "\n".join(out) def main(): ap = argparse.ArgumentParser() ap.add_argument("--typen", default=",".join(TYPEN)) ap.add_argument("--ns", default="12,16,20,24,28") ap.add_argument("--faktoren", default="0.3,0.5,0.7,0.9") ap.add_argument("--inst", type=int, default=60) ap.add_argument("--seed", type=int, default=15) ap.add_argument("--proben", type=int, default=1000) ap.add_argument("--pool", type=int, default=5000) ap.add_argument("--basis_max", type=int, default=400000) ap.add_argument("--jobs", type=int, default=16) ap.add_argument("--tmax", type=int, default=900) ap.add_argument("--csv", default="m15_abschluss.csv") ap.add_argument("--report", default="m15_report.txt") ap.add_argument("--nur_selbsttest", action="store_true") args = ap.parse_args() args.typen = args.typen.split(",") args.ns = [int(x) for x in args.ns.split(",")] args.faktoren = [float(x) for x in args.faktoren.split(",")] t = time.time(); f = selbsttest() print(f"Selbsttest (Ziehen exakt/gleichverteilt, Polymorphismus-Kontrollen = 1): Fehler {f} [{time.time()-t:.1f}s]", flush=True) if f: print("ABBRUCH: Selbsttest fehlgeschlagen."); sys.exit(1) if args.nur_selbsttest: return aufg = [(typ, n, fa, args.seed, i, args.proben, args.pool, args.basis_max, args.tmax) for n in args.ns for typ in args.typen for fa in args.faktoren for i in range(args.inst)] aufg.sort(key=lambda a: -a[1]) zs = []; t0 = time.time() with Pool(args.jobs) as pool, open(args.csv, "w", newline="") as fh: w = None for k, z in enumerate(pool.imap_unordered(aufgabe, aufg), 1): if w is None: w = csv.DictWriter(fh, fieldnames=list(z.keys())); w.writeheader() w.writerow(z); fh.flush(); zs.append(z) if k % 200 == 0 or k == len(aufg): print(f"[{k}/{len(aufg)}] {(time.time()-t0)/60:.1f} Min", flush=True) txt = bericht(zs, args) with open(args.report, "w") as fh: fh.write(txt + "\n") print(txt) if __name__ == "__main__": main()