Skip to content

petri_net_nn.subnets

subnets

Five elemental Petri net subnets from §5 of the architecture spec.

Continuous relaxation of the token firing rule (spec §4.2):

activation(t) = sigmoid( sum_p w(p,t) * a(p) - theta(t) )
a(p)          = sum_{t: (t,p) in F} activation(t) * w(t,p)

Each place's activation a(p) is a scalar in [0,1] (spec §4.1) carried as a tensor of shape (batch,) so that a forward pass can score many process instances at once.

Forward passes return a dict keyed by Petri-net element name (the same names that appear in the diagrams in §5), so a caller can inspect any transition activation or place activation without having to remember a positional convention.

SequentialSubnet 

SequentialSubnet()

Bases: Module

Subnet 1 — sequential execution.

[P_before] --T_step--> [P_after]

Source code in petri_net_nn/subnets.py 
33
34
35
36
37
def __init__(self) -> None:
    super().__init__()
    self.w_in = nn.Parameter(torch.tensor(1.0))
    self.theta_step = nn.Parameter(torch.tensor(0.0))
    self.w_out = nn.Parameter(torch.tensor(1.0))

XORSubnet 

XORSubnet()

Bases: Module

Subnet 2 — exclusive choice.

               --T_route_A--> [P_path_A]
[P_decision] <
               --T_route_B--> [P_path_B]

Both transitions share P_decision as their input place. Soft routing is what the continuous relaxation produces; the structural constraint (only these two declared paths exist) is enforced by construction.

Source code in petri_net_nn/subnets.py 
57
58
59
60
61
62
63
64
65
def __init__(self) -> None:
    super().__init__()
    self.w_A_in = nn.Parameter(torch.tensor(1.0))
    self.theta_A = nn.Parameter(torch.tensor(0.0))
    self.w_A_out = nn.Parameter(torch.tensor(1.0))

    self.w_B_in = nn.Parameter(torch.tensor(-1.0))
    self.theta_B = nn.Parameter(torch.tensor(0.0))
    self.w_B_out = nn.Parameter(torch.tensor(1.0))

AndSplitSubnet 

AndSplitSubnet()

Bases: Module

Subnet 3 — parallel split (AND-gateway).

              --> [P_branch_A]
[P_ready] -- T_spawn
              --> [P_branch_B]

A single transition with two output arcs; both branches activate simultaneously when T_spawn fires.

Source code in petri_net_nn/subnets.py 
91
92
93
94
95
96
def __init__(self) -> None:
    super().__init__()
    self.w_in = nn.Parameter(torch.tensor(1.0))
    self.theta_spawn = nn.Parameter(torch.tensor(0.0))
    self.w_out_A = nn.Parameter(torch.tensor(1.0))
    self.w_out_B = nn.Parameter(torch.tensor(1.0))

AndJoinSubnet 

AndJoinSubnet(sharpness=4.0)

Bases: Module

Subnet 4 — synchronisation (AND-join).

[P_branch_A] --\
                 T_merge --> [P_unified]
[P_branch_B] --/

Spec §5 calls this the hardest subnet to relax: a partial firing (one branch complete, the other not) must not propagate. We bias initialisation toward a high threshold and use a sharpness parameter on the sigmoid so the activation function approaches a step.

Source code in petri_net_nn/subnets.py 
120
121
122
123
124
125
126
def __init__(self, sharpness: float = 4.0) -> None:
    super().__init__()
    self.w_A_in = nn.Parameter(torch.tensor(1.0))
    self.w_B_in = nn.Parameter(torch.tensor(1.0))
    self.theta_merge = nn.Parameter(torch.tensor(1.5))
    self.w_out = nn.Parameter(torch.tensor(1.0))
    self.sharpness = sharpness

SagaSubnet 

SagaSubnet()

Bases: Module

Subnet 5 — saga compensation.

[P_active] --T_succeed--> [P_complete]
[P_active] --T_fail-----> [P_compensating]
[P_compensating] --T_compensate--> [P_initial]

Two competing transitions out of P_active (success / failure), and a compensation transition closing the loop back to P_initial. The whole subnet is a feed-forward DAG when unrolled for one execution step.

Source code in petri_net_nn/subnets.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
def __init__(self) -> None:
    super().__init__()
    self.w_succeed_in = nn.Parameter(torch.tensor(1.0))
    self.theta_succeed = nn.Parameter(torch.tensor(0.0))
    self.w_succeed_out = nn.Parameter(torch.tensor(1.0))

    self.w_fail_in = nn.Parameter(torch.tensor(-1.0))
    self.theta_fail = nn.Parameter(torch.tensor(0.0))
    self.w_fail_out = nn.Parameter(torch.tensor(1.0))

    self.w_compensate_in = nn.Parameter(torch.tensor(1.0))
    self.theta_compensate = nn.Parameter(torch.tensor(0.0))
    self.w_compensate_out = nn.Parameter(torch.tensor(1.0))