from pathlib import Path
from typing import Optional, Union
from multiversx_sdk.abi.address_value import AddressValue
from multiversx_sdk.abi.biguint_value import BigUIntValue
from multiversx_sdk.abi.bytes_value import BytesValue
from multiversx_sdk.abi.interface import ISingleValue
from multiversx_sdk.abi.serializer import Serializer
from multiversx_sdk.abi.small_int_values import U32Value
from multiversx_sdk.core.address import Address
from multiversx_sdk.core.base_factory import BaseFactory
from multiversx_sdk.core.constants import STAKING_SMART_CONTRACT_ADDRESS_HEX
from multiversx_sdk.core.interfaces import IGasLimitEstimator
from multiversx_sdk.core.transaction import Transaction
from multiversx_sdk.core.transactions_factory_config import TransactionsFactoryConfig
from multiversx_sdk.validators.validators_signers import ValidatorsSigners
from multiversx_sdk.wallet.validator_keys import ValidatorPublicKey
[docs]
class ValidatorsTransactionsFactory(BaseFactory):
def __init__(
self,
config: TransactionsFactoryConfig,
gas_limit_estimator: Optional[IGasLimitEstimator] = None,
) -> None:
super().__init__(config, gas_limit_estimator)
self.config = config
self.serializer = Serializer()
[docs]
def create_transaction_for_staking(
self,
sender: Address,
validators_file: Union[Path, ValidatorsSigners],
amount: int,
rewards_address: Optional[Address] = None,
) -> Transaction:
if isinstance(validators_file, Path):
validators_file = ValidatorsSigners.new_from_pem(validators_file)
data_parts = self._prepare_data_parts_for_staking(
node_operator=sender,
validators_file=validators_file,
rewards_address=rewards_address,
)
transaction = Transaction(
sender=sender,
receiver=Address.new_from_hex(STAKING_SMART_CONTRACT_ADDRESS_HEX),
gas_limit=0,
chain_id=self.config.chain_id,
value=amount,
)
self.set_payload(transaction, data_parts)
self.set_gas_limit(
transaction=transaction,
config_gas_limit=self.config.gas_limit_for_staking * validators_file.get_num_of_nodes(),
)
return transaction
def _prepare_data_parts_for_staking(
self,
node_operator: Address,
validators_file: ValidatorsSigners,
rewards_address: Optional[Address] = None,
) -> list[str]:
data_parts = ["stake"]
num_of_nodes = validators_file.get_num_of_nodes()
call_arguments: list[ISingleValue] = []
call_arguments.append(U32Value(num_of_nodes))
validator_signers = validators_file.get_signers()
for validator in validator_signers:
signed_message = validator.sign(node_operator.get_public_key())
call_arguments.append(BytesValue(validator.secret_key.generate_public_key().buffer))
call_arguments.append(BytesValue(signed_message))
if rewards_address:
call_arguments.append(AddressValue.new_from_address(rewards_address))
args = self.serializer.serialize_to_parts(call_arguments)
return data_parts + [arg.hex() for arg in args]
[docs]
def create_transaction_for_topping_up(
self,
sender: Address,
amount: int,
) -> Transaction:
data = ["stake"]
transaction = Transaction(
sender=sender,
receiver=Address.new_from_hex(STAKING_SMART_CONTRACT_ADDRESS_HEX),
gas_limit=0,
chain_id=self.config.chain_id,
value=amount,
)
self.set_payload(transaction, data)
self.set_gas_limit(transaction=transaction, config_gas_limit=self.config.gas_limit_for_topping_up)
return transaction
[docs]
def create_transaction_for_unstaking(
self,
sender: Address,
public_keys: list[ValidatorPublicKey],
) -> Transaction:
data_parts = ["unStake"] + [key.hex() for key in public_keys]
transaction = Transaction(
sender=sender,
receiver=Address.new_from_hex(STAKING_SMART_CONTRACT_ADDRESS_HEX),
gas_limit=0,
chain_id=self.config.chain_id,
)
self.set_payload(transaction, data_parts)
self.set_gas_limit(
transaction=transaction,
config_gas_limit=self.config.gas_limit_for_unstaking * len(public_keys),
)
return transaction
[docs]
def create_transaction_for_unjailing(
self,
sender: Address,
public_keys: list[ValidatorPublicKey],
amount: int,
) -> Transaction:
data_parts = ["unJail"] + [key.hex() for key in public_keys]
transaction = Transaction(
sender=sender,
receiver=Address.new_from_hex(STAKING_SMART_CONTRACT_ADDRESS_HEX),
gas_limit=0,
chain_id=self.config.chain_id,
value=amount,
)
self.set_payload(transaction, data_parts)
self.set_gas_limit(
transaction=transaction,
config_gas_limit=self.config.gas_limit_for_unjailing * len(public_keys),
)
return transaction
[docs]
def create_transaction_for_unbonding(
self,
sender: Address,
public_keys: list[ValidatorPublicKey],
) -> Transaction:
data_parts = ["unBond"] + [key.hex() for key in public_keys]
transaction = Transaction(
sender=sender,
receiver=Address.new_from_hex(STAKING_SMART_CONTRACT_ADDRESS_HEX),
gas_limit=0,
chain_id=self.config.chain_id,
)
self.set_payload(transaction, data_parts)
self.set_gas_limit(
transaction=transaction,
config_gas_limit=self.config.gas_limit_for_unbonding * len(public_keys),
)
return transaction
[docs]
def create_transaction_for_changing_rewards_address(
self,
sender: Address,
rewards_address: Address,
) -> Transaction:
data_parts = ["changeRewardAddress", rewards_address.to_hex()]
transaction = Transaction(
sender=sender,
receiver=Address.new_from_hex(STAKING_SMART_CONTRACT_ADDRESS_HEX),
gas_limit=0,
chain_id=self.config.chain_id,
)
self.set_payload(transaction, data_parts)
self.set_gas_limit(transaction=transaction, config_gas_limit=self.config.gas_limit_for_changing_rewards_address)
return transaction
[docs]
def create_transaction_for_claiming(self, sender: Address) -> Transaction:
data_parts = ["claim"]
transaction = Transaction(
sender=sender,
receiver=Address.new_from_hex(STAKING_SMART_CONTRACT_ADDRESS_HEX),
gas_limit=0,
chain_id=self.config.chain_id,
)
self.set_payload(transaction, data_parts)
self.set_gas_limit(transaction=transaction, config_gas_limit=self.config.gas_limit_for_claiming)
return transaction
[docs]
def create_transaction_for_unstaking_nodes(
self,
sender: Address,
public_keys: list[ValidatorPublicKey],
) -> Transaction:
data_parts = ["unStakeNodes"] + [key.hex() for key in public_keys]
transaction = Transaction(
sender=sender,
receiver=Address.new_from_hex(STAKING_SMART_CONTRACT_ADDRESS_HEX),
gas_limit=0,
chain_id=self.config.chain_id,
)
self.set_payload(transaction, data_parts)
self.set_gas_limit(
transaction=transaction,
config_gas_limit=self.config.gas_limit_for_unstaking_nodes * len(public_keys),
)
return transaction
[docs]
def create_transaction_for_unstaking_tokens(self, sender: Address, amount: int) -> Transaction:
data_parts = ["unStakeTokens", self.serializer.serialize([BigUIntValue(amount)])]
transaction = Transaction(
sender=sender,
receiver=Address.new_from_hex(STAKING_SMART_CONTRACT_ADDRESS_HEX),
gas_limit=0,
chain_id=self.config.chain_id,
)
self.set_payload(transaction, data_parts)
self.set_gas_limit(transaction=transaction, config_gas_limit=self.config.gas_limit_for_unstaking_tokens)
return transaction
[docs]
def create_transaction_for_unbonding_nodes(
self,
sender: Address,
public_keys: list[ValidatorPublicKey],
) -> Transaction:
data_parts = ["unBondNodes"] + [key.hex() for key in public_keys]
transaction = Transaction(
sender=sender,
receiver=Address.new_from_hex(STAKING_SMART_CONTRACT_ADDRESS_HEX),
gas_limit=0,
chain_id=self.config.chain_id,
)
self.set_payload(transaction, data_parts)
self.set_gas_limit(
transaction=transaction,
config_gas_limit=self.config.gas_limit_for_unbonding_nodes * len(public_keys),
)
return transaction
[docs]
def create_transaction_for_unbonding_tokens(self, sender: Address, amount: int) -> Transaction:
data_parts = ["unBondTokens", self.serializer.serialize([BigUIntValue(amount)])]
transaction = Transaction(
sender=sender,
receiver=Address.new_from_hex(STAKING_SMART_CONTRACT_ADDRESS_HEX),
gas_limit=0,
chain_id=self.config.chain_id,
)
self.set_payload(transaction, data_parts)
self.set_gas_limit(transaction=transaction, config_gas_limit=self.config.gas_limit_for_unbonding_tokens)
return transaction
[docs]
def create_transaction_for_cleaning_registered_data(self, sender: Address) -> Transaction:
data_parts = ["cleanRegisteredData"]
transaction = Transaction(
sender=sender,
receiver=Address.new_from_hex(STAKING_SMART_CONTRACT_ADDRESS_HEX),
gas_limit=0,
chain_id=self.config.chain_id,
)
self.set_payload(transaction, data_parts)
self.set_gas_limit(transaction=transaction, config_gas_limit=self.config.gas_limit_for_cleaning_registered_data)
return transaction
[docs]
def create_transaction_for_restaking_unstaked_nodes(
self,
sender: Address,
public_keys: list[ValidatorPublicKey],
) -> Transaction:
data_parts = ["reStakeUnStakedNodes"] + [key.hex() for key in public_keys]
transaction = Transaction(
sender=sender,
receiver=Address.new_from_hex(STAKING_SMART_CONTRACT_ADDRESS_HEX),
gas_limit=0,
chain_id=self.config.chain_id,
)
self.set_payload(transaction, data_parts)
self.set_gas_limit(
transaction=transaction,
config_gas_limit=self.config.gas_limit_for_restaking_unstaked_tokens * len(public_keys),
)
return transaction