gps303/zmsg.py

   1 """ Zeromq messages """
   2
   3 import ipaddress as ip
   4 from struct import pack, unpack
   5
   6 __all__ = "Bcast", "Resp"
   7
   8
   9 def pack_peer(peeraddr):
  10     try:
  11         saddr, port, _x, _y = peeraddr
  12         addr = ip.ip_address(saddr)
  13     except ValueError:
  14         saddr, port = peeraddr
  15         a4 = ip.ip_address(saddr)
  16         addr = ip.IPv6Address(b"\0\0\0\0\0\0\0\0\0\0\xff\xff" + a4.packed)
  17     return addr.packed + pack("!H", port)
  18
  19
  20 def unpack_peer(buffer):
  21     a6 = ip.IPv6Address(buffer[:16])
  22     port = unpack("!H", buffer[16:])[0]
  23     addr = a6.ipv4_mapped
  24     if addr is None:
  25         addr = a6
  26     return (addr, port)
  27
  28
  29 class _Zmsg:
  30     def __init__(self, *args, **kwargs):
  31         if len(args) == 1:
  32             self.decode(args[0])
  33         elif bool(kwargs):
  34             for k, v in self.KWARGS:
  35                 setattr(self, k, kwargs.get(k, v))
  36         else:
  37             raise RuntimeError(
  38                 self.__class__.__name__
  39                 + ": both args "
  40                 + str(args)
  41                 + " and kwargs "
  42                 + str(kwargs)
  43             )
  44
  45     def __repr__(self):
  46         return "{}({})".format(
  47             self.__class__.__name__,
  48             ", ".join(
  49                 [
  50                     "{}={}".format(
  51                         k,
  52                         'bytes.fromhex("{}")'.format(getattr(self, k).hex())
  53                         if isinstance(getattr(self, k), bytes)
  54                         else getattr(self, k),
  55                     )
  56                     for k, _ in self.KWARGS
  57                 ]
  58             ),
  59         )
  60
  61     def __eq__(self, other):
  62         return all(
  63             [getattr(self, k) == getattr(other, k) for k, _ in self.KWARGS]
  64         )
  65
  66     def decode(self, buffer):
  67         raise NotImplementedError(
  68             self.__class__.__name__ + "must implement `decode()` method"
  69         )
  70
  71     @property
  72     def packed(self):
  73         raise NotImplementedError(
  74             self.__class__.__name__ + "must implement `packed()` property"
  75         )
  76
  77
  78 class Bcast(_Zmsg):
  79     """Zmq message to broadcast what was received from the terminal"""
  80
  81     KWARGS = (
  82         ("proto", 256),
  83         ("imei", None),
  84         ("when", None),
  85         ("peeraddr", None),
  86         ("packet", b""),
  87     )
  88
  89     @property
  90     def packed(self):
  91         return (
  92             pack("B", self.proto)
  93             + ("0000000000000000" if self.imei is None else self.imei).encode()
  94             + (
  95                 b"\0\0\0\0\0\0\0\0"
  96                 if self.when is None
  97                 else pack("!d", self.when)
  98             )
  99             + pack_peer(self.peeraddr)
 100             + self.packet
 101         )
 102
 103     def decode(self, buffer):
 104         self.proto = buffer[0]
 105         self.imei = buffer[1:17].decode()
 106         if self.imei == "0000000000000000":
 107             self.imei = None
 108         self.when = unpack("!d", buffer[17:25])[0]
 109         self.peeraddr = unpack_peer(buffer[25:43])
 110         self.packet = buffer[43:]
 111
 112
 113 class Resp(_Zmsg):
 114     """Zmq message received from a third party to send to the terminal"""
 115
 116     KWARGS = (("imei", None), ("packet", b""))
 117
 118     @property
 119     def packed(self):
 120         return (
 121             "0000000000000000" if self.imei is None else self.imei.encode()
 122         ) + self.packet
 123
 124     def decode(self, buffer):
 125         self.imei = buffer[:16].decode()
 126         self.packet = buffer[16:]