popufare/busunit/scripts/format_debug_msg.py

#!/usr/bin/python
#
# Copyright (c) 2019 Clementine Computing LLC.
#
# This file is part of PopuFare.
#
# PopuFare is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# PopuFare is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with PopuFare.  If not, see <https://www.gnu.org/licenses/>.
#

# Messages can be inserted into the IPC hub via the `debug_client` program.
# The `debug_client` requires messages to be of the format:
#
# MAILBOX: XX XX ... XX
#
# Where `MAILBOX` is the mailbox of the message and XX are the hex bytes
# of the payload message in ASCII format.
#
# For example, to send a message to the `BILLING_LOG` mailbox with a message
# of "#testing", the following would be used:
#
# echo 'BILLING_LOG: 23 74 65 73 74 69 6e 67' | /home/bus/bin/debug_client -s
#
# Where the '#' prefix sends the mesage to the `diagnostic_log` table on the server.
#
# See the `MAILBOX_*` defines in `commhub.h` for the full list of mailboxes.
# As of this writing, the list of mailboxes is:
#
# _HELLO        _SUBSCRIBE  _UNSUBSCRIBE    _BROADCAST      _WIRETAP
# _ERROR        _PING       _PONG           _EXIT           _HUP
# STATUS_REQ    PASS_STATUS BILL_STATUS     DRIVER_STATUS   GPS_STATUS
# STOP_STATUS   TOKEN_MAG   TOKEN_RFID      FLUSH_PASSES    UPDATE_PASSES
# DRIVER_NOTIFY BILLING_LOG PIU_MESSAGE     SET_PADDLE      PADDLE_ACK
# NEXT_STOP     PREV_STOP   RULE_CALL       VAULT_DROP      PASSDB_CONSIST
# PASSDB_PULSE
#

import sys
import getopt

VERSION = "0.1.0"

VALID_MAILBOX = [
  "_HELLO", "_SUBSCRIBE", "_UNSUBSCRIBE",
  "_BROADCAST", "_WIRETAP", "_ERROR",
  "_PING", "_PONG", "_EXIT", "_HUP",
  "STATUS_REQ",
  "PASS_STATUS", "BILL_STATUS", "DRIVER_STATUS",
  "GPS_STATUS", "STOP_STATUS",
  "TOKEN_MAG", "TOKEN_RFID",
  "FLUSH_PASSES", "UPDATE_PASSES",
  "DRIVER_NOTIFY", "BILLING_LOG", "PIU_MESSAGE",
  "SET_PADDLE", "PADDLE_ACK",
  "NEXT_STOP", "PREV_STOP",
  "RULE_CALL", "VAULT_DROP",
  "PASSDB_CONSIST", "PASSDB_PULSE" ]


def show_version(ofp):
  ofp.write("Version: " + VERSION + "\n");

def show_help(ofp):
  ofp.write("\n");
  ofp.write("usage:");
  ofp.write("\n");
  ofp.write("\n");
  ofp.write("  format_debug_msg.py [-h] [-F] <MAILBOX> [--asciihex] [-i] <PAYLOAD>\n");
  ofp.write("\n");
  ofp.write("  <MAILBOX>            destination mailbox\n")
  ofp.write("  <PAYLOAD>            payload to convert\n")
  ofp.write("  [--show-mailbox]     print out all valid mailboxes\n")
  ofp.write("  [-F|-force-mailbox]  do not check to see if MAILBOX is in list of valid mailboxes\n")
  ofp.write("  [--asciihex]         payload in ASCII hex format (with spaces) (example: ' 23 74 65 73 74')\n")
  ofp.write("  [-i]                 Read from file instead ('-' for stdin)\n")
  ofp.write("  [-h|-help]           help (this screen)\n")
  ofp.write("\n");

def show_mailboxes():
  for mb in VALID_MAILBOX:
    print mb

mailbox = ""
payload_text = ""
payload_converted = ""
asciihex_flag = False
force_mailbox = False

ifn = ''

argv = sys.argv[1:]
options, rem = getopt.gnu_getopt(sys.argv[1:], "hFi:", ["asciihex", "help", "version", "show-mailbox", "force-mailbox", "input="])
for opt, arg in options:
  if opt in ('-v', '--version'):
    show_version(sys.stdout)
    sys.exit(0)
  if opt in ('-h', '--help'):
    show_help(sys.stdout)
    sys.exit(0)
  elif opt in ('--asciihex'):
    asciihex_flag = True
  elif opt in ('-F', '--force-mailbox'):
    force_mailbox = True
  elif opt in ('--show-mailbox'):
    show_mailboxes()
    sys.exit(0)
  elif opt in ('-i', '--input'):
    ifn = arg
  else:
    show_help(sys.sdterr)
    sys.exit(-1)

if (len(rem) < 2) and ((len(rem) != 1) or (len(ifn)==0)):
  sys.stderr.write("must provide MAILBOX and PAYLOAD\n")
  show_help(sys.stderr)
  sys.exit(-1)

mailbox = rem[0]

if not force_mailbox:
  if not mailbox in VALID_MAILBOX:
    sys.stderr.write("ERROR: could not find '" + mailbox + "' in list of valid mailboxes.\n")
    sys.stderr.write("Use the --force-mailbox option to use this mailbox anyway.\n")
    sys.stderr.write("Use the --show-mailbox option to see a list of vlaid options\n\n")
    sys.exit(-1)

if len(ifn)==0:
  payload_text = rem[1]
else:
  with open(ifn) as fp:
    payload_text = fp.read()

if asciihex_flag:
  payload_converted = payload_text
else:
  for ch in payload_text:
    payload_converted += " " + format(ord(ch), '02x').zfill(2)

print mailbox + ":" + payload_converted