gps303/storage.py

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