src/process.py - master - auzubarev/time-process

src/process.py
92 строки3 KB
Alex Zubarev
add new experiment
18 мар 2026, 12:06
18 мар 2026, 12:06e398619
def build_process(net, run):
    B, E, G = [], [], []
    phi, tau = {}, {}
    post_b = {} 
    pre_e = {}  
    post_e = {} 
    last_b = {}
    for p, count in net["initial_marking"].items():
        if count > 0:
            b_id = (0, p)
            B.append(b_id)
            phi[b_id] = p
            last_b[p] = b_id
            post_b[b_id] = set()
    gt = 0
    for i, (theta, t_id) in enumerate(run, 1):
        e_id = i
        E.append(e_id)
        gt += theta
        tau[e_id] = gt
        phi[e_id] = t_id
        pre_e[e_id] = set()
        post_e[e_id] = set()
        for p_in in net["in_arcs"][t_id]:
            b_in = last_b[p_in]
            G.append((b_in, e_id))
            post_b[b_in].add(e_id)
            pre_e[e_id].add(b_in)
            del last_b[p_in] 
        for p_out in net["out_arcs"][t_id]:
            b_out = (i, p_out)
            B.append(b_out)
            phi[b_out] = p_out
            G.append((e_id, b_out))
            post_b[b_out] = set()
            post_e[e_id].add(b_out)
            last_b[p_out] = b_out
    return {
        "B": B, "E": E, "G": G, 
        "phi": phi, "tau": tau, 
        "post_b": post_b, "pre_e": pre_e, "post_e": post_e
    }

def get_initial_cut(pi):
    return {b for b in pi["B"] if b[0] == 0}

def get_ready_events(pi, cut):
    events = set()
    for b in cut:
        for e in pi["post_b"][b]:
            events.add(e)
    return [e for e in events if pi["pre_e"][e].issubset(cut)]

def change_cut(pi, e, cut):
    return (cut - pi["pre_e"][e]) | pi["post_e"][e]

def get_canonical_lin(pi):
    cut = get_initial_cut(pi)
    canonical_lin = []
    for _ in range(len(pi["E"])):
        ready = get_ready_events(pi, cut)
        ready.sort(key=lambda e: (pi["tau"][e], pi["phi"][e]))
        e = ready[0]
        cut = change_cut(pi, e, cut)
        canonical_lin.append((pi["phi"][e], pi["tau"][e]))
    return tuple(canonical_lin)


def count_lins(pi):
    cuts = {}

    def count_recursive(current_cut):
        cut_tuple = tuple(sorted(list(current_cut)))
        if cut_tuple in cuts:
            return cuts[cut_tuple]
        
        ready = get_ready_events(pi, current_cut)
        if not ready:
            return 1
        min_tau = min(pi["tau"][e] for e in ready)
        valid_events = [e for e in ready if pi["tau"][e] == min_tau]
        
        total = 0
        for e in valid_events:
            new_cut = change_cut(pi, e, current_cut)
            total += count_recursive(new_cut)
        
        cuts[cut_tuple] = total
        return total

    initial_cut = get_initial_cut(pi)
    return count_recursive(initial_cut)