autonomy.deploy.base
Base deployments module.
ENV_
VAR_
AEA_
PASSWORD
nosec
ENV_
VAR_
DEPENDENCIES
nosec
tm_
write_
to_
log
def tm_write_to_log() -> bool
Check the environment variable to see if the user wants to write to log file or not.
NotValidKeysFile Objects
class NotValidKeysFile(Exception)
Raise when provided keys file is not valid.
ResourceValues Objects
class ResourceValues(TypedDict)
Resource type.
Resource Objects
class Resource(TypedDict)
Resource values.
Resources Objects
class Resources(TypedDict)
Deployment resources.
ServiceBuilder Objects
class ServiceBuilder()
Class to assist with generating deployments.
__
init__
def __init__(service: Service,
keys: Optional[List[Union[List[Dict[str, str]],
Dict[str, str]]]] = None,
agent_instances: Optional[List[str]] = None,
apply_environment_variables: bool = False) -> None
Initialize the Base Deployment.
get_
abci_
container_
name
def get_abci_container_name(index: int) -> str
Format ABCI container name.
get_
tm_
container_
name
def get_tm_container_name(index: int) -> str
Format tendermint container name.
try_
get_
all_
participants
def try_get_all_participants() -> Optional[List[str]]
Try get all participants from the ABCI overrides
agent_
instances
@property
def agent_instances() -> Optional[List[str]]
Agent instances.
agent_
instances
@agent_instances.setter
def agent_instances(instances: List[str]) -> None
Agent instances setter.
keys
@property
def keys() -> List[Union[List[Dict[str, str]], Dict[str, str]]]
Keys.
from_
dir
@classmethod
def from_dir(cls,
path: Path,
keys_file: Optional[Path] = None,
number_of_agents: Optional[int] = None,
agent_instances: Optional[List[str]] = None,
apply_environment_variables: bool = False) -> "ServiceBuilder"
Service builder from path.
verify_
agent_
instances
@staticmethod
def verify_agent_instances(addresses: Set[str],
agent_instances: List[str]) -> None
Cross verify agent instances with the keys.
read_
keys
def read_keys(keys_file: Path) -> None
Read in keys from a file on disk.
try_
update_
runtime_
params
def try_update_runtime_params(
multisig_address: Optional[str] = None,
agent_instances: Optional[List[str]] = None,
consensus_threshold: Optional[int] = None) -> None
Try and update setup parameters.
get_
maximum_
participants
def get_maximum_participants() -> int
Returns the maximum number of participants
try_
update_
abci_
connection_
params
def try_update_abci_connection_params() -> None
Try and update ledger connection parameters.
process_
component_
overrides
def process_component_overrides(agent_n: int) -> Dict
Generates env vars based on model overrides.
generate_
agents
def generate_agents() -> List
Generate multiple agent.
generate_
common_
vars
def generate_common_vars(agent_n: int) -> Dict
Retrieve vars common for agent.
generate_
agent
def generate_agent(agent_n: int,
override_idx: Optional[int] = None) -> Dict[Any, Any]
Generate next agent.
BaseDeploymentGenerator Objects
class BaseDeploymentGenerator(abc.ABC)
Base Deployment Class.
__
init__
def __init__(service_builder: ServiceBuilder,
build_dir: Path,
use_tm_testnet_setup: bool = False,
dev_mode: bool = False,
packages_dir: Optional[Path] = None,
open_aea_dir: Optional[Path] = None,
open_autonomy_dir: Optional[Path] = None,
image_author: Optional[str] = None,
resources: Optional[Resources] = None)
Initialise with only kwargs.
generate
@abc.abstractmethod
def generate(image_version: Optional[str] = None,
use_hardhat: bool = False,
use_acn: bool = False) -> "BaseDeploymentGenerator"
Generate the deployment configuration.
generate_
config_
tendermint
@abc.abstractmethod
def generate_config_tendermint() -> "BaseDeploymentGenerator"
Generate the deployment configuration.
populate_
private_
keys
@abc.abstractmethod
def populate_private_keys() -> "BaseDeploymentGenerator"
Populate the private keys to the deployment.
write_
config
def write_config() -> "BaseDeploymentGenerator"
Write output to build dir