Examples¶

Producer/consumer¶

This example implements the simplest producer/consumer network. One end produces tokens which enables the transition that transfers the token to the consumer. Both producer and consumer are instances of soyutnet.place.SpecialPlace class.

It is implemented by the code below which can be found at SoyutNet repo.

def main():
    token_ids = list(range(1, 7))

    async def producer(place):
        try:
            id: id_t = token_ids.pop(0)
            token = (GENERIC_LABEL, id)
            print("Produced:", token)
            return [token]
        except IndexError:
            pass

        return []

    async def consumer(place):
        token = place.get_token(GENERIC_LABEL)
        if token:
            print("Consumed:", token)
        else:
            print("No token in consumer")

The main function starts by defining token IDs of produced token. The producer function is called by \(p_1\) and the consumer is called by \(p_2\).

    place_count = 2

    def on_comparison_ends(observer):
        nonlocal place_count
        place_count -= 1
        if place_count == 0:
            soyutnet.terminate()

    net = SoyutNet()
    net.DEBUG_ENABLED = True
    net.VERBOSE_ENABLED = True
    net.ERROR(
        [
            {
                0: "ABCD",
                1: [
                    1,
                    2,
                    3,
                ],
            },
            (0, "ABCD", [1, 3, 3.0, b"ABCD"]),
        ]

SoyutNet implements observers (soyutnet.observer.Observer) for keeping the record of PT net markings before each firing of a transition.

Currently, observer records has three columns, the time of firing, label and number of tokens with the label (soyutnet.observer.ObserverRecordType).

ComparativeObserver (soyutnet.observer.ComparativeObserver) is used for test purposes. It accepts two additional arguments

expected: A dictionary of token counts with the structure below

expected = {
    record_column_index: [ recorded_value_1, recorded_value_2, ... ],
}

on_comparison_ends: It is called after all entries in expected is compared. In the example above, on_comparison_end is used to termiate the simulation after the test is completed.

The token count in \(p_1\) is observed before each firing of \(t_1\) and compared to the list. If a value does not match, it raises a RuntimeError.

    )
    log_file = Path(__file__).resolve().parent / "log.tmp"
    log_filename = str(log_file)
    net.LOG_FILE = log_filename
    assert net.LOG_FILE == log_filename
    net.SLOW_MOTION = True
    net.LOOP_DELAY = 0
    net.ERROR(
        [
            {
                0: "ABCD",

\(p_1\)’s output is connected to \(t_1\) and \(t_1\)’s output is connected to \(p_2\).

The registry keeps a list of places and transitions and it is provided to the soyutnet.main() function which starts asyncio task loops of PTs.

$ python tests/behavior/simple_example.py
Produced: (0, 1)
No token in consumer
Produced: (0, 2)
Consumed: (0, 1)
Produced: (0, 3)
Consumed: (0, 2)
Produced: (0, 4)
Consumed: (0, 3)
Produced: (0, 5)
Consumed: (0, 4)
Produced: (0, 6)
Consumed: (0, 5)
Consumed: (0, 6)
Simulation is terminated.

n-tester¶

This example implements an n-tester transition which is enabled when input place has \(n\) or more tokens.

It is implemented by the code below which can be found at SoyutNet repo.

import sys
import asyncio

import soyutnet
from soyutnet import SoyutNet
from soyutnet.constants import GENERIC_ID, GENERIC_LABEL, TokenType


def n_tester(n=2):
    with SoyutNet() as net:
        consumed_count = n + 1

        async def consumer(place):
            nonlocal consumed_count
            token: TokenType = place.get_token(GENERIC_LABEL)
            if token:
                net.DEBUG("Consumed:", token)
                consumed_count -= 1
                if consumed_count == 0:
                    soyutnet.terminate()

        o1 = net.ComparativeObserver(
            expected={1: [((GENERIC_LABEL, i),) for i in range(2 * n, n - 1, -1)]},
            verbose=True,
        )
        p1 = net.Place(
            "p1", initial_tokens={GENERIC_LABEL: [GENERIC_ID] * (2 * n)}, observer=o1
        )
        t1 = net.Transition()
        p2 = net.SpecialPlace("p2", consumer=consumer)

        p1.connect(t1, weight=n).connect(p2, weight=1)
        t1.connect(p1, weight=n - 1)

        gv = net.registry.generate_graph()
        return gv


if __name__ == "__main__":
    gv = n_tester(int(sys.argv[1]) + 1)
    with open("test.gv", "w") as fh:
        fh.write(gv)

Usage

$ python3 tests/behavior/n_tester.py 9
$ dot -Tpng test.gv > n_tester_example.png # Generate image from graphviz dot file

Periodic¶

This example implements two transitions that fires at adjustable periods.

It is implemented by the code below which can be found at SoyutNet repo.

import sys
import asyncio

import soyutnet
from soyutnet import SoyutNet
from soyutnet.constants import GENERIC_ID, GENERIC_LABEL


def main(w=2, graph_filename=""):
    async def scheduled():
        await asyncio.sleep(0.005 * w)
        soyutnet.terminate()

    net = SoyutNet()

    reg = net.PTRegistry()
    o1 = net.Observer(verbose=True)
    p1 = net.Place("p1", initial_tokens={GENERIC_LABEL: [GENERIC_ID] * w}, observer=o1)
    o2 = net.Observer(verbose=True)
    p2 = net.Place("p2", initial_tokens={GENERIC_LABEL: [GENERIC_ID] * 0}, observer=o2)
    t1 = net.Transition("t1")
    t2 = net.Transition("t2")
    """Define places and transitions (PTs)"""

    p1.connect(t1, weight=w).connect(p2, weight=w).connect(t2).connect(p1)
    """Connect PTs"""

    reg.register(p1)
    reg.register(p2)
    reg.register(t1)
    reg.register(t2)
    """Save to a list of PTs"""

    soyutnet.run(reg, extra_routines=[scheduled()])
    print("Simulation is terminated.")

    records = reg.get_merged_records()
    for rec in records:
        net.print(rec)

    if graph_filename:
        with open(graph_filename, "w") as fh:
            fh.write(reg.generate_graph(label_names={GENERIC_LABEL: "@"}))

    return records


if __name__ == "__main__":
    main(int(sys.argv[1]), sys.argv[2] if len(sys.argv) > 2 else "")

Usage

It can be seen that t2 fires more frequently than t1 which can be adjusted by the second argument of Python script.

$ python3 tests/behavior/periodic_example.py 2
('p1', (191030.211397, ((0, 2),), 't1'))
('p2', (191030.211567, ((0, 2),), 't2'))
('p2', (191030.211661, ((0, 1),), 't2'))
('p1', (191030.211747, ((0, 2),), 't1'))
('p2', (191030.211865, ((0, 2),), 't2'))
('p2', (191030.211957, ((0, 1),), 't2'))
('p1', (191030.212038, ((0, 2),), 't1'))
('p2', (191030.212158, ((0, 2),), 't2'))
('p2', (191030.212239, ((0, 1),), 't2'))
('p1', (191030.212318, ((0, 2),), 't1'))
('p2', (191030.212428, ((0, 2),), 't2'))

$ python3 tests/behavior/periodic_example.py 3 test.gv
('p1', (190774.824447, ((0, 3),), 't1'))
('p2', (190774.824606, ((0, 3),), 't2'))
('p2', (190774.824707, ((0, 2),), 't2'))
('p2', (190774.824793, ((0, 1),), 't2'))
('p1', (190774.824881, ((0, 3),), 't1'))
('p2', (190774.825015, ((0, 3),), 't2'))
('p2', (190774.825109, ((0, 2),), 't2'))
('p2', (190774.825191, ((0, 1),), 't2'))
('p1', (190774.825274, ((0, 3),), 't1'))
('p2', (190774.825398, ((0, 3),), 't2'))
('p2', (190774.825487, ((0, 2),), 't2'))
('p2', (190774.825568, ((0, 1),), 't2'))
('p1', (190774.825649, ((0, 3),), 't1'))
('p2', (190774.825774, ((0, 3),), 't2'))

$ dot -Tpng test.gv > periodic_example.png

Examples¶

Producer/consumer¶

n-tester¶

Periodic¶

Table of Contents

Previous topic

Next topic

This Page