<commit_before>from typing import Optional

import structlog
from eth_utils import (
    encode_hex,
    event_abi_to_log_topic,
    is_binary_address,
    is_same_address,
    to_canonical_address,
    to_checksum_address,
    to_normalized_address,
)

from raiden.constants import (
    GAS_REQUIRED_FOR_CREATE_ERC20_TOKEN_NETWORK,
    GENESIS_BLOCK_NUMBER,
    NULL_ADDRESS,
)
from raiden.exceptions import InvalidAddress, RaidenRecoverableError, RaidenUnrecoverableError
from raiden.network.proxies.utils import compare_contract_versions
from raiden.network.rpc.client import StatelessFilter, check_address_has_code
from raiden.network.rpc.transactions import check_transaction_threw
from raiden.utils import pex, safe_gas_limit
from raiden.utils.typing import (
    Address,
    Any,
    BlockSpecification,
    List,
    PaymentNetworkID,
    T_TargetAddress,
    TokenAddress,
    TokenAmount,
)
from raiden_contracts.constants import CONTRACT_TOKEN_NETWORK_REGISTRY, EVENT_TOKEN_NETWORK_CREATED
from raiden_contracts.contract_manager import ContractManager

log = structlog.get_logger(__name__)  # pylint: disable=invalid-name


class TokenNetworkRegistry:
    def __init__(
            self,
            jsonrpc_client,
            registry_address: PaymentNetworkID,
            contract_manager: ContractManager,
    ):
        if not is_binary_address(registry_address):
            raise InvalidAddress('Expected binary address format for token network registry')

        check_address_has_code(
            client=jsonrpc_client,
            address=Address(registry_address),
            contract_name=CONTRACT_TOKEN_NETWORK_REGISTRY,
        )

        self.contract_manager = contract_manager
        proxy = jsonrpc_client.new_contract_proxy(
            self.contract_manager.get_contract_abi(CONTRACT_TOKEN_NETWORK_REGISTRY),
            to_normalized_address(registry_address),
        )

        compare_contract_versions(
            proxy=proxy,
            expected_version=contract_manager.contracts_version,
            contract_name=CONTRACT_TOKEN_NETWORK_REGISTRY,
            address=Address(registry_address),
        )

        self.address = registry_address
        self.proxy = proxy
        self.client = jsonrpc_client
        self.node_address = self.client.address

    def get_token_network(
            self,
            token_address: TokenAddress,
            block_identifier: BlockSpecification = 'latest',
    ) -> Optional[Address]:
        """ Return the token network address for the given token or None if
        there is no correspoding address.
        """
        if not isinstance(token_address, T_TargetAddress):
            raise ValueError('token_address must be an address')

        address = self.proxy.contract.functions.token_to_token_networks(
            to_checksum_address(token_address),
        ).call()
        address = to_canonical_address(address)

        if is_same_address(address, NULL_ADDRESS):
            return None

        return address

    def add_token_with_limits(
            self,
            token_address: TokenAddress,
            channel_participant_deposit_limit: TokenAmount,
            token_network_deposit_limit: TokenAmount,
            given_block_identifier: BlockSpecification,
    ) -> Address:
        """
        Register token of `token_address` with the token network.
        The limits apply for version 0.13.0 and above of raiden-contracts,
        since instantation also takes the limits as constructor arguments.<commit_msg>Fix typo<commit_after>from typing import Optional

import structlog
from eth_utils import (
    encode_hex,
    event_abi_to_log_topic,
    is_binary_address,
    is_same_address,
    to_canonical_address,
    to_checksum_address,
    to_normalized_address,
)

from raiden.constants import (
    GAS_REQUIRED_FOR_CREATE_ERC20_TOKEN_NETWORK,
    GENESIS_BLOCK_NUMBER,
    NULL_ADDRESS,
)
from raiden.exceptions import InvalidAddress, RaidenRecoverableError, RaidenUnrecoverableError
from raiden.network.proxies.utils import compare_contract_versions
from raiden.network.rpc.client import StatelessFilter, check_address_has_code
from raiden.network.rpc.transactions import check_transaction_threw
from raiden.utils import pex, safe_gas_limit
from raiden.utils.typing import (
    Address,
    Any,
    BlockSpecification,
    List,
    PaymentNetworkID,
    T_TargetAddress,
    TokenAddress,
    TokenAmount,
)
from raiden_contracts.constants import CONTRACT_TOKEN_NETWORK_REGISTRY, EVENT_TOKEN_NETWORK_CREATED
from raiden_contracts.contract_manager import ContractManager

log = structlog.get_logger(__name__)  # pylint: disable=invalid-name


class TokenNetworkRegistry:
    def __init__(
            self,
            jsonrpc_client,
            registry_address: PaymentNetworkID,
            contract_manager: ContractManager,
    ):
        if not is_binary_address(registry_address):
            raise InvalidAddress('Expected binary address format for token network registry')

        check_address_has_code(
            client=jsonrpc_client,
            address=Address(registry_address),
            contract_name=CONTRACT_TOKEN_NETWORK_REGISTRY,
        )

        self.contract_manager = contract_manager
        proxy = jsonrpc_client.new_contract_proxy(
            self.contract_manager.get_contract_abi(CONTRACT_TOKEN_NETWORK_REGISTRY),
            to_normalized_address(registry_address),
        )

        compare_contract_versions(
            proxy=proxy,
            expected_version=contract_manager.contracts_version,
            contract_name=CONTRACT_TOKEN_NETWORK_REGISTRY,
            address=Address(registry_address),
        )

        self.address = registry_address
        self.proxy = proxy
        self.client = jsonrpc_client
        self.node_address = self.client.address

    def get_token_network(
            self,
            token_address: TokenAddress,
            block_identifier: BlockSpecification = 'latest',
    ) -> Optional[Address]:
        """ Return the token network address for the given token or None if
        there is no correspoding address.
        """
        if not isinstance(token_address, T_TargetAddress):
            raise ValueError('token_address must be an address')

        address = self.proxy.contract.functions.token_to_token_networks(
            to_checksum_address(token_address),
        ).call()
        address = to_canonical_address(address)

        if is_same_address(address, NULL_ADDRESS):
            return None

        return address

    def add_token_with_limits(
            self,
            token_address: TokenAddress,
            channel_participant_deposit_limit: TokenAmount,
            token_network_deposit_limit: TokenAmount,
            given_block_identifier: BlockSpecification,
    ) -> Address:
        """
        Register token of `token_address` with the token network.
        The limits apply for version 0.13.0 and above of raiden-contracts,
        since instantiation also takes the limits as constructor arguments.
{'fraction_non_alphanumeric': 0.055115884680610516, 'fraction_numerical': 0.0016958733747880158, 'mean_word_length': 3.0028280542986425, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7242287', 'n_tokens_mistral': 1940, 'n_tokens_neox': 1901, 'n_words': 406}