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

src/process.py
155 строк4 KB
Alex Zubarev
first_commit
07 май 2026, 09:28
07 май 2026, 09:28933e5c1
def build_process(net, run):
    transitions, thetas = run
    B, E, G = [], [], []
    phi = {}
    main_cuts = {}
    last_b = {}
    pre_e, post_e = {}, {}
    pre_b, post_b = {}, {}
    
    current_cut = set()
    current_predecessors = set()


    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
            pre_b[b_id] = set()
            post_b[b_id] = set() 
            current_cut.add(b_id)

    for i in range(len(thetas)):
        main_cuts[frozenset(current_cut)] = (thetas[i], set(current_predecessors))
        
        if i >= len(transitions):
            break
        t_id = transitions[i]
        e_id = i + 1
        E.append(e_id)
        phi[e_id] = t_id
        pre_e[e_id] = set()
        post_e[e_id] = set()

        current_predecessors.add(e_id)
        
        for p_in in net["in_arcs"][t_id]:
            b_in = last_b[p_in]
            G.append((b_in, e_id))
            pre_e[e_id].add(b_in)
            post_b[b_in].add(e_id) 
            current_cut.remove(b_in)
            del last_b[p_in]
   
        for p_out in net["out_arcs"][t_id]:
            b_out = (e_id, p_out)
            B.append(b_out)
            phi[b_out] = p_out
            G.append((e_id, b_out))
            pre_b[b_out] = {e_id}
            post_b[b_out] = set() 
            post_e[e_id].add(b_out)
            last_b[p_out] = b_out
            current_cut.add(b_out)

    return {
        "B": B, "E": E, "G": G, "phi": phi,
        "pre_e": pre_e, "post_e": post_e,
        "pre_b": pre_b, "post_b": post_b,
        "main_cuts": main_cuts
    }


def tau(pi, C):
    return pi["main_cuts"].get(frozenset(C), (0, set()))[0]


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

def is_parallel(preds1, preds2):
    return not (preds1.issubset(preds2) or preds2.issubset(preds1))

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

    ready_events = set()
    positive_cuts = [m_preds for _, (m_val, m_preds) in pi["main_cuts"].items() if m_val > 0]
    
    for e in potential_events:
        next_cut = change_cut(pi, e, cut)
        next_cut_f = frozenset(next_cut)
        next_preds = preds | {e}

        if next_cut_f in pi["main_cuts"]:
            theta = pi["main_cuts"][next_cut_f][0]
        else:
            is_undefined = False
            for m_preds in positive_cuts:
                if not (next_preds.issubset(m_preds) or m_preds.issubset(next_preds)):
                    is_undefined = True
                    break
            theta = None if is_undefined else 0

        if theta is not None:
            if theta > 0:
                return {e}
            ready_events.add(e)

    return ready_events


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)
    preds = set()
    transitions = []
    thetas = []
    thetas.append(tau(pi, cut))
    
    for _ in range(len(pi["E"])):
        ready = list(get_ready_events(pi, cut, preds))
        ready.sort(key=lambda e: (pi["phi"][e]))
        e = ready[0]
        cut = change_cut(pi, e, cut)
        preds.add(e)
        transitions.append(pi["phi"][e])
        thetas.append(tau(pi, cut))
    return (tuple(transitions), tuple(thetas))


def count_lins(pi):
    cuts = {}

    def count_recursive(current_cut, current_preds):
        cut_tuple = tuple(sorted(list(current_cut)))
        if cut_tuple in cuts:
            return cuts[cut_tuple]
        
        ready = get_ready_events(pi, current_cut, current_preds)
        if not ready:
            return 1
        
        total = 0
        for e in ready:
            new_cut = change_cut(pi, e, current_cut)
            new_preds = current_preds | {e}
            total += count_recursive(new_cut, new_preds)
        
        cuts[cut_tuple] = total
        return total

    initial_cut = get_initial_cut(pi)
    return count_recursive(initial_cut, set())