class_docstr: Base class for P2P RPC servicers (e.g. DHT, averager, MoE server). The interface mimics gRPC servicers.
  
  - ``add_p2p_handlers(self, p2p)`` registers all rpc_* methods of the derived class as P2P handlers, allowing
    other peers to call them. It uses type annotations for the ``request`` parameter and the return value
    to infer protobufs the methods operate with.
  
  - ``get_stub(self, p2p, peer)`` creates a stub with all rpc_* methods. Calls to the stub methods are translated
    to calls to the remote peer.
class_docstr_tok: ["Base","class","for","P2P","RPC","servicers","(","e.g",".","DHT",",","averager",",","MoE","server",")",".","The","interface","mimics","gRPC","servicers",".","-","``","add_p2p_handlers","(","self",",","p2p",")","``","registers","all","rpc_","*","methods","of","the","derived","class","as","P2P","handlers",",","allowing","other","peers","to","call","them",".","It","uses","type","annotations","for","the","``","request","``","parameter","and","the","return","value","to","infer","protobufs","the","methods","operate","with",".","-","``","get_stub","(","self",",","p2p",",","peer",")","``","creates","a","stub","with","all","rpc_","*","methods",".","Calls","to","the","stub","methods","are","translated","to","calls","to","the","remote","peer","."]
code: def _collect_rpc_handlers(cls) -> None:
      if cls._rpc_handlers is not None:
          return
  
      cls._rpc_handlers = []
      for method_name, method in inspect.getmembers(
          cls, predicate=inspect.isfunction
      ):
          if method_name.startswith("rpc_"):
              spec = inspect.getfullargspec(method)
              if len(spec.args) < 3:
                  raise ValueError(
                      f"{method_name} is expected to at least three positional arguments "
                      f"(self, request: TInputProtobuf | AsyncIterator[TInputProtobuf], context: P2PContext)"
                  )
              request_arg = spec.args[1]
              hints = get_type_hints(method)
              try:
                  request_type = hints[request_arg]
                  response_type = hints["return"]
              except KeyError:
                  raise ValueError(
                      f"{method_name} is expected to have type annotations "
                      f"like `dht_pb2.FindRequest` or `AsyncIterator[dht_pb2.FindRequest]` "
                      f"for the `{request_arg}` parameter and the return value"
                  )
              request_type, stream_input = cls._strip_iterator_hint(
                  request_type
              )
              response_type, stream_output = cls._strip_iterator_hint(
                  response_type
              )
  
              cls._rpc_handlers.append(
                  RPCHandler(
                      method_name,
                      request_type,
                      response_type,
                      stream_input,
                      stream_output,
                  )
              )
  
      cls._stub_type = type(
          f"{cls.__name__}Stub",
          (StubBase,),
          {
              handler.method_name: cls._make_rpc_caller(handler)
              for handler in cls._rpc_handlers
          },
      )
  
file_path: hivemind/hivemind/p2p/servicer.py
imports: import asyncio
  import inspect
  from dataclasses import dataclass
  from typing import Any, AsyncIterator, List, Optional, Tuple, Type, get_type_hints
  from hivemind.p2p.p2p_daemon import P2P
  from hivemind.p2p.p2p_daemon_bindings.datastructures import PeerID
label_desc: Use image node_id 1 for calling the ServicerBase obj's underlying member method code with example usage: obj._collect_rpc_handlers(cls) without return types