content: import itertools
  import logging
  from typing import (
      Any,
      Callable,
      NoReturn,
  )
  
  from eth_typing import (
      ABI,
      ABIFunction,
      HexStr,
      TypeStr,
  )
  from eth_utils import (
      abi_to_signature,
      filter_abi_by_type,
      function_abi_to_4byte_selector,
      is_0x_prefixed,
      is_binary_address,
      is_boolean,
      is_bytes,
      is_checksum_address,
      is_dict,
      is_hex_address,
      is_integer,
      is_list_like,
      is_string,
  )
  from eth_utils.curried import (
      apply_formatter_to_array,
  )
  from eth_utils.hexadecimal import (
      encode_hex,
  )
  from eth_utils.toolz import (
      compose,
      groupby,
      valfilter,
      valmap,
  )
  
  from ens.utils import (
      is_valid_ens_name,
  )
  from web3._utils.abi import (
      is_address_type,
      is_array_type,
      is_bool_type,
      is_bytes_type,
      is_int_type,
      is_recognized_type,
      is_string_type,
      is_uint_type,
      length_of_array_type,
      sub_type_of_array_type,
  )
  from web3._utils.formatters import (
      apply_error_formatters,
  )
  from web3.exceptions import (
      BadResponseFormat,
      InvalidAddress,
      MethodUnavailable,
      RequestTimedOut,
      TransactionNotFound,
      Web3RPCError,
      Web3TypeError,
      Web3ValueError,
  )
  from web3.types import (
      RPCResponse,
  )
  
  
  def _prepare_selector_collision_msg(duplicates: dict[HexStr, ABIFunction]) -> str:
      dup_sel = valmap(apply_formatter_to_array(abi_to_signature), duplicates)
      joined_funcs = valmap(lambda funcs: ", ".join(funcs), dup_sel)
      func_sel_msg_list = [
          funcs + " have selector " + sel for sel, funcs in joined_funcs.items()
      ]
      return " and\n".join(func_sel_msg_list)
  
  
  def validate_abi(abi: ABI) -> None:
      """
      Helper function for validating an ABI
      """
      if not is_list_like(abi):
          raise Web3ValueError("'abi' is not a list")
  
      if not all(is_dict(e) for e in abi):
          raise Web3ValueError("'abi' is not a list of dictionaries")
  
      if not all("type" in e for e in abi):
          raise Web3ValueError("'abi' must contain a list of elements each with a type")
  
      functions = filter_abi_by_type("function", abi)
      selectors = groupby(compose(encode_hex, function_abi_to_4byte_selector), functions)
      duplicates = valfilter(lambda funcs: len(funcs) > 1, selectors)
      if duplicates:
          raise Web3ValueError(
              "Abi contains functions with colliding selectors. "
              f"Functions {_prepare_selector_collision_msg(duplicates)}"
          )
  
  
  def validate_abi_type(abi_type: TypeStr) -> None:
      """
      Helper function for validating an abi_type
      """
      if not is_recognized_type(abi_type):
          raise Web3ValueError(f"Unrecognized abi_type: {abi_type}")
  
  
  def validate_abi_value(abi_type: TypeStr, value: Any) -> None:
      """
      Helper function for validating a value against the expected abi_type
      Note: abi_type 'bytes' must either be python3 'bytes' object or ''
      """
      if is_array_type(abi_type) and is_list_like(value):
          # validate length
          specified_length = length_of_array_type(abi_type)
          if specified_length is not None:
              if specified_length < 1:
                  raise Web3TypeError(
                      f"Invalid abi-type: {abi_type}. Length of fixed sized "
                      "arrays must be greater than 0."
                  )
              if specified_length != len(value):
                  raise Web3TypeError(
                      "The following array length does not match the length specified "
                      f"by the abi-type, {abi_type}: {value}"
                  )
  
          # validate sub_types
          sub_type = sub_type_of_array_type(abi_type)
          for v in value:
              validate_abi_value(sub_type, v)
          return
      elif is_bool_type(abi_type) and is_boolean(value):
          return
      elif is_uint_type(abi_type) and is_integer(value) and value >= 0:
          return
      elif is_int_type(abi_type) and is_integer(value):
          return
      elif is_address_type(abi_type):
          validate_address(value)
          return
      elif is_bytes_type(abi_type):
          if is_bytes(value):
              return
          elif is_string(value):
              if is_0x_prefixed(value):
                  return
              else:
                  raise Web3TypeError(
                      "ABI values of abi-type 'bytes' must be either"
                      "a python3 'bytes' object or an '0x' prefixed string."
                  )
      elif is_string_type(abi_type) and is_string(value):
          return
  
      raise Web3TypeError(f"The following abi value is not a '{abi_type}': {value}")
  
  
  def is_not_address_string(value: Any) -> bool:
      return (
          is_string(value)
          and not is_bytes(value)
          and not is_checksum_address(value)
          and not is_hex_address(value)
      )
  
  
  def validate_address(value: Any) -> None:
      """
      Helper function for validating an address
      """
      if is_not_address_string(value):
          if not is_valid_ens_name(value):
              raise InvalidAddress(f"ENS name: '{value}' is invalid.")
          return
      if is_bytes(value):
          if not is_binary_address(value):
              raise InvalidAddress(
                  "Address must be 20 bytes when input type is bytes", value
              )
          return
  
      if not isinstance(value, str):
          raise Web3TypeError(f"Address {value} must be provided as a string")
      if not is_hex_address(value):
          raise InvalidAddress(
              "Address must be 20 bytes, as a hex string with a 0x prefix", value
          )
      if not is_checksum_address(value):
          if value == value.lower():
              raise InvalidAddress(
                  "web3.py only accepts checksum addresses. "
                  "The software that gave you this non-checksum address should be "
                  "considered unsafe, please file it as a bug on their platform. "
                  "Try using an ENS name instead. Or, if you must accept lower safety, "
                  "use Web3.to_checksum_address(lower_case_address).",
                  value,
              )
          else:
              raise InvalidAddress(
                  "Address has an invalid EIP-55 checksum. "
                  "After looking up the address from the original source, try again.",
                  value,
              )
  
  
  def has_one_val(*args: Any, **kwargs: Any) -> bool:
      vals = itertools.chain(args, kwargs.values())
      not_nones = list(filter(lambda val: val is not None, vals))
      return len(not_nones) == 1
  
  
  def assert_one_val(*args: Any, **kwargs: Any) -> None:
      if not has_one_val(*args, **kwargs):
          raise Web3TypeError(
              "Exactly one of the passed values can be specified. "
              f"Instead, values were: {args!r}, {kwargs!r}"
          )
  
  
  # -- RPC Response Validation -- #
  
  KNOWN_REQUEST_TIMEOUT_MESSAGING = {
      # Note: It's important to be very explicit here and not too broad. We don't want
      # to accidentally catch a message that is not for a request timeout. In the worst
      # case, we raise something more generic like `Web3RPCError`. JSON-RPC unfortunately
      # has not standardized error codes for request timeouts.
      "request timed out",  # go-ethereum
  }
  METHOD_NOT_FOUND = -32601
  
  
  def _validate_subscription_fields(response: RPCResponse) -> None:
      params = response["params"]
      subscription = params["subscription"]
      if not isinstance(subscription, str) and not len(subscription) == 34:
          _raise_bad_response_format(
              response, "eth_subscription 'params' must include a 'subscription' field."
          )
  
  
  def _raise_bad_response_format(response: RPCResponse, error: str = "") -> None:
      message = "The response was in an unexpected format and unable to be parsed."
      raw_response = f"The raw response is: {response}"
  
      if error is not None and error != "":
          error = error[:-1] if error.endswith(".") else error
          message = f"{message} {error}. {raw_response}"
      else:
          message = f"{message} {raw_response}"
  
      raise BadResponseFormat(message)
  
  
  def raise_error_for_batch_response(
      response: RPCResponse,
      logger: logging.Logger | None = None,
  ) -> NoReturn:
      error = response.get("error")
      if error is None:
          _raise_bad_response_format(
              response,
              "Batch response must be formatted as a list of responses or "
              "as a single JSON-RPC error response.",
          )
      validate_rpc_response_and_raise_if_error(
          response,
          None,
          is_subscription_response=False,
          logger=logger,
          params=[],
      )
      # This should not be reached, but if it is, raise a generic `BadResponseFormat`
      raise BadResponseFormat(
          "Batch response was in an unexpected format and unable to be parsed."
      )
  
  
  def validate_rpc_response_and_raise_if_error(
      response: RPCResponse,
      error_formatters: Callable[..., Any] | None,
      is_subscription_response: bool = False,
      logger: logging.Logger | None = None,
      params: Any | None = None,
  ) -> None:
      if "jsonrpc" not in response or response["jsonrpc"] != "2.0":
          _raise_bad_response_format(
              response, 'The "jsonrpc" field must be present with a value of "2.0".'
          )
  
      response_id = response.get("id")
      if "id" in response:
          int_error_msg = (
              '"id" must be an integer or a string representation of an integer.'
          )
          if response_id is None and "error" in response:
              # errors can sometimes have null `id`, according to the JSON-RPC spec
              pass
          elif not isinstance(response_id, (str, int)):
              _raise_bad_response_format(response, int_error_msg)
          elif isinstance(response_id, str):
              try:
                  int(response_id)
              except ValueError:
                  _raise_bad_response_format(response, int_error_msg)
      elif is_subscription_response:
          # if `id` is not present, this must be a subscription response
          _validate_subscription_fields(response)
      else:
          _raise_bad_response_format(
              response,
              'Response must include an "id" field or be formatted as an '
              "`eth_subscription` response.",
          )
  
      if all(key in response for key in {"error", "result"}):
          _raise_bad_response_format(
              response, 'Response cannot include both "error" and "result".'
          )
      elif (
          not any(key in response for key in {"error", "result"})
          and not is_subscription_response
      ):
          _raise_bad_response_format(
              response, 'Response must include either "error" or "result".'
          )
      elif "error" in response:
          web3_rpc_error: Web3RPCError | None = None
          error = response["error"]
  
          # raise the error when the value is a string
          if error is None or not isinstance(error, dict):
              _raise_bad_response_format(
                  response,
                  'response["error"] must be a valid object as defined by the '
                  "JSON-RPC 2.0 specification.",
              )
  
          # errors must include a message
          error_message = error.get("message")
          if not isinstance(error_message, str):
              _raise_bad_response_format(
                  response, 'error["message"] is required and must be a string value.'
              )
          elif error_message == "transaction not found":
              transaction_hash = params[0]
              web3_rpc_error = TransactionNotFound(
                  repr(error),
                  rpc_response=response,
                  user_message=(f"Transaction with hash {transaction_hash!r} not found."),
              )
  
          # errors must include an integer code
          code = error.get("code")
          if not isinstance(code, int):
              _raise_bad_response_format(
                  response, 'error["code"] is required and must be an integer value.'
              )
          elif code == METHOD_NOT_FOUND:
              web3_rpc_error = MethodUnavailable(
                  repr(error),
                  rpc_response=response,
                  user_message=(
                      "This method is not available. Check your node provider or your "
                      "client's API docs to see what methods are supported and / or "
                      "currently enabled."
                  ),
              )
          elif any(
              # parse specific timeout messages
              timeout_str in error_message.lower()
              for timeout_str in KNOWN_REQUEST_TIMEOUT_MESSAGING
          ):
              web3_rpc_error = RequestTimedOut(
                  repr(error),
                  rpc_response=response,
                  user_message=(
                      "The request timed out. Check the connection to your node and "
                      "try again."
                  ),
              )
  
          if web3_rpc_error is None:
              # if no condition was met above, raise a more generic `Web3RPCError`
              web3_rpc_error = Web3RPCError(repr(error), rpc_response=response)
  
          response = apply_error_formatters(error_formatters, response)
          if logger is not None:
              logger.debug("RPC error response: %s", response)
  
          raise web3_rpc_error
  
      elif "result" not in response and not is_subscription_response:
          _raise_bad_response_format(response)
  