9.3 情绪与节奏的对齐

#!/usr/bin/env python3
# file: sim/gd_play_09_align.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/gd_play_09_align.py
"""卷二 9.3 情绪与节奏的对齐（S 层）：张力曲线与行动节奏对不对得上。

9.2 量出了张力曲线的形状，但「形状好看」不等于「节奏对得上」。本脚本把**节奏**操作化为
**每回合的行动数** P_t（一回合里成功打出的牌数，来自引擎快照的 played 增量），再问一个可算的
问题：**张力往上走的时候，行动数是不是也在往上走？**

对齐用**同序率 concordance‰**（整数）：对一局里所有回合对 (i<j)，看 ΔT 与 ΔP 是否同号；
同号的对占全部非并列对的比例就是同序率。1000‰ = 完全同向，0‰ = 完全反向，**500‰ 是随机基线**。
对一局里每个回合对计数、跨局求和——这样长局权重更大，且不需要浮点相关。

本脚本比较四种配置：baseline 策略在 energyMax∈{3,4,6}，以及**后置节奏**策略在 energyMax=3
（把动作攒到后面用）。张力与行动数都是**代理指标（proxy）**，同序率只作相对比较。画一条把
两条曲线各自归一化到千分比的纯 xml.etree SVG。全整数、无浮点、无 random。产出 out/play09_align.json。
"""
from __future__ import annotations

import copy
import json
import os
import sys
import time
import xml.etree.ElementTree as ET

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
OUT = os.path.join(ROOT, "out")
CONFIG = os.path.join(ROOT, "config/gd_play_09_curve.json")
sys.path.insert(0, HERE)
import ashlib  # noqa: E402


def run(spec: dict, strategy: list) -> list:
    """逐步重放，返回每回合结束（含决胜回合）的 (张力, 行动数)。规则同 ashlib.run_spec。"""
    p = spec["params"]
    deck: list = []
    for c in spec["cards"]:
        deck.extend([c] * c["count"])
    deck = ashlib.shuffle_deck(deck, spec["seed"])
    hand = deck[: p["handSize"]]
    draw = deck[p["handSize"]:]
    turn, energy = 1, p["energyMax"]
    php, ehp, block, dealt, plays = p["playerHp"], p["enemyHp"], 0, 0, 0
    over = won = False
    rows: list = []

    def snap() -> tuple:
        progress = dealt * 1000 // p["enemyHp"]
        strain = (p["playerHp"] - php) * 1000 // p["playerHp"]
        return ((progress + strain) // 2, plays)

    for cmd in strategy:
        if over:
            break
        if turn > p["turnLimit"]:
            over = True
            break
        if cmd.startswith("p"):
            idx = int(cmd[1:])
            if idx < 0 or idx >= len(hand):
                continue
            c = hand[idx]
            if c["cost"] > energy:
                continue
            hand.pop(idx)
            energy -= c["cost"]
            block += c["block"]
            d = min(c["damage"], ehp)
            ehp -= d
            dealt += d
            plays += 1
            if ehp <= 0:
                over = won = True
        elif cmd == "e":
            absorbed = min(block, p["enemyAttack"])
            block -= absorbed
            php = max(0, php - (p["enemyAttack"] - absorbed))
            rows.append(snap())
            if php <= 0:
                over = True
                break
            turn += 1
            block = 0
            energy = p["energyMax"]
            plays = 0
            while len(hand) < p["handSize"] and draw:
                hand.append(draw.pop(0))
    if won:
        rows.append(snap())
    return rows


def concordance(rows: list) -> tuple[int, int]:
    """同序对与总非并列对：对一局里所有回合对比较 ΔT 与 ΔP 的符号。"""
    conc = tot = 0
    for i in range(len(rows)):
        for j in range(i + 1, len(rows)):
            dt = rows[j][0] - rows[i][0]
            dp = rows[j][1] - rows[i][1]
            if dt == 0 or dp == 0:
                continue
            tot += 1
            if (dt > 0) == (dp > 0):
                conc += 1
    return conc, tot


def avg_curves(spec: dict, seeds: list, strategy: list, over: dict) -> tuple[list, list]:
    """固定池上按回合取**条件均值**（只对活到该回合的局求均值，与 9.2 的释放口径不同）。

    9.2 要画「张力随局结束而释放」的弧线，故把已结束的局按 0 计入；本节要对比**同一批回合里**
    的张力与行动数，若也按 0 计入，行动数会在后半段被「局已结束」拖到 0，看不出节奏。故这里对
    每个回合只统计**活到该回合的局**，两条曲线同口径，形状可比。
    """
    ts: dict[int, int] = {}
    ps: dict[int, int] = {}
    cnt: dict[int, int] = {}
    for sd in seeds:
        s = copy.deepcopy(spec)
        s["seed"] = sd
        s["params"].update(over)
        for i, (t, p_) in enumerate(run(s, strategy)):
            ts[i + 1] = ts.get(i + 1, 0) + t
            ps[i + 1] = ps.get(i + 1, 0) + p_
            cnt[i + 1] = cnt.get(i + 1, 0) + 1
    turns = sorted(cnt)
    return ([ts[t] // cnt[t] for t in turns], [ps[t] // cnt[t] for t in turns])


def draw(tcurve: list, pcurve: list, path: str) -> int:
    W, H, x0, y0, y1 = 620, 280, 70, 44, 220
    root = ashlib.svg_begin(W, H)
    ashlib.svg_text(root, x0, 26, "baseline：张力（绿，上升）与行动数（橙，下降）各归一化到千分比", 14)
    mt = max(tcurve) or 1
    mp = max(pcurve) or 1
    for curve, mx, color in ((tcurve, mt, "#22c55e"), (pcurve, mp, "#f59e0b")):
        pts = " ".join(f"{x0 + i * (W - 110) // max(1, len(curve) - 1)},"
                       f"{y1 - v * 1000 // mx * (y1 - y0) // 1000}" for i, v in enumerate(curve))
        ET.SubElement(root, "polyline", {"points": pts, "fill": "none",
                                         "stroke": color, "stroke-width": "2"})
    return ashlib.svg_save(root, path)


def main() -> int:
    t0 = time.perf_counter_ns()
    spec = ashlib.load_spec()
    with open(CONFIG, encoding="utf-8") as f:
        cfg = json.load(f)
    seeds = cfg["seedPool"][: cfg["cellsPerScan"]]
    base = list(spec["strategy"])
    back = cfg["align"]["backloaded"]
    variants = [(f"baseline em={em}", base, {"energyMax": em}) for em in cfg["align"]["energyMaxCandidates"]]
    variants.append(("backloaded em=3", back, {"energyMax": 3}))

    rows_out = []
    for label, strat, over in variants:
        conc = tot = 0
        for sd in seeds:
            s = copy.deepcopy(spec)
            s["seed"] = sd
            s["params"].update(over)
            c, t = concordance(run(s, strat))
            conc += c
            tot += t
        rate = 1000 * conc // tot if tot else -1
        rows_out.append({"label": label, "concordance": rate, "concordant": conc, "pairs": tot})

    tc, pc = avg_curves(spec, seeds, base, {"energyMax": 3})
    size = draw(tc, pc, os.path.join(OUT, "play09_align.svg"))

    chance = cfg["align"]["chancePermille"]
    print(f"seeds={len(seeds)}  随机基线={chance}‰（>500‰ 同向，<500‰ 反向）")
    print(f"{'配置':<18}{'同序率':>7}{'同序对':>7}{'非并列对':>9}")
    for r in rows_out:
        print(f"{r['label']:<18}{r['concordance']:>7}{r['concordant']:>7}{r['pairs']:>9}")
    print("\nbaseline 平均曲线（回合:张力/行动数）")
    print("  " + " ".join(f"{i + 1}:{t}/{p}" for i, (t, p) in enumerate(zip(tc, pc))))

    elapsed_ms = (time.perf_counter_ns() - t0) // 1000000
    report = {"id": "gd_play_09_align", "seeds": len(seeds), "chancePermille": chance,
              "variants": rows_out, "baselineTension": tc, "baselinePlays": pc,
              "elapsedMs": elapsed_ms}
    os.makedirs(OUT, exist_ok=True)
    with open(os.path.join(OUT, "play09_align.json"), "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2, sort_keys=True)
        f.write("\n")
    ET.parse(os.path.join(OUT, "play09_align.svg"))
    print(f"\nelapsed={elapsed_ms} ms  svg={size} bytes")
    print("wrote out/play09_align.json out/play09_align.svg")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

seeds=32  随机基线=500‰（>500‰ 同向，<500‰ 反向）
配置                    同序率    同序对     非并列对
baseline em=3         290     32      110
baseline em=4         205     14       68
baseline em=6           0      0       43
backloaded em=3       810    180      222

baseline 平均曲线（回合:张力/行动数）
  1:171/2 2:374/2 3:558/2 4:592/1 5:650/1

elapsed=10 ms  svg=613 bytes
wrote out/play09_align.json out/play09_align.svg