import logging import os import sys import time import uuid import gradio as gr from daggr import FnNode, Graph logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", stream=sys.stdout, ) log = logging.getLogger("daggr.micro") import hashlib def _stamp(*parts) -> str: h = hashlib.sha1("||".join(map(str, parts)).encode("utf-8")).hexdigest()[:10] return h def _phase(title: str): log.info("========== %s ==========", title) def _drv(lines, msg: str): log.info("[DRV] %s", msg) lines.append(msg) def _assert_drv(lines, cond: bool, msg_ok: str, msg_bad: str): if cond: _drv(lines, f"OK | {msg_ok}") else: _drv(lines, f"FAIL | {msg_bad}") def make_d(seed_d: int, salt_d: int) -> tuple[int, str]: run_id = uuid.uuid4().hex[:8] t0 = time.time() log.info("[D] start run=%s seed_d=%s salt_d=%s", run_id, seed_d, salt_d) val = (int(seed_d) * 10) + (int(salt_d) % 10) tag = f"D(seed={int(seed_d)},salt={int(salt_d)})" out = (val, tag) dt = (time.time() - t0) * 1000 log.info("[D] done run=%s dt=%.1fms out=%s", run_id, dt, out) return out def make_e(seed_e: int, salt_e: int) -> tuple[int, str]: run_id = uuid.uuid4().hex[:8] t0 = time.time() log.info("[E] start run=%s seed_e=%s salt_e=%s", run_id, seed_e, salt_e) val = (int(seed_e) * 100) + (int(salt_e) % 10) tag = f"E(seed={int(seed_e)},salt={int(salt_e)})" out = (val, tag) dt = (time.time() - t0) * 1000 log.info("[E] done run=%s dt=%.1fms out=%s", run_id, dt, out) return out def combine(d_val: int, d_tag: str, e_val: int, e_tag: str, alpha: float) -> tuple[int, str]: run_id = uuid.uuid4().hex[:8] t0 = time.time() a = float(alpha) log.info("[A] start run=%s d=%s e=%s alpha=%s", run_id, (d_val, d_tag), (e_val, e_tag), a) mix = int((int(d_val) * a) + (int(e_val) * (1.0 - a))) summary = f"mix={mix} | {d_tag} + {e_tag} | alpha={a:.2f}" dt = (time.time() - t0) * 1000 log.info("[A] done run=%s dt=%.1fms mix=%s", run_id, dt, mix) return mix, summary def postprocess(mix: int, a_summary: str, mode: str, bump: int) -> str: run_id = uuid.uuid4().hex[:8] t0 = time.time() bump_i = int(bump) log.info("[B] start run=%s mode=%s bump=%s mix=%s", run_id, mode, bump_i, mix) if mode == "add": mix2 = int(mix) + bump_i elif mode == "mul": mix2 = int(mix) * max(1, bump_i) else: mix2 = int(mix) text = f"{a_summary} || B({mode},{bump_i}) => {mix2}" dt = (time.time() - t0) * 1000 log.info("[B] done run=%s dt=%.1fms mix_out=%s", run_id, dt, mix2) return text def observe(b_text: str, note: str) -> str: run_id = uuid.uuid4().hex[:8] t0 = time.time() log.info("[C] start run=%s note=%s", run_id, note) out = f"[C] note={note} | {b_text}" dt = (time.time() - t0) * 1000 log.info("[C] done run=%s dt=%.1fms", run_id, dt) return out def driver(run_id: int = 1) -> str: """ Driver: exécute une séquence déterministe d'appels Python pour valider la logique (phases 0-3) et produire un rapport. IMPORTANT: ne met pas à jour l'historique UI des nodes D/E/A/B/C. """ lines = [] sid = _stamp("driver", run_id, time.time_ns()) _phase(f"DRIVER start sid={sid}") _drv(lines, f"sid={sid}") _drv(lines, "NOTE: driver calls functions directly; UI history for D/E/A/B/C will NOT change.") # ---------------- # PH0 baseline # ---------------- _phase("PH0 baseline") d0 = make_d(0, 0) e0 = make_e(0, 0) a0 = combine(d0[0], d0[1], e0[0], e0[1], 0.5) b0 = postprocess(a0[0], a0[1], "none", 1) c0 = observe(b0, "hello") _drv(lines, f"PH0 | D={d0} | E={e0}") _drv(lines, f"PH0 | A(mix,summary)={a0}") _drv(lines, f"PH0 | B(text)={b0}") _drv(lines, f"PH0 | C(text)={c0}") # invariants simples _assert_drv(lines, isinstance(a0[0], int), "A.mix is int", f"A.mix not int: {type(a0[0])}") _assert_drv(lines, "alpha=0.50" in a0[1], "A.summary contains alpha", "A.summary missing alpha") _assert_drv(lines, "mix=" in a0[1], "A.summary contains mix", "A.summary missing mix") # ---------------- # PH1A D versions # ---------------- _phase("PH1A D versions (E fixed)") for sd in [1, 2]: d = make_d(sd, 0) a = combine(d[0], d[1], e0[0], e0[1], 0.5) _drv(lines, f"PH1A | seed_d={sd} => D={d} | A={a}") _assert_drv(lines, a[0] == int(d[0] * 0.5 + e0[0] * 0.5), "A.mix matches weighted sum", "A.mix mismatch") # ---------------- # PH1B E versions (D fixed) # ---------------- _phase("PH1B E versions (D fixed to seed=2)") d2 = make_d(2, 0) for se in [1, 2]: e = make_e(se, 0) a = combine(d2[0], d2[1], e[0], e[1], 0.5) _drv(lines, f"PH1B | seed_e={se} => E={e} | A={a}") _assert_drv(lines, a[0] == int(d2[0] * 0.5 + e[0] * 0.5), "A.mix matches weighted sum", "A.mix mismatch") # ---------------- # PH2 alpha variations (same upstream pair) # ---------------- _phase("PH2 alpha variations (D=2,E=1)") d = make_d(2, 0) e = make_e(1, 0) for alpha in [0.2, 0.8, 0.35]: a = combine(d[0], d[1], e[0], e[1], alpha) _drv(lines, f"PH2 | alpha={alpha:.2f} => A={a}") _assert_drv(lines, f"alpha={alpha:.2f}" in a[1], "A.summary alpha matches", "A.summary alpha mismatch") # ---------------- # PH3 downstream noise # ---------------- _phase("PH3 downstream noise") a = combine(d[0], d[1], e[0], e[1], 0.35) for mode, bump in [("add", 1), ("add", 2), ("mul", 2)]: b = postprocess(a[0], a[1], mode, bump) c = observe(b, f"note-{mode}-{bump}") _drv(lines, f"PH3 | B({mode},{bump}) => {b}") _drv(lines, f"PH3 | C(note-{mode}-{bump}) => {c}") _phase(f"DRIVER done sid={sid}") return "\n".join(lines) DRIVER = FnNode( fn=driver, inputs={ "run_id": gr.Slider(label="Driver run_id", minimum=1, maximum=20, step=1, value=1), }, outputs={ "report": gr.Textbox(label="Driver report", lines=30), }, ) D = FnNode( fn=make_d, inputs={ "seed_d": gr.Number(label="D seed", value=0, precision=0), "salt_d": gr.Slider(label="D salt", minimum=0, maximum=9, step=1, value=0), }, outputs={ "d_val": gr.Number(label="D val"), "d_tag": gr.Textbox(label="D tag"), }, ) E = FnNode( fn=make_e, inputs={ "seed_e": gr.Number(label="E seed", value=0, precision=0), "salt_e": gr.Slider(label="E salt", minimum=0, maximum=9, step=1, value=0), }, outputs={ "e_val": gr.Number(label="E val"), "e_tag": gr.Textbox(label="E tag"), }, ) A = FnNode( fn=combine, inputs={ "d_val": D.d_val, "d_tag": D.d_tag, "e_val": E.e_val, "e_tag": E.e_tag, "alpha": gr.Slider(label="A alpha", minimum=0.0, maximum=1.0, step=0.05, value=0.5), }, outputs={ "mix": gr.Number(label="A mix"), "a_summary": gr.Textbox(label="A summary"), }, ) B = FnNode( fn=postprocess, inputs={ "mix": A.mix, "a_summary": A.a_summary, "mode": gr.Radio(label="B mode", choices=["none", "add", "mul"], value="none"), "bump": gr.Slider(label="B bump", minimum=0, maximum=5, step=1, value=1), }, outputs={"b_text": gr.Textbox(label="B text")}, ) C = FnNode( fn=observe, inputs={ "b_text": B.b_text, "note": gr.Textbox(label="C note", value="hello"), }, outputs={"c_text": gr.Textbox(label="C text")}, ) graph = Graph( name="micro-restore-debug-nojson", nodes=[D, E, A, B, C, DRIVER], persist_key=False, ) port = int(os.getenv("PORT", "7860")) graph.launch(host="0.0.0.0", port=port, open_browser=False, access_log=True)