"""Build original Week 7 figures from the lab's generated diagnostic CSVs. Run from the book directory after its lab smoke: conda run -n research python assets/scripts/build-week7-teams.py All inputs are invented. Exact CSV bytes are preserved under assets/data. """ from pathlib import Path import csv import math import os import sys BOOK = Path(__file__).resolve().parents[2] REPO = BOOK.parents[1] LAB = BOOK / "labs/07-teams-peer-effects-communication-and-coordination" MODE = "SYNTHETIC_TEACHING_CONSTRUCTION_NOT_PUBLISHED_REPLICATION" os.environ.setdefault("MPLCONFIGDIR", str(BOOK / ".mpl-cache")) import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt sys.path.insert(0, str(REPO)) from shared.figure_style import COLORS, apply_style, save_figure, style_axis def read_and_preserve(source_name, target_name): source = LAB / "output/diagnose" / source_name target = BOOK / "assets/data" / target_name target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(source.read_bytes()) with target.open(newline="") as stream: rows = list(csv.DictReader(stream)) assert rows and all(r["provenance"] == MODE for r in rows) return rows def save(fig, name): path = BOOK / "assets/figures" / name save_figure(fig, path.with_suffix(".png")) fig.savefig(path.with_suffix(".pdf"), bbox_inches="tight", facecolor="white", metadata={"CreationDate": None, "ModDate": None}) plt.close(fig) def main(): support = read_and_preserve("exposure_support.csv", "07-exposure-support.csv") curve = read_and_preserve("communication_curve.csv", "07-communication-curve.csv") costs = read_and_preserve("benefits_and_costs.csv", "07-benefits-and-costs.csv") assert len(support) == 4 expected = {(0,"L"):(1/3,3/8), (1,"L"):(0,1/8), (0,"H"):(1,1/8), (1,"H"):(2/3,3/8)} observed = {} for row in support: key = int(row["own_aid"]), row["saturation"] values = float(row["peer_share"]), float(row["assignment_probability"]) assert key not in observed and key in expected assert all(math.isclose(a,b,abs_tol=1e-9) for a,b in zip(values,expected[key])) observed[key] = values assert math.isclose(sum(x[1] for x in observed.values()),1) assert not {x[0] for k,x in observed.items() if k[0]==0} & {x[0] for k,x in observed.items() if k[0]==1} x = [float(r["communication"]) for r in curve] y = [float(r["net_value"]) for r in curve] assert all(m>=0 and math.isclose(v,12*math.log1p(m)-2*m-m*m/2,abs_tol=1e-9) for m,v in zip(x,y)) peak = max(range(len(x)),key=lambda i:y[i]) assert x[peak] == 2 and math.isclose(y[peak],12*math.log(3)-6,abs_tol=1e-9) accounting = {r["unit"]:float(r["output_change"]) for r in costs} assert accounting == {"sender":-5,"recipient":8,"combined":3} assert accounting["sender"]+accounting["recipient"] == accounting["combined"] apply_style() fig,ax = plt.subplots(figsize=(10.2,5.5)) for own in [0,1]: for peer in [0,1/3,2/3,1]: if not any(k[0]==own and math.isclose(v[0],peer,abs_tol=1e-9) for k,v in observed.items()): ax.scatter(peer,own,marker="x",s=90,color=COLORS["muted"],linewidths=1.8,zorder=3) ax.text(peer,own-.19,"unsupported",ha="center",va="top",fontsize=10,color=COLORS["muted"]) for (own,saturation),(peer,prob) in observed.items(): color = COLORS["navy"] if saturation=="L" else COLORS["teal"] ax.scatter(peer,own,s=750 if prob==3/8 else 250,color=color,zorder=4) ax.annotate(f"({own}, {saturation}): p = {'3/8' if prob==3/8 else '1/8'}",(peer,own), xytext=(0,23),textcoords="offset points",ha="center",fontsize=11,color=color) ax.set(title="Only four complete exposure states are supported",xlabel="Treated peers / three eligible peers", ylabel="Own task aid",xticks=[0,1/3,2/3,1],xticklabels=["0","1/3","2/3","1"], yticks=[0,1],yticklabels=["0: no aid","1: aid"],xlim=(-.13,1.13),ylim=(-.38,1.43)) style_axis(ax) fig.text(.5,.12,"L: one recipient per team. H: three recipients. Two of four teams receive H.",ha="center",fontsize=10) fig.text(.5,.072,"Labels give unconditional probabilities over both assignment stages; marker area also reflects probability.",ha="center",fontsize=9,color=COLORS["muted"]) fig.text(.5,.023,"Original synthetic design • no own-treatment contrast at a fixed peer share",ha="center",fontsize=10,color=COLORS["muted"]) fig.tight_layout(rect=(0,.16,1,1)) save(fig,"07-exposure-support") fig,axes = plt.subplots(1,2,figsize=(10.5,5.6),gridspec_kw={"width_ratios":[1.15,1]}) ax = axes[0] ax.plot(x,y,color=COLORS["navy"],linewidth=2.4) ax.scatter([x[peak]],[y[peak]],color=COLORS["teal"],s=60,zorder=4) ax.axvline(2,color=COLORS["teal"],linestyle="--",alpha=.55) ax.axhline(0,color=COLORS["muted"],linewidth=.8) ax.annotate("Maximum: m = 2\nV(2) ≈ 7.18",(x[peak],y[peak]),xytext=(2.7,8.6),fontsize=11, arrowprops={"arrowstyle":"->","color":COLORS["teal"]}) ax.set(title="Diminishing gains and rising costs",xlabel="Communication activity, m",ylabel="Net value, V(m)",ylim=(min(y)-1,12)) ax.text(.04,.05,r"$V(m)=12\log(1+m)-2m-m^2/2$",transform=ax.transAxes,fontsize=11) ax = axes[1] values = [accounting[k] for k in ["sender","recipient","combined"]] bars = ax.bar([0,1,2],values,color=[COLORS["rust"],COLORS["teal"],COLORS["navy"]],width=.58) ax.bar_label(bars,labels=[f"{v:+.0f}" for v in values],padding=5,fontsize=12) ax.axhline(0,color=COLORS["muted"],linewidth=.9) ax.set(title="Count both sides of helping",ylabel="Change in comparable output units",xticks=[0,1,2], xticklabels=["Sender","Recipient","Joint"],ylim=(-7.5,11)) for ax in axes: style_axis(ax) fig.text(.5,.09,"Separate original synthetic examples: a communication objective and sender–recipient accounting.",ha="center",fontsize=9.5) fig.text(.5,.033,"Invented units; no published estimate or calibrated recommendation.",ha="center",fontsize=10,color=COLORS["muted"]) fig.tight_layout(rect=(0,.14,1,1),w_pad=3) save(fig,"07-communication-value") print("PASS: support, probabilities, curve and accounting verified; three exact CSV inputs and two PNG/PDF figures saved.") if __name__ == "__main__": main()