symaware.base.utils package

Submodules

symaware.base.utils.async_loop_lock module

class symaware.base.utils.async_loop_lock.AsyncLoopLock(loop=None)

Bases: ABC

Generic interface for a lock used to synchronize an async loop. Used by AsyncLoopLockable objects, in order to synchronize themselves in an async loop. The next iteration of the loop will run only after the object has completed its task and the lock has been released. In other words, the time between each iteration of the loop it determined by which of the two takes longer.

Example

A lock that synchronizes an async loop at a fixed time interval of 1. Until the task takes less than 1 second, the next loop will wait for the lock before starting, meaning that the task will be executed at most once per second. If the task takes more than 1 second, the next loop will start immediately after the task is completed.

>>> import asyncio
>>> from symaware.base.utils import TimeIntervalAsyncLoopLock
>>> # Create the lock
>>> lock = TimeIntervalAsyncLoopLock(1)
>>>
>>> async def task(delay: float):
...     asyncio.sleep(delay)
...
>>> # This could be an object that implements the AsyncLoopLockable interface
>>> async def run():
...     task_time = 0
...     while True:
...         await task(task_time) # Do something that takes "task_time" seconds
...         await lock.next_loop() # Wait for the next loop to start
...         task_time += 0.1
...

Parameters:

loop (AbstractEventLoop | None, default: None) – event loop. If none is provided, the default one will be used

property loop: AbstractEventLoop

Event loop

abstract async next_loop()

Wait for the next loop to start.

Example

>>> async def run(self):
...     while True:
...         # Do something
...         await self.next_loop() # Wait for the next loop to start

async release_loop()

Release the lock, causing the object to complete all future loops immediately.

property released: bool

Whether the lock has been released. It usually means that the simulation is stopping.

class symaware.base.utils.async_loop_lock.ConditionAsyncLoopLock(loop=None)

Bases: AsyncLoopLock

Lock used to synchronize an async loop using a condition.

Parameters:

loop (AbstractEventLoop | None, default: None) – event loop. If none is provided, the default one will be used

property condition: Condition

async next_loop()

Wait for the next loop to start.

Example

>>> async def run(self):
...     while True:
...         # Do something
...         await self.next_loop() # Wait for the next loop to start

async release_loop()

Release the lock, causing the object to complete all future loops immediately.

async trigger()

Trigger the event, causing the next loop to start.

class symaware.base.utils.async_loop_lock.DefaultAsyncLoopLock(loop=None)

Bases: AsyncLoopLock

Default lock used to synchronize an async loop. It does not wait for the next loop to start.

Parameters:

loop (AbstractEventLoop | None, default: None) – event loop. If none is provided, the default one will be used

async next_loop()

Wait for the next loop to start.

Example

>>> async def run(self):
...     while True:
...         # Do something
...         await self.next_loop() # Wait for the next loop to start

class symaware.base.utils.async_loop_lock.EventAsyncLoopLock(loop=None)

Bases: AsyncLoopLock

Lock used to synchronize an async loop using an event.

Parameters:

loop (AbstractEventLoop | None, default: None) – event loop. If none is provided, the default one will be used

clear()

Clear the event, resetting it to the non-set state. The next loop will not start until the event is triggered again.

property event: Event

async next_loop()

Wait for the next loop to start.

Example

>>> async def run(self):
...     while True:
...         # Do something
...         await self.next_loop() # Wait for the next loop to start

async release_loop()

Release the lock, causing the object to complete all future loops immediately.

trigger()

Trigger the event, causing the next loop to start.

class symaware.base.utils.async_loop_lock.TimeIntervalAsyncLoopLock(time_interval, loop=None)

Bases: AsyncLoopLock

Lock used to synchronize an async loop at a fixed time interval.

Parameters:

• time_interval (float) – time interval in seconds

• loop (AbstractEventLoop | None, default: None) – event loop. If none is provided, the default one will be used

property last_timestamp: float

async next_loop()

Wait for the next loop to start.

Example

>>> async def run(self):
...     while True:
...         # Do something
...         await self.next_loop() # Wait for the next loop to start

property time_interval: float

symaware.base.utils.async_loop_lockable module

class symaware.base.utils.async_loop_lockable.AsyncLoopLockable(async_loop_lock=None)

Bases: Generic[Tasynclooplock]

Generic interface for objects that need to be synchronized in an async loop.

Parameters:

async_loop_lock (Optional[TypeVar(Tasynclooplock, bound= AsyncLoopLock)], default: None) – Lock used to synchronize the component in an async loop

property async_loop_lock: Tasynclooplock

Async loop lock

async async_stop()

Release the async loop lock. It means that the object will complete the current loop immediately and then stop.

Example

This function is used by the Agent when it stops the async loop.

property can_loop: bool

async next_loop()

Wait for the next loop to start

Example

This function is used by the Agent to synchronize the Component in an async loop.

symaware.base.utils.logging module

symaware.base.utils.logging.get_logger(file_name, class_name=None)

Returns a logger with an appropriate name. If no class name is provided, the last part of the file name is capitalized and used instead.

Example

The logger utility can be used in any module. Is is recommended give the logger the name of the module and the class that uses it.

>>> from symaware.base import get_logger
>>> logger = get_logger(__name__, "MyClass")
>>> logger.info("Hello world!")

Parameters:

• file_name (str) – The name of the file where the logger is defined

• class_name (str | None, default: None) – The name of the class that will use the logger

Returns:

Logger – The logger

symaware.base.utils.logging.initialize_logger(level=20)

Set the logging level for all loggers

Parameters:

level (str | int, default: 20) – The logging level to set

symaware.base.utils.logging.log(logger=<Logger symaware.base.utils.Logging (WARNING)>, level=10)

Decorator to log the input and output of a function.

Examples

Using the decorator to log the input and output of a function

>>> import logging
>>> from symaware.base import log, get_logger, initialize_logger
>>>
>>> initialize_logger(logging.INFO)
>>> logger = get_logger(__name__)
>>>
>>> @log(logger, logging.INFO)
... def my_function(x):
...     return x + 1
>>>
>>> # INFO:__main__:my_function(1, {})
>>> # INFO:__main__:my_function(...) -> 2
>>> my_function(1)
2

>>> import logging
>>> from symaware.base import log, get_logger, initialize_logger
>>>
>>> initialize_logger(logging.DEBUG)
>>> logger = get_logger(__name__)
>>>
>>> class MyClass:
...     __LOGGER = get_logger(__name__, "MyClass")
...     @log(__LOGGER)
...     def my_function(self, x):
...         return x + 2
>>>
>>> my_instance = MyClass()
>>> # DEBUG:__main__.MyClass:my_function(4, {})
>>> # DEBUG:__main__.MyClass:my_function(...) -> 6
>>> my_instance.my_function(4)
6

Parameters:

• logger (Logger, default: <Logger symaware.base.utils.Logging (WARNING)>) – what logger to use. If not provided, a default logger from this module is used

• level (int, default: 10) – what logging level to use

Returns:

Callable[[Callable[[ParamSpec(Param, bound= None)], TypeVar(Return)]], Callable[[ParamSpec(Param, bound= None)], TypeVar(Return)]] – Log function decorator

symaware.base.utils.null_object module

class symaware.base.utils.null_object.NullObject

Bases: object

Object implementing the null object pattern. It is used as a placeholder when an object is needed but no actual object is available. Using the object in any way will raise an exception.

Example

Using a NullObject results in an exception being raised.

>>> from symaware.base.utils import NullObject
>>> null_object = NullObject()
>>> try:
...     null_object.attribute
... except NotImplementedError as e:
...     print(e)
NullObject is a NullObject, it is used as a placeholder. Make sure to instantiate a valid instance of this class

_instance: NullObject | None = None

_null_error()

Return type:

NoReturn

classmethod instance()

Get the instance of the NullObject with a singleton pattern.

Returns:

TypeVar(Nullobjectself, bound= NullObject) – The singleton instance of the NullObject.

symaware.base.utils.publisher module

class symaware.base.utils.publisher.Publisher

Bases: ABC

The Publisher interface declares a set of methods for managing subscribers (observers) and notifying them of state changes, implementing the Publisher-Subscriber pattern. The publisher will invoke the callback provided by the subscriber when a determinate event occurs.

Other Packages can use this class to get notified of events, exchanging information and triggering actions.

_callbacks

A dictionary of set of callbacks added to this publisher. The key for each set is the identifier of the event the callback is interested in. Note that each set does not allow duplicates.

_add(event, callback)

Add a callback to the internal dictionary of callbacks.

Parameters:

• event (str) – Identifier of the event the callback will be invoked for

• callback (Callable) – The callback to attach. It is a callable object that will be invoked when the publisher’s state changes

_notify(event, *data)

Notify the subscribers of the event.

Example

Notifying the subscribers of an event will invoke the callbacks attached to that event.

>>> from symaware.base.utils import Publisher
>>> phrases = []
>>> publisher = Publisher() # concrete subclass of Publisher
>>> publisher.add("event", lambda: phrases.append("Hello World!"))
>>> publisher.add("event", lambda: phrases.append("Hello"))
>>> publisher.add("another_event", lambda: phrases.append("OK"))
>>> publisher._notify("event")
>>> print(" - ".join(sorted(phrases)))
Hello - Hello World!

Parameters:

• event (str) – event identifier

• data (Any) – data to invoke the callbacks with

_remove(event, callback)

Remove a callback from the internal dictionary of callbacks.

Parameters:

• event (str) – Identifier of the event the callback was attached to

• callback (Callable) – The callback to remove

add(event, callback)

Add a callback to the publisher. From now on, every time the publisher’s state changes, it will invoke the callback. Keep in mind that duplicate callbacks are not allowed, so calling this method multiple times with the same one will not add it multiple times. This said, it is easy to create multiple semantically equals callbacks, for example by using lambda functions.

Warning

Using the add() method directly is discouraged, for no static type checking is performed. It is recommended to use the specific method provided by the concrete subclass of Publisher.

Example

Adding callbacks to the publisher and notifying them of an event. Only the callbacks added to that specific event will be invoked.

>>> from symaware.base.utils import Publisher
>>> publisher = Publisher() # concrete subclass of Publisher
>>> publisher.add("event", lambda: print("Hello World!"))
>>> publisher.add("another_event", lambda: print("OK"))
>>> publisher._notify("event")
Hello World!

Parameters:

• event (str) – Identifier of the event the callback will be invoked for.

• callback (Callable) – The callback to attach. It is a callable object that will be invoked when the publisher’s state changes.

remove(event, callback)

Remove a callback from the publisher. It won’t be invoked anymore when a notification is sent.

Warning

Using the remove() method directly is discouraged, for no static type checking is performed. It is recommended to use the specific method provided by the concrete subclass of Publisher.

Example

Removing a callbacks from the publisher and notifying the remaining ones of an event.

>>> from symaware.base.utils import Publisher
>>> publisher = Publisher() # concrete subclass of Publisher
>>> fun = lambda: print("Hello World!")
>>> publisher.add("event", fun)
>>> publisher.add("event", lambda: print("Hello"))
>>> publisher.remove("event", fun)
>>> publisher._notify("event")
Hello

Parameters:

• event (str) – Identifier of the event the callback was attached to.

• callback (Callable) – The callback to remove.

Module contents