loctrkd/protomodule.py

   1 """ Things the module implementing a protocol exports """
   2
   3 from typing import (
   4     Any,
   5     Callable,
   6     Dict,
   7     List,
   8     Optional,
   9     Protocol,
  10     _ProtocolMeta,  # How not to cheat here?!
  11     Tuple,
  12     Type,
  13     TYPE_CHECKING,
  14     Union,
  15 )
  16
  17
  18 class MetaPkt(type):
  19     """
  20     For each class corresponding to a message, automatically create
  21     two nested classes `In` and `Out` that also inherit from their
  22     "nest". Class attribute `IN_KWARGS` defined in the "nest" is
  23     copied to the `In` nested class under the name `KWARGS`, and
  24     likewise, `OUT_KWARGS` of the nest class is copied as `KWARGS`
  25     to the nested class `Out`. In addition, methods `encode` and
  26     `decode` are defined in both classes equal to `in_{en|de}code()`
  27     and `out_{en|de}code()` respectively.
  28     """
  29
  30     if TYPE_CHECKING:
  31
  32         def __getattr__(self, name: str) -> Any:
  33             pass
  34
  35         def __setattr__(self, name: str, value: Any) -> None:
  36             pass
  37
  38     def in_decode(self, *args: Any) -> None:
  39         ...
  40
  41     def out_decode(self, *args: Any) -> None:
  42         ...
  43
  44     def in_encode(self, *args: Any) -> Any:
  45         ...
  46
  47     def out_encode(self, *args: Any) -> Any:
  48         ...
  49
  50     def __new__(
  51         cls: Type["MetaPkt"],
  52         name: str,
  53         bases: Tuple[type, ...],
  54         attrs: Dict[str, Any],
  55     ) -> "MetaPkt":
  56         newcls = super().__new__(cls, name, bases, attrs)
  57         newcls.In = super().__new__(
  58             cls,
  59             name + ".In",
  60             (newcls,) + bases,
  61             {
  62                 "KWARGS": newcls.IN_KWARGS,
  63                 "decode": newcls.in_decode,
  64                 "encode": newcls.in_encode,
  65             },
  66         )
  67         newcls.Out = super().__new__(
  68             cls,
  69             name + ".Out",
  70             (newcls,) + bases,
  71             {
  72                 "KWARGS": newcls.OUT_KWARGS,
  73                 "decode": newcls.out_decode,
  74                 "encode": newcls.out_encode,
  75             },
  76         )
  77         return newcls
  78
  79
  80 # Have to do this to prevent incomprehensible error message:
  81 # TypeError: metaclass conflict: the metaclass of a derived class \
  82 #     must be a (non-strict) subclass of the metaclasses of all its bases
  83 class _MetaProto(_ProtocolMeta, MetaPkt):
  84     pass
  85
  86
  87 class ProtoClass(Protocol, metaclass=_MetaProto):
  88     IN_KWARGS: Tuple[Tuple[str, Callable[[Any], Any], Any], ...] = ()
  89     OUT_KWARGS: Tuple[Tuple[str, Callable[[Any], Any], Any], ...] = ()
  90
  91     @classmethod
  92     def proto_name(cls) -> str:
  93         ...
  94
  95     class In:
  96         def __init__(self, *args: Any, **kwargs: Any) -> None:
  97             ...
  98
  99         def encode(self) -> bytes:
 100             ...
 101
 102         def decode(self, *args: Any, **kwargs: Any) -> None:
 103             ...
 104
 105         @property
 106         def packed(self) -> bytes:
 107             ...
 108
 109     class Out:
 110         def __init__(self, *args: Any, **kwargs: Any) -> None:
 111             ...
 112
 113         def encode(self) -> bytes:
 114             ...
 115
 116         def decode(self, *args: Any, **kwargs: Any) -> None:
 117             ...
 118
 119         @property
 120         def packed(self) -> bytes:
 121             ...
 122
 123
 124 class ProtoModule:
 125     __name__: str
 126
 127     class Stream:
 128         def recv(self, segment: bytes) -> List[Union[bytes, str]]:
 129             ...
 130
 131         def close(self) -> bytes:
 132             ...
 133
 134     @staticmethod
 135     def enframe(buffer: bytes, imei: Optional[str] = None) -> bytes:
 136         ...
 137
 138     class DecodeError(Exception):
 139         ...
 140
 141     @staticmethod
 142     def exposed_protos() -> List[Tuple[str, bool]]:
 143         ...
 144
 145     @staticmethod
 146     def probe_buffer(buffer: bytes) -> bool:
 147         ...
 148
 149     @staticmethod
 150     def parse_message(packet: bytes, is_incoming: bool = True) -> Any:
 151         ...
 152
 153     @staticmethod
 154     def inline_response(packet: bytes) -> Optional[bytes]:
 155         ...
 156
 157     @staticmethod
 158     def is_goodbye_packet(packet: bytes) -> bool:
 159         ...
 160
 161     @staticmethod
 162     def imei_from_packet(packet: bytes) -> Optional[str]:
 163         ...
 164
 165     @staticmethod
 166     def proto_of_message(packet: bytes) -> str:
 167         ...
 168
 169     @staticmethod
 170     def proto_handled(proto: str) -> bool:
 171         ...
 172
 173     @staticmethod
 174     def class_by_prefix(prefix: str) -> Union[Type[ProtoClass], List[str]]:
 175         ...
 176
 177     @staticmethod
 178     def make_response(
 179         cmd: str, imei: str, **kwargs: Any
 180     ) -> Optional[ProtoClass.Out]:
 181         ...