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 def pack_peer(peeraddr):
   9     saddr, port, _x, _y = peeraddr
  10     addr6 = ip.ip_address(saddr)
  11     addr = addr6.ipv4_mapped
  12     if addr is None:
  13         addr = addr6
  14     return pack("B", addr.version) + (addr.packed + b"\0\0\0\0\0\0\0\0\0\0\0\0")[:16] + pack("!H", port)
  15
  16 def unpack_peer(buffer):
  17     version = buffer[0]
  18     if version not in (4, 6):
  19         return None
  20     if version == 4:
  21         addr = ip.IPv4Address(buffer[1:5])
  22     else:
  23         addr = ip.IPv6Address(buffer[1:17])
  24     port = unpack("!H", buffer[17:19])[0]
  25     return (addr, port)
  26
  27
  28 class _Zmsg:
  29     def __init__(self, *args, **kwargs):
  30         if len(args) == 1:
  31             self.decode(args[0])
  32         elif bool(kwargs):
  33             for k, v in self.KWARGS:
  34                 setattr(self, k, kwargs.get(k, v))
  35         else:
  36             raise RuntimeError(
  37                 self.__class__.__name__
  38                 + ": both args "
  39                 + str(args)
  40                 + " and kwargs "
  41                 + str(kwargs)
  42             )
  43
  44     def decode(self, buffer):
  45         raise RuntimeError(
  46             self.__class__.__name__ + "must implement `encode()` method"
  47         )
  48
  49     @property
  50     def packed(self):
  51         raise RuntimeError(
  52             self.__class__.__name__ + "must implement `encode()` method"
  53         )
  54
  55
  56 class Bcast(_Zmsg):
  57     """Zmq message to broadcast what was received from the terminal"""
  58
  59     KWARGS = (
  60         ("proto", 256),
  61         ("imei", None),
  62         ("when", None),
  63         ("peeraddr", None),
  64         ("packet", b""),
  65     )
  66
  67     @property
  68     def packed(self):
  69         return (
  70             pack("B", self.proto)
  71             + ("0000000000000000" if self.imei is None else self.imei).encode()
  72             + (b"\0\0\0\0\0\0\0\0" if self.when is None else pack("!d", self.when))
  73             + pack_peer(self.peeraddr)
  74             + self.packet
  75         )
  76
  77     def decode(self, buffer):
  78         self.proto = buffer[0]
  79         self.imei = buffer[1:17].decode()
  80         if self.imei == "0000000000000000":
  81             self.imei = None
  82         self.when = unpack("!d", buffer[17:25])[0]
  83         self.peeraddr = unpack_peer(buffer[25:44])
  84         self.packet = buffer[44:]
  85
  86
  87 class Resp(_Zmsg):
  88     """Zmq message received from a third party to send to the terminal"""
  89
  90     KWARGS = (("imei", None), ("packet", b""))
  91
  92     @property
  93     def packed(self):
  94         return (
  95             "0000000000000000" if self.imei is None else self.imei.encode()
  96         ) + self.packet
  97
  98     def decode(self, buffer):
  99         self.imei = buffer[:16].decode()
 100         self.packet = buffer[16:]