gps303/storage.py

   1 """ Store zmq broadcasts to sqlite """
   2
   3 from datetime import datetime, timezone
   4 from logging import getLogger
   5 import zmq
   6
   7 from . import common
   8 from .evstore import initdb, stow
   9 from .gps303proto import proto_of_message
  10 from .zmsg import Bcast
  11
  12 log = getLogger("gps303/storage")
  13
  14
  15 def runserver(conf):
  16     dbname = conf.get("storage", "dbfn")
  17     log.info('Using Sqlite3 database "%s"', dbname)
  18     initdb(dbname)
  19     zctx = zmq.Context()
  20     zsub = zctx.socket(zmq.SUB)
  21     zsub.connect(conf.get("collector", "publishurl"))
  22     zsub.setsockopt(zmq.SUBSCRIBE, b"")
  23
  24     try:
  25         while True:
  26             zmsg = Bcast(zsub.recv())
  27             log.debug(
  28                 "%s IMEI %s from %s at %s: %s",
  29                 "I" if zmsg.is_incoming else "O",
  30                 zmsg.imei,
  31                 zmsg.peeraddr,
  32                 datetime.fromtimestamp(zmsg.when).astimezone(tz=timezone.utc),
  33                 zmsg.packet.hex(),
  34             )
  35             if zmsg.peeraddr is not None:
  36                 addr, port = zmsg.peeraddr
  37                 peeraddr = str((str(addr), port))
  38             else:
  39                 peeraddr = None
  40             stow(
  41                 is_incoming=zmsg.is_incoming,
  42                 peeraddr=peeraddr,
  43                 when=zmsg.when,
  44                 imei=zmsg.imei,
  45                 proto=proto_of_message(zmsg.packet),
  46                 packet=zmsg.packet,
  47             )
  48     except KeyboardInterrupt:
  49         pass
  50
  51
  52 if __name__.endswith("__main__"):
  53     runserver(common.init(log))