168 lines
9 KiB
Python
168 lines
9 KiB
Python
#!/usr/bin/env python3
|
||
"""Полигон, пакет-8 (замер 3): сетка kwic × подача — счёт ИЗ СЫРЬЯ.
|
||
|
||
Мера — совпадение консолидированного dst с ПОДПИСЬЮ владельца. Нормализация обеих сторон
|
||
симметричная и объявлена в §0.4: регистр, ё/е, пробелы, дефисы. Разбор ответа — боевым
|
||
`terminology.ParseReply` через `termharness -mode parse`.
|
||
|
||
Правило чтения разрыва (куплено фазой B′ пакета-7 за $0.024): разрыв между точками засчитывается,
|
||
только если он больше ВНУТРИТОЧЕЧНОГО размаха по повторам. Поэтому печатаются обе величины, и ни
|
||
один вывод не строится на одной точке.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import glob
|
||
import json
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
from collections import defaultdict
|
||
from pathlib import Path
|
||
|
||
|
||
def norm_dst(s: str) -> str:
|
||
s = (s or "").strip().lower().replace("ё", "е")
|
||
s = re.sub(r"[‐-―−]", "-", s)
|
||
s = re.sub(r"\s+", " ", s)
|
||
return s.strip(" .,:;!?«»\"'()")
|
||
|
||
|
||
def parse_reply(harness: Path, reply: str, keys: list[str]) -> dict:
|
||
with tempfile.TemporaryDirectory() as td:
|
||
rp, kp, op = Path(td) / "r", Path(td) / "k", Path(td) / "o"
|
||
rp.write_text(reply, encoding="utf-8")
|
||
kp.write_text(json.dumps(keys, ensure_ascii=False), encoding="utf-8")
|
||
subprocess.run([str(harness), "-mode", "parse", "-reply", str(rp), "-keys", str(kp),
|
||
"-out", str(op)], check=True, capture_output=True)
|
||
return json.loads(op.read_text(encoding="utf-8"))
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--raw-dir", required=True, type=Path)
|
||
ap.add_argument("--sample", required=True, type=Path, help="JSON выборки [{src,dst,type,occ}]")
|
||
ap.add_argument("--harness", required=True, type=Path)
|
||
ap.add_argument("--out", type=Path)
|
||
a = ap.parse_args()
|
||
|
||
sample = json.loads(a.sample.read_text(encoding="utf-8"))
|
||
gold = {s["src"]: s["dst"] for s in sample}
|
||
|
||
# tag → rep → {key: dst|""|None}
|
||
got: dict[str, dict[int, dict]] = defaultdict(lambda: defaultdict(dict))
|
||
money: dict[str, dict] = defaultdict(lambda: {"usd": 0.0, "in": 0, "out": 0, "cached": 0, "calls": 0})
|
||
errors = []
|
||
for f in sorted(glob.glob(str(a.raw_dir / "*.json"))):
|
||
d = json.loads(Path(f).read_text(encoding="utf-8"))
|
||
if "error" in d:
|
||
errors.append((Path(f).name, d["error"]))
|
||
continue
|
||
tag, rep = d["tag"], d["rep"]
|
||
g = parse_reply(a.harness, d["response"], d["keys"])
|
||
for k, v in g["answered"].items():
|
||
got[tag][rep][k] = v
|
||
for k in g["declined"]:
|
||
got[tag][rep][k] = ""
|
||
m = money[tag]
|
||
m["usd"] += d.get("usd", 0.0)
|
||
u = d.get("usage") or {}
|
||
m["in"] += u.get("in", 0); m["out"] += u.get("out", 0); m["cached"] += u.get("cached", 0)
|
||
m["calls"] += 1
|
||
|
||
if errors:
|
||
print(f"ВЫЗОВОВ С ОШИБКОЙ: {len(errors)} — {errors[:3]}", file=sys.stderr)
|
||
|
||
# ключи = нормализованные src; сид даёт src, ключ строим тем же харнессом
|
||
p = subprocess.run([str(a.harness), "-mode", "norm"], input="\n".join(sorted(gold)),
|
||
capture_output=True, text=True, check=True)
|
||
key_of = {l.split("\t")[0]: l.split("\t")[1] for l in p.stdout.split("\n") if l}
|
||
gold_by_key = {key_of[s]: d for s, d in gold.items()}
|
||
src_by_key = {key_of[s]: s for s in gold}
|
||
|
||
def sort_key(t: str):
|
||
m = re.fullmatch(r"g(\d+)x(\d+)", t)
|
||
return (0, int(m.group(1)), int(m.group(2))) if m else (1, 0, 0) # именные армы — в конец
|
||
|
||
order = sorted(got, key=sort_key)
|
||
print(f"выборка: {len(gold)} подписанных термов; точек сетки: {len(order)}\n")
|
||
header = f"{'точка':8s} {'вызовов':>7s} {'$':>9s} {'точн r0':>8s} {'r1':>6s} {'r2':>6s} {'размах':>7s} {'отказ':>6s} {'молч':>5s}"
|
||
print(header)
|
||
print("-" * len(header))
|
||
summary = {}
|
||
for tag in order:
|
||
accs, decl, sil = [], [], []
|
||
for rep in sorted(got[tag]):
|
||
g = got[tag][rep]
|
||
hit = sum(1 for k, d in gold_by_key.items() if k in g and g[k] and norm_dst(g[k]) == norm_dst(d))
|
||
accs.append(hit / len(gold_by_key))
|
||
decl.append(sum(1 for k in gold_by_key if g.get(k) == ""))
|
||
sil.append(sum(1 for k in gold_by_key if k not in g))
|
||
m = money[tag]
|
||
span = max(accs) - min(accs)
|
||
cells = [f"{x:.3f}" for x in accs] + ["—"] * (3 - len(accs))
|
||
print(f"{tag:8s} {m['calls']:7d} {m['usd']:9.5f} {cells[0]:>8s} {cells[1]:>6s} {cells[2]:>6s} "
|
||
f"{span:7.3f} {sum(decl)/len(decl):6.1f} {sum(sil)/len(sil):5.1f}")
|
||
summary[tag] = {"acc": accs, "span": span, "usd": m["usd"], "calls": m["calls"],
|
||
"in": m["in"], "out": m["out"], "cached": m["cached"],
|
||
"declined": decl, "silent": sil}
|
||
|
||
if summary:
|
||
spans = [v["span"] for v in summary.values()]
|
||
within = max(spans)
|
||
print(f"\nмаксимальный ВНУТРИТОЧЕЧНЫЙ размах: {within:.3f} — разрыв между точками "
|
||
f"засчитывается только если он БОЛЬШЕ этой величины")
|
||
means = {t: sum(v["acc"]) / len(v["acc"]) for t, v in summary.items()}
|
||
best = max(means, key=lambda t: means[t])
|
||
base = "g3x40"
|
||
if base in means:
|
||
print(f"боевой дефолт {base}: средняя точность {means[base]:.3f}; лучшая точка {best}: {means[best]:.3f}; "
|
||
f"разница {means[best]-means[base]:+.3f}")
|
||
print("\nсредняя точность по точкам:", ", ".join(f"{t}={means[t]:.3f}" for t in order))
|
||
|
||
# развилка владельца: широкая подача × узкий контекст vs узкая × широкий
|
||
if "g8x40" in means and "g3x120" in means:
|
||
d = means["g8x40"] - means["g3x120"]
|
||
verdict = "РАЗЛИЧИМЫ" if abs(d) > within else "НЕРАЗЛИЧИМЫ на этой выборке"
|
||
print(f"\nразвилка 8×40 (много узких) vs 3×120 (мало широких): {means['g8x40']:.3f} vs "
|
||
f"{means['g3x120']:.3f}, разница {d:+.3f} при размахе {within:.3f} → {verdict}")
|
||
|
||
# Позиции, на которых точки РАСХОДЯТСЯ (норма D39.46). Считается ПОКАЖДОМУ повтору отдельно:
|
||
# подмножество, определённое по одному прогону, само есть выбор по случайности сэмплирования,
|
||
# и точность на нём тогда завышается у той точки, чей ответ в этот прогон был удачным.
|
||
reps = sorted(set.intersection(*[set(got[t]) for t in order]))
|
||
div_acc: dict[str, list[float]] = {t: [] for t in order}
|
||
div_sizes = []
|
||
for rep in reps:
|
||
div = [k for k in gold_by_key
|
||
if len({norm_dst(got[t][rep].get(k) or "") for t in order}) > 1]
|
||
div_sizes.append(len(div))
|
||
for t in order:
|
||
g = got[t][rep]
|
||
hit = sum(1 for k in div if g.get(k) and norm_dst(g[k]) == norm_dst(gold_by_key[k]))
|
||
div_acc[t].append(hit / len(div) if div else float("nan"))
|
||
print(f"\nрасходящихся позиций по повторам: {div_sizes} из {len(gold_by_key)}")
|
||
for t in order:
|
||
v = div_acc[t]
|
||
print(f" {t:9s} " + " ".join(f"r{r}={x:.3f}" for r, x in zip(reps, v))
|
||
+ f" среднее {sum(v)/len(v):.3f} размах {max(v)-min(v):.3f}")
|
||
within_div = max(max(v) - min(v) for v in div_acc.values())
|
||
print(f" максимальный внутриточечный размах на расходящихся: {within_div:.3f}")
|
||
summary["_divergent"] = {t: div_acc[t] for t in order} | {"sizes": div_sizes, "within": within_div}
|
||
|
||
rep0 = reps[0]
|
||
div0 = [k for k in gold_by_key if len({norm_dst(got[t][rep0].get(k) or "") for t in order}) > 1]
|
||
print("\n сами расходящиеся позиции (прогон 0):")
|
||
for k in div0:
|
||
row = " | ".join(f"{t[1:]}:{(got[t][rep0].get(k) or '—')}" for t in order)
|
||
print(f" {src_by_key[k]} (подпись: {gold_by_key[k]}) → {row}")
|
||
|
||
if a.out:
|
||
a.out.write_text(json.dumps({"summary": summary, "errors": errors}, ensure_ascii=False, indent=1),
|
||
encoding="utf-8")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|