gps303/zmsg.py

   1 """ Zeromq messages """
   2
   3 import ipaddress as ip
   4 from struct import pack, unpack
   5 from typing import Any, cast, Optional, Tuple, Type, Union
   6
   7 __all__ = "Bcast", "Resp", "topic"
   8
   9
  10 def pack_peer(
  11     peeraddr: Union[None, Tuple[str, int], Tuple[str, int, Any, Any]]
  12 ) -> bytes:
  13     if peeraddr is None:
  14         addr: Union[ip.IPv4Address, ip.IPv6Address] = ip.IPv6Address(0)
  15         port = 0
  16     elif len(peeraddr) == 2:
  17         peeraddr = cast(Tuple[str, int], peeraddr)
  18         saddr, port = peeraddr
  19         addr = ip.ip_address(saddr)
  20     elif len(peeraddr) == 4:
  21         peeraddr = cast(Tuple[str, int, Any, Any], peeraddr)
  22         saddr, port, _x, _y = peeraddr
  23         addr = ip.ip_address(saddr)
  24     if isinstance(addr, ip.IPv4Address):
  25         addr = ip.IPv6Address(b"\0\0\0\0\0\0\0\0\0\0\xff\xff" + addr.packed)
  26     return addr.packed + pack("!H", port)
  27
  28
  29 def unpack_peer(
  30     buffer: bytes,
  31 ) -> Tuple[str, int]:
  32     a6 = ip.IPv6Address(buffer[:16])
  33     port = unpack("!H", buffer[16:])[0]
  34     a4 = a6.ipv4_mapped
  35     if a4 is not None:
  36         return (str(a4), port)
  37     elif a6 == ip.IPv6Address("::"):
  38         return ("", 0)
  39     return (str(a6), port)
  40
  41
  42 class _Zmsg:
  43     KWARGS: Tuple[Tuple[str, Any], ...]
  44
  45     def __init__(self, *args: Any, **kwargs: Any) -> None:
  46         if len(args) == 1:
  47             self.decode(args[0])
  48         elif bool(kwargs):
  49             for k, v in self.KWARGS:
  50                 setattr(self, k, kwargs.get(k, v))
  51         else:
  52             raise RuntimeError(
  53                 self.__class__.__name__
  54                 + ": both args "
  55                 + str(args)
  56                 + " and kwargs "
  57                 + str(kwargs)
  58             )
  59
  60     def __repr__(self) -> str:
  61         return "{}({})".format(
  62             self.__class__.__name__,
  63             ", ".join(
  64                 [
  65                     "{}={}".format(
  66                         k,
  67                         'bytes.fromhex("{}")'.format(getattr(self, k).hex())
  68                         if isinstance(getattr(self, k), bytes)
  69                         else getattr(self, k),
  70                     )
  71                     for k, _ in self.KWARGS
  72                 ]
  73             ),
  74         )
  75
  76     def __eq__(self, other: object) -> bool:
  77         if isinstance(other, self.__class__):
  78             return all(
  79                 [getattr(self, k) == getattr(other, k) for k, _ in self.KWARGS]
  80             )
  81         return NotImplemented
  82
  83     def decode(self, buffer: bytes) -> None:
  84         raise NotImplementedError(
  85             self.__class__.__name__ + "must implement `decode()` method"
  86         )
  87
  88     @property
  89     def packed(self) -> bytes:
  90         raise NotImplementedError(
  91             self.__class__.__name__ + "must implement `packed()` property"
  92         )
  93
  94
  95 def topic(
  96     proto: int, is_incoming: bool = True, imei: Optional[str] = None
  97 ) -> bytes:
  98     return pack("BB", is_incoming, proto) + (
  99         b"" if imei is None else pack("16s", imei.encode())
 100     )
 101
 102
 103 class Bcast(_Zmsg):
 104     """Zmq message to broadcast what was received from the terminal"""
 105
 106     KWARGS = (
 107         ("is_incoming", True),
 108         ("proto", 256),
 109         ("imei", None),
 110         ("when", None),
 111         ("peeraddr", None),
 112         ("packet", b""),
 113     )
 114
 115     @property
 116     def packed(self) -> bytes:
 117         return (
 118             pack(
 119                 "BB16s",
 120                 int(self.is_incoming),
 121                 self.proto,
 122                 b"0000000000000000"
 123                 if self.imei is None
 124                 else self.imei.encode(),
 125             )
 126             + (
 127                 b"\0\0\0\0\0\0\0\0"
 128                 if self.when is None
 129                 else pack("!d", self.when)
 130             )
 131             + pack_peer(self.peeraddr)
 132             + self.packet
 133         )
 134
 135     def decode(self, buffer: bytes) -> None:
 136         self.is_incoming = bool(buffer[0])
 137         self.proto = buffer[1]
 138         self.imei: Optional[str] = buffer[2:18].decode()
 139         if self.imei == "0000000000000000":
 140             self.imei = None
 141         self.when = unpack("!d", buffer[18:26])[0]
 142         self.peeraddr = unpack_peer(buffer[26:44])
 143         self.packet = buffer[44:]
 144
 145
 146 class Resp(_Zmsg):
 147     """Zmq message received from a third party to send to the terminal"""
 148
 149     KWARGS = (("imei", None), ("when", None), ("packet", b""))
 150
 151     @property
 152     def packed(self) -> bytes:
 153         return (
 154             pack(
 155                 "16s",
 156                 "0000000000000000"
 157                 if self.imei is None
 158                 else self.imei.encode(),
 159             )
 160             + (
 161                 b"\0\0\0\0\0\0\0\0"
 162                 if self.when is None
 163                 else pack("!d", self.when)
 164             )
 165             + self.packet
 166         )
 167
 168     def decode(self, buffer: bytes) -> None:
 169         self.imei = buffer[:16].decode()
 170         self.when = unpack("!d", buffer[16:24])[0]
 171         self.packet = buffer[24:]