<commit_before>#
# Copyright (c) 2019 UAVCAN Development Team
# This software is distributed under the terms of the MIT License.
# Author: <NAME> <<EMAIL>>
#

from __future__ import annotations
import re
import errno
import typing
import socket
import logging
import pyuavcan
from ._network_map import NetworkMap


_logger = logging.getLogger(__name__)


class NetworkMapIPv4(NetworkMap):
    """
    In IPv4 networks, the node-ID of zero cannot be used because it represents the subnet address;
    the maximum node-ID can only be used if it is not the same as the broadcast address for the subnet.
    """

    def __init__(self, ip_address: str):
        self._local = IPv4Address.parse(ip_address)
        if self._local.netmask == 0 or self._local.hostmask == 0:
            raise ValueError(f'The subnet mask in {ip_address} is invalid or missing. '
                             f'It is needed to determine the subnet broadcast address.')

        # Say, if the node-ID bit length is 12, the maximum node-ID would be 4095,
        # but it won't be usable if the subnet mask is 20 bits wide because it would be the
        # broadcast address for the subnet. In this example, we can use the full range of node-ID
        # values if the subnet mask is 19 bits wide or less.
        self._max_nodes: int = min(2 ** self.NODE_ID_BIT_LENGTH, self._local.hostmask)

        self._local_node_id = int(self._local) - int(self._local.subnet_address)
        if self._local_node_id >= self._max_nodes:
            raise pyuavcan.transport.InvalidMediaConfigurationError(
                f'The local IP address/mask {self._local} resolves to a node-ID value {self._local_node_id} '
                f'which is invalid.'
            )

        assert (int(self._local.subnet_address) + self._local_node_id) in self._local
        assert self._local_node_id < self._max_nodes

        # Test the address configuration to detect configuration errors early.
        # I suppose we could also set up a pair of sockets and send a test datagram just for extra paranoia?
        for s in [
            self.make_output_socket(None, 65535),
            self.make_output_socket(1, 65535),
            self.make_input_socket(0),
        ]:
            # This invariant is supposed to be upheld by the OS, so we use an assertion check.
            assert IPv4Address.parse(s.getsockname()[0]) == self._local.host_address
            s.close()

        self._ip_to_nid_cache: typing.Dict[str, typing.Optional[int]] = {}

    @property
    def max_nodes(self) -> int:
        return self._max_nodes

    @property
    def local_node_id(self) -> int:
        return self._local_node_id

    def map_ip_address_to_node_id(self, ip: str) -> typing.Optional[int]:<commit_msg>Stray assertion fix<commit_after>#
# Copyright (c) 2019 UAVCAN Development Team
# This software is distributed under the terms of the MIT License.
# Author: <NAME> <<EMAIL>>
#

from __future__ import annotations
import re
import errno
import typing
import socket
import logging
import pyuavcan
from ._network_map import NetworkMap


_logger = logging.getLogger(__name__)


class NetworkMapIPv4(NetworkMap):
    """
    In IPv4 networks, the node-ID of zero cannot be used because it represents the subnet address;
    the maximum node-ID can only be used if it is not the same as the broadcast address for the subnet.
    """

    def __init__(self, ip_address: str):
        self._local = IPv4Address.parse(ip_address)
        if self._local.netmask == 0 or self._local.hostmask == 0:
            raise ValueError(f'The subnet mask in {ip_address} is invalid or missing. '
                             f'It is needed to determine the subnet broadcast address.')

        # Say, if the node-ID bit length is 12, the maximum node-ID would be 4095,
        # but it won't be usable if the subnet mask is 20 bits wide because it would be the
        # broadcast address for the subnet. In this example, we can use the full range of node-ID
        # values if the subnet mask is 19 bits wide or less.
        self._max_nodes: int = min(2 ** self.NODE_ID_BIT_LENGTH, self._local.hostmask)

        self._local_node_id = int(self._local) - int(self._local.subnet_address)
        if self._local_node_id >= self._max_nodes:
            raise pyuavcan.transport.InvalidMediaConfigurationError(
                f'The local IP address/mask {self._local} resolves to a node-ID value {self._local_node_id} '
                f'which is invalid.'
            )

        assert (int(self._local.subnet_address) + self._local_node_id) in self._local
        assert self._local_node_id < self._max_nodes

        # Test the address configuration to detect configuration errors early.
        # I suppose we could also set up a pair of sockets and send a test datagram just for extra paranoia?
        self.make_input_socket(0).close()
        for s in [
            self.make_output_socket(None, 65535),
            self.make_output_socket(1, 65535),
        ]:
            # This invariant is supposed to be upheld by the OS, so we use an assertion check.
            assert IPv4Address.parse(s.getsockname()[0]) == self._local.host_address
            s.close()

        self._ip_to_nid_cache: typing.Dict[str, typing.Optional[int]] = {}

    @property
    def max_nodes(self) -> int:
        return self._max_nodes

    @property
    def local_node_id(self) -> int:
        return self._local_node_id

    def map_ip_address_to_node_id(self, ip: str) -> typing.Optional[int]:
{'fraction_non_alphanumeric': 0.07651858567543064, 'fraction_numerical': 0.012330009066183137, 'mean_word_length': 2.6970509383378016, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1450150', 'n_tokens_mistral': 1651, 'n_tokens_neox': 1557, 'n_words': 569}