gpsd/clients/gpsplot.py.in

#!@PYSHEBANG@
# -*- coding: utf-8 -*-
# @GENERATED@
# This file is Copyright 2020 by the GPSD project
# SPDX-License-Identifier: BSD-2-clause

# This code runs compatibly under Python 2 and 3.x for x >= 2.
# Preserve this property!
#
# Codacy D203 and D211 conflict, I choose D203
# Codacy D212 and D213 conflict, I choose D212
"""gpsplot -- plot gpsd data in real time with matplotlib."""

from __future__ import print_function

import argparse
import datetime
import os
import socket
import sys
import time            # for time.time()

if 'DISPLAY' not in os.environ or not os.environ['DISPLAY']:
    have_display = False
else:
    have_display = True
    try:
        # Don't fail until after check for -h or -V.
        from matplotlib import pyplot as plt
        have_matplotlib = True
    except (ImportError, RuntimeError):
        have_matplotlib = False

# pylint wants local modules last
try:
    import gps
    import gps.clienthelpers
except ImportError as e:
    sys.stderr.write(
        "%s: can't load Python gps libraries -- check PYTHONPATH.\n" %
        (sys.argv[0]))
    sys.stderr.write("%s\n" % e)
    sys.exit(1)

gps_version = '@VERSION@'
if gps.__version__ != gps_version:
    sys.stderr.write("%s: ERROR: need gps module version %s, got %s\n" %
                     (sys.argv[0], gps_version, gps.__version__))
    sys.exit(1)


def _do_one_line(data):
    """Dump one report line."""
    if options.stripchart:
        if 'time' not in data:
            plt.pause(0.05)
            return

        timestr = datetime.datetime.strptime(data['time'],
                                          '%Y-%m-%dT%H:%M:%S.%fZ')

        if fields[0] in data and fields[2] in data:
            axs[0].scatter(timestr, data[fields[0]])
            axs[1].scatter(timestr, data[fields[2]])
        if fields[4] in data:
            axs[2].scatter(timestr, data[fields[4]] * conversion.altfactor)
    else:
        # must be scatterplot
        if fields[0] in data and fields[2] in data:
            axs[0].scatter(data[fields[2]], data[fields[0]])
        if fields[4] in data:
            axs[1].scatter(0.5, data[fields[4]] * conversion.altfactor)

    plt.pause(0.05)
    try:
        # try to tighten up the plot.  let it fail silently
        plt.tight_layout(h_pad=.2)
    except ValueError:
        pass


# get default units from the environment
# GPSD_UNITS, LC_MEASUREMENT and LANG
default_units = gps.clienthelpers.unit_adjustments()

description = 'Create dynamic plots from gpsd with matplotlib.'
usage = '%(prog)s [OPTIONS] [host[:port[:device]]]'
epilog = ('BSD terms apply: see the file COPYING in the distribution root'
          ' for details.')

parser = argparse.ArgumentParser(
             description=description,
             epilog=epilog,
             formatter_class=argparse.RawDescriptionHelpFormatter,
             usage=usage)
parser.add_argument(
    '-?',
    action="help",
    help='show this help message and exit'
)
parser.add_argument(
    '-D',
    '--debug',
    default=0,
    dest='debug',
    help='Set level of debug. Must be integer. [Default %(default)s)]',
    type=int,
)
parser.add_argument(
    '--device',
    default='',
    dest='device',
    help='The device to connect. [Default %(default)s)]',
)
parser.add_argument(
    '--fields',
    choices=['llh', 'llm'],
    default='llh',
    dest='fields',
    help='Fields to plot. [Default %(default)s)]',
)
parser.add_argument(
    '--host',
    default='localhost',
    dest='host',
    help='The host to connect. [Default %(default)s)]',
)
parser.add_argument(
    '-n',
    '--count',
    default=0,
    dest='count',
    help='Count of messages to parse. 0 to disable. [Default %(default)s)]',
    type=int,
)
parser.add_argument(
    '--port',
    default=gps.GPSD_PORT,
    dest='port',
    help='The port to connect. [Default %(default)s)]',
)
parser.add_argument(
    '--plottype',
    choices=['scatterplot', 'stripchart'],
    default='scatterplot',
    dest='plottype',
    help='Plot type. [Default %(default)s)]',
)
parser.add_argument(
    '-u', '--units',
    choices=['i', 'imperial', 'n', 'nautical', 'm', 'metric'],
    default=default_units.name,
    dest='units',
    help='Units [Default %(default)s)]',
)
parser.add_argument(
    '-V', '--version',
    action='version',
    help='Output version to stderr, then exit',
    version="%(prog)s: Version " + gps_version + "\n",
)
parser.add_argument(
    '-x',
    '--seconds',
    default=0,
    dest='seconds',
    help='Seconds of messages to parse. 0 to disable. [Default %(default)s)]',
    type=int,
)
parser.add_argument(
    'target',
    help='[host[:port[:device]]]',
    nargs='?',
)
options = parser.parse_args()

# allow -V and -h, above, before exiting on matplotlib, or DISPLAY, not found
if not have_display:
    # matplotlib will not import w/o DISPLAY
    sys.stderr.write("gpsplot: ERROR: $DISPLAY not set\n")
    sys.exit(1)
if not have_matplotlib:
    sys.stderr.write("gpsplot: ERROR: required Python module "
                     "matplotlib not found\n")
    sys.exit(1)

# get conversion factors
conversion = gps.clienthelpers.unit_adjustments(units=options.units)
flds = {'llh': ('lat', 'Latitude', 'lon', 'Longitude', 'altHAE', 'altHAE'),
        'llm': ('lat', 'Latitude', 'lon', 'Longitude', 'altMSL', 'altMSL'),
        }

if options.fields not in flds:
    sys.stderr.write("gpsplot: Invalid --fields argument %s\n" %
                     options.fields)
    sys.exit(1)

fields = flds[options.fields]

# the options host, port, device are set by the defaults
if options.target:
    # override host, port and device with target
    arg = options.target.split(':')
    len_arg = len(arg)
    if len_arg == 1:
        (options.host,) = arg
    elif len_arg == 2:
        (options.host, options.port) = arg
    elif len_arg == 3:
        (options.host, options.port, options.device) = arg
    else:
        parser.print_help()
        sys.exit(0)

if not options.port:
    options.port = gps.GPSD_PORT

options.scatterplot = False
options.stripchart = False
if 'scatterplot' == options.plottype.lower():
    # scatterplot
    options.scatterplot = True
else:
    # stripchart
    options.stripchart = True

options.mclass = 'TPV'
# Fields to parse
# autodetect, read one message, use those fields
options.json_fields = None

options.frames = None
options.subclass = None

try:
    session = gps.gps(host=options.host, port=options.port,
                      verbose=options.debug)
except socket.error:
    sys.stderr.write("gpsplot: Could not connect to gpsd daemon\n")
    sys.exit(1)

session.stream(gps.WATCH_ENABLE | gps.WATCH_SCALED, devpath=options.device)


plt.ion()

x = []
y = []

if options.scatterplot:
    fig = plt.figure(figsize=(7, 7))
    # x/y
    ax = fig.add_axes([0.18, 0.17, 0.65, 0.65])
    plt.xticks(rotation=30, ha='right')
    # z
    ax1 = fig.add_axes([0.85, 0.17, 0.02, 0.65])
    axs = [ax, ax1]

    for ax in axs:
        ax.ticklabel_format(useOffset=False)
    axs[0].tick_params(direction='in', top=True, right=True)
    axs[0].set_xlabel(fields[3])
    axs[0].set_ylabel(fields[1])
    axs[1].set_title("%s (%s)" % (fields[5], conversion.altunits))
    axs[1].tick_params(bottom=False, left=False, right=True,
                       labeltop=True, labelbottom=False)
    axs[1].xaxis.set_visible(False)
    axs[1].yaxis.tick_right()

elif options.stripchart:
    fig, axs = plt.subplots(3, sharex=True, figsize=(7, 7))
    for ax in axs:
        ax.ticklabel_format(useOffset=False)
        ax.tick_params(direction='in', top=True, right=True)
    axs[0].set_title(fields[1])
    axs[1].set_title(fields[3])
    axs[2].set_title("%s (%s)" % (fields[5], conversion.altunits))
    # plt.xticks(rotation=30, ha='right')
    fig.autofmt_xdate(rotation=30, ha='right')
    plt.subplots_adjust(left=0.16, bottom=0.10)
else:
    sys.stderr.write("Error: Unknown plot type\n")
    sys.exit(1)


count = 0
if 0 < options.seconds:
    end_seconds = time.time() + options.seconds
else:
    end_seconds = 0

try:
    while True:
        report = session.next()
        if report['class'] != options.mclass:
            continue

        _do_one_line(report)

        if 0 < options.count:
            count += 1
            if count >= options.count:
                break

        if 0 < options.seconds:
            if time.time() > end_seconds:
                break

except KeyboardInterrupt:
    # caught control-C
    # FIXME: plot is in different process, does not come here...
    print()
    sys.exit(1)

# make the plot persist until window closed.
plt.show(block=True)