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¶
- 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.
- PTRegistry(*args: Any, **kwargs: Any) PTRegistry¶
- SLOW_MOTION: bool¶
If set, task loops are delayed for
soyutnet.SoyutNet.LOOP_DELAYseconds
- TokenRegistry(*args: Any, **kwargs: Any) TokenRegistry¶
- 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.
- 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:
objectDefines 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.starttosoyutnet.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.endto detemine the transition is enabled or not.- Returns:
Trueif 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.endafter 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.weighttokens fromsoyutnet.pt_common.Arc.startand yields them tosoyutnet.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:
TokenBase 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 > p2pt_arcis aPTCommoninstance:pt1 > pt2, callspt1.connect(pt2)and returnspt1.
pt_arcis anArcinstance:pt > arc, setsarc.start = selfand returnsarc.
pt_arcis aset: 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 << p2pt_arcis aPTCommoninstance:pt1 << pt2, callspt2.connect(pt1)and returnspt2.
pt_arcis anArcinstance:pt << arc, setsarc.start = selfand returnsarc.
pt_arcis aset: 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 < p2pt_arcis aPTCommoninstance:pt1 < pt2, callspt1.connect(pt2)and returnspt1.
pt_arcis anArcinstance:pt < arc, setsarc.start = selfand returnsarc.
pt_arcis aset: 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 >> p2pt_arcis aPTCommoninstance:pt1 >> pt2, callspt1.connect(pt2)and returnspt2.
pt_arcis anArcinstance:pt >> arc, setsarc.start = selfand returnsarc.
pt_arcis aset: 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_arcis aPTCommoninstance.
- 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.
- _name: str¶
Name of the PT
- async _observe(requester: str = '') None¶
Dummy observer.
- async _process_input_arcs() bool¶
Acquires tokens from enabled input arcs and stores them.
- Returns:
If
Trueproceeds 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:
Trueby default, else goes back tosoyutnet.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:
token – A label and ID pair.
strict – If set the label of token must already be in
soyutnet.pt_common.PTCommon._tokensdictionary.
- 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
otherreferences.
- 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.
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:
PTCommonDefines 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:
PlaceCustom 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
Truecontinues 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
Nonecallssoyutnet.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:
PTCommonDefines 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:
Trueif 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:
ObserverThis 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
Falsewhen there are no values left to compare insoyutnet.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:
BaseObjectCan 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.Placethat 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.ObserverRecordTypealias 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:
RegistryKeeps track of PTCommon instances.
- 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_ttypes.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.
- 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:
BaseObjectRegistry 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.
- 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:
objectBase 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.
- 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:
ExceptionGeneric 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