Skip to content

aea.decision_maker.base

This module contains the decision maker class.

OwnershipState Objects

class OwnershipState(ABC)

Represent the ownership state of an agent (can proxy a ledger).

set

@abstractmethod
def set(**kwargs: Any) -> None

Set values on the ownership state.

Arguments:

  • kwargs: the relevant keyword arguments

apply_delta

@abstractmethod
def apply_delta(**kwargs: Any) -> None

Apply a state update to the ownership state.

This method is used to apply a raw state update without a transaction.

Arguments:

  • kwargs: the relevant keyword arguments

is_initialized

@property
@abstractmethod
def is_initialized() -> bool

Get the initialization status.

is_affordable_transaction

@abstractmethod
def is_affordable_transaction(terms: Terms) -> bool

Check if the transaction is affordable (and consistent).

Arguments:

  • terms: the transaction terms

Returns:

True if the transaction is legal wrt the current state, false otherwise.

apply_transactions

@abstractmethod
def apply_transactions(list_of_terms: List[Terms]) -> "OwnershipState"

Apply a list of transactions to (a copy of) the current state.

Arguments:

  • list_of_terms: the sequence of transaction terms.

Returns:

the final state.

__copy__

@abstractmethod
def __copy__() -> "OwnershipState"

Copy the object.

Preferences Objects

class Preferences(ABC)

Class to represent the preferences.

set

@abstractmethod
def set(**kwargs: Any) -> None

Set values on the preferences.

Arguments:

  • kwargs: the relevant key word arguments

is_initialized

@property
@abstractmethod
def is_initialized() -> bool

Get the initialization status.

Returns True if exchange_params_by_currency_id and utility_params_by_good_id are not None.

marginal_utility

@abstractmethod
def marginal_utility(ownership_state: OwnershipState, **kwargs: Any) -> float

Compute the marginal utility.

Arguments:

  • ownership_state: the ownership state against which to compute the marginal utility.
  • kwargs: optional keyword arguments

Returns:

the marginal utility score

utility_diff_from_transaction

@abstractmethod
def utility_diff_from_transaction(ownership_state: OwnershipState,
                                  terms: Terms) -> float

Simulate a transaction and get the resulting utility difference (taking into account the fee).

Arguments:

  • ownership_state: the ownership state against which to apply the transaction.
  • terms: the transaction terms.

Returns:

the score.

__copy__

@abstractmethod
def __copy__() -> "Preferences"

Copy the object.

ProtectedQueue Objects

class ProtectedQueue(Queue)

A wrapper of a queue to protect which object can read from it.

__init__

def __init__(access_code: str) -> None

Initialize the protected queue.

Arguments:

  • access_code: the access code to read from the queue

put

def put(internal_message: Optional[Message],
        block: bool = True,
        timeout: Optional[float] = None) -> None

Put an internal message on the queue.

If optional args block is true and timeout is None (the default), block if necessary until a free slot is available. If timeout is a positive number, it blocks at most timeout seconds and raises the Full exception if no free slot was available within that time. Otherwise (block is false), put an item on the queue if a free slot is immediately available, else raise the Full exception (timeout is ignored in that case).

Arguments:

  • internal_message: the internal message to put on the queue
  • block: whether to block or not
  • timeout: timeout on block

Raises:

  • ValueError: if the item is not an internal message

put_nowait

def put_nowait(internal_message: Optional[Message]) -> None

Put an internal message on the queue.

Equivalent to put(item, False).

Arguments:

  • internal_message: the internal message to put on the queue

Raises:

  • ValueError: if the item is not an internal message

get

def get(block: bool = True, timeout: Optional[float] = None) -> None

Inaccessible get method.

Arguments:

  • block: whether to block or not
  • timeout: timeout on block

Raises:

  • ValueError: access not permitted.

get_nowait

def get_nowait() -> None

Inaccessible get_nowait method.

Raises:

  • ValueError: access not permitted.

protected_get

def protected_get(access_code: str,
                  block: bool = True,
                  timeout: Optional[float] = None) -> Optional[Message]

Access protected get method.

Arguments:

  • access_code: the access code
  • block: If optional args block is true and timeout is None (the default), block if necessary until an item is available.
  • timeout: If timeout is a positive number, it blocks at most timeout seconds and raises the Empty exception if no item was available within that time.

Raises:

  • ValueError: if caller is not permitted

Returns:

internal message

DecisionMakerHandler Objects

class DecisionMakerHandler(WithLogger, ABC)

This class implements the decision maker.

__init__

def __init__(identity: Identity, wallet: Wallet, config: Dict[str, Any],
             **kwargs: Any) -> None

Initialize the decision maker handler.

Arguments:

  • identity: the identity
  • wallet: the wallet
  • config: the user defined configuration of the handler
  • kwargs: the key word arguments

agent_name

@property
def agent_name() -> str

Get the agent name.

identity

@property
def identity() -> Identity

Get identity of the agent.

wallet

@property
def wallet() -> Wallet

Get wallet of the agent.

config

@property
def config() -> Dict[str, Any]

Get user defined configuration

context

@property
def context() -> SimpleNamespace

Get the context.

message_out_queue

@property
def message_out_queue() -> AsyncFriendlyQueue

Get (out) queue.

handle

@abstractmethod
def handle(message: Message) -> None

Handle an internal message from the skills.

Arguments:

  • message: the internal message

DecisionMaker Objects

class DecisionMaker(WithLogger)

This class implements the decision maker.

__init__

def __init__(decision_maker_handler: DecisionMakerHandler) -> None

Initialize the decision maker.

Arguments:

  • decision_maker_handler: the decision maker handler

agent_name

@property
def agent_name() -> str

Get the agent name.

message_in_queue

@property
def message_in_queue() -> ProtectedQueue

Get (in) queue.

message_out_queue

@property
def message_out_queue() -> AsyncFriendlyQueue

Get (out) queue.

decision_maker_handler

@property
def decision_maker_handler() -> DecisionMakerHandler

Get the decision maker handler.

start

def start() -> None

Start the decision maker.

stop

def stop() -> None

Stop the decision maker.

execute

def execute() -> None

Execute the decision maker.

Performs the following while not stopped:

  • gets internal messages from the in queue and calls handle() on them

handle

def handle(message: Message) -> None

Handle an internal message from the skills.

Arguments:

  • message: the internal message