238 lines
8.1 KiB
Python
238 lines
8.1 KiB
Python
"""Ensemble grammar inference — run multiple algorithms, pick best by scoring."""
|
|
|
|
from .crx import CRX
|
|
from .idregex import idregex
|
|
from .grammar import alphabet, match as grammar_match, Empty, Epsilon
|
|
from .mdl import score_grammar
|
|
|
|
|
|
def _matches(grammar, sequence):
|
|
"""Check if a sequence matches the grammar."""
|
|
if grammar is None or isinstance(grammar, (Empty, Epsilon)):
|
|
return not sequence if isinstance(grammar, Epsilon) else False
|
|
try:
|
|
return grammar_match(grammar, sequence)
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _fit_score(grammar, seq):
|
|
"""Score how tightly a sequence fits the grammar core."""
|
|
if not seq:
|
|
return 0.0
|
|
if grammar is None or isinstance(grammar, Empty):
|
|
return 0.0
|
|
try:
|
|
if not grammar_match(grammar, seq):
|
|
return 0.0
|
|
alpha = alphabet(grammar)
|
|
if not alpha:
|
|
return 0.5
|
|
unique_syms = len(set(seq))
|
|
total_syms = len(seq)
|
|
return max(0.0, 1.0 - (total_syms - unique_syms) / max(total_syms, 1))
|
|
except Exception:
|
|
return 0.0
|
|
|
|
|
|
def _symbol_rarity_score(seq, all_sequences):
|
|
"""Score a sequence by how rare its symbols are across the dataset.
|
|
1.0 = all symbols are common, 0.0 = mostly rare symbols.
|
|
"""
|
|
from collections import Counter
|
|
all_syms = Counter()
|
|
for s in all_sequences:
|
|
all_syms.update(s)
|
|
n = len(all_sequences)
|
|
scores = []
|
|
for sym in seq:
|
|
freq = all_syms.get(sym, 0) / n
|
|
scores.append(min(freq, 1.0))
|
|
return sum(scores) / len(scores) if scores else 0.0
|
|
|
|
|
|
def _find_core(sequences, min_coverage=0.8):
|
|
"""Find the core subset of sequences by iterative CRX + outlier removal."""
|
|
if not sequences or min_coverage >= 1.0:
|
|
crx_g = CRX().infer(sequences)
|
|
return crx_g, sequences, [], []
|
|
|
|
from collections import Counter
|
|
all_syms = Counter()
|
|
for s in sequences:
|
|
all_syms.update(s)
|
|
n = len(sequences)
|
|
|
|
def _rarity(seq):
|
|
rare_count = sum(1 for sym in seq if all_syms.get(sym, 0) / n < 0.3)
|
|
return rare_count / max(len(seq), 1)
|
|
|
|
working = list(sequences)
|
|
removed_indices = []
|
|
crx = CRX()
|
|
|
|
for _ in range(50):
|
|
if len(working) < 3:
|
|
break
|
|
target = max(int(len(sequences) * min_coverage), 1)
|
|
if len(working) <= target:
|
|
break
|
|
scores = [(i, _rarity(seq)) for i, seq in enumerate(working)]
|
|
scores.sort(key=lambda x: -x[1])
|
|
if len(scores) < 2 or scores[0][1] == scores[-1][1]:
|
|
break
|
|
worst_idx = scores[0][0]
|
|
removed_indices.append(working[worst_idx])
|
|
working = [s for i, s in enumerate(working) if i != worst_idx]
|
|
|
|
core_g = crx.infer(working) if working else None
|
|
return core_g, working, removed_indices, []
|
|
|
|
|
|
def mdl_score_simple(grammar, sequences, method='langsize'):
|
|
"""Score a grammar. Default: Language Size (Bex et al., arXiv:1004.2372)."""
|
|
return score_grammar(grammar, sequences, method=method)
|
|
|
|
|
|
def _run_idregex(sequences, kmax, N, method='langsize'):
|
|
"""Run standalone iDRegEx, return (grammar, score) or (None, inf)."""
|
|
g = idregex(sequences, kmax=kmax, N=N)
|
|
if g and not isinstance(g, Empty):
|
|
return g, mdl_score_simple(g, sequences, method=method)
|
|
return None, float('inf')
|
|
|
|
|
|
_ALGO_NAMES = {
|
|
'crx': 'CRX',
|
|
'idregex': 'iDRegEx',
|
|
}
|
|
|
|
|
|
_ALGORITHMS = {
|
|
'crx': lambda s, k, n, m='langsize': (CRX().infer(s), mdl_score_simple(CRX().infer(s), s, method=m)),
|
|
'idregex': _run_idregex,
|
|
}
|
|
|
|
|
|
def _run_kore(sequences, kmax, N, method='langsize'):
|
|
"""Run kOREInference, return (grammar, score) or (None, inf)."""
|
|
from .kore import kOREInference
|
|
kore = kOREInference(k_max=kmax, N=N)
|
|
result = kore.infer(sequences)
|
|
if result:
|
|
_, expr, _ = result
|
|
if not isinstance(expr, Empty):
|
|
return expr, mdl_score_simple(expr, sequences, method=method)
|
|
return None, float('inf')
|
|
|
|
|
|
def infer_ensemble(sequences, kmax=2, N=3, prefer=None, min_coverage=1.0, include_kore=False, include_idregex=False, method='langsize'):
|
|
"""Run all applicable algorithms and return the best by scoring.
|
|
|
|
Args:
|
|
sequences: List of sequences, each a list of strings.
|
|
kmax: Maximum k for k-ORE inference (iDRegEx, kOREInference).
|
|
N: Number of random trials for k-ORE inference.
|
|
prefer: Optional — 'crx', 'idregex', or 'koreinference' to skip
|
|
ensemble and return only that algorithm's result.
|
|
min_coverage: When < 1.0, also runs CRX on the tightest core subset.
|
|
include_idregex: Run iDRegEx (slow, opt-in).
|
|
method: Scoring method — 'langsize' (default) or 'mdl' (fallback).
|
|
|
|
Returns:
|
|
dict with keys: best, all, why, core (optional)
|
|
"""
|
|
if prefer and prefer.lower() in _ALGORITHMS:
|
|
key = prefer.lower()
|
|
fn = _ALGORITHMS[key]
|
|
algo_name = _ALGO_NAMES.get(key, key)
|
|
g, score = fn(sequences, kmax, N, method)
|
|
if g and not isinstance(g, Empty):
|
|
return {
|
|
'best': {'algorithm': algo_name, 'grammar': g, 'mdl_score': round(score, 2)},
|
|
'all': [{'algorithm': algo_name, 'grammar': g, 'mdl_score': round(score, 2)}],
|
|
'why': f"Requested {algo_name} only.",
|
|
}
|
|
return {
|
|
'best': None,
|
|
'all': [],
|
|
'why': f"{algo_name} returned empty (no grammar found).",
|
|
}
|
|
|
|
results = []
|
|
|
|
crx_g = CRX().infer(sequences)
|
|
crx_score = mdl_score_simple(crx_g, sequences, method=method) if crx_g and not isinstance(crx_g, Empty) else float('inf')
|
|
results.append(('CRX', crx_g if crx_g and not isinstance(crx_g, Empty) else Empty(), crx_score))
|
|
|
|
if include_idregex:
|
|
idr_g, idr_score = _run_idregex(sequences, kmax, N, method=method)
|
|
if idr_g:
|
|
results.append(('iDRegEx', idr_g, idr_score))
|
|
|
|
if include_kore:
|
|
kore_g, kore_score = _run_kore(sequences, kmax, N, method=method)
|
|
if kore_g:
|
|
results.append(('kOREInference', kore_g, kore_score))
|
|
|
|
results = [r for r in results if r[1] and not isinstance(r[1], Empty)]
|
|
if not results:
|
|
base = {
|
|
'best': None,
|
|
'all': [],
|
|
'why': "No algorithm produced a non-empty grammar.",
|
|
}
|
|
if min_coverage < 1.0:
|
|
core_g, core_seqs, outliers, _ = _find_core(sequences, min_coverage)
|
|
base['core'] = {
|
|
'grammar': core_g,
|
|
'coverage': round(len(core_seqs) / max(len(sequences), 1), 2) if sequences else 0,
|
|
'outliers': outliers,
|
|
}
|
|
return base
|
|
|
|
results.sort(key=lambda x: x[2])
|
|
best = results[0]
|
|
all_results = [
|
|
{'algorithm': a, 'grammar': g, 'mdl_score': round(s, 2)}
|
|
for a, g, s in results
|
|
]
|
|
|
|
why_parts = []
|
|
if len(results) == 1:
|
|
why_parts.append(f"Only {results[0][0]} produced a result.")
|
|
else:
|
|
scores_str = ', '.join(f"{r[0]}={r[2]:.1f}" for r in results)
|
|
why_parts.append(f"Scores: {scores_str}.")
|
|
|
|
match_strs = []
|
|
for r_algo, r_grammar, _ in results:
|
|
if r_grammar and not isinstance(r_grammar, Empty):
|
|
m = sum(1 for s in sequences if _matches(r_grammar, s))
|
|
match_strs.append(f"{r_algo}={m}/{len(sequences)}")
|
|
if match_strs:
|
|
why_parts.append(f"Match rates: {', '.join(match_strs)}.")
|
|
|
|
why_parts.append(f"{best[0]} selected (MDL score {best[2]}).")
|
|
|
|
result = {
|
|
'best': {
|
|
'algorithm': best[0],
|
|
'grammar': best[1],
|
|
'mdl_score': round(best[2], 2),
|
|
},
|
|
'all': all_results,
|
|
'why': ' '.join(why_parts),
|
|
}
|
|
|
|
if min_coverage < 1.0:
|
|
core_g, core_seqs, outliers, _ = _find_core(sequences, min_coverage)
|
|
result['core'] = {
|
|
'grammar': core_g,
|
|
'coverage': round(len(core_seqs) / max(len(sequences), 1), 2) if sequences else 0,
|
|
'outlier_count': len(outliers),
|
|
'outliers': outliers,
|
|
}
|
|
result['why'] += f' Core CRX ({min_coverage:.0%} coverage, {len(outliers)} outliers): {core_g}'
|
|
|
|
return result
|