check_clipped_cone_events.py

#!/usr/bin/env python3
"""Finite semantics for query-relative clipped-cone events.

Integer calendar keys, comparisons, max, sets and literal cone expansion are
oracles. This is not a native scheduler, producer or pointer-routing model.
"""
from itertools import product
import json


def literal(q, a, b):
    return frozenset(q + ''.join(tail) for n in range(a, b + 1)
                     for tail in product('01', repeat=n-len(q)))


def stream(rows, word):
    calendar = {}
    maximum = 0
    answers = []
    for n in range(1, len(word)+1):
        for q, a, b in rows:
            if len(q) == n-1 and word[:n-1] == q:
                calendar.setdefault(a, []).append(b)
        for b in calendar.pop(n, []):
            maximum = max(maximum, b)
        answers.append(n <= maximum)
    return tuple(answers)


def main():
    rows = (('',1,2), ('',4,5), ('0',2,2), ('0',3,5),
            ('00',3,4), ('1',2,5), ('01',3,3), ('000',4,5))
    words = [''.join(w) for n in range(1,6) for w in product('01',repeat=n)]
    sets = [literal(*r) for r in rows]
    counts = dict(versions=0,query_prefixes=0,negative=0,zero_slack_controls=0,
                  early_activation_controls=0,raw_batch_flip_controls=0)
    snapshots = []
    for mask in range(1 << len(rows)):
        chosen = tuple(r for i,r in enumerate(rows) if mask >> i & 1)
        expected = frozenset().union(*(s for i,s in enumerate(sets) if mask >> i & 1))
        snap = (chosen,expected)
        snapshots.append(snap)
        for word in words:
            got = stream(chosen,word)
            truth = tuple(word[:n] in expected for n in range(1,len(word)+1))
            assert got == truth
            counts['query_prefixes'] += len(got)
            counts['negative'] += got.count(False)
        counts['versions'] += 1
    # Reading other versions does not change retained semantic snapshots.
    for chosen,expected in snapshots:
        assert stream(chosen,'00101') == tuple('00101'[:n] in expected for n in range(1,6))
    # A one-symbol activation delay loses an answer at its final symbol.
    assert stream((('0',2,2),),'00') == (False,True)
    counts['zero_slack_controls'] += 1
    # Taking upper endpoints before lower endpoints become due is unsound.
    assert stream((('0',4,4),),'000') == (False,False,False)
    counts['early_activation_controls'] += 1
    # Same-prefix raw records: every unchecked record could change a negative.
    for h in (1,2,4,8,16,32,64):
        base = [('0',3,3)]*h
        assert not stream(base,'00')[-1]
        for i in range(h):
            changed = base.copy();changed[i]=('0',2,2)
            assert stream(changed,'00')[-1]
            counts['raw_batch_flip_controls'] += 1
    # Existing dictionary primitives: fixed synchronous alias controls only.
    # These finite maps are semantic, not a pointer-level implementation.
    names = ('0','1','00','01','10','11')
    def alias(v, kind):
        state = 0
        out = ''
        for bit in v:
            state ^= int(bit)
            out += bit if kind == 0 else str(state) if kind == 1 else '0'
        return out
    counts['synchronous_alias_queries'] = 0
    for kind in range(3):
        for ready_mask in range(64):
            for selected_mask in range(64):
                flags = {}
                truth = set()
                for i,v in enumerate(names):
                    if not selected_mask >> i & 1:
                        continue
                    q = alias(v,kind)
                    ready = bool(ready_mask >> i & 1)
                    flags[q] = flags.get(q,False) or ready
                    if ready:
                        truth.update((q+'0',q+'1'))
                for n in range(1,4):
                    for bits in product('01',repeat=n):
                        w = ''.join(bits)
                        assert flags.get(w[:-1],False) == (w in truth)
                        counts['synchronous_alias_queries'] += 1
    print(json.dumps(counts,sort_keys=True))


if __name__ == '__main__':
    main()