101 lines
4.0 KiB
Python
101 lines
4.0 KiB
Python
"""Overlap check between a new training task and the eval tasks (spec text and object names).
|
|
|
|
A training task is rejected when its spec or its object names are too close to an eval task.
|
|
"""
|
|
import glob
|
|
import json
|
|
import math
|
|
import os
|
|
import re
|
|
from collections import Counter
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
STOP = set("the a an of to in is are be and or for with that this it as on by at from not no if then each "
|
|
"all any must shall can will when which its into than one two use used using value values "
|
|
"object objects method methods class classes table tables field fields return returns".split())
|
|
SPEC_LIMIT = 0.75 # cosine of word counts of the whole spec (same-category specs share boilerplate: 0.63-0.67 seen)
|
|
GOAL_LIMIT = 0.60 # cosine of the Goal and Business rules sections
|
|
NAME_LIMIT = 0.60 # Jaccard of name tokens (placeholder and Z prefix removed)
|
|
|
|
|
|
def words(text):
|
|
text = re.sub(r"\{\{[Pp]\}\}", " ", text)
|
|
return [w for w in re.findall(r"[a-z][a-z0-9]{2,}", text.lower()) if w not in STOP]
|
|
|
|
|
|
def cosine(a, b):
|
|
ca, cb = Counter(a), Counter(b)
|
|
dot = sum(ca[k] * cb[k] for k in ca if k in cb)
|
|
na = math.sqrt(sum(v * v for v in ca.values()))
|
|
nb = math.sqrt(sum(v * v for v in cb.values()))
|
|
return dot / (na * nb) if na and nb else 0.0
|
|
|
|
|
|
def section(spec, start, stop=None):
|
|
m = re.search(rf"(?:^|\n)#*\s*\d*\.?\s*{start}.*?\n(.*?)(?=\n#*\s*\d*\.?\s*(?:{stop})|\Z)" if stop
|
|
else rf"(?:^|\n)#*\s*\d*\.?\s*{start}.*?\n(.*)", spec, re.S | re.I)
|
|
return m.group(1) if m else ""
|
|
|
|
|
|
def core_text(spec):
|
|
return section(spec, "Goal", "Open questions|Context") + "\n" + section(spec, "Business rules", "Constraints")
|
|
|
|
|
|
def name_tokens(name):
|
|
n = re.sub(r"\{\{[Pp]\}\}", "", name).lower()
|
|
n = re.sub(r"^(?:zcl_|zif_|zcx_|z|/\w+/)", "", n)
|
|
return {t for t in re.split(r"[_\W]+", n) if len(t) > 2 and t not in ("test", "tab", "cls", "intf")}
|
|
|
|
|
|
def task_names(task):
|
|
return [o["name"] for k in ("contract", "seed", "reference") for o in task.get(k, []) if o.get("name")]
|
|
|
|
|
|
def load_task(d):
|
|
task = json.load(open(os.path.join(d, "task.json")))
|
|
spec = open(os.path.join(d, "spec.md")).read()
|
|
return {"id": task.get("id", os.path.basename(d)), "spec": words(spec), "core": words(core_text(spec)),
|
|
"names": [name_tokens(n) for n in task_names(task)],
|
|
"main": [name_tokens(o["name"]) for o in task.get("contract", []) if o.get("name")]}
|
|
|
|
|
|
def load_pool(pool):
|
|
out = []
|
|
for d in sorted(glob.glob(os.path.join(ROOT, "tasks_gen", pool, "G*")) + glob.glob(os.path.join(ROOT, "tasks", "T*"))):
|
|
try:
|
|
out.append(load_task(d))
|
|
except (OSError, ValueError):
|
|
pass
|
|
return out
|
|
|
|
|
|
def jaccard(a, b):
|
|
return len(a & b) / len(a | b) if a and b else 0.0
|
|
|
|
|
|
def compare(x, y):
|
|
"""Scores of two loaded tasks: spec cosine, core cosine, best name Jaccard (contract names)."""
|
|
name = max([jaccard(a, b) for a in x["main"] for b in y["main"]] or [0.0])
|
|
return {"spec": cosine(x["spec"], y["spec"]), "core": cosine(x["core"], y["core"]), "name": name}
|
|
|
|
|
|
def check(task_dir_or_loaded, pool_tasks):
|
|
"""Return a list of problems (empty = no overlap) against a list of loaded tasks."""
|
|
x = task_dir_or_loaded if isinstance(task_dir_or_loaded, dict) else load_task(task_dir_or_loaded)
|
|
problems = []
|
|
for y in pool_tasks:
|
|
if y["id"] == x["id"]:
|
|
continue
|
|
s = compare(x, y)
|
|
if s["spec"] >= SPEC_LIMIT or s["core"] >= GOAL_LIMIT or s["name"] >= NAME_LIMIT:
|
|
problems.append((y["id"], s))
|
|
return problems
|
|
|
|
|
|
def load_bundle(b):
|
|
"""Loaded form of a generated bundle ({"task", "files"}) before it is written to disk."""
|
|
task, spec = b["task"], b["files"].get("spec.md", "")
|
|
return {"id": task.get("id", "(new)"), "spec": words(spec), "core": words(core_text(spec)),
|
|
"names": [name_tokens(n) for n in task_names(task)],
|
|
"main": [name_tokens(o["name"]) for o in task.get("contract", []) if o.get("name")]}
|