a Calculus of Communicating System(CCS)のプロセス関係判定のために書いたプログラム (某講義用につくったもの)

ccs_random-process.py

# encoding: utf-8
# a Calculus of Communicating System(CCS)に関して、
# プロセスの関係を判定するために書いたプログラム
# たぶんあってる
import random
random.seed()
randint = random.randint

# τ(プロセスの内部action)をプロセス生成に含めるか(ここでは't'で表現)
tau = True
# プロセスに含めるaction('t'以外)
code = ['a', 'b']

# ここらへんはmake_processのパラメタ
branch = 2
lev = 1
c_param = 8

# ランダムにプロセス生成
def make_process():
    def dfs(level = 0):
        if level > lev:
            return []
        res = []
        if randint(0, c_param) or level == 0:
            for i in xrange(randint(1, branch)):
                val = randint(0, len(code))
                if tau and val < 1:
                    res.append(('t', dfs(level + 1)))
                else:
                    res.append((code[val-tau], dfs(level + 1)))
        return res
    tree = []
    while not tree:
        tree = dfs()
    return tree

# Trace生成
# tau_ignore: τをトレースに含めないか
def get_trace(tree, tau_ignore = 1):
    traces = set()
    def dfs(tree, trace = []):
        traces.add("".join(trace))

        for e in tree:
            if e[0] is "t" and tau_ignore:
                dfs(e[1], trace)
            else:
                dfs(e[1], trace + [e[0]])
    dfs(tree)
    return traces

# プロセスの失敗集合(failure set)を生成する
def get_failure(trace):
    traces = []
    def dfs(tree, trace = [], tau = 0):
        can = set()
        for e in tree:
            if e[0] is "t":
                can.update(dfs(e[1], trace, 1))
            else:
                dfs(e[1], trace + [e[0]], 0)
                can.add(e[0])
        if tau:
            return can
        traces.append([trace, can])
    dfs(trace)

    fset = {}
    for tr, can in traces:
        for c in code:
            if c not in can:
                fset.setdefault("".join(tr), set()).add(c)
    return fset
# 失敗集合が等しいか判定する
# (failure equivalenceの判定)
def eq_failure(S, T):
    for s in S:
        if s not in T or S[s] != T[s]:
            return 0
    return 1

# ２つのプロセスにある関係があるかを求める
# 各(tau_ignore, epsilon)の値
# 強双模倣関係(s.b.) : (0, 0)
# 観測合同関係(o.c.) : (0, 1)
# 弱双模倣関係(w.b.) : (1, 1)
# トレース等価(t.e.) (τなし) : otherwise
def is_equiv(S, T, tau_ignore, epsilon_trans):
    def check(target, trace):
        if not trace:
            return 1
        ok = 0
        cur = trace[0]
        nxt = trace[1:]
        if epsilon_trans:
            for e in target:
                if e[0] is cur:
                    ok = 1
                    if not check(e[1], nxt):
                        return 0
                elif e[0] is "t":
                    ok = 1
                    if not check(e[1], trace):
                        return 0
        else:
            for e in target:
                if e[0] is cur:
                    ok = 1
                    if not check(e[1], nxt):
                        return 0
        return ok

    def dfs(tree, target, trace = []):
        #print "check", trace, "->", target
        if not check(target, trace):
            #print "failed"
            return 0
        for e in tree:
            if e[0] is "t":
                if tau_ignore:
                    if not dfs(e[1], target, trace):
                        return 0
                else:
                    if not dfs(e[1], target, trace + ["t"]):
                        return 0
            else:
                if not dfs(e[1], target, trace + [e[0]]):
                    return 0
        return 1
    return dfs(S, T) and dfs(T, S)

# 以下のプログラムでは、ランダムにプロセスを生成し、
# そのプロセスがある関係を満たしているかを判定

# 失敗具合等価性(failure equivalence)を満たすか
fe = 1
# トレース等価(τ含む)を満たすか
te_with_tau = 1

while 1:
    S = make_process()
    T = make_process()
    traceS = get_trace(S)
    traceT = get_trace(T)
    if traceS != traceT:
        continue
    #if is_equiv(S, T, 0, 0):
    #if not is_equiv(S, T, 0, 0) and is_equiv(S, T, 0, 1):
    #if not is_equiv(S, T, 0, 1) and is_equiv(S, T, 1, 1):
    if not is_equiv(S, T, 1, 1):
        failS = get_failure(S)
        failT = get_failure(T)
        if (fe and not eq_failure(failS, failT)) or (not fe and eq_failure(failS, failT)):
            continue
        traceStau = get_trace(S, 0)
        traceTtau = get_trace(T, 0)
        if (te_with_tau and traceStau != traceTtau) or (not te_with_tau and traceStau == traceTtau):
            continue
        print "process pair found!"
        print "="*10
        print "trace set:", traceS
        print "-- process S:", S
        print "trace set including tau:", traceStau
        print "FS(S):", failS
        print "-- process T:", T
        print "trace set including tau:", traceTtau
        print "FS(T):", failT
        print "="*10
        break