tobi/grammar-inference-engine
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

wip: AST foundation — grammar.py, expr.py→AST, soa.py→AST labels

Browse source
This commit is contained in:
tobjend 2026-07-12 20:59:53 +02:00
parent 9d206f586f
commit ea6cac53e3
26 changed files with 1423 additions and 1816 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
bex/__init__.py
View file

@ -19,7 +19,8 @@ from .rwrsq import rwr_sq
from .idregex import idregex
from .kore import kOREInference, validate_k_ore
from .koa import KOA, build_complete_koa
from .expr import concat, disj, star, optional, alphabet, strip_k
from .expr import concat, disj, star, optional, alphabet
from .koa import strip_k
from .marking import mark_koa
from .tokenizer import YAMLTokenizer
from .ensemble import infer_ensemble

6
bex/baum_welch.py
View file

@ -37,7 +37,8 @@ def init_probabilities(G, sequences):
v = empty_count / total
else:
lab = G.label(t)
base = lab.rsplit('_', 1)[0] if '_' in lab else lab
lab_val = lab.value if hasattr(lab, 'value') else str(lab)
base = lab_val.rsplit('_', 1)[0] if '_' in lab_val else lab_val
count = start_counts.get(base, 0)
copies = sum(1 for u in succ if G.label(u) == lab)
v = (count / total) / max(copies, 1)
@ -75,7 +76,8 @@ def bw_iteration(prob, sequences, node_to_idx, n_states, all_nodes, G):
for n in all_nodes:
lab = G.label(n)
if lab:
base = lab.rsplit('_', 1)[0] if '_' in lab else lab
lab_val = lab.value if hasattr(lab, 'value') else str(lab)
base = lab_val.rsplit('_', 1)[0] if '_' in lab_val else lab_val
emit.setdefault(base, []).append(n)
# sink emits nothing
sink = G.sink

14
bex/cli.py
View file

@ -16,6 +16,8 @@ from .tokenizer import YAMLTokenizer
from .kore import kOREInference
from .template import generate_template
from .ilocal import iLocal, extract_contexts_from_file, reduce_contexts
from .grammar import Empty
from .ensemble import infer_ensemble
def find_yaml_files(directory):
@ -115,16 +117,20 @@ def main():
kore = kOREInference(k_max=args.k_max)
if args.crx:
result = kore.infer_with_crx(all_sequences)
_, expr, method = result
print(f" Method: {method}", file=sys.stderr)
result = infer_ensemble(all_sequences, method='crx')
if result['best'] is not None:
expr = result['best']['grammar']
print(f" Method: {result['best']['algorithm']}", file=sys.stderr)
else:
expr = Empty()
print(" Kein Ergebnis", file=sys.stderr)
else:
result = kore.infer(all_sequences)
if result:
_, expr, k = result
print(f" Bestes k: {k}", file=sys.stderr)
else:
expr = "∅"
expr = Empty()
print(" Kein Ergebnis", file=sys.stderr)
print(f" Inferred expression: {expr}", file=sys.stderr)

58
bex/crx.py
View file

@ -1,41 +1,39 @@
"""CRX — Direct CHARE inference (Algorithm 7, TODS 2010)."""
"""CRX — Direct CHARE inference (Algorithm 7, TODS 2010).
Produces AST nodes directly — no SORE string intermediate.
"""
from collections import defaultdict
from .expr import concat
from .grammar import Symbol, Concat, Alt, Plus, Optional, Star, Epsilon, Empty
class CRX:
"""
|———— Algorithm 7: CRX ————|
Input: sample S (list of token lists)
Output: CHARE r such that S ⊆ L(r)
Output: AST node r such that S ⊆ L(r)
"""
def infer(self, sequences):
S = [list(s) for s in sequences if s]
if not S:
return 'ε'
return Epsilon()
sigma = set()
for w in S:
for a in w:
sigma.add(a)
if not sigma:
return 'ε'
return Epsilon()
# Step 1: Compute ImmedPred and equivalence classes ≈_S
immed = set()
for w in S:
for i in range(len(w) - 1):
immed.add((w[i], w[i + 1]))
# Reachability: →_S (reflexive, transitive closure)
closure = self._transitive_closure(sigma, immed)
# Equivalence: a ≈_S b iff a →*_S b and b →*_S a
eq = self._equivalence(sigma, closure)
# Build class map: symbol → class index
sym_to_cls = {}
classes = []
for cls_syms in eq:
@ -44,23 +42,11 @@ class CRX:
sym_to_cls[sym] = idx
classes.append(set(cls_syms))
# Step 2-3: Preserve only singleton nodes? No, the algorithm says merge singletons
# that share Pred/Succ in the Hasse diagram. But actually, looking at the algorithm
# more carefully:
#
# "while a maximal set of singleton nodes γ₁,...,γ_ℓ such that
# Pred_HS(γ₁)=···=Pred_HS(γ_ℓ) and Succ_HS(γ₁)=···=Succ_HS(γ_ℓ) exists do
# Replace γ₁,...,γ_ℓ by γ := ∪ⱼ γⱼ"
#
# This merges singleton equivalence classes (classes with exactly one symbol)
# that have the same Pred and Succ sets in the Hasse diagram.
changed = True
while changed:
changed = False
singleton_ids = [i for i, c in enumerate(classes) if len(c) == 1]
# Compute Pred and Succ for each singleton (considering ALL symbols in each class)
hs_pred = {}
hs_succ = {}
for i in singleton_ids:
@ -75,7 +61,6 @@ class CRX:
if any((sym_i, sym_j) in immed for sym_j in c):
hs_succ[i].add(j)
# Group by same (Pred, Succ)
groups = defaultdict(list)
for i in singleton_ids:
groups[(frozenset(hs_pred[i]), frozenset(hs_succ[i]))].append(i)
@ -92,13 +77,11 @@ class CRX:
changed = True
break
# After merging, rebuild sym_to_cls to map to new class indices
sym_to_cls = {}
for idx, cls in enumerate(classes):
for sym in cls:
sym_to_cls[sym] = idx
# Step 5: Topological sort of the Hasse diagram
adj = {i: set() for i in range(len(classes))}
indeg = {i: 0 for i in range(len(classes))}
for a, b in immed:
@ -108,7 +91,6 @@ class CRX:
adj[ca].add(cb)
indeg[cb] += 1
# Topological sort (Kahn's algorithm)
order = []
q = [i for i in range(len(classes)) if indeg[i] == 0]
while q:
@ -121,7 +103,6 @@ class CRX:
remaining = set(range(len(classes))) - set(order)
order.extend(remaining)
# Step 6-16: Assign chain factors (Algorithm 7 lines 7-14)
def count_in_class(w, syms):
return sum(1 for a in w if a in syms)
@ -136,27 +117,27 @@ class CRX:
some_two_or_more = any(c >= 2 for c in counts)
sym_list = sorted(syms)
factor = '+'.join(sym_list)
if len(sym_list) > 1:
factor = '(' + factor + ')'
alt_node = Alt([Symbol(s) for s in sym_list])
else:
alt_node = Symbol(sym_list[0])
if all_exactly_one:
pass # (a₁+···+aₙ)
parts.append(alt_node)
elif all_at_most_one:
factor += '?' # (a₁+···+aₙ)?
parts.append(Optional(alt_node))
elif all_at_least_one and some_two_or_more:
factor += '+' # (a₁+···+aₙ)+
parts.append(Plus(alt_node))
else:
factor += '+?' # (a₁+···+aₙ)+?
parts.append(factor)
parts.append(Plus(Optional(alt_node)))
if not parts:
return 'ε'
return '.'.join(parts)
return Epsilon()
if len(parts) == 1:
return parts[0]
return Concat(parts)
def _transitive_closure(self, sigma, immed):
"""Compute reflexive, transitive closure of immed relation."""
closure = {(a, b) for (a, b) in immed}
for a in sigma:
closure.add((a, a))
@ -172,7 +153,6 @@ class CRX:
return closure
def _equivalence(self, sigma, closure):
"""Compute equivalence classes of ≈_S."""
remaining = set(sigma)
classes = []
while remaining:

121
bex/crx_refined.py
View file

@ -1,43 +1,16 @@
"""CRX Refined — Cluster-then-infer CRX with over-approximation detection.
Standard CRX produces over-approximated grammars when the Hasse diagram is
non-linear (branching). This module fixes that by:
1. Clustering similar sequences before inference
2. Inferring per-cluster (tighter grammars)
3. Reporting confidence metrics
Key insight: code call sequences have branching patterns that CHAREs
(linear chain expressions) can't represent. Clustering reduces branching
within each group, making CRX's linear assumption more valid.
Tradeoff: cluster granularity.
- Too coarse → over-approximation (standard CRX)
- Too fine → no generalization (one grammar per sequence)
- Sweet spot → cluster by structural features
"""
"""CRX Refined — Cluster-then-infer CRX with over-approximation detection."""
from collections import defaultdict
from .crx import CRX
from .grammar import Epsilon
def _cluster_by_structure(sequences):
"""Cluster sequences by structural features.
Uses (first_symbol, last_symbol, length_bucket) as the cluster key.
This groups sequences that start and end the same way, which
typically means they follow the same calling pattern.
Args:
sequences: list of token lists
Returns:
dict mapping cluster_key → list of sequences
"""
"""Cluster sequences by structural features."""
groups = defaultdict(list)
for seq in sequences:
if not seq:
continue
# Length bucket: short (1-3), medium (4-8), long (9+)
length = len(seq)
if length <= 3:
length_bucket = 'short'
@ -51,92 +24,58 @@ def _cluster_by_structure(sequences):
def crx_refined(sequences, min_cluster=2):
"""Cluster-then-infer CRX.
Clusters sequences by (first, last, length), infers CRX per cluster,
and returns the grammar from the largest cluster. Falls back to standard
CRX if no cluster has enough sequences.
Args:
sequences: list of token lists
min_cluster: minimum cluster size to infer from (default: 2)
Returns:
CHARE expression string
"""
"""Cluster-then-infer CRX."""
S = [list(s) for s in sequences if s]
if not S:
return 'ε'
return Epsilon()
clusters = _cluster_by_structure(S)
# Find clusters large enough to infer from
valid_clusters = {k: v for k, v in clusters.items() if len(v) >= min_cluster}
if not valid_clusters:
# No cluster large enough — fall back to standard CRX
return CRX().infer(S)
# Pick the largest cluster
best_key = max(valid_clusters, key=lambda k: len(valid_clusters[k]))
best_cluster = valid_clusters[best_key]
return CRX().infer(best_cluster)
def crx_with_confidence(sequences, min_cluster=2):
"""Cluster-then-infer CRX with confidence metrics.
Returns:
dict with keys:
grammar: str — the refined grammar
confidence: float — fraction of training pairs captured by grammar structure
n_clusters: int — number of clusters
largest_cluster: int — size of largest cluster
n_sequences: int — total sequences
"""
"""Cluster-then-infer CRX with confidence metrics."""
S = [list(s) for s in sequences if s]
if not S:
return {
'grammar': 'ε',
'grammar': Epsilon(),
'confidence': 1.0,
'n_clusters': 0,
'largest_cluster': 0,
'n_sequences': 0,
}
clusters = _cluster_by_structure(S)
valid_clusters = {k: v for k, v in clusters.items() if len(v) >= min_cluster}
if valid_clusters:
if not valid_clusters:
g = CRX().infer(S)
return {
'grammar': g,
'confidence': 0.5,
'n_clusters': len(clusters),
'largest_cluster': max((len(v) for v in clusters.values()), default=0),
'n_sequences': len(S),
}
best_key = max(valid_clusters, key=lambda k: len(valid_clusters[k]))
best_cluster = valid_clusters[best_key]
grammar = CRX().infer(best_cluster)
else:
best_cluster = S
grammar = CRX().infer(S)
# Compute confidence: what fraction of consecutive pairs in the training
# data are "captured" by the grammar's structure
all_pairs = set()
for w in S:
for i in range(len(w) - 1):
all_pairs.add((w[i], w[i + 1]))
# Pairs in the best cluster that appear in grammar structure
cluster_pairs = set()
for w in best_cluster:
for i in range(len(w) - 1):
cluster_pairs.add((w[i], w[i + 1]))
# How many training pairs are cluster pairs? (coverage)
captured = len(all_pairs & cluster_pairs)
confidence = captured / len(all_pairs) if all_pairs else 1.0
g = CRX().infer(best_cluster)
captured = sum(1 for s in S if any(_matches_cluster(g, s) for c in valid_clusters.values() for s in c))
confidence = captured / max(len(S), 1)
return {
'grammar': grammar,
'confidence': confidence,
'n_clusters': len(clusters),
'grammar': g,
'confidence': round(confidence, 3),
'n_clusters': len(valid_clusters),
'largest_cluster': len(best_cluster),
'n_sequences': len(S),
}
def _matches_cluster(grammar, seq):
from .grammar import match as grammar_match
try:
return grammar_match(grammar, seq)
except Exception:
return False

342
bex/ensemble.py
View file

@ -1,290 +1,36 @@
"""Ensemble grammar inference — run multiple algorithms, pick best by MDL scoring."""
"""Ensemble grammar inference — run multiple algorithms, pick best by scoring."""
import re
from .crx import CRX
from .idregex import idregex
from .expr import alphabet
from .mdl import model_cost, mdl_score, lang_size_score, score_grammar
def _parse_parts(expr):
"""Parse expression into a list of tokens for matching.
Each token: (type, value, quantifier)
type: 'symbol' | 'disj' | 'concat' | 'empty'
quantifier: '' | '?' | '+' | '+?'
"""
if not expr or expr == '∅':
return [('empty', '', '')]
if expr == 'ε':
return [('empty', '', '+?')]
# 1. Check if it's a concatenation (split outermost by '.')
# Must check BEFORE stripping trailing quantifier, because
# quantifiers belong to individual parts (e.g., a?.b+)
concat_parts = _split_outer(expr.strip(), '.')
if len(concat_parts) > 1:
children = []
for p in concat_parts:
children.extend(_parse_parts(p.strip()))
return [('concat', children, '')]
# 2. Now handle quantifier suffix on this single part
quantifier = ''
if expr.endswith('+?'):
quantifier = '+?'
expr = expr[:-2]
elif expr.endswith('*'):
quantifier = '*'
expr = expr[:-1]
elif expr.endswith('?'):
quantifier = '?'
expr = expr[:-1]
elif expr.endswith('+'):
quantifier = '+'
expr = expr[:-1]
# 3. Disjunction group: (a+b+c) for CRX or (a|b|c) for iDRegEx
if expr.startswith('(') and expr.endswith(')'):
inner = expr[1:-1]
# Try CRX-style (+) first, then iDRegEx-style (|)
disj_parts = _split_outer(inner, '+')
if len(disj_parts) <= 1:
disj_parts = _split_outer(inner, '|')
if len(disj_parts) > 1:
children = []
for p in disj_parts:
p = p.strip()
# Parse as a flat symbol (don't split dots — they're part of
# the symbol name, e.g. "community.docker.docker_image")
children.append(_parse_flat_symbol(p))
return [('disj', children, quantifier)]
# Single element inside parens: treat as flat symbol
return [_parse_flat_symbol(inner)]
# 4. Single symbol
if expr and expr not in ('∅', 'ε'):
return [('symbol', expr, quantifier)]
return []
def _parse_flat_symbol(s):
"""Parse a single symbol with optional quantifier, no dot splitting.
Unlike _parse_parts, this treats dots as part of the symbol name
(e.g. 'community.docker.docker_image' stays as one symbol).
"""
s = s.strip()
quantifier = ''
if s.endswith('+?'):
quantifier = '+?'
s = s[:-2]
elif s.endswith('*'):
quantifier = '*'
s = s[:-1]
elif s.endswith('?'):
quantifier = '?'
s = s[:-1]
elif s.endswith('+'):
quantifier = '+'
s = s[:-1]
if s and s not in ('∅', 'ε'):
return ('symbol', s, quantifier)
return ('empty', '', quantifier)
def _split_outer(s, sep):
"""Split on `sep` at the top level (not inside parentheses)."""
depth = 0
parts = []
cur = []
for ch in s:
if ch == '(':
depth += 1
cur.append(ch)
elif ch == ')':
depth -= 1
cur.append(ch)
elif ch == sep and depth == 0:
parts.append(''.join(cur))
cur = []
else:
cur.append(ch)
parts.append(''.join(cur))
return parts
def _match_possible(token, seq, pos):
"""Return all possible end positions after matching this token starting at pos."""
ttype, tval, tquant = token
positions = []
if ttype == 'empty':
positions.append(pos)
elif ttype == 'symbol':
if tquant in ('', '?'):
if pos < len(seq) and seq[pos] == tval:
positions.append(pos + 1)
if tquant == '?':
positions.append(pos)
elif tquant in ('+?', '*'):
positions.append(pos)
cnt = pos
while cnt < len(seq) and seq[cnt] == tval:
cnt += 1
positions.append(cnt)
elif tquant == '+':
if pos < len(seq) and seq[pos] == tval:
cnt = pos + 1
positions.append(cnt)
while cnt < len(seq) and seq[cnt] == tval:
cnt += 1
positions.append(cnt)
elif ttype == 'disj':
if tquant in ('', '?'):
for child in tval:
for ep in _match_possible(child, seq, pos):
positions.append(ep)
if tquant == '?':
positions.append(pos)
elif tquant in ('+?', '*'):
positions.append(pos)
for child in tval:
for ep in _match_possible(child, seq, pos):
if ep > pos:
positions.append(ep)
# After consuming one, recurse to try more
for ep2 in _match_possible(token, seq, ep):
if ep2 > ep:
positions.append(ep2)
elif tquant == '+':
for child in tval:
for ep in _match_possible(child, seq, pos):
if ep > pos:
positions.append(ep)
for ep2 in _match_possible(token, seq, ep):
if ep2 > ep:
positions.append(ep2)
elif ttype == 'concat':
# Match all children sequentially
def _match_seq(children, start):
cur = [start]
for child in children:
next_cur = []
for p in cur:
next_cur.extend(_match_possible(child, seq, p))
cur = next_cur
if not cur:
break
return cur
if tquant in ('', '?'):
positions.extend(_match_seq(tval, pos))
if tquant == '?':
positions.append(pos)
elif tquant in ('+?', '*'):
positions.append(pos)
inner_end = _match_seq(tval, pos)
for ep in inner_end:
if ep > pos:
positions.append(ep)
for ep2 in _match_possible(token, seq, ep):
if ep2 > ep:
positions.append(ep2)
elif tquant == '+':
inner_end = _match_seq(tval, pos)
for ep in inner_end:
if ep > pos:
positions.append(ep)
for ep2 in _match_possible(token, seq, ep):
if ep2 > ep:
positions.append(ep2)
return positions
def _match_tokens(tokens, seq, pos=0):
"""Try to match tokens against seq starting at pos. Returns max position or None."""
cur = [pos]
for token in tokens:
next_cur = []
for p in cur:
next_cur.extend(_match_possible(token, seq, p))
cur = next_cur
if not cur:
return None
return max(cur) if cur else pos
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:
tokens = _parse_parts(grammar.strip())
if not tokens:
return False
end = _match_tokens(tokens, sequence)
if end is None:
return False
return end == len(sequence)
return grammar_match(grammar, sequence)
except Exception:
return False
def _fit_score(grammar, seq):
"""Score how tightly a sequence fits: 1.0 = perfect match to core,
0.0 = mostly uses optional/repeated parts.
Instead of trying to parse the grammar structure (which is fragile),
this measures how well seq matches against the grammatical core by
comparing its symbol positions to the grammar's 'spine' — the symbols
that appear in all sequences.
"""
"""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:
# Strategy: parse grammar tokens, match seq, count what fraction
# of seq length is consumed by obligatory (non-?, non-+?) tokens.
tokens = _parse_parts(grammar.strip())
if not tokens or tokens[0][0] == 'empty':
if not grammar_match(grammar, seq):
return 0.0
def _classify_tokens(node):
"""Return (obligatory_count, optional_count) for this node."""
tt, tv, tq = node
if tt == 'symbol':
if tq in ('', '+'):
return (1, 0)
return (0, 1)
if tt == 'concat':
ob, op = 0, 0
for c in tv:
if c[0] == 'empty':
continue
o1, o2 = _classify_tokens(c)
ob += o1
op += o2
return (ob, op)
if tt == 'disj':
# Any alternative counts as optional
return (0, len(tv))
return (0, 0)
ob, op = _classify_tokens(tokens[0])
total = ob + op
if total == 0:
alpha = alphabet(grammar)
if not alpha:
return 0.5
# Match seq and see how many symbols are actually consumed
end = _match_tokens(tokens, seq)
if end is None or end != len(seq):
return 0.0
# Fit = fraction of mandatory symbols / total mandatory+optional
# Penalizes sequences that lean heavily on optional parts
return max(0.0, 1.0 - (op / total))
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
@ -306,14 +52,7 @@ def _symbol_rarity_score(seq, all_sequences):
def _find_core(sequences, min_coverage=0.8):
"""Find the core subset of sequences by iterative CRX + outlier removal.
Outlier detection uses symbol rarity: sequences with rare symbols
(appearing in few other sequences) are removed first.
Returns:
(core_grammar, core_sequences, outliers, fit_scores)
"""
"""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, [], []
@ -335,19 +74,13 @@ def _find_core(sequences, min_coverage=0.8):
for _ in range(50):
if len(working) < 3:
break
target = max(int(len(sequences) * min_coverage), 1)
if len(working) <= target:
break
# Score by rarity: most rare symbol → worst fit
scores = [(i, _rarity(seq)) for i, seq in enumerate(working)]
scores.sort(key=lambda x: -x[1]) # most rare first
# If all sequences have the same score, stop (no outliers to remove)
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]
@ -357,17 +90,14 @@ def _find_core(sequences, min_coverage=0.8):
def mdl_score_simple(grammar, sequences, method='langsize'):
"""Score a grammar. Default: Language Size (Bex et al., arXiv:1004.2372).
Lower is better. Use method='mdl' for the old MDL fallback.
"""
"""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 g != '∅':
if g and not isinstance(g, Empty):
return g, mdl_score_simple(g, sequences, method=method)
return None, float('inf')
@ -391,6 +121,7 @@ def _run_kore(sequences, kmax, N, method='langsize'):
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')
@ -404,28 +135,19 @@ def infer_ensemble(sequences, kmax=2, N=3, prefer=None, min_coverage=1.0, includ
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
of sequences. Outliers (worst-fitting) are iteratively
removed until at least this fraction remains. The core
grammar and outlier list are included in the response.
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, Bex et al. arXiv:1004.2372)
or 'mdl' (fallback).
method: Scoring method — 'langsize' (default) or 'mdl' (fallback).
Returns:
dict with keys:
best: {algorithm, grammar, mdl_score}
all: [{algorithm, grammar, mdl_score}, ...]
why: str explaining the choice
core: (optional) {grammar, coverage, outliers} — only when
min_coverage < 1.0
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 g != '∅':
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)}],
@ -434,29 +156,26 @@ def infer_ensemble(sequences, kmax=2, N=3, prefer=None, min_coverage=1.0, includ
return {
'best': None,
'all': [],
'why': f"{algo_name} returned ∅ (no grammar found).",
'why': f"{algo_name} returned empty (no grammar found).",
}
results = []
# 1. CRX (always fast, always produces a result)
crx_g = CRX().infer(sequences)
crx_score = mdl_score_simple(crx_g, sequences, method=method) if crx_g and crx_g != '∅' else float('inf')
results.append(('CRX', crx_g if crx_g and crx_g != '∅' else '∅', crx_score))
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))
# 2. iDRegEx (opt-in via include_idregex=True — slow on large groups)
if include_idregex:
idr_g, idr_score = _run_idregex(sequences, kmax, N, method=method)
if idr_g:
results.append(('iDRegEx', idr_g, idr_score))
# 3. kOREInference (opt-in via include_kore=True)
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 r[1] != '∅']
results = [r for r in results if r[1] and not isinstance(r[1], Empty)]
if not results:
base = {
'best': None,
@ -479,8 +198,6 @@ def infer_ensemble(sequences, kmax=2, N=3, prefer=None, min_coverage=1.0, includ
for a, g, s in results
]
active = {r[0] for r in results}
why_parts = []
if len(results) == 1:
why_parts.append(f"Only {results[0][0]} produced a result.")
@ -490,7 +207,7 @@ def infer_ensemble(sequences, kmax=2, N=3, prefer=None, min_coverage=1.0, includ
match_strs = []
for r_algo, r_grammar, _ in results:
if r_grammar and r_grammar != '∅':
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:
@ -508,7 +225,6 @@ def infer_ensemble(sequences, kmax=2, N=3, prefer=None, min_coverage=1.0, includ
'why': ' '.join(why_parts),
}
# Core analysis when min_coverage < 1.0
if min_coverage < 1.0:
core_g, core_seqs, outliers, _ = _find_core(sequences, min_coverage)
result['core'] = {

165
bex/expr.py
View file

@ -1,164 +1,55 @@
"""Expression utilities for SOREs and k-OREs."""
"""Expression utilities — all functions return grammar.py AST nodes."""
import re
from .grammar import Symbol, Concat, Alt, Plus, Optional, Star, Epsilon, Empty
def sym(s):
"""Create a simple symbol expression."""
return s
"""Create a symbol node."""
return Symbol(s)
def concat(*parts):
"""Create concatenation expression."""
parts = [p for p in parts if p and p != 'ε']
"""Create concatenation AST node."""
parts = [p for p in parts if p and not isinstance(p, Epsilon)]
if not parts:
return 'ε'
return Epsilon()
if len(parts) == 1:
return parts[0]
return '.'.join(parts)
return Concat(parts)
def disj(*parts):
"""Create disjunction expression."""
parts = [p for p in parts if p and p != '∅']
"""Create disjunction AST node."""
parts = [p for p in parts if p and not isinstance(p, Empty)]
if not parts:
return '∅'
return Empty()
if len(parts) == 1:
return parts[0]
return '(' + '|'.join(parts) + ')'
return Alt(parts)
def star(expr):
"""Create iteration expression (one or more, r+)."""
if not expr or expr in ('∅', 'ε'):
return expr
if len(expr) == 1 or (expr.startswith('(') and expr.endswith(')')):
return expr + '+'
return '(' + expr + ')+'
"""Create one-or-more repetition AST node (r+)."""
if not expr or isinstance(expr, Empty):
return expr or Empty()
if isinstance(expr, Epsilon):
return Epsilon()
return Plus(expr)
def optional(expr):
"""Create optional expression (r?)."""
if not expr or expr in ('∅', 'ε'):
return 'ε'
if len(expr) == 1 or (expr.startswith('(') and expr.endswith(')')):
return expr + '?'
return '(' + expr + ')?'
"""Create optional AST node (r?)."""
if not expr or isinstance(expr, Empty):
return Epsilon()
if isinstance(expr, Epsilon):
return Epsilon()
return Optional(expr)
def alphabet(expr):
"""Return set of alphabet symbols in expression."""
cleaned = re.sub(r'[+?*().|]', ' ', expr)
result = set()
for token in cleaned.split():
token = token.strip('_0123456789')
if token and token not in ('ε', '∅'):
result.add(token)
return result
def alphabet(node):
"""Return set of alphabet symbols in AST node."""
from .grammar import alphabet as _grammar_alphabet
return _grammar_alphabet(node)
def strip_k(s):
"""Remove k-ORE markers: a_1 → a, b^(2) → b."""
result = re.sub(r'_\d+', '', s)
result = re.sub(r'\^\(\d+\)', '', result)
result = re.sub(r'^\(|\)$', '', result)
return result
def has_repeats(expr, symbol):
"""Check if a symbol appears more than once in expression."""
return expr.count(symbol) > 1
def lang_size_at_most(expr, n, alphabet_symbols=None):
"""Compute |L(r)<=n| — number of words of length ≤ n in L(r)."""
if alphabet_symbols is None:
alphabet_symbols = alphabet(expr)
if not alphabet_symbols:
return 1 if 'ε' in expr else 0
size = 0
for length in range(n + 1):
size += _count_words(expr, length, alphabet_symbols)
return size
def _count_words(expr, length, alphabet_symbols):
if length < 0:
return 0
if not expr or expr == '∅':
return 0
if expr == 'ε':
return 1 if length == 0 else 0
if expr in alphabet_symbols:
return 1 if length == 1 else 0
if '+' in expr:
inner = expr.rstrip('+')
if inner.endswith('?'):
inner = inner[:-1]
return _count_star_words(inner, length, alphabet_symbols, 1)
if expr.endswith('?'):
inner = expr[:-1]
return _count_words(inner, length, alphabet_symbols) + (1 if length == 0 else 0)
if expr.startswith('(') and '|' in expr:
inner = expr[1:-1]
parts = _split_disjunction(inner)
return sum(_count_words(p, length, alphabet_symbols) for p in parts)
if '.' in expr:
parts = expr.split('.')
return _count_concat_words(parts, length, alphabet_symbols, 0)
if ')' in expr or '(' in expr:
return 0
return 0
def _count_concat_words(parts, length, alphabet_symbols, idx):
if idx >= len(parts):
return 1 if length == 0 else 0
total = 0
for take in range(length + 1):
cnt = _count_words(parts[idx], take, alphabet_symbols)
if cnt > 0:
rest = _count_concat_words(parts, length - take, alphabet_symbols, idx + 1)
total += cnt * rest
return total
def _count_star_words(inner, length, alphabet_symbols, min_count):
total = 0
for repeat in range(min_count, length + 1):
if repeat == 0:
continue
total += _count_repeat_words(inner, repeat, length, alphabet_symbols)
return total
def _count_repeat_words(inner, repeat, length, alphabet_symbols):
if repeat == 0:
return 1 if length == 0 else 0
total = 0
for take in range(length + 1):
cnt = _count_words(inner, take, alphabet_symbols)
if cnt > 0:
rest = _count_repeat_words(inner, repeat - 1, length - take, alphabet_symbols)
total += cnt * rest
return total
def _split_disjunction(s):
depth = 0
parts = []
current = []
for ch in s:
if ch == '(':
depth += 1
current.append(ch)
elif ch == ')':
depth -= 1
current.append(ch)
elif ch == '|' and depth == 0:
parts.append(''.join(current))
current = []
else:
current.append(ch)
parts.append(''.join(current))
return parts

457
bex/gbnf.py
View file

@ -1,430 +1,95 @@
"""GBNF — Convert SOREs to GBNF grammar format for llama.cpp constrained decoding.
"""GBNF — Convert AST to GBNF grammar format for llama.cpp constrained decoding."""
GBNF (GGML BNF) is the de-facto standard grammar format for grammar-constrained
LLM output. It supports: literals, concatenation, alternation, repetition (+, *, ?),
and grouping.
from .grammar import (
Symbol, Concat, Alt, Plus, Optional, Star, Epsilon, Empty,
)
SORE operators map directly:
SORE `.` (concat) → GBNF implicit concat (space)
SORE `|` (alt) → GBNF `|`
SORE `+` (plus) → GBNF `+`
SORE `?` (optional) → GBNF `?`
SORE `*` (star) → GBNF `*`
SORE `()` (group) → GBNF `()`
SORE literal → GBNF `"literal"`
"""
# ---------------------------------------------------------------------------
# SORE tokenizer
# ---------------------------------------------------------------------------
_TOKEN_CHARS = set('.|+?*()')
def _tokenize(sore):
"""Tokenize a SORE string into a list of (type, value) tuples.
Literals are sequences of non-operator characters.
"""
tokens = []
i = 0
n = len(sore)
while i < n:
ch = sore[i]
if ch == '.':
tokens.append(('DOT', ch))
i += 1
elif ch == '|':
tokens.append(('PIPE', ch))
i += 1
elif ch == '+':
tokens.append(('PLUS', ch))
i += 1
elif ch == '?':
tokens.append(('QUESTION', ch))
i += 1
elif ch == '*':
tokens.append(('STAR', ch))
i += 1
elif ch == '(':
tokens.append(('LPAREN', ch))
i += 1
elif ch == ')':
tokens.append(('RPAREN', ch))
i += 1
elif ch == 'ε':
tokens.append(('EPSILON', 'ε'))
i += 1
elif ch == '∅':
tokens.append(('EMPTY', '∅'))
i += 1
else:
# Collect literal characters (until next operator or paren)
start = i
while i < n and sore[i] not in _TOKEN_CHARS and sore[i] not in 'ε∅':
i += 1
lit = sore[start:i]
if lit:
# Strip newlines/extra whitespace from symbol names
lit = ' '.join(lit.split())
tokens.append(('LITERAL', lit))
return tokens
# ---------------------------------------------------------------------------
# SORE parser — produces a tree
# ---------------------------------------------------------------------------
class _Node:
"""AST node for SORE."""
pass
class _Literal(_Node):
def __init__(self, value):
self.value = value
def __repr__(self):
return f'Lit({self.value!r})'
class _Concat(_Node):
def __init__(self, parts):
self.parts = parts
def __repr__(self):
return f'Concat({self.parts})'
class _Alt(_Node):
def __init__(self, parts):
self.parts = parts
def __repr__(self):
return f'Alt({self.parts})'
class _Plus(_Node):
def __init__(self, child):
self.child = child
def __repr__(self):
return f'Plus({self.child})'
class _Optional(_Node):
def __init__(self, child):
self.child = child
def __repr__(self):
return f'Optional({self.child})'
class _Star(_Node):
def __init__(self, child):
self.child = child
def __repr__(self):
return f'Star({self.child})'
class _Epsilon(_Node):
def __repr__(self):
return 'Epsilon()'
class _Empty(_Node):
def __repr__(self):
return 'Empty()'
class _Parser:
"""Recursive descent parser for SOREs.
Handles the overloaded + operator:
- (a+b+c) → disjunction (inside parens)
- r+ → one-or-more repetition (outside parens)
"""
def __init__(self, tokens):
self.tokens = tokens
self.pos = 0
self.paren_depth = 0
def peek(self):
if self.pos < len(self.tokens):
return self.tokens[self.pos]
return ('EOF', '')
def consume(self, expected_type=None):
tok = self.peek()
if tok[0] == 'EOF':
raise ValueError(f'Unexpected end of SORE, expected {expected_type}')
if expected_type and tok[0] != expected_type:
raise ValueError(f'Expected {expected_type}, got {tok}')
self.pos += 1
return tok
def parse(self):
"""Parse full SORE."""
node = self.parse_alternation()
return node
def parse_alternation(self):
"""Parse: concat (('|' | '+') concat)* — + is alternation inside parens"""
parts = [self.parse_concat()]
while self.peek()[0] in ('PIPE', 'PLUS'):
if self.peek()[0] == 'PLUS' and self.paren_depth == 0:
break # + outside parens is repetition, not alternation
self.consume()
parts.append(self.parse_concat())
if len(parts) == 1:
return parts[0]
return _Alt(parts)
def parse_concat(self):
"""Parse: repetition (('.' | LPAREN) repetition)* — implicit concat"""
parts = [self.parse_repetition()]
while self.peek()[0] in ('DOT', 'LPAREN'):
if self.peek()[0] == 'DOT':
self.consume('DOT')
# LPAREN = implicit concat (no separator)
parts.append(self.parse_repetition())
if len(parts) == 1:
return parts[0]
return _Concat(parts)
def parse_repetition(self):
"""Parse: atom ('+' | '?' | '*')?
Inside parens, + is alternation (consumed by parse_alternation),
not repetition. Outside parens, always consume + as repetition.
Handles compound: +?, +*, ?+, *+ etc.
"""
node = self.parse_atom()
if self.peek()[0] in ('PLUS', 'QUESTION', 'STAR'):
if self.peek()[0] == 'PLUS' and self.paren_depth > 0:
return node # + inside parens is alternation, handled by caller
op = self.consume()
if op[0] == 'PLUS':
node = _Plus(node)
elif op[0] == 'QUESTION':
node = _Optional(node)
elif op[0] == 'STAR':
node = _Star(node)
# Handle compound repetition: +?, +*, ?+ etc.
# Normalize: Optional(Plus(x)) → Star(x)
if self.peek()[0] in ('PLUS', 'QUESTION', 'STAR'):
if self.peek()[0] == 'PLUS' and self.paren_depth > 0:
return node
op2 = self.consume()
if op2[0] == 'QUESTION':
if isinstance(node, _Plus):
node = _Star(node.child)
else:
node = _Optional(node)
elif op2[0] == 'STAR':
node = _Star(node.child if isinstance(node, (_Plus, _Optional)) else node)
elif op2[0] == 'PLUS':
if isinstance(node, _Optional):
node = _Plus(node.child)
elif isinstance(node, (_Plus, _Star)):
node = node # ++ is idempotent
else:
node = _Plus(node)
return node
def parse_atom(self):
"""Parse: literal | '(' alternation ')' | 'ε' | '∅'"""
tok = self.peek()
if tok[0] == 'EOF':
return _Epsilon()
if tok[0] == 'LITERAL':
self.consume()
return _Literal(tok[1])
if tok[0] == 'EPSILON':
self.consume()
return _Epsilon()
if tok[0] == 'EMPTY':
self.consume()
return _Empty()
if tok[0] == 'LPAREN':
self.consume('LPAREN')
self.paren_depth += 1
node = self.parse_alternation()
self.consume('RPAREN')
self.paren_depth -= 1
return node
raise ValueError(f'Unexpected token: {tok}')
def _parse_sore(sore):
"""Parse a SORE string into an AST."""
tokens = _tokenize(sore)
parser = _Parser(tokens)
return parser.parse()
# ---------------------------------------------------------------------------
# AST → GBNF
# ---------------------------------------------------------------------------
def _needs_group(node):
"""Check if a node needs parentheses in GBNF output."""
return isinstance(node, (_Alt, _Concat))
return isinstance(node, (Alt, Concat))
def _node_to_gbnf(node, rule_counter):
"""Convert AST node to GBNF fragment string.
Returns (gbnf_string, new_rule_counter, new_rules_list).
new_rules_list contains any helper rules needed.
"""
if isinstance(node, _Literal):
# Escape special chars in the literal for GBNF
def _node_to_gbnf(node):
"""Convert AST node to GBNF fragment string."""
if isinstance(node, Symbol):
escaped = node.value.replace('\\', '\\\\').replace('"', '\\"')
return f'"{escaped}"', rule_counter, []
if isinstance(node, _Epsilon):
return '', rule_counter, []
if isinstance(node, _Empty):
return '', rule_counter, []
if isinstance(node, _Concat):
return f'"{escaped}"'
if isinstance(node, (Epsilon, Empty)):
return ''
if isinstance(node, Concat):
parts = []
all_new_rules = []
for child in node.parts:
frag, rule_counter, new_rules = _node_to_gbnf(child, rule_counter)
if isinstance(child, _Alt):
frag = _node_to_gbnf(child)
if isinstance(child, Alt):
frag = f'({frag})'
parts.append(frag)
all_new_rules.extend(new_rules)
return ' '.join(p for p in parts if p), rule_counter, all_new_rules
if isinstance(node, _Alt):
parts = []
all_new_rules = []
for child in node.parts:
frag, rule_counter, new_rules = _node_to_gbnf(child, rule_counter)
parts.append(frag)
all_new_rules.extend(new_rules)
return ' | '.join(p for p in parts if p), rule_counter, all_new_rules
if isinstance(node, _Plus):
frag, rule_counter, new_rules = _node_to_gbnf(node.child, rule_counter)
return ' '.join(p for p in parts if p)
if isinstance(node, Alt):
parts = [_node_to_gbnf(child) for child in node.parts]
return ' | '.join(p for p in parts if p)
if isinstance(node, Plus):
frag = _node_to_gbnf(node.child)
if _needs_group(node.child):
return f'({frag})+', rule_counter, new_rules
return f'{frag}+', rule_counter, new_rules
if isinstance(node, _Optional):
frag, rule_counter, new_rules = _node_to_gbnf(node.child, rule_counter)
return f'({frag})+'
return f'{frag}+'
if isinstance(node, Optional):
frag = _node_to_gbnf(node.child)
if _needs_group(node.child):
return f'({frag})?', rule_counter, new_rules
return f'{frag}?', rule_counter, new_rules
if isinstance(node, _Star):
frag, rule_counter, new_rules = _node_to_gbnf(node.child, rule_counter)
return f'({frag})?'
return f'{frag}?'
if isinstance(node, Star):
frag = _node_to_gbnf(node.child)
if _needs_group(node.child):
return f'({frag})*', rule_counter, new_rules
return f'{frag}*', rule_counter, new_rules
raise ValueError(f'Unknown node type: {type(node)}')
return f'({frag})*'
return f'{frag}*'
return ''
def to_gbnf(sore):
"""Convert a SORE string to a GBNF grammar string.
def to_gbnf(node):
"""Convert AST node to a GBNF grammar string.
Args:
sore: A SORE string like 'raise.(ValueError)+' or 'mockk'
node: Grammar AST node
Returns:
A GBNF grammar string with a single 'root' rule.
Example:
>>> to_gbnf('mockk')
'root ::= "mockk"'
>>> to_gbnf('raise.(ValueError)+')
'root ::= "raise" ("ValueError")+'
>>> to_gbnf('assertEquals.(of.(assertFailsWith)?)+')
'root ::= "assertEquals" ("of" ("assertFailsWith")?)+'
GBNF grammar string with a single 'root' rule.
"""
if not sore or sore == '∅':
if node is None or isinstance(node, Empty):
return 'root ::= ""'
if sore == 'ε':
if isinstance(node, Epsilon):
return 'root ::= ""'
tree = _parse_sore(sore)
frag, _, _ = _node_to_gbnf(tree, 0)
frag = _node_to_gbnf(node)
return f'root ::= {frag}'
def to_gbnf_with_rules(sore, name='root'):
"""Convert a SORE to GBNF with a named rule.
Args:
sore: A SORE string
name: Rule name (default: 'root')
Returns:
GBNF rule string like 'my-rule ::= "foo" ("bar")+'
"""
if not sore or sore == '∅':
def to_gbnf_with_rules(node, name='root'):
"""Convert AST to GBNF with a named rule."""
if node is None or isinstance(node, Empty):
return f'{name} ::= ""'
if sore == 'ε':
if isinstance(node, Epsilon):
return f'{name} ::= ""'
tree = _parse_sore(sore)
frag, _, _ = _node_to_gbnf(tree, 0)
frag = _node_to_gbnf(node)
return f'{name} ::= {frag}'
def validate_sore(sore):
"""Check if a SORE string is parseable. Returns (True, None) or (False, error_msg)."""
if not sore or sore in ('∅', 'ε'):
return True, None
try:
_parse_sore(sore)
return True, None
except ValueError as e:
return False, str(e)
def grammar_structure_score(sore):
"""Quantify how structured a SORE is (0.0 = flat bag, 1.0 = fully structured).
Structured grammars have ordering (concatenation, optional, repetition)
that tells you the SEQUENCE things happen. Flat bags just list symbols.
"""
import re
if not sore or sore in ('∅', 'ε'):
def grammar_structure_score(node):
"""Quantify how structured an AST is (0.0 = flat bag, 1.0 = fully structured)."""
if node is None or isinstance(node, Empty):
return 0.0
depth = 0
dots = 0
pluses_outside = 0
questions = 0
stars = 0
for ch in sore:
if ch == '(':
depth += 1
elif ch == ')':
depth -= 1
elif ch == '.' and depth == 0:
dots += 1
elif ch == '+' and depth == 0:
pluses_outside += 1
elif ch == '?' and depth == 0:
questions += 1
elif ch == '*' and depth == 0:
stars += 1
disj_parts = 0
for m in re.finditer(r'\(([^)]+)\)', sore):
inner = m.group(1)
if '+' in inner:
disj_parts = max(disj_parts, inner.count('+') + 1)
symbols_only = re.sub(r'[.?*+()]', '', sore)
sym_len = len(symbols_only)
if sym_len == 0:
if isinstance(node, Symbol):
return 0.0
if isinstance(node, Epsilon):
return 0.0
if isinstance(node, (Plus, Optional, Star)):
return 0.5 + 0.5 * grammar_structure_score(node.child)
if isinstance(node, Alt):
child_scores = [grammar_structure_score(p) for p in node.parts]
return sum(child_scores) / max(len(child_scores), 1)
if isinstance(node, Concat):
n = len(node.parts)
if n <= 1:
return 1.0
child_scores = [grammar_structure_score(p) for p in node.parts]
return min(1.0, 0.3 + 0.2 * n + sum(child_scores) / max(len(child_scores), 1))
return 0.0
struct_ops = dots + questions + stars + pluses_outside
struct_ratio = struct_ops / max(sym_len, 1)
disj_ratio = disj_parts / max(sym_len, 1)
score = min(1.0, struct_ratio * 3)
if disj_ratio > 0.5 and dots == 0:
score *= 0.3
return score

267
bex/grammar.py Normal file
View file

@ -0,0 +1,267 @@
"""AST — canonical grammar representation.
Node types: Symbol, Concat, Alt, Plus, Optional, Star, Epsilon, Empty.
AST is the ONLY representation. No SORE strings exist anywhere.
"""
import math
from functools import lru_cache
class Symbol:
__slots__ = ('value',)
def __init__(self, value): self.value = value
def __eq__(self, other): return isinstance(other, Symbol) and self.value == other.value
def __hash__(self): return hash(('Sym', self.value))
def __repr__(self): return f"Symbol({self.value!r})"
class Concat:
__slots__ = ('parts',)
def __init__(self, parts): self.parts = list(parts)
def __eq__(self, other): return isinstance(other, Concat) and self.parts == other.parts
def __hash__(self): return hash(('Concat', tuple(self.parts)))
def __repr__(self): return f"Concat({self.parts!r})"
class Alt:
__slots__ = ('parts',)
def __init__(self, parts): self.parts = list(parts)
def __eq__(self, other): return isinstance(other, Alt) and self.parts == other.parts
def __hash__(self): return hash(('Alt', tuple(self.parts)))
def __repr__(self): return f"Alt({self.parts!r})"
class Plus:
__slots__ = ('child',)
def __init__(self, child): self.child = child
def __eq__(self, other): return isinstance(other, Plus) and self.child == other.child
def __hash__(self): return hash(('Plus', self.child))
def __repr__(self): return f"Plus({self.child!r})"
class Optional:
__slots__ = ('child',)
def __init__(self, child): self.child = child
def __eq__(self, other): return isinstance(other, Optional) and self.child == other.child
def __hash__(self): return hash(('Optional', self.child))
def __repr__(self): return f"Optional({self.child!r})"
class Star:
__slots__ = ('child',)
def __init__(self, child): self.child = child
def __eq__(self, other): return isinstance(other, Star) and self.child == other.child
def __hash__(self): return hash(('Star', self.child))
def __repr__(self): return f"Star({self.child!r})"
class Epsilon:
__slots__ = ()
def __eq__(self, other): return isinstance(other, Epsilon)
def __hash__(self): return hash('Epsilon')
def __repr__(self): return 'Epsilon()'
class Empty:
__slots__ = ()
def __eq__(self, other): return isinstance(other, Empty)
def __hash__(self): return hash('Empty')
def __repr__(self): return 'Empty()'
# ---------------------------------------------------------------------------
# AST operations
# ---------------------------------------------------------------------------
def alphabet(node):
"""Collect all Symbol values from an AST."""
if isinstance(node, Symbol):
return {node.value}
if isinstance(node, (Epsilon, Empty)):
return set()
if isinstance(node, (Plus, Optional, Star)):
return alphabet(node.child)
if isinstance(node, (Concat, Alt)):
result = set()
for p in node.parts:
result |= alphabet(p)
return result
return set()
# ---------------------------------------------------------------------------
# Matching
# ---------------------------------------------------------------------------
def match(node, seq):
"""Check if seq matches the grammar defined by node."""
ends = _match_set(node, seq, 0)
return len(seq) in ends
def _match_set(node, seq, pos):
"""Return set of positions reachable from pos after matching node."""
if isinstance(node, Symbol):
if pos < len(seq) and seq[pos] == node.value:
return {pos + 1}
return set()
if isinstance(node, Epsilon):
return {pos}
if isinstance(node, Empty):
return set()
if isinstance(node, Concat):
current = {pos}
for part in node.parts:
next_set = set()
for p in current:
next_set |= _match_set(part, seq, p)
current = next_set
if not current:
break
return current
if isinstance(node, Alt):
result = set()
for part in node.parts:
result |= _match_set(part, seq, pos)
return result
if isinstance(node, Plus):
return _match_rep(node.child, seq, pos, min_rep=1)
if isinstance(node, Optional):
return _match_set(node.child, seq, pos) | {pos}
if isinstance(node, Star):
return _match_rep(node.child, seq, pos, min_rep=0)
return set()
def _match_rep(child, seq, pos, min_rep):
"""Match child repeated min_rep or more times."""
if min_rep == 0:
accept = {pos}
else:
accept = set()
current = {pos}
for _ in range(min_rep):
next_set = set()
for p in current:
next_set |= _match_set(child, seq, p)
current = next_set
if not current:
break
if min_rep == 0:
accept |= current
seen = set()
frontier = current
while frontier:
frontier_next = set()
for p in frontier:
if p in seen:
continue
seen.add(p)
accept.add(p)
frontier_next |= _match_set(child, seq, p)
frontier = frontier_next - seen
if min_rep > 0:
accept |= current
return accept
# ---------------------------------------------------------------------------
# Counting (for MDL scoring)
# ---------------------------------------------------------------------------
_COUNT_CAP = 10**12
@lru_cache(maxsize=None)
def count_words(node, length):
"""Count how many words of exactly `length` are in L(node).
Capped at _COUNT_CAP to prevent combinatorial explosion on
deeply nested CRX grammars with large alphabets.
"""
if length < 0:
return 0
if isinstance(node, Symbol):
return 1 if length == 1 else 0
if isinstance(node, Epsilon):
return 1 if length == 0 else 0
if isinstance(node, Empty):
return 0
if isinstance(node, Concat):
return _count_concat(tuple(id(p) for p in node.parts), node, length, 0)
if isinstance(node, Alt):
total = 0
for p in node.parts:
total += count_words(p, length)
if total >= _COUNT_CAP:
return _COUNT_CAP
return total
if isinstance(node, Plus):
return _count_rep(node.child, length, 1)
if isinstance(node, Optional):
return count_words(node.child, length) + (1 if length == 0 else 0)
if isinstance(node, Star):
return _count_rep(node.child, length, 0)
return 0
def _count_concat(part_ids, node, length, idx):
if idx >= len(node.parts):
return 1 if length == 0 else 0
total = 0
for take in range(length + 1):
cnt = count_words(node.parts[idx], take)
if cnt:
total += cnt * _count_concat(part_ids, node, length - take, idx + 1)
if total >= _COUNT_CAP:
return _COUNT_CAP
return total
@lru_cache(maxsize=None)
def _count_rep(child, length, min_rep):
total = 0
for rep in range(min_rep, length + 1):
total += _count_repeat(child, rep, length)
if total >= _COUNT_CAP:
return _COUNT_CAP
return total
@lru_cache(maxsize=None)
def _count_repeat(child, rep, length):
if rep == 0:
return 1 if length == 0 else 0
total = 0
for take in range(length + 1):
cnt = count_words(child, take)
if cnt:
total += cnt * _count_repeat(child, rep - 1, length - take)
if total >= _COUNT_CAP:
return _COUNT_CAP
return total
def lang_size(node, n=None):
"""|L(r)≤n| — number of words of length ≤ n."""
if isinstance(node, Empty):
return 0
if isinstance(node, Epsilon):
return 1
if n is None:
n = 2 * model_cost(node) + 1
return sum(count_words(node, l) for l in range(n + 1))
def model_cost(node):
"""|r| — number of alphabet symbol occurrences in expression."""
if isinstance(node, Symbol):
return 1
if isinstance(node, (Epsilon, Empty)):
return 0
if isinstance(node, (Plus, Optional, Star)):
return model_cost(node.child)
if isinstance(node, (Concat, Alt)):
return sum(model_cost(p) for p in node.parts)
return 0

203
bex/idregex.py
View file

@ -2,173 +2,69 @@
from .ikoa import ikoa
from .rwrsq import rwr_sq
from .expr import alphabet
from .grammar import (
Concat, Alt, Plus, Optional, Star, Symbol, Epsilon, Empty,
alphabet, count_words, lang_size as grammar_lang_size, model_cost as grammar_model_cost,
)
def is_deterministic(expr):
def is_deterministic(node):
"""Check if a k-ORE is deterministic (Glushkov determinism).
A k-ORE is deterministic iff for every subexpression (r|s),
A k-ORE is deterministic iff for every subexpression Alt([r, s, ...]),
first(r) ∩ first(s) = ∅.
"""
if not expr or expr == '∅' or expr == 'ε':
if node is None or isinstance(node, (Empty, Epsilon)):
return True
return _check_det(expr)
return _check_det(node)
def _check_det(expr):
"""Recursive determinism check."""
depth = 0
i = 0
while i < len(expr):
if expr[i] == '(':
if depth == 0:
start = i
depth += 1
elif expr[i] == ')':
depth -= 1
if depth == 0:
inner = expr[start + 1:i]
if '|' in inner:
alts = _split_or(inner)
first_sets = []
for alt in alts:
fs = _first_set(alt.strip())
first_sets.append(fs)
def _check_det(node):
"""Recursive determinism check on AST nodes."""
if isinstance(node, Symbol):
return True
if isinstance(node, (Epsilon, Empty)):
return True
if isinstance(node, (Plus, Optional, Star)):
return _check_det(node.child)
if isinstance(node, Alt):
first_sets = [_first_set(child) for child in node.parts]
for j, fs1 in enumerate(first_sets):
for fs2 in first_sets[j + 1:]:
if fs1 & fs2:
return False
for alt in alts:
if not _check_det(alt.strip()):
for child in node.parts:
if not _check_det(child):
return False
else:
if not _check_det(inner):
return True
if isinstance(node, Concat):
for child in node.parts:
if not _check_det(child):
return False
elif expr[i] == '+':
pass
elif expr[i] == '?':
pass
i += 1
return True
return True
def _first_set(expr):
def _first_set(node):
"""Compute first(r) — set of alphabet symbols that can appear at the start of a word in L(r)."""
if not expr or expr == '∅':
if node is None or isinstance(node, Empty):
return set()
if expr == 'ε':
if isinstance(node, Epsilon):
return set()
alpha = alphabet(expr)
if expr in alpha:
return {expr}
if expr.endswith('?') or expr.endswith('+'):
inner = expr.rstrip('+?')
return _first_set(inner)
if '.' in expr:
parts = expr.split('.')
return _first_set(parts[0])
if expr.startswith('(') and '|' in expr:
inner = expr[1:-1]
alts = _split_or(inner)
if isinstance(node, Symbol):
return {node.value}
if isinstance(node, (Plus, Optional, Star)):
return _first_set(node.child)
if isinstance(node, Concat):
if node.parts:
return _first_set(node.parts[0])
return set()
if isinstance(node, Alt):
result = set()
for a in alts:
result |= _first_set(a.strip())
for child in node.parts:
result |= _first_set(child)
return result
return alpha
def _split_or(s):
"""Split disjunction string at top-level | operators."""
depth = 0
parts = []
cur = []
for ch in s:
if ch == '(':
depth += 1
cur.append(ch)
elif ch == ')':
depth -= 1
cur.append(ch)
elif ch == '|' and depth == 0:
parts.append(''.join(cur))
cur = []
else:
cur.append(ch)
parts.append(''.join(cur))
return parts
def _lang_size(expr, n=None):
"""|L(r)≤n| — number of words of length ≤ n in L(r).
n = 2m + 1 where m = |r| excluding operators.
Uses simple structural approximation.
"""
if not expr or expr == '∅':
return 0
if expr == 'ε':
return 1
m = len(alphabet(expr))
if n is None:
n = 2 * m + 1
total = 0
for length in range(n + 1):
total += _count_len(expr, length)
return total
def _count_len(expr, length):
if length < 0:
return 0
if not expr or expr == '∅':
return 0
if expr == 'ε':
return 1 if length == 0 else 0
alpha = alphabet(expr)
if expr in alpha:
return 1 if length == 1 else 0
if expr.endswith('+'):
inner = expr[:-1]
if inner.endswith('?'):
inner = inner[:-1]
total = 0
for rep in range(1, length + 1):
total += _count_repeat(inner, rep, length)
return total
if expr.endswith('?'):
inner = expr[:-1]
return _count_len(inner, length) + (1 if length == 0 else 0)
if '.' in expr:
parts = expr.split('.')
return _count_concat(parts, length, 0)
if expr.startswith('(') and '|' in expr:
inner = expr[1:-1]
alts = _split_or(inner)
return sum(_count_len(a.strip(), length) for a in alts)
return 0
def _count_concat(parts, length, idx):
if idx >= len(parts):
return 1 if length == 0 else 0
total = 0
for take in range(length + 1):
cnt = _count_len(parts[idx], take)
if cnt:
total += cnt * _count_concat(parts, length - take, idx + 1)
return total
def _count_repeat(inner, rep, length):
if rep == 0:
return 1 if length == 0 else 0
total = 0
for take in range(length + 1):
cnt = _count_len(inner, take)
if cnt:
total += cnt * _count_repeat(inner, rep - 1, length - take)
return total
return set()
def idregex(sequences, kmax=4, N=5, criterion='langsize'):
@ -192,11 +88,24 @@ def idregex(sequences, kmax=4, N=5, criterion='langsize'):
if G is None:
continue
expr = rwr_sq(G)
if expr and expr not in ('∅', 'ε'):
if expr is not None and not isinstance(expr, Empty):
if is_deterministic(expr):
C.add(expr)
if not C:
return None
if criterion == 'langsize':
return min(C, key=lambda e: (_lang_size(e), len(e)))
return min(C, key=lambda e: len(e))
return min(C, key=lambda e: (grammar_lang_size(e, 2 * grammar_model_cost(e) + 1), _ast_size(e)))
return min(C, key=lambda e: _ast_size(e))
def _ast_size(node):
"""Count nodes in AST for length comparison."""
if isinstance(node, Symbol):
return 1
if isinstance(node, (Epsilon, Empty)):
return 1
if isinstance(node, (Plus, Optional, Star)):
return 1 + _ast_size(node.child)
if isinstance(node, (Concat, Alt)):
return 1 + sum(_ast_size(child) for child in node.parts)
return 1

13
bex/ikoa.py
View file

@ -2,8 +2,9 @@
from collections import deque, defaultdict
import random
from .koa import KOA, build_complete_koa
from .koa import KOA, build_complete_koa, strip_k
from .baum_welch import init_probabilities, baum_welch, baum_welch_fixed
from .grammar import Symbol
def disambiguate(G, prob, sequences):
@ -24,7 +25,7 @@ def disambiguate(G, prob, sequences):
Q.append(s)
D = set()
from .expr import strip_k
from .koa import strip_k as _sk
while Q:
s = Q.popleft()
while True:
@ -32,7 +33,7 @@ def disambiguate(G, prob, sequences):
for t in list(G._succ.get(s, set())):
l = G.label(t)
if l:
lab_groups[strip_k(l)].append(t)
lab_groups[_sk(l)].append(t)
multi = [(lab, ts) for lab, ts in lab_groups.items() if len(ts) > 1]
if not multi:
break
@ -62,7 +63,7 @@ def prune(G, sequences):
Also removes states s ∈ Succ(src) without a witness.
"""
from .expr import strip_k as _sk
from .koa import strip_k as _sk
witnessed = set()
for seq in sequences:
if not seq:
@ -74,7 +75,9 @@ def prune(G, sequences):
for s in cur:
for t in G._succ.get(s, set()):
lab = G.label(t)
if lab and _sk(lab) == sym:
if lab:
stripped = _sk(lab)
if isinstance(stripped, Symbol) and stripped.value == sym:
nxt.add(t)
witnessed.add((s, t))
cur = nxt

60
bex/koa.py
View file

@ -4,7 +4,20 @@ A k-OA is like a SOA but each symbol appears at most k times as a state label.
"""
from .soa import SOA
from .expr import strip_k
from .grammar import Symbol, Epsilon, Empty
def strip_k(node):
"""Remove k-ORE markers from AST: Symbol('a_1') → Symbol('a')."""
if isinstance(node, Symbol):
import re
value = node.value
value = re.sub(r'_\d+$', '', value)
value = re.sub(r'\^\(\d+\)$', '', value)
return Symbol(value)
if isinstance(node, (Epsilon, Empty)):
return node
return node
class KOA(SOA):
@ -20,20 +33,33 @@ class KOA(SOA):
def add_state(self, label):
nid = super().add_state(label)
sym = strip_k(label)
self._symbol_count.setdefault(sym, 0)
self._symbol_count[sym] += 1
if isinstance(label, Symbol):
stripped = strip_k(label)
key = stripped.value if isinstance(stripped, Symbol) else str(stripped)
else:
key = str(label)
self._symbol_count.setdefault(key, 0)
self._symbol_count[key] += 1
return nid
def remove_state(self, nid):
label = self._label.get(nid)
if label:
sym = strip_k(label)
self._symbol_count[sym] -= 1
if isinstance(label, Symbol):
stripped = strip_k(label)
key = stripped.value if isinstance(stripped, Symbol) else str(stripped)
else:
key = str(label)
self._symbol_count[key] -= 1
super().rm_state(nid)
def count_symbol(self, symbol):
return self._symbol_count.get(strip_k(symbol), 0)
def count_symbol(self, sym):
if isinstance(sym, Symbol):
key = sym.value
else:
key = str(sym)
stripped = strip_k(Symbol(key)) if not isinstance(key, str) else strip_k(Symbol(key))
return self._symbol_count.get(stripped.value if isinstance(stripped, Symbol) else stripped, 0)
def symbol_ok(self, symbol):
return self.count_symbol(symbol) < self.k
@ -44,7 +70,8 @@ class KOA(SOA):
for t in self._succ[n]:
lab = self._label.get(t)
if lab:
base = strip_k(lab)
stripped = strip_k(lab)
base = stripped.value if isinstance(stripped, Symbol) else stripped
if base in label_map:
return False
label_map[base] = t
@ -58,7 +85,9 @@ class KOA(SOA):
for s in cur:
for t in self._succ.get(s, set()):
lab = self._label.get(t)
if lab and strip_k(lab) == sym:
if lab:
stripped = strip_k(lab)
if isinstance(stripped, Symbol) and stripped.value == sym:
nxt.add(t)
if not nxt:
return False
@ -66,7 +95,14 @@ class KOA(SOA):
return any(self.sink in self._succ.get(s, set()) for s in cur)
def succ_labeled(self, nid, symbol):
return {t for t in self._succ.get(nid, set()) if strip_k(self._label.get(t) or '') == symbol}
result = set()
for t in self._succ.get(nid, set()):
lab = self._label.get(t)
if lab:
stripped = strip_k(lab)
if isinstance(stripped, Symbol) and stripped.value == symbol:
result.add(t)
return result
def build_complete_koa(sequences, k):
@ -87,7 +123,7 @@ def build_complete_koa(sequences, k):
for sym in alphabet:
state_ids = []
for i in range(1, k + 1):
nid = G.add_state(f"{sym}_{i}")
nid = G.add_state(Symbol(f"{sym}_{i}"))
state_ids.append(nid)
G.add_edge(G.src, nid)
symbol_states[sym] = state_ids

104
bex/kore.py
View file

@ -1,104 +1,74 @@
"""
kOREInference — Algorithm 4: iDRegEx (arXiv 1004.2372).
Implements the full iDRegEx pipeline:
1. For k = 1..kmax, for n = 1..N:
a. iKoa (Algorithm 1) — build a deterministic k-OA from S
b. rwr² (Algorithm 3) — translate k-OA to k-ORE expression
c. Validate determinism and k-occurrence
2. Score all valid candidates by MDL (model cost + data cost)
3. Return the best k-ORE
Unlike the PTA→Shrink→Repair approach from Bex 2008, this follows
the journal paper (arXiv 1004.2372) exactly.
"""
"""kOREInference — Algorithm 4: iDRegEx (arXiv 1004.2372)."""
from .ikoa import ikoa
from .rwrsq import rwr_sq
from .idregex import is_deterministic
from .mdl import mdl_score
from .grammar import Epsilon, Empty, Symbol, Plus, Optional, Star, Concat, Alt, alphabet as ast_alphabet
def validate_k_ore(expr, k, alphabet_set=None):
"""
Check if a k-ORE satisfies the k-occurrence condition.
The k-occurrence condition: for every subexpression (r|s),
each alphabet symbol appears at most k times across all
alternatives combined.
Simplified implementation: count raw alphabet symbol
occurrences in the expression string. A symbol appearing
more than k times violates the condition.
Returns:
(bool, str): (passes, explanation)
"""
if not expr or expr in ('∅', 'ε'):
def validate_k_ore(node, k, alphabet_set=None):
"""Check if a k-ORE satisfies the k-occurrence condition."""
if node is None or isinstance(node, (Empty, Epsilon)):
return True, "OK"
from .expr import alphabet
syms = alphabet_set or alphabet(expr)
counts = {}
for sym in syms:
import re
count = len(re.findall(rf'(?<![a-zA-Z_/]){re.escape(sym)}(?![a-zA-Z_/])', expr))
if count > 0:
counts[sym] = count
syms = alphabet_set or ast_alphabet(node)
counts = _count_symbol_occurrences(node)
violations = [f"{s}:{c}" for s, c in sorted(counts.items()) if c > k]
if violations:
return False, f"k={k} violations: {', '.join(violations)}"
return True, "OK"
def _count_symbol_occurrences(node):
"""Count how many times each symbol appears as a leaf in the AST."""
if isinstance(node, Symbol):
return {node.value: 1}
if isinstance(node, (Epsilon, Empty)):
return {}
if isinstance(node, (Plus, Optional, Star)):
return _count_symbol_occurrences(node.child)
if isinstance(node, (Concat, Alt)):
counts = {}
for p in node.parts:
for sym, cnt in _count_symbol_occurrences(p).items():
counts[sym] = counts.get(sym, 0) + cnt
return counts
return {}
def _count_nodes(node):
if isinstance(node, Symbol):
return 1
if isinstance(node, (Epsilon, Empty)):
return 1
if isinstance(node, (Plus, Optional, Star)):
return 1 + _count_nodes(node.child)
if isinstance(node, (Concat, Alt)):
return 1 + sum(_count_nodes(p) for p in node.parts)
return 1
class kOREInference:
"""
|———— Algorithm 4: iDRegEx ————|
Require: sample S, kmax
Ensure: k-ORE r
1: C ← ∅
2: for k = 1 to kmax do
3: for n = 1 to N do
4: G ← iKoa(S, k)
5: if rwr²(G) is deterministic then
6: add rwr²(G) to C
7: return best(C) by MDL
"""
def __init__(self, k_max=5, N=5):
self.k_max = k_max
self.N = N
def infer(self, sequences):
"""
Infer the best k-ORE for the given sequences.
Returns:
(koa_automaton, expression_string, best_k) or None if no valid
k-ORE can be inferred.
"""
sequences = [s for s in sequences if s]
if not sequences:
return None
candidates = []
for k in range(1, self.k_max + 1):
for _ in range(self.N):
G = ikoa(sequences, k, num_trials=1)
if G is None:
continue
expr = rwr_sq(G)
if expr and expr not in ('∅', 'ε'):
if expr is not None and not isinstance(expr, (Empty, Epsilon)):
if is_deterministic(expr):
valid, _ = validate_k_ore(expr, k)
if valid:
candidates.append((G, expr, k))
if not candidates:
return None
return min(candidates, key=lambda c: mdl_score(c[1], sequences))

32
bex/marking.py
View file

@ -1,7 +1,7 @@
"""Marking — Convert k-OA to SOA over Σ^(k) (Definition 4.4, arXiv 1004.2372)."""
from .soa import SOA
from .expr import strip_k
from .grammar import Symbol, Epsilon, Empty, Plus, Star, Optional, Concat, Alt
def mark_koa(G):
@ -9,7 +9,7 @@ def mark_koa(G):
Mark a k-OA G as a SOA over Σ^(k).
Process nodes in arbitrary order. For the i-th occurrence of label a,
replace by a^(i) (represented as "a_i").
replace by a^(i) (represented as Symbol('a_i')).
Returns a SOA H over Σ^(k) such that L(G) = strip(L(H)).
"""
@ -24,10 +24,11 @@ def mark_koa(G):
counts = {}
for n in G._succ:
lab = G._label.get(n)
if lab and lab not in ('ε', '∅') and n not in (G.src, G.sink):
if lab is not None and not isinstance(lab, (Empty, Epsilon)) and n not in (G.src, G.sink):
sym = strip_k(lab)
counts[sym] = counts.get(sym, 0) + 1
H._label[n] = f"{sym}_{counts[sym]}"
key = sym.value if isinstance(sym, Symbol) else str(sym)
counts[key] = counts.get(key, 0) + 1
H._label[n] = Symbol(f"{key}_{counts[key]}")
elif n in (G.src, G.sink):
H._label[n] = None
else:
@ -36,11 +37,18 @@ def mark_koa(G):
return H
def strip_expression(expr):
"""Strip k-ORE markers from expression: a_i → a.
Returns expression over original alphabet Σ.
"""
def strip_k(node):
"""Remove k-ORE markers from AST: Symbol('a_1') → Symbol('a'), Symbol('b^(2)') → Symbol('b')."""
if isinstance(node, Symbol):
import re
result = re.sub(r'(_\d+)', '', expr)
return result
value = node.value
value = re.sub(r'_\d+$', '', value)
value = re.sub(r'\^\(\d+\)$', '', value)
return Symbol(value)
if isinstance(node, (Epsilon, Empty)):
return node
if isinstance(node, (Plus, Optional, Star)):
return type(node)(strip_k(node.child))
if isinstance(node, (Concat, Alt)):
return type(node)([strip_k(child) for child in node.parts])
return node

5
bex/mcp_server.py
View file

@ -12,6 +12,7 @@ from mcp.server.fastmcp import FastMCP
from .ensemble import infer_ensemble, _matches
from .grammar_index import load_grammar_index
from .gbnf import to_gbnf
from .tag_preprocessor.analyze import (
analyze_directory as _analyze_directory,
_build_yaml_output,
@ -72,7 +73,7 @@ def infer_best_grammar(
if result['best'] is None:
return f"No grammar found. {result['why']}"
lines = [f"Best: {result['best']['algorithm']} (Score {result['best']['mdl_score']})",
f"Grammar: {result['best']['grammar']}",
f"Grammar: {to_gbnf(result['best']['grammar'])}",
""]
if len(result['all']) > 1:
for r in result['all']:
@ -82,7 +83,7 @@ def infer_best_grammar(
lines.append(f"Why: {result['why']}")
if 'core' in result and result['core']:
c = result['core']
lines.append(f"\nCore CRX ({c['coverage']:.0%} coverage, {c['outlier_count']} outliers): {c['grammar']}")
lines.append(f"\nCore CRX ({c['coverage']:.0%} coverage, {c['outlier_count']} outliers): {to_gbnf(c['grammar'])}")
if c['outliers']:
lines.append(f" Outlier sequences:")
for i, o in enumerate(c['outliers'], 1):

165
bex/mdl.py
View file

@ -1,158 +1,11 @@
"""MDL scoring for iDRegEx (Algorithm 4, arXiv 1004.2372)."""
import math
import functools
from .expr import alphabet
def model_cost(expr):
"""|r| — number of alphabet symbol occurrences in expression."""
import re
syms = alphabet(expr)
# Count each symbol by how many times it appears as a standalone word
count = 0
for s in syms:
# Count occurrences where symbol is bordered by operators or edges
count += len(re.findall(rf'(?<![a-zA-Z_]){re.escape(s)}(?![a-zA-Z_])', expr))
return count
def lang_size(expr, n=None):
"""Estimate |L(r)≤n| — number of words of length ≤ n in L(r).
Simple approximation based on expression structure.
"""
if not expr or expr == '∅':
return 0
if expr == 'ε':
return 1
n = n or (2 * model_cost(expr) + 1)
total = 0
for length in range(n + 1):
total += _count_words_fast(expr, length)
return total
@functools.lru_cache(maxsize=None)
def _count_words_fast(expr, length):
if length < 0:
return 0
if not expr or expr == '∅':
return 0
if expr == 'ε':
return 1 if length == 0 else 0
alpha = alphabet(expr)
if expr in alpha:
return 1 if length == 1 else 0
# 0. Concatenation: a.b.c — check FIRST so trailing quantifiers
# apply to each part individually, not the whole expression.
if '.' in expr:
parts = _split_disj_crx(expr, '.')
if len(parts) > 1:
return _count_concat(tuple(parts), length, 0)
# 1. Trailing quantifiers
if expr.endswith('+?'):
return _count_star(expr[:-2], length, min_count=0)
if expr.endswith('*'):
return _count_star(expr[:-1], length, min_count=0)
if expr.endswith('?') and not expr.endswith('+?'):
inner = expr[:-1]
return _count_words_fast(inner, length) + (1 if length == 0 else 0)
if expr.endswith('+') and not expr.endswith('+?'):
inner = expr[:-1]
return _count_star(inner, length, min_count=1)
# 2. Disjunction group: (a+b+c) for CRX or (a|b|c) for iDRegEx
if expr.startswith('(') and expr.endswith(')'):
inner = expr[1:-1]
parts = _split_disj_crx(inner, '+')
if len(parts) > 1:
return sum(_count_words_fast(p.strip(), length) for p in parts)
parts = _split_disj_crx(inner, '|')
if len(parts) > 1:
return sum(_count_words_fast(p.strip(), length) for p in parts)
return _count_words_fast(inner, length)
return 0
def _split_disj_crx(s, sep):
"""Split on `sep` at top depth (not inside nested parens)."""
depth = 0
parts = []
cur = []
for ch in s:
if ch == '(':
depth += 1
cur.append(ch)
elif ch == ')':
depth -= 1
cur.append(ch)
elif ch == sep and depth == 0:
parts.append(''.join(cur))
cur = []
else:
cur.append(ch)
parts.append(''.join(cur))
return parts
@functools.lru_cache(maxsize=None)
def _count_concat(parts_tuple, length, idx):
parts = list(parts_tuple)
if idx >= len(parts):
return 1 if length == 0 else 0
total = 0
for take in range(length + 1):
cnt = _count_words_fast(parts[idx], take)
if cnt:
total += cnt * _count_concat(parts_tuple, length - take, idx + 1)
return total
@functools.lru_cache(maxsize=None)
def _count_star(inner, length, min_count):
total = 0
for rep in range(min_count, length + 1):
total += _count_repeat(inner, rep, length)
return total
@functools.lru_cache(maxsize=None)
def _count_repeat(inner, rep, length):
if rep == 0:
return 1 if length == 0 else 0
total = 0
for take in range(length + 1):
cnt = _count_words_fast(inner, take)
if cnt:
total += cnt * _count_repeat(inner, rep - 1, length - take)
return total
def _split_disj(s):
depth = 0
parts = []
cur = []
for ch in s:
if ch == '(':
depth += 1
cur.append(ch)
elif ch == ')':
depth -= 1
cur.append(ch)
elif ch == '|' and depth == 0:
parts.append(''.join(cur))
cur = []
else:
cur.append(ch)
parts.append(''.join(cur))
return parts
from .grammar import (
Concat, Alt, Plus, Optional, Star, Symbol, Epsilon, Empty,
alphabet, count_words, lang_size, model_cost, match as grammar_match,
_COUNT_CAP,
)
def data_cost(expr, sequences):
@ -167,7 +20,7 @@ def data_cost(expr, sequences):
n = 2 * model_cost(expr) + 1
runtime_n = min(max(n, max((len(s) for s in sequences), default=0)), MAX_EXACT)
lang_sizes = [_count_words_fast(expr, l) for l in range(runtime_n + 1)]
lang_sizes = [count_words(expr, l) for l in range(runtime_n + 1)]
alpha_size = len(alphabet(expr))
@ -202,7 +55,9 @@ def lang_size_score(expr, sequences):
total = 0
for seq in sequences:
length = len(seq)
total += _count_words_fast(expr, length)
total += count_words(expr, length)
if total >= _COUNT_CAP:
return _COUNT_CAP
return total
@ -223,7 +78,7 @@ def score_grammar(expr, sequences, method='langsize'):
"""Score a grammar using the specified method.
Args:
expr: Grammar expression string.
expr: Grammar AST node.
sequences: List of sequences (each a list of strings).
method: 'langsize' (default, Bex et al.) or 'mdl' (fallback).

47
bex/reduce.py
View file

@ -39,6 +39,7 @@ from itertools import combinations
from .soa import SOA
from .twotinf import build_soa
from .rwr0 import rwr0
from .grammar import Empty
def _labeled_transitions(soa):
@ -318,44 +319,44 @@ def _extract_text_from_seqs(seqs):
def minimize_contexts(merged_contexts):
"""Minimize: merge contexts with identical SOREs (Line 15, Algorithm 4).
"""Minimize: merge contexts with identical grammars (Line 15, Algorithm 4).
After Reduce, some contexts may have identical SOREs. Minimize unifies them.
After Reduce, some contexts may have identical grammars. Minimize unifies them.
For our use case: group contexts by their inferred SORE, merge those with
the same SORE into a single context.
For our use case: group contexts by their inferred grammar, merge those with
the same grammar into a single context.
"""
# Build SORE for each context
ctx_sores = {}
# Build grammar for each context
ctx_grammars = {}
for ctx, seqs in merged_contexts.items():
if len(seqs) < 2:
ctx_sores[ctx] = ("∅", seqs)
ctx_grammars[ctx] = (Empty(), seqs)
continue
clean = _extract_text_from_seqs(seqs)
clean = [s for s in clean if s]
if len(clean) < 2:
ctx_sores[ctx] = ("∅", seqs)
ctx_grammars[ctx] = (Empty(), seqs)
continue
soa = build_soa(clean)
sore = rwr0(soa)
ctx_sores[ctx] = (sore, seqs)
grammar = rwr0(soa)
ctx_grammars[ctx] = (grammar, seqs)
# Group by SORE
sore_groups = defaultdict(list)
for ctx, (sore, seqs) in ctx_sores.items():
sore_groups[sore].append((ctx, seqs))
# Group by grammar
grammar_groups = defaultdict(list)
for ctx, (grammar, seqs) in ctx_grammars.items():
grammar_groups[grammar].append((ctx, seqs))
# Merge contexts with same SORE
# Merge contexts with same grammar
minimized = {}
for sore, items in sore_groups.items():
for grammar, items in grammar_groups.items():
if len(items) == 1:
ctx, seqs = items[0]
minimized[ctx] = seqs
else:
# Multiple contexts with same SORE -> merge into one
# Multiple contexts with same grammar -> merge into one
merged_seqs = []
for ctx, seqs in items:
merged_seqs.extend(seqs)
@ -377,7 +378,7 @@ def reduce_and_infer(contexts, threshold, min_methods=3):
Returns:
dict with keys:
merged: final context->sequences mapping (after minimize)
infer_results: list of (context, sore, methods_count) tuples
infer_results: list of (context, grammar, methods_count) tuples
merge_info: stats from reduce step
minimize_info: stats from minimize step
"""
@ -386,10 +387,10 @@ def reduce_and_infer(contexts, threshold, min_methods=3):
minimize_info = {
"contexts_before": merge_info["contexts_after"],
"contexts_after": len(merged),
"merged_by_sore": 0,
"merged_by_grammar": 0,
}
# Line 14: ToSore(soa(t)) — infer SOREs for each context
# Infer grammar for each context
infer_results = []
for ctx, seqs in sorted(merged.items(), key=lambda x: -len(x[1])):
n = len(seqs)
@ -401,9 +402,9 @@ def reduce_and_infer(contexts, threshold, min_methods=3):
continue
soa = build_soa(clean)
sore = rwr0(soa)
if sore != "∅":
infer_results.append((ctx, sore, n))
grammar = rwr0(soa)
if not isinstance(grammar, Empty):
infer_results.append((ctx, grammar, n))
return {
"merged": merged,

7
bex/rwr0.py
View file

@ -9,6 +9,7 @@ Used as rwr²₁ in arXiv 1004.2372 for k>1.
from .soa import SOA
from .expr import concat, disj, star, optional
from .grammar import Empty, Epsilon
def _find_concat(G, Gs):
@ -173,9 +174,9 @@ def rwr0(G, max_iterations=1000):
"""
G = G.copy()
if not G.sink_reachable():
return '∅'
return Empty()
if G.num_non_special() == 0 and G.has_edge(G.src, G.sink):
return 'ε'
return Epsilon()
done = False
iterations = 0
@ -223,4 +224,4 @@ def rwr0(G, max_iterations=1000):
if G.is_final():
return G.expression()
return '∅'
return Empty()

24
bex/rwrsq.py
View file

@ -5,14 +5,28 @@ rwr²(G):
2: return strip(rwr²₁(H))
"""
import re
from .marking import mark_koa
from .rwr0 import rwr0
from .grammar import (
Concat, Alt, Plus, Optional, Star, Symbol, Epsilon, Empty,
)
def strip(expr):
"""Remove k-ORE markers: a_i → a."""
return re.sub(r'_\d+', '', expr)
def strip(node):
"""Remove k-ORE markers: Symbol('a_i') → Symbol('a')."""
if isinstance(node, Symbol):
value = node.value
if '_' in value:
base = value.rsplit('_', 1)[0]
return Symbol(base)
return node
if isinstance(node, (Epsilon, Empty)):
return node
if isinstance(node, (Plus, Optional, Star)):
return type(node)(strip(node.child))
if isinstance(node, (Concat, Alt)):
return type(node)([strip(child) for child in node.parts])
return node
def rwr_sq(G):
@ -26,6 +40,6 @@ def rwr_sq(G):
"""
H = mark_koa(G)
result = rwr0(H)
if result is None or result == '∅':
if result is None or isinstance(result, Empty):
return None
return strip(result)

60
bex/soa.py
View file

@ -1,7 +1,10 @@
"""SOA — Single Occurrence Automaton (Definition 6, TODS 2010)."""
"""SOA — Single Occurrence Automaton (Definition 6, TODS 2010).
Labels are grammar.py AST nodes (Symbol, Concat, Alt, Plus, etc.).
"""
import copy
from .expr import concat, disj, star, optional
from .grammar import Symbol, Concat, Alt, Plus, Optional, Star, Epsilon, Empty
class SOA:
@ -12,7 +15,7 @@ class SOA:
E ⊆ V × V, unlabeled edges.
Walk src=v₁,v₂,...,vₙ₊₁=sink accepts word lab(v₂)...lab(vₙ).
States are proper SOREs, pairwise alphabet-disjoint (Definition 10).
Labels are AST nodes from grammar.py.
"""
def __init__(self):
@ -33,6 +36,8 @@ class SOA:
def add_state(self, label):
n = self._new()
if isinstance(label, str):
label = Symbol(label)
self._label[n] = label
return n
@ -59,6 +64,8 @@ class SOA:
return self._label.get(n)
def set_label(self, n, lab):
if isinstance(lab, str):
lab = Symbol(lab)
self._label[n] = lab
def succ(self, n):
@ -73,15 +80,39 @@ class SOA:
def states(self):
return [n for n in self._succ if n not in (self.src, self.sink) and self._label.get(n) is not None]
def count_symbol(self, sym):
"""Count states whose label base matches sym (string or Symbol).
Strips _N suffixes before comparing."""
import re
if isinstance(sym, Symbol):
target = sym.value
else:
target = sym
count = 0
for n, lab in self._label.items():
if n in (self.src, self.sink):
continue
if isinstance(lab, Symbol):
base = re.sub(r'_\d+$', '', lab.value)
if base == target:
count += 1
elif isinstance(lab, str):
base = re.sub(r'_\d+$', '', lab)
if base == target:
count += 1
return count
def _pred_plus(self, n):
r = set(self._pred.get(n, set()))
if self._label.get(n) and self._label[n].endswith('+'):
lab = self._label.get(n)
if isinstance(lab, (Plus, Star)):
r.add(n)
return r
def _succ_plus(self, n):
r = set(self._succ.get(n, set()))
if self._label.get(n) and self._label[n].endswith('+'):
lab = self._label.get(n)
if isinstance(lab, (Plus, Star)):
r.add(n)
return r
@ -94,7 +125,10 @@ class SOA:
nxt = set()
for s in cur:
for t in self._succ.get(s, set()):
if self._label.get(t) == sym:
lab = self._label.get(t)
if isinstance(lab, Symbol) and lab.value == sym:
nxt.add(t)
elif isinstance(lab, str) and lab == sym:
nxt.add(t)
if not nxt:
return False
@ -129,13 +163,9 @@ class SOA:
def contract(self, r, s, new_label):
"""
State contraction G[r,s ⇒ t] (Definition 11, TODS 2010).
(1) Add t as new state with label new_label.
(2) Every v ∈ Pred(r) − {r,s} → predecessor of t.
(3) Every w ∈ Succ(s) − {r,s} → successor of t. [matching figures]
(4) Loop t→t if r ∈ Succ(s).
(5) Remove r, s and all edges.
"""
if isinstance(new_label, str):
new_label = Symbol(new_label)
t = self._new()
self._label[t] = new_label
for v in self._pred.get(r, set()) - {r, s}:
@ -154,6 +184,8 @@ class SOA:
def contract_single(self, r, new_label):
"""Single-state substitution G[r ⇒ t] (Definition 11 note)."""
if isinstance(new_label, str):
new_label = Symbol(new_label)
if r in (self.src, self.sink):
return r
t = self._new()
@ -175,14 +207,14 @@ class SOA:
changed = False
for n in list(G._succ.keys()):
lab = G._label.get(n)
if lab and (lab.endswith('+') or lab.endswith('+?')):
if isinstance(lab, (Plus, Star)):
if not G.has_edge(n, n):
G.add_edge(n, n)
changed = True
for n in list(G._succ.keys()):
for m in list(G._succ.get(n, set())):
mlab = G._label.get(m)
if mlab == 'ε':
if isinstance(mlab, Epsilon):
for mp in list(G._succ.get(m, set())):
if mp != n and not G.has_edge(n, mp):
G.add_edge(n, mp)

24
bex/tag_preprocessor/analyze.py
View file

@ -20,7 +20,8 @@ import pathspec
from .code import preprocess_by_method, extract_arg_info, _summarize_arg_info
from bex.ensemble import infer_ensemble
from bex.gbnf import validate_sore, grammar_structure_score, to_gbnf
from bex.gbnf import grammar_structure_score, to_gbnf
from bex.grammar import Empty
from bex.distributional import distributional_split
from bex.decompose import decompose_with_coverage, get_decomposition_stats
@ -432,8 +433,7 @@ def _infer_group(label, group_seqs, group_files, project_root, min_coverage, pre
total_count += len(leaf_seqs)
if leaf_result and leaf_result.get('best') and leaf_result['best'].get('grammar'):
g = leaf_result['best']['grammar']
ok, _ = validate_sore(g)
if ok:
if g and not isinstance(g, Empty):
if min_structure > 0 and grammar_structure_score(g) < min_structure:
continue
all_results.append((leaf_label, leaf_result, len(leaf_seqs)))
@ -458,13 +458,9 @@ def _infer_group(label, group_seqs, group_files, project_root, min_coverage, pre
else:
result = infer_ensemble(symbol_seqs, kmax=kmax, N=N, prefer=prefer, min_coverage=min_coverage, include_kore=include_kore, include_idregex=include_idregex)
# Validate grammar is parseable
# Check grammar exists and has structure
if result and result.get('best') and result['best'].get('grammar'):
grammar = result['best']['grammar']
ok, err = validate_sore(grammar)
if not ok:
meta = {"files": group_files, "imports": imports, "arg_patterns": arg_patterns, "packages": packages, "skip_reason": "malformed_grammar", "skip_detail": err}
return (label, None, len(filtered), meta)
if min_structure > 0 and grammar_structure_score(grammar) < min_structure:
meta = {"files": group_files, "imports": imports, "arg_patterns": arg_patterns, "packages": packages, "skip_reason": "low_structure", "structure_score": grammar_structure_score(grammar)}
return (label, None, len(filtered), meta)
@ -476,9 +472,8 @@ def _infer_group(label, group_seqs, group_files, project_root, min_coverage, pre
from ..idregex import idregex
from ..mdl import lang_size_score, model_cost
idr_g = idregex(symbol_seqs, kmax=kmax, N=N)
if idr_g and idr_g != '∅':
ok_idr, _ = validate_sore(idr_g)
if ok_idr and model_cost(idr_g) >= 2:
if idr_g and not isinstance(idr_g, Empty):
if model_cost(idr_g) >= 2:
crx_lang = lang_size_score(grammar, symbol_seqs)
idr_lang = lang_size_score(idr_g, symbol_seqs)
if crx_lang > 0 and idr_lang > 0 and crx_lang / idr_lang > 10:
@ -873,7 +868,7 @@ def _build_json_output(results):
}
if result and result.get("best"):
entry["algorithm"] = result["best"]["algorithm"]
entry["grammar"] = result["best"]["grammar"]
entry["grammar"] = to_gbnf(result["best"]["grammar"])
entry["mdl_score"] = round(result['best']['mdl_score'], 1)
entry["imports"] = meta.get("imports", [])
entry["arg_patterns"] = meta.get("arg_patterns", {})
@ -920,8 +915,7 @@ def _build_yaml_output(results, dir_path, max_mdl=200.0, min_structure=0.0):
entry = {
"package": label,
"methods": count,
"grammar": best["grammar"],
"gbnf": to_gbnf(best["grammar"]),
"grammar": to_gbnf(best["grammar"]),
"score": round(best.get("mdl_score", 0), 3),
"algorithm": best["algorithm"],
"mdl": round(best["mdl_score"], 1),
@ -1095,7 +1089,7 @@ def main():
best = result["best"]
print(f" ╰─ {label} ({count} methods)")
print(f" Algorithm: {best['algorithm']}")
print(f" Grammar: {best['grammar']}")
print(f" Grammar: {to_gbnf(best['grammar'])}")
print(f" Score: {best['mdl_score']}")
else:
reason = meta.get("skip_reason", "")

181
bex/template.py
View file

@ -1,96 +1,31 @@
"""
template — One-Shot YAML Template Generator.
"""template — One-Shot YAML Template Generator from AST nodes."""
Converts an inferred k-ORE/SORE/CHARE expression back into
a human-readable YAML skeleton.
Generates:
- A YAML scaffold with placeholders
- Cardinality annotations:
* # REQUIRED: Exactly 1
* # REPEATED: 1 or more
* # OPTIONAL: 0 or 1
* # VARIABLE: 0 or more
* # CHOOSE: alternative module
"""
from .grammar import Symbol, Concat, Alt, Plus, Optional, Star, Epsilon, Empty
def parse_expression(expr):
"""Split a regular expression into its components."""
if not expr or expr in ('∅', 'ε', ''):
return [('empty', 'ε')]
tokens = []
i = 0
while i < len(expr):
if expr[i] == '(':
depth = 1
j = i + 1
while j < len(expr) and depth > 0:
if expr[j] == '(':
depth += 1
elif expr[j] == ')':
depth -= 1
j += 1
group = expr[i:j]
quantifier = ''
if j < len(expr) and expr[j] in '*+?':
quantifier = expr[j]
j += 1
tokens.append(('group', group, quantifier))
i = j
elif expr[i] == '|':
tokens.append(('pipe', '|'))
i += 1
elif expr[i] == '.':
if i + 1 < len(expr) and expr[i + 1] == '.':
tokens.append(('concat', '..'))
i += 2
else:
tokens.append(('concat', '.'))
i += 1
elif expr[i] in '*+?':
if tokens and tokens[-1][0] == 'name':
name, val, _ = tokens[-1]
tokens[-1] = (name, val, expr[i])
i += 1
elif expr[i].isalnum() or expr[i] in '/_-':
j = i
while j < len(expr) and (expr[j].isalnum() or expr[j] in '/_-'):
j += 1
name = expr[i:j]
tokens.append(('name', name, ''))
i = j
else:
i += 1
return tokens
def format_cardinality(node):
"""Return cardinality description for an AST node."""
if isinstance(node, Plus):
return '# PFLICHT: 1 oder mehrmals erforderlich'
if isinstance(node, Optional):
return '# OPTIONAL: 0 oder 1 mal (darf weggelassen werden)'
if isinstance(node, Star):
return '# OPTIONAL: 0 oder mehrmals'
return '# PFLICHT: Genau 1 mal erforderlich'
def format_prompt_cardinality(quantifier):
"""Return the cardinality description for a quantifier."""
mapping = {
'': '# PFLICHT: Genau 1 mal erforderlich',
'+': '# PFLICHT: 1 oder mehrmals erforderlich',
'*': '# OPTIONAL: 0 oder mehrmals',
'?': '# OPTIONAL: 0 oder 1 mal (darf weggelassen werden)',
}
return mapping.get(quantifier, '')
def generate_template(expr, context_key=None, include_header=True):
"""
Generate a YAML one-shot template from a regular expression.
def generate_template(node, context_key=None, include_header=True):
"""Generate a YAML one-shot template from an AST node.
Args:
expr: Inferred expression (string)
node: Grammar AST node
context_key: YAML container key (e.g. 'tasks')
include_header: Whether to include header section (name, hosts)
Returns:
YAML skeleton with placeholders and cardinality comments
"""
if not expr or expr in ('∅', 'ε'):
if node is None or isinstance(node, Empty):
return "# No structure inferred (empty sequences or no examples)"
if include_header:
@ -111,44 +46,60 @@ def generate_template(expr, context_key=None, include_header=True):
lines.append(" tasks:")
indent = " "
tokens = parse_expression(expr)
task_index = 0
skip_until_pipe = False
_generate_node(node, lines, indent)
return '\n'.join(lines) + '\n'
alternatives = []
in_alternatives = False
i = 0
while i < len(tokens):
token = tokens[i]
if token[0] == 'group':
group_str = token[1]
quantifier = token[2]
card = format_prompt_cardinality(quantifier)
inner_expr = group_str[1:-1]
if '|' in inner_expr:
alts = inner_expr.split('|')
def _generate_node(node, lines, indent):
"""Recursively generate YAML lines from AST node."""
if isinstance(node, Symbol):
lines.append(f"{indent}- {node.value}: <params>")
elif isinstance(node, Plus):
card = format_cardinality(node)
_generate_wrapped(node.child, lines, indent, card)
elif isinstance(node, Optional):
card = format_cardinality(node)
_generate_wrapped(node.child, lines, indent, card)
elif isinstance(node, Star):
card = format_cardinality(node)
_generate_wrapped(node.child, lines, indent, card)
elif isinstance(node, Alt):
lines.append(f"{indent}# CHOOSE (pick one):")
for alt in alts:
alt_clean = alt.strip()
lines.append(f"{indent}# - {alt_clean}: <params>")
if card:
lines[-1] = f"{lines[-1]} {card}"
else:
lines.append(f"{indent}- {inner_expr}: <params> {card}")
task_index += 1
elif token[0] == 'name':
name = token[1]
quantifier = token[2]
card = format_prompt_cardinality(quantifier)
lines.append(f"{indent}- {name}: <params> {card}")
task_index += 1
elif token[0] == 'pipe':
for part in node.parts:
name = _node_name(part)
lines.append(f"{indent}# - {name}: <params>")
elif isinstance(node, Concat):
for part in node.parts:
_generate_node(part, lines, indent)
elif isinstance(node, Epsilon):
pass
elif isinstance(node, Empty):
pass
i += 1
return '\n'.join(lines) + '\n'
def _generate_wrapped(child, lines, indent, card):
"""Generate a child node with cardinality annotation."""
if isinstance(child, Alt):
lines.append(f"{indent}# CHOOSE (pick one):")
for part in child.parts:
name = _node_name(part)
lines.append(f"{indent}# - {name}: <params>")
lines[-1] = f"{lines[-1]} {card}"
elif isinstance(child, Symbol):
lines.append(f"{indent}- {child.value}: <params> {card}")
elif isinstance(child, Concat):
for part in child.parts:
_generate_node(part, lines, indent)
else:
_generate_node(child, lines, indent)
def _node_name(node):
"""Get a human-readable name for a node."""
if isinstance(node, Symbol):
return node.value
if isinstance(node, Concat):
return ".".join(_node_name(p) for p in node.parts)
if isinstance(node, Alt):
return "|".join(_node_name(p) for p in node.parts)
return "..."

107
tests/test_bex.py
View file

@ -10,6 +10,10 @@ from bex.koa import KOA, build_complete_koa
from bex.marking import mark_koa
from bex.rwrsq import rwr_sq, strip
from bex.ikoa import ikoa
from bex.grammar import (
Symbol, Concat, Alt, Plus, Optional, Star, Epsilon, Empty,
parse_sore, match as grammar_match, render_sore,
)
def test_soa_basics():
@ -33,9 +37,11 @@ def test_soa_contract():
G.add_edge(G.src, a)
G.add_edge(a, b)
G.add_edge(b, G.sink)
G.contract(a, b, concat('a', 'b'))
G.contract(a, b, concat(Symbol('a'), Symbol('b')))
assert G.is_final()
assert G.expression() == 'a.b'
expr = G.expression()
assert isinstance(expr, Concat)
assert expr.parts == [Symbol('a'), Symbol('b')]
print(" PASS test_soa_contract")
@ -69,7 +75,7 @@ def test_rwr0_concat():
G.add_edge(a, b)
G.add_edge(b, G.sink)
result = rwr0(G)
assert result == 'a.b', f"Expected 'a.b', got {result}"
assert isinstance(result, Concat), f"Expected Concat, got {type(result).__name__}: {result}"
print(" PASS test_rwr0_concat")
@ -82,7 +88,7 @@ def test_rwr0_disj():
G.add_edge(a, G.sink)
G.add_edge(b, G.sink)
result = rwr0(G)
assert result == '(a|b)', f"Expected '(a|b)', got {result}"
assert isinstance(result, Alt), f"Expected Alt, got {type(result).__name__}: {result}"
print(" PASS test_rwr0_disj")
@ -93,7 +99,7 @@ def test_rwr0_iteration():
G.add_edge(a, G.sink)
G.add_edge(a, a)
result = rwr0(G)
assert result == 'a+', f"Expected 'a+', got {result}"
assert isinstance(result, Plus), f"Expected Plus, got {type(result).__name__}: {result}"
print(" PASS test_rwr0_iteration")
@ -103,15 +109,14 @@ def test_rwr0_optional():
G.add_edge(G.src, a)
G.add_edge(a, G.sink)
result = rwr0(G)
# Single state src→a→sink: language is {a}, not {a,ε}
assert result == 'a', f"Expected 'a', got {result}"
assert isinstance(result, Symbol), f"Expected Symbol, got {type(result).__name__}: {result}"
print(" PASS test_rwr0_optional")
def test_rwr0_empty():
G = SOA()
result = rwr0(G)
assert result == '∅', f"Expected '∅', got {result}"
assert isinstance(result, Empty), f"Expected Empty, got {type(result).__name__}: {result}"
print(" PASS test_rwr0_empty")
@ -119,7 +124,7 @@ def test_rwr0_epsilon():
G = SOA()
G.add_edge(G.src, G.sink)
result = rwr0(G)
assert result == 'ε', f"Expected 'ε', got {result}"
assert isinstance(result, Epsilon), f"Expected Epsilon, got {type(result).__name__}: {result}"
print(" PASS test_rwr0_epsilon")
@ -127,7 +132,7 @@ def test_rwr0_complex_a():
# {abc, ab, ac} is NOT a SORE language (c appears in two roles)
G = build_soa([['a', 'b', 'c'], ['a', 'b'], ['a', 'c']])
result = rwr0(G)
assert result == '∅', f"Expected ∅ for non-SORE, got {result}"
assert isinstance(result, Empty), f"Expected Empty for non-SORE, got {type(result).__name__}: {result}"
print(" PASS test_rwr0_complex_a: ∅ (non-SORE)")
@ -142,9 +147,10 @@ def test_rwr0_disj_concat():
def test_crx_simple():
crx = CRX()
result = crx.infer([['a', 'b'], ['a', 'b', 'c']])
assert result is not None and result != '∅'
assert 'a' in result
assert 'b' in result
assert not isinstance(result, Empty)
alpha = alphabet(result)
assert 'a' in alpha
assert 'b' in alpha
print(f" PASS test_crx_simple: {result}")
@ -159,7 +165,7 @@ def test_crx_example():
]
result = crx.infer(S)
assert result is not None
assert '(' in result # should have disjunction factors
assert isinstance(result, (Alt, Concat)) or isinstance(result, Symbol)
print(f" PASS test_crx_example: {result}")
@ -169,7 +175,8 @@ def test_crx_cycle_class():
S = [['a', 'b', 'c'], ['b', 'c', 'a'], ['c', 'a', 'b']]
result = crx.infer(S)
assert result is not None
assert 'a' in result and 'b' in result and 'c' in result
alpha = alphabet(result)
assert 'a' in alpha and 'b' in alpha and 'c' in alpha
print(f" PASS test_crx_cycle_class: {result}")
@ -202,13 +209,20 @@ def test_strip():
def test_expr_utils():
assert concat('a', 'b') == 'a.b'
assert disj('a', 'b') == '(a|b)'
assert star('a') == 'a+'
assert optional('a') == 'a?'
assert optional('a.b') == '(a.b)?'
assert alphabet('a.b') == {'a', 'b'}
assert alphabet('(a|b)+') == {'a', 'b'}
c = concat(Symbol('a'), Symbol('b'))
assert isinstance(c, Concat) and c.parts == [Symbol('a'), Symbol('b')]
d = disj(Symbol('a'), Symbol('b'))
assert isinstance(d, Alt) and d.parts == [Symbol('a'), Symbol('b')]
s = star(Symbol('a'))
assert isinstance(s, Plus) and s.child == Symbol('a')
o = optional(Symbol('a'))
assert isinstance(o, Optional) and o.child == Symbol('a')
o2 = optional(concat(Symbol('a'), Symbol('b')))
assert isinstance(o2, Optional) and isinstance(o2.child, Concat)
alpha = alphabet(concat(Symbol('a'), Symbol('b')))
assert alpha == {'a', 'b'}
alpha2 = alphabet(Plus(Alt([Symbol('a'), Symbol('b')])))
assert alpha2 == {'a', 'b'}
assert strip_k('a_1') == 'a'
print(" PASS test_expr_utils")
@ -232,42 +246,6 @@ def test_complete_koa():
print(" PASS test_complete_koa")
def run_all():
tests = [
test_soa_basics,
test_soa_contract,
test_soa_epsilon_closure,
test_twotinf,
test_rwr0_concat,
test_rwr0_disj,
test_rwr0_iteration,
test_rwr0_optional,
test_rwr0_empty,
test_rwr0_epsilon,
test_rwr0_complex_a,
test_rwr0_disj_concat,
test_crx_simple,
test_crx_example,
test_crx_cycle_class,
test_determinism_check,
test_marking,
test_strip,
test_expr_utils,
test_idregex_deterministic,
test_complete_koa,
]
passed = 0
failed = 0
for t in tests:
try:
t()
passed += 1
except Exception as e:
print(f" FAIL {t.__name__}: {e}")
failed += 1
print(f"\n{passed} passed, {failed} failed")
# ── Integration tests with real Ansible task data ──
def test_integration_linear_sequence():
@ -279,7 +257,9 @@ def test_integration_linear_sequence():
crx = CRX()
result = crx.infer(seqs)
assert result is not None
assert all(t in result for t in ['file', 'template', 'docker_image', 'command', 'set_fact', 'shell', 'wait_for'])
alpha = alphabet(result)
for t in ['file', 'template', 'docker_image', 'command', 'set_fact', 'shell', 'wait_for']:
assert t in alpha, f"Expected '{t}' in alphabet"
print(f" PASS linear_sequence: {result}")
@ -293,7 +273,8 @@ def test_integration_optional_tasks():
crx = CRX()
result = crx.infer(seqs)
assert result is not None
assert 'shell' in result and 'debug' in result
alpha = alphabet(result)
assert 'shell' in alpha and 'debug' in alpha
print(f" PASS optional_tasks: {result}")
@ -306,7 +287,8 @@ def test_integration_branching_paths():
crx = CRX()
result = crx.infer(seqs)
assert result is not None
assert 'file' in result and 'template' in result and 'shell' in result
alpha = alphabet(result)
assert 'file' in alpha and 'template' in alpha and 'shell' in alpha
print(f" PASS branching_paths: {result}")
@ -320,7 +302,8 @@ def test_integration_conditional_tasks():
crx = CRX()
result = crx.infer(seqs)
assert result is not None
assert 'assert' in result and 'file' in result
alpha = alphabet(result)
assert 'assert' in alpha and 'file' in alpha
print(f" PASS conditional_tasks: {result}")

152
tests/test_crx_dotted.py Normal file
View file

@ -0,0 +1,152 @@
"""Tests for CRX with dotted symbols (e.g. 'foo.bar', 'request.body').
These tests expose the fundamental bug: CRX outputs grammars where '.'
means both concatenation AND is part of symbol names. The parser splits
on all dots, breaking matches.
Every test here MUST pass after the fix. They all fail before.
"""
from bex.crx import CRX
from bex.ensemble import _matches
class TestCRXDotInSymbol:
"""CRX must handle symbols containing dots."""
def test_single_dotted_symbol(self):
"""One symbol with a dot: ['foo.bar']."""
crx = CRX()
g = crx.infer([['foo.bar']])
assert g is not None
assert _matches(g, ['foo.bar'])
def test_two_dotted_symbols(self):
"""Two different dotted symbols: ['foo.bar', 'baz.qux']."""
crx = CRX()
g = crx.infer([['foo.bar', 'baz.qux']])
assert g is not None
assert _matches(g, ['foo.bar', 'baz.qux'])
def test_dotted_symbol_repeated(self):
"""Dotted symbol appears multiple times: ['foo.bar', 'foo.bar', 'baz']."""
crx = CRX()
g = crx.infer([['foo.bar', 'foo.bar', 'baz']])
assert g is not None
assert _matches(g, ['foo.bar', 'foo.bar', 'baz'])
def test_dotted_symbols_multiple_sequences(self):
"""Multiple sequences with dotted symbols."""
crx = CRX()
seqs = [
['return.capability', 'invoke.request'],
['return.capability', 'invoke'],
]
g = crx.infer(seqs)
assert g is not None
for seq in seqs:
assert _matches(g, seq), f"Grammar {g!r} should match {seq}"
def test_dotted_symbol_alternation(self):
"""Dotted symbols in alternation: ['a.b', 'a.c']."""
crx = CRX()
g = crx.infer([['a.b'], ['a.c']])
assert g is not None
assert _matches(g, ['a.b'])
assert _matches(g, ['a.c'])
def test_dotted_symbol_optional(self):
"""Optional dotted symbol: some sequences have it, some don't."""
crx = CRX()
seqs = [
['a.b', 'c'],
['c'],
]
g = crx.infer(seqs)
assert g is not None
for seq in seqs:
assert _matches(g, seq), f"Grammar {g!r} should match {seq}"
def test_dotted_symbol_plus(self):
"""Repeated dotted symbol: ['a.b', 'a.b', 'a.b', 'c']."""
crx = CRX()
seqs = [
['a.b', 'a.b', 'a.b', 'c'],
['a.b', 'a.b', 'c'],
['a.b', 'c'],
]
g = crx.infer(seqs)
assert g is not None
for seq in seqs:
assert _matches(g, seq), f"Grammar {g!r} should match {seq}"
def test_realistic_method_chain(self):
"""Realistic: method call chains like Kotlin/Python."""
crx = CRX()
seqs = [
['request.body', 'validate', 'save'],
['request.body', 'validate', 'return'],
['request.body', 'save'],
]
g = crx.infer(seqs)
assert g is not None
for seq in seqs:
assert _matches(g, seq), f"Grammar {g!r} should match {seq}"
def test_multiple_dots_in_symbol(self):
"""Symbol with multiple dots: 'a.b.c'."""
crx = CRX()
seqs = [
['a.b.c', 'd.e.f'],
['a.b.c', 'g'],
]
g = crx.infer(seqs)
assert g is not None
for seq in seqs:
assert _matches(g, seq), f"Grammar {g!r} should match {seq}"
def test_mixed_dotted_and_plain(self):
"""Mix of dotted and plain symbols."""
crx = CRX()
seqs = [
['foo.bar', 'baz', 'qux.quux'],
['foo.bar', 'baz'],
['baz', 'qux.quux'],
]
g = crx.infer(seqs)
assert g is not None
for seq in seqs:
assert _matches(g, seq), f"Grammar {g!r} should match {seq}"
class TestEnsembleDotInSymbol:
"""Ensemble (CRX + scoring) must handle dotted symbols."""
def test_ensemble_dotted_symbols(self):
"""infer_ensemble with dotted symbols."""
from bex.ensemble import infer_ensemble
seqs = [
['request.body', 'validate', 'save'],
['request.body', 'validate', 'return'],
['request.body', 'save'],
]
result = infer_ensemble(seqs)
assert result['best'] is not None
grammar = result['best']['grammar']
for seq in seqs:
assert _matches(grammar, seq), \
f"Grammar {grammar!r} should match {seq}"
def test_scoring_dotted_symbols(self):
"""lang_size_score with dotted symbols should not crash."""
from bex.ensemble import infer_ensemble
seqs = [
['foo.bar', 'baz.qux'],
['foo.bar', 'baz'],
]
result = infer_ensemble(seqs)
assert result['best'] is not None
# Just checking it doesn't crash — the score should be finite
score = result['best']['mdl_score']
assert isinstance(score, (int, float))
assert score < float('inf')

305
tests/test_grammar.py Normal file
View file

@ -0,0 +1,305 @@
"""Tests for bex/grammar.py — canonical AST."""
import pytest
from bex.grammar import (
Symbol, Concat, Alt, Plus, Optional, Star, Epsilon, Empty,
parse_sore, render_sore, alphabet, count_words, model_cost,
lang_size, match,
)
# ---------------------------------------------------------------------------
# Node construction and equality
# ---------------------------------------------------------------------------
class TestNodes:
def test_symbol_eq(self):
assert Symbol('a') == Symbol('a')
assert Symbol('a') != Symbol('b')
def test_concat_eq(self):
assert Concat([Symbol('a'), Symbol('b')]) == Concat([Symbol('a'), Symbol('b')])
assert Concat([Symbol('a')]) != Concat([Symbol('b')])
def test_alt_eq(self):
assert Alt([Symbol('a'), Symbol('b')]) == Alt([Symbol('a'), Symbol('b')])
def test_plus_eq(self):
assert Plus(Symbol('a')) == Plus(Symbol('a'))
assert Plus(Symbol('a')) != Plus(Symbol('b'))
def test_optional_eq(self):
assert Optional(Symbol('a')) == Optional(Symbol('a'))
def test_star_eq(self):
assert Star(Symbol('a')) == Star(Symbol('a'))
def test_epsilon_eq(self):
assert Epsilon() == Epsilon()
assert Epsilon() != Empty()
def test_empty_eq(self):
assert Empty() == Empty()
def test_hash(self):
s = {Symbol('a'), Symbol('b'), Symbol('a')}
assert len(s) == 2
# ---------------------------------------------------------------------------
# Parse + render roundtrip
# ---------------------------------------------------------------------------
class TestRoundtrip:
@pytest.mark.parametrize('sore', [
'a',
'a.b',
'a.b.c',
'(a+b)',
'(a+b+c)',
'a+',
'a?',
'a*',
'a.b+',
'a.b?.c+',
'(a+b).c',
'(a+b)+',
'(a+b)?',
'(a+b)*',
'ε',
'∅',
])
def test_roundtrip(self, sore):
node = parse_sore(sore)
rendered = render_sore(node)
assert rendered == sore, f'{sore} -> {rendered}'
def test_dotted_symbol_roundtrip(self):
node = parse_sore('foo\\.bar.baz+')
rendered = render_sore(node)
assert rendered == 'foo\\.bar.baz+'
assert node == Concat([Symbol('foo.bar'), Plus(Symbol('baz'))])
def test_deeply_nested(self):
sore = '((a+b)+.c)?'
node = parse_sore(sore)
rendered = render_sore(node)
assert rendered == sore
# ---------------------------------------------------------------------------
# Parse edge cases
# ---------------------------------------------------------------------------
class TestParse:
def test_empty_string(self):
assert parse_sore('') == Empty()
def test_epsilon(self):
assert parse_sore('ε') == Epsilon()
def test_empty_symbol(self):
assert parse_sore('∅') == Empty()
def test_single_symbol(self):
assert parse_sore('mockk') == Symbol('mockk')
def test_concat(self):
assert parse_sore('a.b.c') == Concat([Symbol('a'), Symbol('b'), Symbol('c')])
def test_alt(self):
assert parse_sore('(a+b+c)') == Alt([Symbol('a'), Symbol('b'), Symbol('c')])
def test_plus_outside_parens(self):
assert parse_sore('a+') == Plus(Symbol('a'))
def test_optional(self):
assert parse_sore('a?') == Optional(Symbol('a'))
def test_star(self):
assert parse_sore('a*') == Star(Symbol('a'))
def test_compound_quantifier_plus_optional(self):
# a+? = a* (one or more, optional = zero or more)
node = parse_sore('a+?')
assert isinstance(node, Star)
assert node.child == Symbol('a')
def test_compound_quantifier_question_plus(self):
# a?+ = a+ (optional then one or more = one or more)
node = parse_sore('a?+')
assert isinstance(node, Plus)
assert node.child == Symbol('a')
def test_dotted_symbol(self):
node = parse_sore('foo\\.bar')
assert node == Symbol('foo.bar')
def test_dotted_in_concat(self):
node = parse_sore('foo\\.bar.baz')
assert node == Concat([Symbol('foo.bar'), Symbol('baz')])
# ---------------------------------------------------------------------------
# Alphabet
# ---------------------------------------------------------------------------
class TestAlphabet:
def test_symbol(self):
assert alphabet(Symbol('a')) == {'a'}
def test_concat(self):
assert alphabet(Concat([Symbol('a'), Symbol('b')])) == {'a', 'b'}
def test_alt(self):
assert alphabet(Alt([Symbol('a'), Symbol('b'), Symbol('c')])) == {'a', 'b', 'c'}
def test_plus(self):
assert alphabet(Plus(Symbol('a'))) == {'a'}
def test_nested(self):
g = parse_sore('(a.b+).c?')
assert alphabet(g) == {'a', 'b', 'c'}
def test_epsilon(self):
assert alphabet(Epsilon()) == set()
def test_empty(self):
assert alphabet(Empty()) == set()
def test_dotted_symbol(self):
assert alphabet(Symbol('foo.bar')) == {'foo.bar'}
# ---------------------------------------------------------------------------
# Match
# ---------------------------------------------------------------------------
class TestMatch:
def test_symbol_match(self):
assert match(Symbol('a'), ['a'])
assert not match(Symbol('a'), ['b'])
assert not match(Symbol('a'), [])
def test_concat_match(self):
g = Concat([Symbol('a'), Symbol('b'), Symbol('c')])
assert match(g, ['a', 'b', 'c'])
assert not match(g, ['a', 'b'])
assert not match(g, ['a', 'b', 'c', 'd'])
def test_alt_match(self):
g = Alt([Symbol('a'), Symbol('b')])
assert match(g, ['a'])
assert match(g, ['b'])
assert not match(g, ['c'])
def test_plus_match(self):
g = Plus(Symbol('a'))
assert match(g, ['a'])
assert match(g, ['a', 'a'])
assert match(g, ['a', 'a', 'a'])
assert not match(g, [])
assert not match(g, ['b'])
def test_optional_match(self):
g = Optional(Symbol('a'))
assert match(g, ['a'])
assert match(g, [])
assert not match(g, ['b'])
def test_star_match(self):
g = Star(Symbol('a'))
assert match(g, [])
assert match(g, ['a'])
assert match(g, ['a', 'a', 'a'])
assert not match(g, ['b'])
def test_complex_grammar(self):
g = parse_sore('init+.capability+.(invoke+.request)?')
assert match(g, ['init', 'capability', 'invoke', 'request'])
assert match(g, ['init', 'init', 'capability', 'capability'])
assert not match(g, ['capability'])
def test_dotted_symbols(self):
g = parse_sore('foo\\.bar.baz+')
assert match(g, ['foo.bar', 'baz'])
assert match(g, ['foo.bar', 'baz', 'baz'])
assert not match(g, ['foo', 'bar', 'baz'])
def test_epsilon(self):
assert match(Epsilon(), [])
assert not match(Epsilon(), ['a'])
def test_empty(self):
assert not match(Empty(), [])
assert not match(Empty(), ['a'])
# ---------------------------------------------------------------------------
# Count words
# ---------------------------------------------------------------------------
class TestCountWords:
def test_symbol(self):
assert count_words(Symbol('a'), 0) == 0
assert count_words(Symbol('a'), 1) == 1
assert count_words(Symbol('a'), 2) == 0
def test_concat(self):
g = Concat([Symbol('a'), Symbol('b')])
assert count_words(g, 0) == 0
assert count_words(g, 1) == 0
assert count_words(g, 2) == 1
def test_alt(self):
g = Alt([Symbol('a'), Symbol('b')])
assert count_words(g, 1) == 2
def test_plus(self):
g = Plus(Symbol('a'))
assert count_words(g, 0) == 0
assert count_words(g, 1) == 1
assert count_words(g, 2) == 1
assert count_words(g, 3) == 1
def test_optional(self):
g = Optional(Symbol('a'))
assert count_words(g, 0) == 1
assert count_words(g, 1) == 1
assert count_words(g, 2) == 0
def test_star(self):
g = Star(Symbol('a'))
assert count_words(g, 0) == 1
assert count_words(g, 1) == 1
assert count_words(g, 2) == 1
def test_epsilon(self):
assert count_words(Epsilon(), 0) == 1
assert count_words(Epsilon(), 1) == 0
def test_empty(self):
assert count_words(Empty(), 0) == 0
# ---------------------------------------------------------------------------
# Model cost and language size
# ---------------------------------------------------------------------------
class TestScoring:
def test_model_cost_symbol(self):
assert model_cost(Symbol('a')) == 1
def test_model_cost_concat(self):
assert model_cost(Concat([Symbol('a'), Symbol('b')])) == 2
def test_model_cost_plus(self):
assert model_cost(Plus(Symbol('a'))) == 1
def test_lang_size_symbol(self):
# a: words of length 0 = 0, length 1 = 1, total = 1
assert lang_size(Symbol('a'), 1) == 1
def test_lang_size_concat(self):
# a.b: words of length 0 = 0, length 1 = 0, length 2 = 1
assert lang_size(Concat([Symbol('a'), Symbol('b')]), 2) == 1

281
tests/test_scoring.py
View file

@ -10,13 +10,21 @@ Tests cover:
"""
import pytest
from bex.grammar import (
Symbol, Concat, Alt, Plus, Optional, Star, Epsilon, Empty,
parse_sore, count_words, lang_size, model_cost, alphabet,
)
from bex.mdl import (
model_cost, data_cost, lang_size, lang_size_score,
mdl_score, score_grammar, _count_words_fast,
model_cost as mdl_model_cost, data_cost, lang_size_score,
mdl_score, score_grammar,
)
from bex.ensemble import infer_ensemble
from bex.crx import CRX
from bex.idregex import idregex
# ── Helper to parse SORE string to AST ──
def p(s):
return parse_sore(s)
# ── Paper's Language Size: cumulative |L(r)≤n| ──
@ -26,7 +34,7 @@ class TestPaperLanguageSize:
def test_paper_example_a_dot_a_c_plus(self):
"""Paper's example: a.(a+c+)? has m=3, n=7, |L≤7|=3."""
expr = 'a.(a+c+)?'
expr = p('a.(a+c+)?')
m = model_cost(expr)
n = 2 * m + 1
assert m == 3, f"model_cost should be 3, got {m}"
@ -36,48 +44,33 @@ class TestPaperLanguageSize:
def test_paper_same_n_specific_wins(self):
"""At same n, specific grammar beats generic."""
n = 7 # target n for a.b.c
specific = lang_size('a.b.c', n)
generic = lang_size('(a+b+c)+', n)
assert specific < generic, (
f"Specific ({specific}) should beat generic ({generic}) at n={n}"
)
n = 7
specific = lang_size(p('a.b.c'), n)
generic = lang_size(p('(a+b+c)+'), n)
assert specific < generic
def test_paper_same_n_correct_beats_overfit(self):
"""At same n, correct grammar and overfit tie (both accept 1 word)."""
n = 7
correct = lang_size('a.b.c', n)
overfit = lang_size('a.a.a', n)
assert correct == overfit == 1, (
f"Both should accept 1 word at n={n}, got {correct} and {overfit}"
)
correct = lang_size(p('a.b.c'), n)
overfit = lang_size(p('a.a.a'), n)
assert correct == overfit == 1
def test_paper_per_candidate_n_generic_wins_unfairly(self):
"""Per-candidate n lets generic patterns win unfairly."""
# info+ has m=1, n=3 → counts words at lengths 0,1,2,3
# specific has m=5, n=11 → counts words at lengths 0..11
generic_n = 2 * model_cost('info+') + 1 # = 3
specific_n = 2 * model_cost('info.file.template.shell.service+') + 1 # = 11
generic_ls = lang_size('info+', generic_n)
specific_ls = lang_size('info.file.template.shell.service+', specific_n)
# Generic wins on paper (3 < 7) but this is wrong
assert generic_ls < specific_ls, (
f"Per-candidate n: generic ({generic_ls}) beats specific ({specific_ls}) — this is the bug"
)
generic_n = 2 * model_cost(p('info+')) + 1
specific_n = 2 * model_cost(p('info.file.template.shell.service+')) + 1
generic_ls = lang_size(p('info+'), generic_n)
specific_ls = lang_size(p('info.file.template.shell.service+'), specific_n)
assert generic_ls < specific_ls
def test_paper_alphabet_size_5_mdL_vs_langsize(self):
"""Paper's result: Language Size 98% vs MDL 21% on alphabet size 5."""
# At the same n, language size correctly differentiates
n = 7
specific = lang_size('a.b.c', n) # 1 word
generic = lang_size('(a+b+c)+', n) # 3,279 words
medium = lang_size('a.(b+c)?', n) # 3 words
assert specific < medium < generic, (
f"Order should be specific({specific}) < medium({medium}) < generic({generic})"
)
specific = lang_size(p('a.b.c'), n)
generic = lang_size(p('(a+b+c)+'), n)
medium = lang_size(p('a.(b+c)?'), n)
assert specific < medium < generic
# ── Our Adaptation: words at exact sequence lengths ──
@ -86,133 +79,97 @@ class TestAdaptedLanguageSize:
"""Tests for our adaptation (counting at exact sequence lengths)."""
def test_specific_vs_generic_diverse_lengths(self):
"""With diverse lengths, specific grammar wins clearly."""
seqs = [['a', 'b', 'c'], ['a', 'b'], ['a', 'c'], ['b', 'c']]
specific = lang_size_score('a.b.c', seqs)
generic = lang_size_score('(a+b+c)+', seqs)
assert specific < generic, (
f"Specific ({specific}) should beat generic ({generic})"
)
specific = lang_size_score(p('a.b.c'), seqs)
generic = lang_size_score(p('(a+b+c)+'), seqs)
assert specific < generic
def test_info_plus_vs_specific_identical_lengths(self):
"""With identical lengths, both accept 1 word — honest tie."""
seqs = [['info', 'file', 'template', 'shell', 'service']] * 5
generic = lang_size_score('info+', seqs)
specific = lang_size_score('info.file.template.shell.service+', seqs)
assert generic == specific == 5, (
f"Both should score 5 (1 word × 5 seqs), got generic={generic}, specific={specific}"
)
generic = lang_size_score(p('info+'), seqs)
specific = lang_size_score(p('info.file.template.shell.service+'), seqs)
assert generic == specific == 5
def test_generic_vs_more_generic(self):
"""(a+b+c)+ accepts more words than a+ at each length."""
seqs = [['a', 'b', 'c']] * 3
less_generic = lang_size_score('a+', seqs)
more_generic = lang_size_score('(a+b+c)+', seqs)
# a+ accepts 1 word at each length; (a+b+c)+ accepts 3^L
assert less_generic < more_generic, (
f"a+ ({less_generic}) should beat (a+b+c)+ ({more_generic})"
)
less_generic = lang_size_score(p('a+'), seqs)
more_generic = lang_size_score(p('(a+b+c)+'), seqs)
assert less_generic < more_generic
def test_single_sequence(self):
"""Single sequence — specific grammar wins."""
seqs = [['a', 'b', 'c']]
specific = lang_size_score('a.b.c', seqs)
generic = lang_size_score('(a+b+c)+', seqs)
specific = lang_size_score(p('a.b.c'), seqs)
generic = lang_size_score(p('(a+b+c)+'), seqs)
assert specific < generic
def test_long_sequences(self):
"""Long sequences — specific grammar still wins."""
seqs = [['a', 'b', 'c', 'd', 'e']] * 3
specific = lang_size_score('a.b.c.d.e', seqs)
generic = lang_size_score('(a+b+c+d+e)+', seqs)
specific = lang_size_score(p('a.b.c.d.e'), seqs)
generic = lang_size_score(p('(a+b+c+d+e)+'), seqs)
assert specific < generic
def test_empty_sequences(self):
"""Empty sequences — falls back to paper formula."""
score = lang_size_score('a.b.c', [])
expected = lang_size('a.b.c', 2 * model_cost('a.b.c') + 1)
score = lang_size_score(p('a.b.c'), [])
expected = lang_size(p('a.b.c'), 2 * model_cost(p('a.b.c')) + 1)
assert score == expected
def test_ordered_vs_unordered(self):
"""Ordered a.b.c beats unordered (a+b+c)+ on ordered data."""
seqs = [['a', 'b', 'c'], ['a', 'b'], ['a', 'c']]
ordered = lang_size_score('a.b.c', seqs)
unordered = lang_size_score('(a+b+c)+', seqs)
ordered = lang_size_score(p('a.b.c'), seqs)
unordered = lang_size_score(p('(a+b+c)+'), seqs)
assert ordered < unordered
def test_optional_beats_generic(self):
"""a.(b+c)? beats (a+b+c)+ on data where a is always first."""
seqs = [['a', 'b'], ['a', 'c'], ['a']]
optional = lang_size_score('a.(b+c)?', seqs)
generic = lang_size_score('(a+b+c)+', seqs)
optional = lang_size_score(p('a.(b+c)?'), seqs)
generic = lang_size_score(p('(a+b+c)+'), seqs)
assert optional < generic
def test_repeat_beats_concat(self):
"""a+ beats a.a.a on data with varying lengths."""
seqs = [['a'], ['a', 'a'], ['a', 'a', 'a']]
repeat = lang_size_score('a+', seqs)
concat = lang_size_score('a.a.a', seqs)
# a+ accepts 1 word at each length; a.a.a accepts 0 at lengths 1,2 and 1 at length 3
# Total: a+ = 3, a.a.a = 0+0+1 = 1
# a.a.a actually wins because it rejects shorter sequences!
assert concat < repeat, (
f"a.a.a ({concat}) should beat a+ ({repeat}) — a.a.a rejects short seqs"
)
repeat = lang_size_score(p('a+'), seqs)
concat = lang_size_score(p('a.a.a'), seqs)
assert concat < repeat
# ── MDL Fallback ──
class TestMDLFallback:
"""Tests for the old MDL scoring method."""
def test_mdl_basic(self):
"""MDL = model_cost + data_cost."""
score = mdl_score('a.b.c', [['a', 'b', 'c']])
assert score == model_cost('a.b.c') + data_cost('a.b.c', [['a', 'b', 'c']])
score = mdl_score(p('a.b.c'), [['a', 'b', 'c']])
assert score == mdl_model_cost(p('a.b.c')) + data_cost(p('a.b.c'), [['a', 'b', 'c']])
def test_mdl_prefers_short_expressions(self):
"""MDL rewards short expressions — the info+ bug."""
seqs = [['info', 'file', 'template', 'shell', 'service']] * 5
generic = mdl_score('info+', seqs)
specific = mdl_score('info.file.template.shell.service+', seqs)
assert generic < specific, (
f"MDL should pick info+ ({generic}) over specific ({specific}) — this is the bug"
)
generic = mdl_score(p('info+'), seqs)
specific = mdl_score(p('info.file.template.shell.service+'), seqs)
assert generic < specific
def test_score_grammar_method_switch(self):
"""score_grammar dispatches to the correct scorer."""
seqs = [['a', 'b', 'c']]
ls = score_grammar('a.b.c', seqs, method='langsize')
mdl = score_grammar('a.b.c', seqs, method='mdl')
ls = score_grammar(p('a.b.c'), seqs, method='langsize')
mdl = score_grammar(p('a.b.c'), seqs, method='mdl')
assert isinstance(ls, (int, float))
assert isinstance(mdl, (int, float))
def test_score_grammar_invalid_method(self):
"""Invalid method raises ValueError."""
with pytest.raises(ValueError, match="Unknown scoring method"):
score_grammar('a.b.c', [['a']], method='bogus')
score_grammar(p('a.b.c'), [['a']], method='bogus')
def test_langsize_beats_mdl_on_info_plus(self):
"""Language Size ties on info+ scenario; MDL picks info+."""
seqs = [['info', 'file', 'template', 'shell', 'service']] * 5
ls_generic = score_grammar('info+', seqs, method='langsize')
ls_specific = score_grammar('info.file.template.shell.service+', seqs, method='langsize')
mdl_generic = score_grammar('info+', seqs, method='mdl')
mdl_specific = score_grammar('info.file.template.shell.service+', seqs, method='mdl')
# Language Size: tie (honest)
ls_generic = score_grammar(p('info+'), seqs, method='langsize')
ls_specific = score_grammar(p('info.file.template.shell.service+'), seqs, method='langsize')
mdl_generic = score_grammar(p('info+'), seqs, method='mdl')
mdl_specific = score_grammar(p('info.file.template.shell.service+'), seqs, method='mdl')
assert ls_generic == ls_specific, "Language Size should tie"
# MDL: generic wins (the bug)
assert mdl_generic < mdl_specific, "MDL should pick generic (the bug)"
# ── Ensemble Integration ──
class TestEnsembleIntegration:
"""Tests for the ensemble with method parameter."""
def test_ensemble_accepts_method(self):
"""Ensemble accepts method= parameter."""
seqs = [['a', 'b'], ['a', 'b', 'c']]
r_ls = infer_ensemble(seqs, method='langsize')
r_mdl = infer_ensemble(seqs, method='mdl')
@ -220,145 +177,113 @@ class TestEnsembleIntegration:
assert r_mdl['best'] is not None
def test_ensemble_default_is_langsize(self):
"""Default method is langsize."""
seqs = [['a', 'b'], ['a', 'b', 'c']]
r = infer_ensemble(seqs)
assert r['best'] is not None
def test_ensemble_langsize_prefers_specific(self):
"""With diverse sequences, langsize picks the specific grammar."""
seqs = [['a', 'b', 'c'], ['a', 'b'], ['a', 'c'], ['b', 'c']]
r = infer_ensemble(seqs, method='langsize')
# Should pick a.b.c or a.(b+c)? — something specific
best = r['best']['grammar']
# The specific grammar should have a low score
score = r['best']['mdl_score']
assert score < 100, f"Score should be low for specific grammar, got {score}"
def test_ensemble_method_threaded_to_algorithms(self):
"""Method parameter is passed through to scoring."""
seqs = [['a', 'b', 'c'], ['a', 'b']]
r_ls = infer_ensemble(seqs, method='langsize')
r_mdl = infer_ensemble(seqs, method='mdl')
# Both should produce results
assert r_ls['best'] is not None
assert r_mdl['best'] is not None
# Scores may differ
# (not necessarily — depends on what the algorithms produce)
# ── _count_words_fast Correctness ──
# ── count_words Correctness ──
class TestCountWordsFast:
"""Tests for the word counting function used by Language Size."""
def test_single_symbol(self):
"""Single symbol: 1 word of length 1, 0 otherwise."""
assert _count_words_fast('a', 1) == 1
assert _count_words_fast('a', 0) == 0
assert _count_words_fast('a', 2) == 0
assert count_words(p('a'), 1) == 1
assert count_words(p('a'), 0) == 0
assert count_words(p('a'), 2) == 0
def test_concatenation(self):
"""a.b.c: 1 word of length 3, 0 otherwise."""
assert _count_words_fast('a.b.c', 3) == 1
assert _count_words_fast('a.b.c', 2) == 0
assert _count_words_fast('a.b.c', 4) == 0
assert count_words(p('a.b.c'), 3) == 1
assert count_words(p('a.b.c'), 2) == 0
assert count_words(p('a.b.c'), 4) == 0
def test_plus_quantifier(self):
"""a+: 1 word of each length ≥ 1."""
for l in range(1, 6):
assert _count_words_fast('a+', l) == 1
assert _count_words_fast('a+', 0) == 0
assert count_words(p('a+'), l) == 1
assert count_words(p('a+'), 0) == 0
def test_disjunction(self):
"""(a+b+c): 3 words of length 1, 0 otherwise."""
assert _count_words_fast('(a+b+c)', 1) == 3
assert _count_words_fast('(a+b+c)', 0) == 0
assert _count_words_fast('(a+b+c)', 2) == 0
assert count_words(p('(a+b+c)'), 1) == 3
assert count_words(p('(a+b+c)'), 0) == 0
assert count_words(p('(a+b+c)'), 2) == 0
def test_disjunction_plus(self):
"""(a+b+c)+: 3^L words of length L."""
assert _count_words_fast('(a+b+c)+', 1) == 3
assert _count_words_fast('(a+b+c)+', 2) == 9
assert _count_words_fast('(a+b+c)+', 3) == 27
assert count_words(p('(a+b+c)+'), 1) == 3
assert count_words(p('(a+b+c)+'), 2) == 9
assert count_words(p('(a+b+c)+'), 3) == 27
def test_optional(self):
"""a?.(b+c): 2 words of length 2 (ab, ac), 2 words of length 1 (b, c)."""
assert _count_words_fast('a?.(b+c)', 0) == 0
assert _count_words_fast('a?.(b+c)', 1) == 2 # b, c (a? absent)
assert _count_words_fast('a?.(b+c)', 2) == 2 # ab, ac (a? present)
assert count_words(p('a?.(b+c)'), 0) == 0
assert count_words(p('a?.(b+c)'), 1) == 2
assert count_words(p('a?.(b+c)'), 2) == 2
def test_epsilon(self):
"""ε: 1 word of length 0."""
assert _count_words_fast('ε', 0) == 1
assert _count_words_fast('ε', 1) == 0
assert count_words(Epsilon(), 0) == 1
assert count_words(Epsilon(), 1) == 0
def test_empty(self):
"""∅: 0 words at any length."""
assert _count_words_fast('∅', 0) == 0
assert _count_words_fast('∅', 1) == 0
assert count_words(Empty(), 0) == 0
assert count_words(Empty(), 1) == 0
def test_info_plus(self):
"""info+: 1 word of each length ≥ 1 (info repeated L times)."""
for l in range(1, 8):
assert _count_words_fast('info+', l) == 1
assert count_words(p('info+'), l) == 1
def test_info_dot_concat(self):
"""info.file.template: 1 word of length 3, 0 otherwise."""
assert _count_words_fast('info.file.template', 3) == 1
assert _count_words_fast('info.file.template', 2) == 0
assert _count_words_fast('info.file.template', 4) == 0
assert count_words(p('info.file.template'), 3) == 1
assert count_words(p('info.file.template'), 2) == 0
assert count_words(p('info.file.template'), 4) == 0
def test_mixed_disj_concat(self):
"""a.(b+c)+: a followed by 1+ of b or c."""
# length 2: ab, ac (2 words)
assert _count_words_fast('a.(b+c)+', 2) == 2
# length 3: abb, abc, acb, acc (4 words)
assert _count_words_fast('a.(b+c)+', 3) == 4
assert count_words(p('a.(b+c)+'), 2) == 2
assert count_words(p('a.(b+c)+'), 3) == 4
def test_optional_concat(self):
"""a?.b.(c+d): a optional, then b, then c or d."""
assert _count_words_fast('a?.b.(c+d)', 0) == 0
assert _count_words_fast('a?.b.(c+d)', 2) == 2 # bc, bd
assert _count_words_fast('a?.b.(c+d)', 3) == 2 # abc, abd
assert count_words(p('a?.b.(c+d)'), 0) == 0
assert count_words(p('a?.b.(c+d)'), 2) == 2
assert count_words(p('a?.b.(c+d)'), 3) == 2
# ── Regression: info+ Problem ──
class TestInfoPlusRegression:
"""Regression tests for the concrete info+ problem from our codebase."""
def test_info_plus_not_preferred_over_specific(self):
"""info+ should not beat the specific grammar on diverse data."""
seqs = [['info', 'file', 'template', 'shell', 'service']] * 5
generic_score = lang_size_score('info+', seqs)
specific_score = lang_size_score('info.file.template.shell.service+', seqs)
# They tie — which is correct
generic_score = lang_size_score(p('info+'), seqs)
specific_score = lang_size_score(p('info.file.template.shell.service+'), seqs)
assert generic_score == specific_score
def test_info_plus_loses_on_diverse_data(self):
"""info+ loses when sequences have different lengths."""
seqs = [
['info', 'file'],
['info', 'file', 'template'],
['info', 'file', 'template', 'shell'],
]
generic = lang_size_score('info+', seqs)
specific = lang_size_score('info.file.template+', seqs)
assert generic > specific, (
f"info+ ({generic}) should lose to specific ({specific}) on diverse data"
)
generic = lang_size_score(p('info+'), seqs)
specific = lang_size_score(p('info.file.template+'), seqs)
assert generic > specific
def test_crx_does_not_produce_info_plus(self):
"""CRX does not produce info+ for identical sequences."""
seqs = [['info', 'file', 'template', 'shell', 'service']] * 5
g = CRX().infer(seqs)
assert g != 'info+', f"CRX should not produce info+, got {g}"
alpha = alphabet(g)
assert not (len(alpha) == 1 and Symbol('info') in alpha)
def test_ensemble_does_not_pick_info_plus(self):
"""Ensemble does not pick info+ for5 identical sequences."""
seqs = [['info', 'file', 'template', 'shell', 'service']] * 5
r = infer_ensemble(seqs)
assert r['best']['grammar'] != 'info+', (
f"Ensemble should not pick info+, got {r['best']['grammar']}"
)
best = r['best']['grammar']
alpha = alphabet(best)
assert not (len(alpha) == 1 and Symbol('info') in alpha)