atomspace-layer-discovery / experiments / ladder / port / statsect.py

#!/usr/bin/env python3
# Statistical check of stage-C parses (see STAGE_C_CHECK.md): G0 (Linas recorded), LF (Linas fresh), LS (lgsect).
#   statsect.py <pairmi.tsv> <G0 parses-C.tsv> <LF.tsv> <LS.tsv>
import math, re, sys
from collections import Counter, defaultdict

PAIR = re.compile(r'^\(EdgeLink \(BondNode "ANY"\) \(ListLink \(WordNode "([^"]*)"\) \(WordNode "([^"]*)"\)\)\)\t\(PredicateNode "\*-Mutual Info Key-\*"\)\t\S+ (\S+)$')
BOND = re.compile(r'^\(EdgeLink \(BondNode "([^"]*)"\) \(ListLink \(WordNode "([^"]*)"\) \(WordNode "([^"]*)"\)\)\)$')

fmi = {}
for line in open(sys.argv[1]):
    m = PAIR.match(line.rstrip('\n'))
    if m: fmi[(m.group(1), m.group(2))] = float(m.group(3))

def load(path):
    parses = defaultdict(lambda: {'links': 0, 'any': 0, 'fmi': [], 'conn': []})
    sects = Counter()
    for line in open(path):
        parts = line.rstrip('\n').split('\t', 2)
        if len(parts) != 3 or not parts[0].isdigit(): continue
        key, atom = (int(parts[0]), int(parts[1])), parts[2]
        p = parses[key]
        if atom.startswith('(Section '):
            sects[atom] += 1
            p['conn'].append(atom.count('(Connector '))
            continue
        m = BOND.match(atom)
        if not m: continue
        p['links'] += 1
        if m.group(1) == 'ANY': p['any'] += 1
        else: p['fmi'].append(fmi.get((m.group(2), m.group(3)), float('nan')))
    return parses, sects

def metrics(parses):
    ps = list(parses.values())
    links = sum(p['links'] for p in ps)
    f = [x for p in ps for x in p['fmi'] if not math.isnan(x)]
    c = [x for p in ps for x in p['conn']]
    return {
        'parses': len(ps), 'links/parse': links / len(ps), 'ANY share': sum(p['any'] for p in ps) / links,
        'mean FMI (non-ANY)': sum(f) / len(f), 'connectors/section': sum(c) / len(c),
    }

def cosine(a, b):
    dot = sum(a[k] * b[k] for k in a if k in b)
    return dot / math.sqrt(sum(v * v for v in a.values()) * sum(v * v for v in b.values()))

runs = {name: load(path) for name, path in zip(('G0', 'LF', 'LS'), sys.argv[2:5])}
M = {k: metrics(v[0]) for k, v in runs.items()}
ok = True
print(f'{"metric":22s} {"G0":>10s} {"LF":>10s} {"LS":>10s}   |LF-G0|   |LS-G0|  pass')
for m in M['G0']:
    g, lf, ls = M['G0'][m], M['LF'][m], M['LS'][m]
    if m == 'parses':
        print(f'{m:22s} {g:10d} {lf:10d} {ls:10d}'); continue
    p = abs(ls - g) <= 2 * abs(lf - g) + 0.01 * abs(g)
    ok &= p
    print(f'{m:22s} {g:10.4f} {lf:10.4f} {ls:10.4f} {abs(lf - g):9.4f} {abs(ls - g):9.4f}  {"PASS" if p else "FAIL"}')
cf, cs = cosine(runs['LF'][1], runs['G0'][1]), cosine(runs['LS'][1], runs['G0'][1])
p = cs >= cf - 0.02
ok &= p
print(f'section-type cosine to G0: LF {cf:.4f}  LS {cs:.4f}  {"PASS" if p else "FAIL"}  (LF vs LS {cosine(runs["LF"][1], runs["LS"][1]):.4f})')
print('STAGE C STATISTICAL CHECK:', 'PASS' if ok else 'FAIL')
sys.exit(0 if ok else 1)