SoyutNet

soyutnet module

class soyutnet.SoyutNet(extra_routines: list[Coroutine[Any, Any, None]] = [])

Bases: object

property AUTO_REGISTER: bool
Arc(start: PTCommon | None = None, end: PTCommon | None = None, weight: int = 1, labels: Sequence[int] = (0,)) Arc
exception Break

Bases: Exception

Raised from bye() to exit SoyutNet context prematurely.

ComparativeObserver(*args: Any, **kwargs: Any) ComparativeObserver
DEBUG(*args: Any) None

Print debug messages when soyutnet.SoyutNet.DEBUG_ENABLED.

DEBUG_ENABLED: bool

if set, soyutnet.SoyutNet.DEBUG() will print.

DEBUG_V(*args: Any) None

Print debug messages when soyutnet.SoyutNet.VERBOSE_ENABLED.

ERROR(*args: Any) None

Print error messages.

ERROR_V(*args: Any) None

Print error messages when soyutnet.SoyutNet.VERBOSE_ENABLED.

FLOAT_DECIMAL_PLACE_FORMAT: int

Number of decimal places of floats in debug prints

property LOG_FILE: str
property LOOP_DELAY: float

Asyncio tasks main loop delay for debugging.

Returns:

Delay amount in seconds.

Observer(*args: Any, **kwargs: Any) Observer
PTRegistry(*args: Any, **kwargs: Any) PTRegistry
Place(*args: Any, **kwargs: Any) PTCommon
SLOW_MOTION: bool

If set, task loops are delayed for soyutnet.SoyutNet.LOOP_DELAY seconds

SpecialPlace(*args: Any, **kwargs: Any) PTCommon
Token(*args: Any, **kwargs: Any) Token
TokenRegistry(*args: Any, **kwargs: Any) TokenRegistry
Transition(*args: Any, **kwargs: Any) PTCommon
property VERBOSE_ENABLED: bool
_AUTO_REGISTER: bool

Automatically register when a new PT created.

_LOG_FILE: str

Name of the log file

_VERBOSE_ENABLED: bool

if set, soyutnet.SoyutNet.DEBUG_V() will print.

__print(*args: ~typing.Any, file: ~typing.TextIO = <_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>, **kwargs: ~typing.Any) None
__sprint(*args: Any, depth: int = 0, separator: str = ' ') str
static _auto_register(func: Callable[[Any, Any], PTCommon]) Callable[[Any, Any], PTCommon]

Decorator for automatically registering a new PT instance.

e.g. .. code:: python

p = net.Place() t = net.Transition()

automatically calls

net.registry.register(p)
net.registry.register(t)
_error(*args: ~typing.Any, file: ~typing.TextIO = <_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>, **kwargs: ~typing.Any) None
_extra_routines: list[Coroutine[Any, Any, None]]

List of additional task functions to be run in a SoyutNet context.

_logger: Logger | None

Log handler

_print(*args: ~typing.Any, file: ~typing.TextIO = <_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>, **kwargs: ~typing.Any) None
_reg: PTRegistry | None

Auto created PT registry if AUTO_REGISTER is enabled.

bye() None

Prematurely exits the SoyutNet context when called.

Raises:

Break

get_loop_name() str

Get the name of current loop which this function is called from.

Returns:

Name of the loop.

print(*args: Any, **kwargs: Any) None
property registry: PTRegistry
async sleep(amount: float = 0.0) None

Wrapper for task sleep function.

Parameters:

amount – Sleep amount in seconds.

time() float

Get current time since the program starts.

Returns:

Current time in seconds.

soyutnet._add_int_handlers(pt_registry: PTRegistry) None
soyutnet._cancel_all_tasks() None
soyutnet._int_handler(signame: str, loop: AbstractEventLoop, pt_registry: PTRegistry) None
async soyutnet.main(pt_registry: PTRegistry, extra_routines: list[Coroutine[Any, Any, None]] = []) None

Main entry point of PT net simulation.

Runs the tasks assigned to places and transitions registered in pt_registry.

Parameters:
  • pt_registry – Registry object keeping all places and transitions in the model.

  • extra_routines – Asyncio task functions to be run additional to the PT net loops.

soyutnet.run(*args: Any, ignore_cancelled_exception: bool = True, **kwargs: Any) None
soyutnet.terminate() None

Terminates PT net simulation.

soyutnet.pt_common module

class soyutnet.pt_common.Arc(start: Any, end: Any, weight: int = 1, labels: Sequence[int] = (0,))

Bases: object

Defines a generic labeled PT net arc which connects places to transitions or vice versa.

__gt__(*args: Any, **kwargs: Any) Any

Return self>value.

__lshift__(*args: Any, **kwargs: Any) Any
__lt__(*args: Any, **kwargs: Any) Any

Return self<value.

__rshift__(*args: Any, **kwargs: Any) Any
_end: ReferenceType[Any] | None

Output place/transition

_labels: tuple[int, ...]

The list of arc labels

_queue: Queue

Input/output queue for transmitting tokens from soyutnet.pt_common.Arc.start to soyutnet.pt_common.Arc.end

_start: ReferenceType[Any] | None

Input place/transition

static _validate(func: Any) Any

Decorator to automatically validate a value provided to a setter is correct.

property end: Any
get_graphviz_definition(t: int = 0, label_names: Dict[int, str] = {}) str

Generates graphviv DOT formated edge definition for the arc.

Parameters:

t – Event index for clustering multiple steps of PT net simulation.

Returns:

Edge definition.

index_at_end: int

Index in the list of output arcs of soyutnet.pt_common.Arc.end

index_at_start: int

Index in the list of output arcs of soyutnet.pt_common.Arc.start

is_enabled() bool

It is checked by the output transition at soyutnet.pt_common.Arc.end to detemine the transition is enabled or not.

Returns:

True if enabled.

labels(remember_last_processed: bool = False) Generator[int, None, None]

Generator to iterate through arc labels, _labels.

Parameters:

remember_last_processed – Continue iteration from the last index recorded a previous iteration.

async notify_observer(label: int, increment: int = -1) None

Called from soyutnet.transition.Transition._process_input_arcs() when the transition is fired.

async observe_input_places(requester: str = '') None

It is called by the output transition at soyutnet.pt_common.Arc.end after the transition is enabled.

It records the tokens counts just before the transition happens.

Parameters:

requester – The identity of caller.

async send(token: Tuple[int, int] | Tuple[Never, ...]) None

Puts a token to the output arc.

Parameters:

token – Token.

property start: Any
async wait() AsyncGenerator[Tuple[int, int] | Tuple[Never, ...], None]

Acquires soyutnet.pt_common.Arc.weight tokens from soyutnet.pt_common.Arc.start and yields them to soyutnet.pt_common.Arc.end

Returns:

Tokens.

weight: int

Arc weight

class soyutnet.pt_common.PTCommon(name: str = '', initial_tokens: Dict[int, list[int]] = {}, processor: Callable[[PTCommon], Awaitable[bool]] | None = None, **kwargs: Any)

Bases: Token

Base class implementing shared properties of places and transitions.

__gt__(pt_arc: Self | Arc | Set[Self], arc: Arc | None = None) Self | Arc

Used for readable connection syntax

e.g. p1 > t > p2

  • pt_arc is a PTCommon instance:

    pt1 > pt2, calls pt1.connect(pt2) and returns pt1.

  • pt_arc is an Arc instance:

    pt > arc, sets arc.start = self and returns arc.

  • pt_arc is a set: Runs,

    out = self
    for entry in pt_arc:
        out = out > entry
    assert out == self
    

    and, returns out.

Examples: PT connection syntax

Parameters:

pt_arc – PTCommon instance, Arc or set of PTCommon instances

Returns:

self if pt_arc is a PTCommon or Arc instance, self if pt_arc is a set.

__lshift__(pt_arc: Self | Arc | Set[Self], arc: Arc | None = None) Self | Arc

Used for readable connection syntax

e.g. p1 << t << p2

  • pt_arc is a PTCommon instance:

    pt1 << pt2, calls pt2.connect(pt1) and returns pt2.

  • pt_arc is an Arc instance:

    pt << arc, sets arc.start = self and returns arc.

  • pt_arc is a set: Runs,

    out = self
    for entry in pt_arc:
        out = out << entry
    assert out == entry
    

    and, returns out.

Examples: PT connection syntax

Parameters:

pt_arc – PTCommon instance, Arc or set of PTCommon instances

Returns:

pt_arc if pt_arc is a PTCommon or Arc instance, last entry of the set if pt_arc is a set.

__lt__(pt_arc: Self | Arc | Set[Self], arc: Arc | None = None) Self | Arc

Used for readable connection syntax

e.g. p1 < t < p2

  • pt_arc is a PTCommon instance:

    pt1 < pt2, calls pt1.connect(pt2) and returns pt1.

  • pt_arc is an Arc instance:

    pt < arc, sets arc.start = self and returns arc.

  • pt_arc is a set: Runs,

    out = self
    for entry in pt_arc:
        out = out < entry
    assert out == self
    

    and, returns out.

Examples: PT connection syntax

Parameters:

pt_arc – PTCommon instance, Arc or set of PTCommon instances

Returns:

self if pt_arc is a PTCommon or Arc instance, self if pt_arc is a set.

__rshift__(pt_arc: Self | Arc | Set[Self], arc: Arc | None = None) Self | Arc

Used for readable connection syntax

e.g. p1 >> t >> p2

  • pt_arc is a PTCommon instance:

    pt1 >> pt2, calls pt1.connect(pt2) and returns pt2.

  • pt_arc is an Arc instance:

    pt >> arc, sets arc.start = self and returns arc.

  • pt_arc is a set: Runs,

    out = self
    for entry in pt_arc:
        out = out >> entry
    assert out == entry
    

    and, returns out.

Examples: PT connection syntax

Parameters:
  • pt_arc – PTCommon instance, Arc or set of PTCommon instances

  • arc – Provides weight and labels of the arc when pt_arc is a PTCommon instance.

Returns:

pt_arc if pt_arc is a PTCommon or Arc instance, last entry of the set if pt_arc is a set.

async _get_input_arcs() AsyncGenerator[Arc, None]

Generator to iterate through input arcs.

Returns:

Input arcs.

async _get_output_arcs() AsyncGenerator[Arc, None]

Generator to iterate through output arcs.

Returns:

Output arcs.

_get_token(label: int) Tuple[int, int] | Tuple[Never, ...]

Gets the first token with the given label from a FIFO list.

Parameters:

label – Label.

Returns:

Token.

_get_token_count(label: int) int

Get the number of tokens with the given label.

Parameters:

label – Label.

Returns:

Number of tokens with the given label.

_input_arcs: list[Arc]

List of input arcs

_name: str

Name of the PT

async _observe(requester: str = '') None

Dummy observer.

_observer: Observer | None

Observes the tokens before each firing of output transitions

_output_arcs: list[Arc]

List of output arcs

async _process_input_arcs() bool

Acquires tokens from enabled input arcs and stores them.

Returns:

If True proceeds to processing tokens and output arcs, else continues waiting for enabled arcs.

async _process_output_arcs() None

Sends tokens to the output PTs.

async _process_tokens() bool

Processes input tokens before sending if required.

Returns:

True by default, else goes back to soyutnet.pt_common.PTCommon._process_input_arcs().

_processor: Callable[['PTCommon'], Awaitable[bool]] | None

Custom token processing function that is called between processing input and output arcs

_put_token(token: Tuple[int, int] | Tuple[Never, ...], strict: bool = True) int

Places tokens into a list based on its label.

Parameters:
Returns:

Number of tokens with the given label.

async _set_initial_marking() None
_tokens: TokenWalletType

Keeps tokens

connect(other: Self, weight: int = 1, labels: Sequence[int] = [0]) Self

Connects the output of self to the input of an other PT by creating an Arc in between.

Parameters:
  • other – The place/transition which it will be connected to.

  • weight – Arc weight.

  • labels – List of arc labels.

Returns:

Output place or transition that other references.

get_sorted_input_arcs() list[Arc]
get_token(label: int) Tuple[int, int] | Tuple[Never, ...]

Gets the first token with the given label from a FIFO list.

Parameters:

label – Label.

Returns:

A token if exists else empty token which is a null tuple (tuple()).

get_token_count(label: int) int

Get the number of tokens with the given label.

Parameters:

label – Label.

Returns:

Number of tokens.

ident() str

Returns the unique identifier of PT.

Returns:

Unique identifier.

is_dangling() bool
async observe(requester: str = '') None

Public observe function called by output arcs.

Parameters:

requester – The identity of caller.

put_token(label: int = 0, id: int = 0) int

Places tokens into a list based on its label.

Parameters:
  • label – Label.

  • id – ID.

Returns:

Number of tokens with the given label.

async should_continue() bool

Main loop of async task assigned to the PT.

Returns:

Continues task if True.

async soyutnet.pt_common._loop(pt: PTCommon) None

Task function assigned to the PT.

Parameters:

pt – PTCommon instance.

soyutnet.place module

class soyutnet.place.Place(name: str = '', observer: Observer | None = None, observer_record_limit: int = 0, observer_verbose: bool = False, **kwargs: Any)

Bases: PTCommon

Defines PTNet places.

async _observe(requester: str = '') None

Save token counts for each label to the observer’s records.

Also, add the number of tokens sent to the caller arc.

Parameters:

token_count_in_arc – Number of tokens in the output arc of places must also be added to the count.

class soyutnet.place.SpecialPlace(name: str = '', consumer: Callable[[SpecialPlace], Awaitable[None]] | None = None, producer: Callable[[SpecialPlace], Awaitable[list[Tuple[int, int] | Tuple[Never, ...]]]] | None = None, **kwargs: Any)

Bases: Place

Custom place class whose token processing methods can be overriden.

_consumer: Callable[['SpecialPlace'], Awaitable[None]] | None

Custom soyutnet.pt_common.PTCommon._process_input_arcs() function.

async _process_input_arcs() bool

Calls custom producer function after the default soyutnet.pt_common.PTCommon._process_input_arcs().

Returns:

If True continues to processing tokens and output arcs, else loops back to processing input arcs.

async _process_output_arcs() None

Handles consumer function first, if it is defined.

See, soyutnet.place.Place._process_output_arcs().

async _process_tokens() bool

Calls custom consumer function. If it is None calls soyutnet.pt_common.PTCommon._process_tokens() function.

_producer: Callable[['SpecialPlace'], Awaitable[list[TokenType]]] | None

Custom soyutnet.pt_common.PTCommon._process_output_arcs() function.

async observe(requester: str = '') None

Public observe function called by output arcs.

Parameters:

requester – The identity of caller.

soyutnet.transition module

soyutnet.transition.FiringHistoryType

Type for list of firing records

alias of list[Tuple[float]]

soyutnet.transition.FiringRecordType

Firing record type

alias of Tuple[float]

class soyutnet.transition.Transition(name: str = '', record_firing: bool = False, **kwargs: Any)

Bases: PTCommon

Defines PTNet transitions.

_firing_records: FiringHistoryType

Keeps timestampts of each firing of the transition: py:attr:soyutnet.transition.FiringRecordType

_new_firing_record() None
_no_of_times_enabled: int

Counts the number of time the transition is enabled

_notifier: asyncio.Condition

Firing notifier

async _process_input_arcs() bool

Acquires and stores tokens.

Returns:

True if the transition is enabled, else goes back to waiting input arcs to be enabled.

async _process_output_arcs() None

Fires the transition.

NOTE: sum of w(p_prev, self) == sum of w(self, p_next) must satisfy for each label.

Sends tokens to the output places when required conditions are satisfied.

_record_firing: bool

Enables recording firings of transitions

get_firing_records() list[Tuple[float]]

Returns all firing records. soyutnet.transition.Transition._firing_records

Returns:

Firing records.

get_no_of_times_enabled() int

Returns number of times the transition is enabled.

Returns:

Number of times the transition is enabled.

async wait_for_firing() bool

soyutnet.token module

class soyutnet.token.Token(label: int = 0, binding: Any = None, **kwargs: Any)

Bases: BaseObject

_binding: Any

Binded object

_id: id_t

Unique token ID

_label: label_t

Token label

get_binding() Any

Returns the object binded to the token.

Returns:

Binded object.

get_id() int

Returns token’s unique ID.

Returns:

ID.

get_label() int

Returns the label of token

Returns:

Label.

soyutnet.observer module

class soyutnet.observer.ComparativeObserver(expected: Dict[int, list[Any]], on_comparison_ends: Callable[[ComparativeObserver], None] | None = None, **kwargs: Any)

Bases: Observer

This observer compares the records to the provided list for test purposes.

_expected: Dict[int, list[Any]]

The list of values to be compared to

_expected_index: int

Index of the last value compared in the list

_is_comparing: bool

Set to False when there are no values left to compare in soyutnet.observer.ComparativeObserver._expected

_on_comparison_ends: Callable[['ComparativeObserver'], None] | None

Callback function to be called when comparison ends

async _save(record: Tuple[float, Tuple[Tuple[int, int] | Tuple[Never, ...], ...], str]) None

Save a new record after comparing.

Parameters:

record – Record.

Returns:

Number of records.

soyutnet.observer.MergedRecordsType

A type for list of all records of all observers

alias of list[Tuple[str, Tuple[float, Tuple[Tuple[int, int] | Tuple[Never, …], …], str] | Tuple[float]]]

class soyutnet.observer.Observer(record_limit: int = 0, verbose: bool = False, place: Any | None = None, **kwargs: Any)

Bases: BaseObject

Can be assigned to a place to observe its state.

_add_record(record: Tuple[float, Tuple[Tuple[int, int] | Tuple[Never, ...], ...], str]) int

Adds a new record.

Parameters:

record – Record.

Returns:

Number of records.

async _clean_records() int

Cleans records when it is required.

Returns:

Number of records kept after cleaning the old ones.

async _display_records() None

Print records to the stdout.

_hysteresis_bounds: Tuple[int, int]

Deleting old records are optimized by an hysteresis comparator

_lock: asyncio.Lock

Async lock for safe access to records

_place: ReferenceType[Place] | None

Weak reference to the soyutnet.place.Place that is observed.

_record_limit: int

Maximum number of records to be kept.

_records: ObserverHistoryType

Records of observations. soyutnet.observer.ObserverHistoryType

async _save(record: Tuple[float, Tuple[Tuple[int, int] | Tuple[Never, ...], ...], str]) None

Save a new record.

Parameters:

record – Record.

_set_place(place: Any | None) None
_verbose: bool

Prints the observations if True

get_records(column: int = -1) list[Any]

Returns the records at the specified column.

Parameters:

column – The column index of records requested. It returns all columns if it it -1.

Returns:

List of records.

ident() str

Get object’s unique identifier

Returns:

Identifier string.

async inc_token_count(label: int, inc: int = 1) None

Adds to the token count with the given label.

It is called by places when a new token acquired or sent.

Parameters:
  • label – Token label.

  • inc – The value to be added to the count.

async save(requester: str = '') None

Save counted tokens to the list of records.

It is called by output transitions when they are enabled.

Parameters:

requester – The identity of the caller.

soyutnet.observer.ObserverHistoryType

Type for a list of soyutnet.observer.ObserverRecordType

alias of list[Tuple[float, Tuple[Tuple[int, int] | Tuple[Never, …], …], str]]

soyutnet.observer.ObserverRecordType

Observer records: (observation time, label, no of tokens with the label, identity of requester of the record)

alias of Tuple[float, Tuple[Tuple[int, int] | Tuple[Never, …], …], str]

soyutnet.registry module

soyutnet.registry.DirectoryType

Registry directory type

alias of Dict[int, list[Tuple[int, Any]]]

class soyutnet.registry.PTRegistry(**kwargs: Any)

Bases: Registry

Keeps track of PTCommon instances.

_get_graphviz_node_definition(pt: PTCommon, t: int) str
generate_graph(net_name: str = 'Net', indent: str = '\t', label_names: Dict[int, str] = {}, ignore_dangling_pts: bool = True) str

Generated graph definition in Graphviz dot text format.

Parameters:
  • net – Given name of the PT net

  • indent – Indentation string used in sub-blocks of dot text format

  • label_names – Readable version of label_t types.

  • ignore_dangling_pts – The PTs with no input/output connections are ignored when it is set.

get_loops() Generator[Coroutine[Any, Any, None], None, None]

Yields asyncio task functions assigned to the PT.

Returns:

Asyncio task function.

get_merged_records(ignore_special_places: bool = True, place_names: list[str] = []) list[Tuple[str, Tuple[float, Tuple[Tuple[int, int] | Tuple[Never, ...], ...], str] | Tuple[float]]]

Merges all observer records and sorts by their timestamps.

Returns:

Merged and sorted observer records.

register(pt: PTCommon) int

Registers a PT.

Parameters:

pt – PTCommon instance.

Returns:

Unique ID assigned to the PT.

soyutnet.registry.PostRegisterCallbackType

Type of callbacks run after an object is registered

alias of Callable[[int, Any], None]

class soyutnet.registry.Registry(**kwargs: Any)

Bases: BaseObject

Registry keeps track of (label, id) tuples and the objects assigned to them. It generates unique ids for new objects.

_directory: DirectoryType

Keeps all objects categorized by labels

_id_counter: id_t

Auto-incrementing id assigned to new objects

_lock: asyncio.Lock

Locks access to soyutnet.registry.Registry._directory

_new_id() int

Creates new ids for new objects.

Returns:

Unique id

entries(label: int | None = None) Generator[Any, None, None]

Iterates through all entries with the given label.

Parameters:

label – Label. If it is None, iterates through all labels.

Returns:

Yields entries

get_entry(label: int, id: int | None = None, del_entry: bool = False) Any

Returns a token with given label and ID.

Parameters:
  • label – Label.

  • id – ID. If None, returns the first entry with the given label.

  • del_entry – Removes the entry from the registry if True.

Returns:

The registered token.

get_entry_count(label: int = 0) int

Returns the number of entries with the given label.

Parameters:

label – Label.

Returns:

Number of entries.

get_first_entry(label: int = 0) Tuple[int, Any]

Returns first entry with given label. First entry is the one registered first.

Parameters:

label – Label.

Returns:

Entry.

pop_entry(label: int, id: int | None = None) Any

Returns a token with given label and ID and removes it from the registry.

Parameters:
  • label – Label.

  • id – ID. If None, returns the first entry with the given label.

Returns:

The registered token.

register(obj: ~typing.Any, post_register_callback: ~typing.Callable[[int, ~typing.Any], None] = <function _default_post_register_callback>) int

Register a new object

Parameters:
  • obj – New object of any type.

  • post_register_callback – Called after object is registered.

Returns:

Assigned unique ID.

class soyutnet.registry.TokenRegistry(**kwargs: Any)

Bases: Registry

register(token: Token) int

Register a new token

Parameters:

token – New token.

Returns:

Assigned unique ID.

soyutnet.registry._default_post_register_callback(dummy1: Any, dummy2: int) None

Default dummy soyutnet.registry.PostRegisterCallbackType

soyutnet.constants module

class soyutnet.constants.BaseObject(net: SoyutNet)

Bases: object

Base SoyutNet object inherited by all classes.

_net: ReferenceType['SoyutNet']

Reference to the creator SoyutNet instance.

ident() str

Get object’s unique identifier

Returns:

Identifier string.

property net: SoyutNet

Get reference of the SoyutNet instance assigned to instances of all object types, for a particular simulation.

Returns:

Reference to the creator SoyutNet instance.

soyutnet.constants.GENERIC_ID: int = 0

Generic ID

soyutnet.constants.GENERIC_LABEL: int = 0

Generic label

soyutnet.constants.INVALID_ID: int = -11

Invalid id

soyutnet.constants.INVALID_LABEL: int = -10

Invalid label

exception soyutnet.constants.SoyutNetError(message: str = 'An error occured.')

Bases: Exception

Generic error class left as future work.

soyutnet.constants.TokenType

Token type

alias of Tuple[int, int] | Tuple[Never, …]

soyutnet.constants.TokenWalletType

Token container type

alias of Dict[int, list[int]]

soyutnet.constants.id_t

ID type

soyutnet.constants.label_t

Label type

soyutnet.constants.random_identifier(N: int = 5) str

Generates a random string.

Parameters:

N – Length of random string

Returns:

Random string