sedbgmux/sedbgmux-dump.py

#!/usr/bin/env python3

# This file is a part of sedbgmux, an open source DebugMux client.
# Copyright (c) 2023  Vadim Yanitskiy <fixeria@osmocom.org>
#
# SPDX-License-Identifier: GPL-3.0-or-later
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program 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
# General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

import logging as log
import argparse
import math
import sys

from sedbgmux.io import *
from sedbgmux import DbgMuxFrame


def dumpio_auto(fname: str, *args, **kw) -> DumpIO:
    ''' Detect dump format automatically '''
    if fname.endswith('.socat.dump'):
        return DumpIOSocat(fname, *args, **kw)
    elif fname.endswith('.dump'):
        return DumpIONative(fname, *args, **kw)
    raise DumpIOError('Automatic format detection failed')


class SEDbgMuxDumpApp:
    FORMATS = {
        'auto'      : (dumpio_auto, 'Automatic dump format detection (by filename)'),
        'native'    : (DumpIONative, 'Native binary dump format for this package'),
        'socat'     : (DumpIOSocat, 'ASCII hexdump generated by socat (-x option)'),
    }

    def __init__(self, argv) -> None:
        if argv.verbose > 0:
            log.root.setLevel(log.DEBUG)
        if argv.command == 'parse':
            self.do_parse(argv)
        elif argv.command == 'convert':
            self.do_convert(argv)
        elif argv.command == 'list-formats':
            self.do_list_formats()

    def do_parse(self, argv):
        dump = self.FORMATS[argv.format][0](argv.input, readonly=True)
        num_records: int = 0
        while num_records < argv.num_records:
            try:
                record: dict = dump.read()
                record['nr'] = num_records
                num_records += 1
            except DumpIOEndOfFile:
                break
            self._print_record(record)

    def do_convert(self, argv):
        di = self.FORMATS[argv.input_format][0](argv.input, readonly=True)
        do = self.FORMATS[argv.output_format][0](argv.output, readonly=False)
        num_records: int = 0
        while num_records < argv.num_records:
            try:
                record: dict = di.read()
                do.write(record)
                num_records += 1
            except DumpIOEndOfFile:
                break
        log.info('Converted %u records', num_records)

    def do_list_formats(self) -> None:
        for name, desc in self.FORMATS.items():
            print('%s\t\t%s' % (name, desc[1]))

    def _print_record(self, record: dict) -> None:
        frame = DbgMuxFrame.Frame.parse(record['data'])
        print('Record #{nr:04d} @ {timestamp:f} {dir} frame'.format(**record),
              '(Ns={TxCount:03d}, Nr={RxCount:03d}, fcs=0x{FCS:04x})'.format(**frame),
              '{} {}'.format(frame['MsgType'], frame['MsgData'].hex()))
        fcs: int = DbgMuxFrame.fcs_func(record['data'][:-2])
        if fcs != frame['FCS']:
            msg: str = 'Indicated 0x{ind:04x} != calculated 0x{calc:04x}' \
                       .format(ind=frame['FCS'], calc=fcs)
            if not argv.ignore_bad_fcs:
                raise DumpIOFcsError(msg)
            print('Bad FCS: ' + msg)
        if argv.decode_payload:
            msg = DbgMuxFrame.Msg.parse(frame['MsgData'], MsgType=frame['MsgType'])
            if msg != b'':
                print('  {}'.format(msg))


ap = argparse.ArgumentParser(prog='sedbgmux-dump',
                             description='DebugMux dump management utility')
sp = ap.add_subparsers(dest='command', metavar='command', required=True,
                       help='sub-command help')

ap.add_argument('-v', '--verbose', action='count', default=0,
                help='print debug logging')

parse = sp.add_parser('parse', help='parse a dump file')
parse.add_argument('input', metavar='INPUT', type=str,
                   help='input file to be parsed')
parse.add_argument('-f', '--format', type=str, default='auto',
                   choices=SEDbgMuxDumpApp.FORMATS.keys(),
                   help='input file format (default: %(default)s)')
parse.add_argument('-dp', '--decode-payload', action='store_true',
                   help='decode DebugMux frame contents')
parse.add_argument('-nr', '--num-records', type=int, default=math.inf,
                   help='number of records to parse (default: all)')
parse.add_argument('--ignore-bad-fcs', action='store_true',
                   help='do not abort parsing on FCS mismatch')

convert = sp.add_parser('convert', help='convert between different formats')
convert.add_argument('input', metavar='INPUT', type=str,
                     help='input file to be converted')
convert.add_argument('output', metavar='OUTPUT', type=str,
                     help='output file')
convert.add_argument('-if', '--input-format', type=str, required=True,
                     choices=SEDbgMuxDumpApp.FORMATS.keys(),
                     help='input file format')
convert.add_argument('-of', '--output-format', type=str, required=True,
                     choices=SEDbgMuxDumpApp.FORMATS.keys(),
                     help='output file format')
convert.add_argument('-nr', '--num-records', type=int, default=math.inf,
                     help='number of records to convert (default: all)')

sp.add_parser('list-formats', help='list all supported formats')

log.basicConfig(
    format='[%(levelname)s] %(filename)s:%(lineno)d %(message)s', level=log.INFO)

if __name__ == '__main__':
    argv = ap.parse_args()
    app = SEDbgMuxDumpApp(argv)