gpsd/clients/gpscat.py.in

#!@PYSHEBANG@
#
# @GENERATED@
# This file is Copyright 2010 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
"""Display GPS output.  Hexify it if necessary."""

from __future__ import absolute_import, print_function, division

import argparse
import os
import select
import socket
import string
import sys
import termios

# pylint wants local modules last
try:
    import gps
    import gps.packet as sniffer
    from gps.misc import BINARY_ENCODING
except ImportError as e:
    sys.stderr.write(
        "gpscat: can't load Python gps libraries -- check PYTHONPATH.\n")
    sys.stderr.write("%s\n" % e)
    sys.exit(1)

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


# The spec says 82, but some receivers (TN-200, GSW 2.3.2) output 86 characters
# the Skyrtaq S2525F8 emits 100 chars
NMEA_MAX = 102

# Lowest debug level at which packet getter begins to emit messages, minus one
BASELEVEL = sniffer.LOG_IO

highhalf_latch = True

PRINTABLE = set(bytearray(string.printable, encoding=BINARY_ENCODING))


def hexdump(st):
    """Convert string to hex string."""
    dmp = ""
    for ch in bytearray(st):  # bytearray gets array of ints in Python 2 and 3
        if ch in PRINTABLE:
            dmp += chr(ch)
        else:
            dmp += "\\x%02x" % ch
    return dmp


def reporter(errlevel, mesg):
    """Report errors, depending on log level."""
    if errlevel <= options.debug:
        sys.stdout.write(mesg)


if __name__ == '__main__':
    usage = '%(prog)s [OPTIONS]'
    epilog = ('BSD terms apply: see the file COPYING in the distribution root'
              ' for details.')

    parser = argparse.ArgumentParser(usage=usage, epilog=epilog)
    parser.add_argument(
        '-?',
        action="help",
        help='show this help message and exit'
    )
    parser.add_argument(
        '-D',
        '--debug',
        dest='debug',
        default=0,
        type=int,
        help='Set level of debug. Must be integer. [Default %(default)s]',
    )
    parser.add_argument(
        '-p', '--packetizer',
        action='store_true',
        dest='packetizer',
        default=False,
        help="Set packetizer mode",
    )
    parser.add_argument(
        '-s', '--speed',
        dest='speed_str',
        default=None,
        help=('Set port speed. '
              ' Optionally parity (N, O, E) and stop bits (1, 2)'),
    )
    parser.add_argument(
        '-t', '--typeflag',
        action='store_true',
        dest='typeflag',
        default=False,
        help="Select packetizer mode with type and length",
    )
    parser.add_argument(
        '-V', '--version',
        action='version',
        version="%(prog)s: Version " + gps_version + "\n",
        help='Output version to stderr, then exit',
    )
    parser.add_argument(
        'target',
        help='File or Serial port to read from',
    )

    options = parser.parse_args()

    options.speed = None
    options.parity = None
    options.stopbits = None
    if options.speed_str:
        if options.speed_str[-2] in ('N', 'E', 'O'):
            options.parity = options.speed_str[-2]
            options.stopbits = int(options.speed_str[-1])
            options.speed_str = options.speed_str[:-2]
        options.speed = int(options.speed_str)

    # adjust debug level
    options.debug += BASELEVEL
    if options.packetizer or options.typeflag:
        options.rawmode = False
    else:
        options.rawmode = True

    try:
        if "rfcomm" in options.target:     # Bluetooth special case
            s = socket.socket(socket.AF_BLUETOOTH, socket.SOCK_STREAM,
                              socket.BTPROTO_RFCOMM)
            s.connect((options.target, 1))
            tty = s.fileno()
        else:                            # Ordinary device
            tty = os.open(options.target, os.O_RDWR)

        if options.speed is not None:
            (iflag, oflag, cflag, lflag, ispeed, ospeed, cc) = \
                termios.tcgetattr(tty)
            try:
                ispeed = ospeed = eval("termios.B%d" % options.speed)
            except AttributeError:
                sys.stderr.write("gpscat: unknown baud rate %d\n" %
                                 options.speed)
                raise SystemExit(1)
            if options.stopbits:
                cflag &= ~termios.CSIZE
                cflag |= (termios.CS8, termios.CS7)[options.stopbits - 1]
            if options.parity:
                if options.parity == 'N':
                    iflag &= ~termios.PARENB
                    iflag &= ~termios.INPCK
                elif options.parity == 'O':
                    iflag |= termios.INPCK
                    cflag |= termios.PARENB
                    cflag |= termios.PARODD
                elif options.parity == 'E':
                    iflag |= termios.INPCK
                    cflag |= termios.PARENB
                    cflag &= ~termios.PARODD
            termios.tcsetattr(tty, termios.TCSANOW,
                              [iflag, oflag, cflag, lflag, ispeed, ospeed, cc])

        if not options.rawmode:
            getter = sniffer.new()
            sniffer.register_report(reporter)
        seqno = 0
        while True:
            rlist, wlist, xlist = select.select([tty], [], [])
            if rlist == [tty]:
                if options.rawmode:
                    buf = os.read(tty, NMEA_MAX)
                    if not buf:
                        break
                    sys.stdout.write(hexdump(buf))
                else:
                    (length, ptype, packet, counter) = getter.get(tty)
                    seqno += 1
                    if length == 0:
                        break
                    if options.typeflag:
                        sys.stdout.write('%d (%d@%d): %s\n' % (ptype, length,
                                         counter - length, hexdump(packet)))
                    else:
                        sys.stdout.write(hexdump(packet) + "\n")
    except KeyboardInterrupt:
        if options.rawmode:
            sys.stdout.write("\n")
        raise SystemExit(0)

# Local variables:
# mode: python
# end:
# vim: set expandtab shiftwidth=4