python-guppy 0.1.9-2ubuntu4 / usr / share / pyshared / guppy / etc / KanExtension.py

#._cv_part guppy.etc.KanExtension

class LeftKanExtension:

    # Implementation of algorithms described by Brown and Heyworth (ref.251)
    # and Heyworth (ref.253).

    def __init__(self, mod, A, B, R, X, F):
	# External subsystem dependencies
	#    mod.KnuthBendix
	#    mod.FiniteAutomaton
	#    mod.SolveFSA
	#    mod.Cat
	#	mod.Cat.Function
	#	mod.Cat.Functor
	#	mod.Cat.check_graph
	#	mod.Cat.check_rules

	self.mod = mod
	self.Cat = mod.Cat

	#

	self.Cat.check_graph(A)
	self.Cat.check_graph(B)
	self.Cat.check_rules(R, B)
	
	#

	self.A = A
	self.B = B
	self.R = [(tuple(g), tuple(h)) for (g, h) in R]
	self.X = X
	self.F = F

	self.general_procedure()


    def general_procedure(self):
	self.initialize_tables()
	self.make_confluent_system()
	if 0:
	    self.make_catalogue()
	else:
	    self.make_automaton()
	self.make_natural_transformation()
	
    def initialize_tables(self):
	self.obj_to_str_table = {}
	self.str_to_obj_table = {}
	self.make_initial_rules()

    def make_initial_rules(self):

	# Algorithm 6.1 in (251)

	Re = []

	def add_rule(a, b):
	    aw = self.make_word(a)
	    bw = self.make_word(b)
	    if aw != bw:
		Re.append( ( aw, bw ))

	for a in self.A.arrows:
	    srca = self.A.source(a)
	    tgta = self.A.target(a)
	    XA = self.X.fo(srca)
	    Xa = self.X.fa(a)
	    Fa = tuple(self.F.fa(a))
	    Fsrca = self.F.fo(srca)
	    Ftgta = self.F.fo(tgta)
	    if Fa:
		t = Fsrca
		for b in Fa:
		    srcb = self.B.source(b)
		    if srcb != t:
			raise ValueError, \
'Arrow [%s] with source %s does not compose with target %s'%(b, srcb, t)
		    t = self.B.target(b)
		if t != Ftgta:
		    raise ValueError, \
'Arrow %s with target %s does not compose with %s'%(Fa, t, Ftgta)
	    else:
		if Fsrca != Ftgta:
		    raise ValueError, \
'Source %s does not match target %s'%(Fsrca, Ftgta)
	    for x in XA:
		add_rule(((srca, x),) + Fa , ((tgta, Xa(x)),) )

	Rk = [(self.make_word(x), self.make_word(y)) for (x, y) in self.R]
		    

	self.Re = Re
	self.Rk = Rk
	self.Rinit = Re + Rk

    def make_confluent_system(self):
	self.rs = self.mod.KnuthBendix(self.Rinit, delim='.')
	self.Rconf = self.rs.reductions

    def make_automaton(self):
	# Make nondeterministic finite automaton
	
	def target(e):
	    if len(e) == 1 and isinstance(e[0], tuple):
		return self.F.fo(e[0][0])
	    else:
		return self.B.target(e[-1])

	XA = []
	for A in self.A.objects:
	    for x in self.X.fo(A):
		XA.append(((A, x),))

	follows = dict([(B, []) for B in self.B.objects])
	for b, (srcb, tgtb) in self.B.arrows.items():
	    follows[srcb].append((b, tgtb))

	IR = dict([(self.make_term(u), self.make_term(v)) for u, v in self.Rconf])
		   
	pplR = {}
	for l, r in self.Rconf:
	    t = self.make_term(l)
	    for i in range(1, len(t)):
		pplR[t[:i]] = 1

	s0 = ('s0',)

	fsa = self.mod.FiniteAutomaton(s0)

	for xi in XA:
	    if xi not in IR:
		fsa.add_transition(s0, xi[0], xi)

	for xi in XA:
	    for b, tgtb in follows[target(xi)]:
		bterm = (b,)
		xib = xi + bterm
		if xib in pplR:
		    fsa.add_transition(xi, b, xib, tgtb)
		elif (bterm in pplR and xib not in IR):
		    fsa.add_transition(xi, b, bterm, tgtb)
		elif xib not in IR:
		    fsa.add_transition(xi, b, tgtb)

	for Bi in self.B.objects:
	    for b, tgtb in follows[Bi]:
		bterm = (b,)
		if bterm in pplR:
		    fsa.add_transition(Bi, b, bterm, tgtb)
		elif bterm not in IR:
		    fsa.add_transition(Bi, b, tgtb)

	for u in pplR:
	    if u in XA:
		continue
	    for b, tgtb in follows[target(u)]:
		bterm = (b,)
		ub = u + bterm
		if ub in pplR:
		    fsa.add_transition(u, b, ub, tgtb)
		elif self.irreducible(ub): # ub not in IR:
		    fsa.add_transition(u, b, tgtb)
		
	def get_RS(Bi):
	    finals = {}
	    finals[Bi] = 1
	    for xi in XA:
		if self.F.fo(xi[0][0]) == Bi:
		    finals[xi] = 1
	    for u in pplR:
		if target(u) == Bi:
		    finals[u] = 1
		    
	    for c in fsa.get_composites():
		for s in c:
		    if s not in finals:
			break
		else:
		    finals[c] = 1

	    dfa = fsa.get_minimized_dfa(finals)
	    regexp = self.mod.SolveFSA(dfa)
	    return RegularSet(regexp)

	KB = self.Cat.Function(get_RS, self.B.objects, None)

	Kb = self.Cat.Function(
		      lambda a:KanAction(self.B, KB, a, target, self.irreducible, self.reduce),
		      self.B.arrows,
		      KanAction,
		      )
	    

	self.KB = KB
	self.Kb = Kb
	self.K = self.Cat.Functor(KB, Kb)


    def make_catalogue(self):
	# Catalogue the elements of the sets pointed to by extension functor K,
	# according to algorithm described in 7.1 in (251).

	# Precondition:

	# Tables initialized and a confluent system created.
	# The system is assumed to be finite, otherwise we won't terminate.

	# Postcondition:

	# Functor self.K represented as:
	#
	# self.K.tabo = self.KB = dict mapping,
        #		source: {each B in self.B.objects}
	#		target: sets represented as lists
	# self.K.taba = self.Kb = dict, mapping
	#		source: {each a in self.B.arrows}
	#		target: tabulated function, mapping
	#			source: KB[source of a]
	#			target: KB[target of a]

	def target(e):
	    if len(e) == 1:
		return self.F.fo(e[0][0])
	    else:
		return self.B.target(e[-1])

	def add_element(e):
	    if self.irreducible(e):
		block.append(e)
		KB[target(e)].append(e)
	    else:
		pass
		#print e, self
		#pdb.set_trace()

	KB = dict([(B, []) for B in self.B.objects])
	block = []

	for A in self.A.objects:
	    for x in self.X.fo(A):
		add_element(((A, x),))
		    
	while block:
	    oblock = block
	    block = []
	    for e in oblock:
		tgt = target(e)
		for a in self.B.arrows:
		    if self.B.source(a) == tgt:
			add_element( e + (a,) )

	Kb = {}

	for a in self.B.arrows:
	    src = KB[self.B.source(a)]
	    tgt = KB[self.B.target(a)]
	    tab = dict([(s, self.reduce(s + (a,))) for s in src])
	    Kb[a] = self.Cat.Function(tab, src, tgt)

	KB = self.Cat.Function(KB, self.B.objects, KB.values())
	Kb = self.Cat.Function(Kb, self.B.arrows, Kb.values())

	self.KB = KB
	self.Kb = Kb
	self.K = self.Cat.Functor(KB, Kb)

	
    def make_natural_transformation(self):
	
	# Precondition:
	# initial tables should be initialized
	# self.K.fo should exist

	# Postcondition:
	#
	# self.nat[A] for A in self.A.objects

	get_nat_memo = {}
	def get_nat(A):
	    if A in get_nat_memo:
		return get_nat_memo[A]


	    src = self.X.fo(A)
	    tgt = self.K.fo(self.F.fo(A))
	    tab = dict([(x, self.reduce( ((A, x),) )) for x in src])
	    get_nat_memo[A] = self.Cat.Function(tab, src, tgt)
	    return get_nat_memo[A]

	self.nat = self.Cat.Function(get_nat, self.A.objects, None)
		   


    def make_word(self, x):
	ots = self.obj_to_str
	return '.'.join([ots(e) for e in x if e != ''])

    def obj_to_str(self, x):
	otn = self.obj_to_str_table
	try:
	    return otn[x]
	except KeyError:
	    assert not (isinstance(x, tuple) and len(x) > 2)
	    n = str(len(otn))
	    #n = '%d:%s'%(len(otn), x)
	    #n = str(x)
	    otn[x] = n
	    self.str_to_obj_table[n] = x
	    return n

    def str_to_obj(self, x):
	return self.str_to_obj_table[x]

    def irreducible(self, x):
	tx = self.make_word(x)
	return tx == self.rs.reduce(tx)

    def reduce(self, x):
	w = self.rs.reduce(self.make_word(x))
	return self.make_term(w)

    def make_term(self, word):
	sto = self.str_to_obj_table
	return tuple( [sto[s] for s in word.split('.') if s] )

class KanAction:
    def __init__(self, B, KB, a, targetof, irreducible, reduce):
	srca = B.source(a)
	tgta = B.target(a)
	self.src = KB(srca)
	self.tgt = KB(tgta)
	self.a = a
	self.srca = srca
	self.targetof = targetof
	self.irreducible = irreducible
	self.reduce = reduce

    def __call__(self, s):
	if self.targetof(s) != self.srca:
	    raise TypeError, '''\
Target of %r (= %r) does not match source of %r (= %r)'''%(
    s, self.targetof(s), self.a, self.srca)
	if not self.irreducible(s):
	    raise TypeError, '''\
Argument %r is reducible to %r; and is thus not in the source set K.fo(%r)'''%(
	    s, self.reduce(s),self.srca)
	return self.reduce(s + (self.a,))


class RegularSet:
    # Wraps a regular expression;
    # provides a set protocol for the underlying set of sequences:
    #  o If the RE specifies a finite language, iteration over its strings
    #  [ o set inclusion ]

    is_simplified = 0
    def __init__(self, re):
	self.re = re
    
    def __iter__(self):
	return iter(self.uniform)

    def __getitem__(self, x):
	return self.uniform[x]

    def __len__(self):
	return len(self.uniform)

    def get_xs_covered(self, coverage):
	N = coverage
	X = self.re.limited(coverage)
	xs = X.sequni()
	return [tuple(x) for x in xs]

    def get_uniform(self):
	self.simplify()
	return self.re.sequni()
	
    uniform = property(fget=get_uniform)

    def simplify(self):
	if not self.is_simplified:
	    self.re = self.re.simplified()
	    self.is_simplified = 1

		
class ObjectTester:
    def __init__(self, category_tester, object, code):
	self.category_tester = category_tester
	self.functor = category_tester.functor
	self.object = object
	self.code = code
	
    def get_all_arrows(self):
	return self.category_tester.arrows[self.object]

    def get_intermediate_test_code(self):
	return self.code

    def get_python_test_source_code(self):
	cmap = {
	    'aseq':'assert e[%r] == e[%r]',
	    'evalfa':'e[%r] = fa[%r](e[%r])',
	    'asfo':'assert fo[%r](e[%r])'
	    }

	return '\n'.join([cmap[c[0]]%c[1:] for c in self.code])

    def execode(self, arg):
	code = self.get_python_test_source_code()

	e = {'arg':arg}
	d = {'fa':self.functor.fa,
	     'fo':self.functor.fo,
	     'e':e,
	     }
	exec code in d
	return e
    
    def intercode(self, arg):
	e = {'arg':arg}
	fa = self.functor.fa
	fo = self.functor.fo
	for c in self.code:
	    a = c[0]
	    if a == 'evalfa':
		dst, ar, src = c[1:]
		e[dst] = fa[ar](e[src])
	    elif a == 'asfo':
		ob, src = c[1:]
		if not fo[ob](e[src]):
		    raise ValueError, 'Predicate failed'
	    elif a == 'aseq':
		na, nb = c[1:]
		if e[na] != e[nb]:
		    raise ValueError, 'e[%r] != e[%r]'%(na, nb)
	    else:
		raise ValueError, 'Invalid code: %r'%(a,)


    def test(self, arg):
	return self.intercode(arg)

class CategoryTester:
    def __init__(self, mod, functor, arrows, get_arrow_name=None):
	self.mod = mod
	self.cat = functor.src
	self.functor = functor
	self.arrows = arrows
	if get_arrow_name is not None:
	    self.get_arrow_name = get_arrow_name


    def get_arrow_name(self, a):
	return '.'.join(a)


    def get_eval_arrows_code(self, object, argname):
	fa = self.functor.fa

	name = argname
	memo = {():name}
	memolist = [((),name)]

	codes = []

	def eval_arrow(a):
	    if a in memo:
		return memo[a]
	    a0 = a[:-1]
	    a1 = a[-1]
	    name = self.get_arrow_name(a)
	    na0 = eval_arrow(a0)
	    #codes.append('%s = fa[%r](%s)'%(name, a1, na0))
	    codes.append(('evalfa', name, a1, na0))
	    memo[a] = name
	    memolist.append((a, name))
	    return name

	for ar in self.arrows[object]:
	    eval_arrow(ar)
	
	return codes, memolist

    def get_object_tester(self, object):
	code = self.get_test_object_code(object)
	return ObjectTester(self, object, code)

    def get_test_inclusion_code(self, object, ml):
	codes = []
	src = self.functor.fo.src
	for arrow, value in ml:
	    ob = object
	    if arrow:
		ob = self.cat.graph.target(arrow[-1])
	    #codes.append('assert fo[%r](%s)'%(ob, value))
	    if src is None or ob in src:
		codes.append(('asfo', ob, value))
	return codes

    def get_test_object_code(self, object):
	argname = 'arg'
	evalcodes, memolist = self.get_eval_arrows_code(object, argname)

	relcodes = self.get_test_relations_code(object, memolist)

	incodes = self.get_test_inclusion_code(object, memolist)

	return evalcodes+relcodes+incodes
	

    def get_test_relations_code(self, object, memolist):
	codes = []
	cat = self.cat
	fa = self.functor.fa
	memo = dict(memolist)

	def teval_arrow(ar):
	    if ar in memo:
		return memo[ar]
	    a0 = teval_arrow(ar[:-1])
	    name = self.get_arrow_name(ar)
	    #codes.append('%s = fa[%r](%s)'%(name, ar[-1], a0))
	    codes.append(('evalfa', name, ar[-1], a0))
	    memo[ar] = name
	    return name

	# Check that the equality relations really match up
	# for all arrows in old memolist, i.e. original unique arrows
	# which is arguably overkill sometimes?..
	for a, b in cat.relations:
	    a = tuple(a)
	    b = tuple(b)
	    src = cat.graph.source(a[0])
	    for (arr, val) in memolist:
		if arr:
		    tgt = cat.graph.target(arr[-1])
		else:
		    tgt = object
		if src == tgt:
		    ara = arr + a
		    arb = arr + b
		    if ara != arb:
			va = teval_arrow(ara)
			vb = teval_arrow(arb)
			assert va != vb
			#codes.append('assert %s == %s'%(va, vb))
			codes.append(('aseq', va, vb))
	return codes
	    
    def test_object(self, object, value):
	tester = self.get_object_tester(object)
	tester.test(value)
	return tester

    def test_object_fail(self, object, value):
	try:
	    self.test_object(object, value)
	except:
	    pass
	else:
	    raise Exception, 'Exception excepted'

	    




class _GLUECLAMP_:
    # 'imports'

    def _get_KnuthBendix(self):    	return self._parent.KnuthBendix.KnuthBendix
    def _get_FiniteAutomaton(self):    	return self._parent.FSA.FiniteAutomaton
    def _get_SolveFSA(self):    	return self._parent.RE.SolveFSA
    def _get_Cat(self):		    	return self._parent.Cat

    # Main exported interface is the lke method
    # which provides a context for the LeftKanExtension class.

    def lke(self, A, B, R, X, F):
	return LeftKanExtension(self, A, B, R, X, F)

    # Other functions - examples of applications of Kan extension
    # in alphabetic order

    def arrows_map(self, cat, from_objects=0, coverage=1):
	if from_objects:
	    cat = cat.get_dual()

	A = self.Cat.Graph(cat.graph.objects, [])
	B = cat.graph
	R = cat.relations
	X = self.Cat.Functor(lambda x: [1], lambda x: lambda y:y)
	F = self.Cat.Functor(lambda x: x, lambda x: [])
	ke = self.lke(A, B, R, X, F)

	memo = {}

	def get_arrows(object):
	    if object in memo:
		return memo[object]
	    re = ke.K.fo[object].re.rempretup()
	    if from_objects:
		re = re.reversed()
	    if str(coverage).startswith('length'):
		maxlen = int(coverage[6:])
		ar = []
		xs = re.get_words_memo()
		for i in range(1, maxlen+1):
		    ar.extend([tuple(x) for x in xs.get_words_of_length(i)])
	    else:
		re = re.limited(coverage)
		xs = re.sequni()
		ar = [tuple(x) for x in xs]
	    memo[object] = ar
	    return ar

	return self.Cat.Function(
	    get_arrows,
	    src = ke.K.fo.src,
	    tgt = None
	    )

    def category_tester(self, functor, arrows=None, coverage=1):
	if isinstance(functor, tuple):
	    fo, fa, src = functor
	    if fo is None:
		fo = lambda x:lambda y:1
	    functor = self.Cat.Functor(fo, fa, src)
	if arrows is None:
	    arrows = self.arrows_map(functor.src, from_objects=1, coverage=coverage)
	return CategoryTester(self, functor, arrows)

    def coequalizer(self, S0, S1, f0, f1):
	# Given
	#
	# S0, S1 sets (objects that can be iterated over)
	# f0, f1 functions from S0 to S1
	#
	# Return a coequalizing function,
	# such that in the following diagram:

	#
	#  S0 ===== S0
	#  |	    |
	#  | f0	    | f1
	#  |	    |
	#  V	    V
	#  S1 ===== S1 ==== coequalizing_function.src
	#  |
	#  | coequalizing_function
	#  |
	#  V
	#  coequalizing_function.tgt

	# both paths from S0 to coequalizing_function.tgt will be equivalent,
	# and coequalizing_function.tgt is a colimit of all such sets.
	#
	# The coequalizing_function object is callable with
	# an argument from S1, and has the following attributes:
	#   .src		is identical to S1
	#   .tgt		is a set in iterable form
	#   .asdict()		returns a dict representing the mapping

	objects = [0, 1]
	arrows = {'a0':(0, 1), 'a1': (0, 1)}
	A = self.Cat.Graph(objects, arrows)

	Xo = self.Cat.Function({0:S0, 1:S1}, objects, [S0,S1])
	Xa = self.Cat.Function({'a0':f0, 'a1':f1}, arrows, [f0,f1])
	X = self.Cat.Functor(Xo, Xa)

	colimit_object, colimit_functions = self.colimit(A, X)
	return colimit_functions[1]

    def colimit(self, A, X):
	# According to 9.6 in (ref.251)

	B = self.Cat.Graph([0], {})
	R = []
	F = self.Cat.Functor(lambda x: 0, lambda x: ())

	lka = self.lke(A, B, R, X, F)

	colimit_object = lka.KB[0]
	colimit_functions = lka.nat

	# Reduce elements to a smaller (but isomorphic) form
	# I.E since elements are all of the form
	#	((A, X),)
	# they can be reduced to the form
	#   	(A, X)
	#

	colimit_object = [x[0] for x in colimit_object]

	colimit_functions = dict([
	    (A, self.Cat.Function(
		dict([(a, k[0])
		      for (a, k) in cof.items()]),
		cof.src,
		colimit_object,
		)
	     )
	    for (A, cof) in colimit_functions.items()])


	return colimit_object, colimit_functions

    def test_arrows(self, functor, object, value):
	# Application of arrow listing to test sequencing
	# Discussed in Notes Mar 9 2005

	tester = self.category_tester(functor)
	return tester.test_object(object, value)