loctrkd/mkgpx.py

   1 """ Example that produces gpx from events in evstore """
   2
   3 # run as:
   4 # python -m loctrkd.mkgpx <IMEI>
   5 # Generated gpx is emitted to stdout
   6
   7 from configparser import ConfigParser
   8 from datetime import datetime, timezone
   9 from getopt import getopt
  10 from importlib import import_module
  11 from logging import getLogger
  12 from sqlite3 import connect
  13 from sys import argv
  14 from typing import Any, cast, List, Tuple
  15
  16 from . import common
  17 from .protomodule import ProtoModule
  18
  19 log = getLogger("loctrkd/mkgpx")
  20
  21
  22 pmods: List[ProtoModule] = []
  23
  24
  25 def main(
  26     conf: ConfigParser, opts: List[Tuple[str, str]], args: List[str]
  27 ) -> None:
  28     global pmods
  29     pmods = [
  30         cast(ProtoModule, import_module("." + modnm, __package__))
  31         for modnm in conf.get("common", "protocols").split(",")
  32     ]
  33     db = connect(conf.get("storage", "dbfn"))
  34     c = db.cursor()
  35     c.execute(
  36         """select tstamp, is_incoming, proto, packet from events
  37            where imei = ?  and is_incoming = true
  38            and proto in (?, ?)
  39            order by tstamp""",
  40         (args[0], "BS:UD", "BS:UD2"),
  41     )
  42     print(
  43         """<?xml version="1.0"?>
  44     <gpx version="1.1"
  45     creator="loctrkd"
  46     xmlns="http://www.topografix.com/GPX/1/1">
  47       <name>Location Data</name>
  48       <trk>
  49         <name>Location Data</name>
  50         <trkseg>
  51     """
  52     )
  53
  54     for tstamp, is_incoming, proto, packet in c:
  55         for pmod in pmods:
  56             if pmod.proto_handled(proto):
  57                 msg = pmod.parse_message(packet, is_incoming=is_incoming)
  58         lat, lon = msg.latitude, msg.longitude
  59         isotime = (
  60             datetime.fromtimestamp(tstamp)
  61             .astimezone(tz=timezone.utc)
  62             .isoformat()
  63         )
  64         isotime = isotime[: isotime.rfind(".")] + "Z"
  65         trkpt = """      <trkpt lat="{}" lon="{}">
  66               <time>{}</time>
  67           </trkpt>""".format(
  68             lat, lon, isotime
  69         )
  70         print(trkpt)
  71         if False:
  72             print(
  73                 datetime.fromtimestamp(tstamp)
  74                 .astimezone(tz=timezone.utc)
  75                 .isoformat(),
  76                 msg,
  77             )
  78     print(
  79         """    </trkseg>
  80       </trk>
  81     </gpx>"""
  82     )
  83
  84
  85 if __name__.endswith("__main__"):
  86     opts, args = getopt(argv[1:], "o:c:d")
  87     main(common.init(log, opts=opts), opts, args)