Added commandline argument -q to add QSO(s) and --stdin to read QSO(s) from stdin

Author: gitandy

README.md

@@ -202,6 +202,27 @@ To leave the event mode for the following QSOs type a single `$` followed by a S
 For xOTA just enter one of SOTA, POTA e.g. `$pota` instead of the contest ID. 
 Then set your own xOTA reference with `-Nxx-999` and track the QSO partners reference with `%xx-999`.
 
+Commandline support
+-------------------
+HamCC is also able to import QSOs from STDIN.
+
+    # echo -e "8 s df1asc 'Andreas 20241105d\n4 f df1asc 1202d" | hamcc --stdin
+
+or put the QSOs in a file (one QSO per line) 
+
+e.g. `qsos.txt` 
+
+    8 s df1asc 'Andreas 20241105d
+    4 f df1asc 1202d
+
+Commandline invokation
+
+    # hamcc --stdin < qsos.txt
+
+Another possibility is to use argument `-q` per QSO
+
+    # hamcc -q 8 s df1asc 'Andreas 20241105d -q 4 f df1asc 1202d 
+
 Source Code
 -----------
 The source code is available at [GitHub](https://github.com/gitandy/HamCC)

src/hamcc/__main__.py

@@ -1,105 +1,51 @@
+# Copyright 2024 by Andreas Schawo, licensed under CC BY-SA 4.0
+
 """Log Ham Radio QSOs via console"""
 
 import os
 import sys
-import time
 import logging
-from platform import python_implementation
-from curses import wrapper, error
-if python_implementation() == 'PyPy':
-    # noinspection PyUnresolvedReferences,PyPep8Naming
-    from curses import Window as window
-else:
-    # noinspection PyUnresolvedReferences
-    from curses import window
+from collections.abc import Iterator
+from typing import TextIO
 
+LOG_FMT = '%(asctime)s %(levelname)-8s %(name)s: %(message)s'
 logger = logging.getLogger('HamCC')
-logging.basicConfig(filename='./hamcc.log', filemode='w', format='%(asctime)s %(levelname)-8s %(name)s: %(message)s',
+logging.basicConfig(filename='./hamcc.log', filemode='w', format=LOG_FMT,
                     level=logging.INFO)
 
 from adif_file import adi
 from adif_file import __version_str__ as __version_adif_file__
 
 from . import __proj_name__, __version_str__, __author_name__, __copyright__
-from .hamcc import CassiopeiaConsole, adif_date2iso, adif_time2iso
-
-PROMPT = 'QSO> '
-LN_MYDATA = 0
-LN_QSODATA = 1
-LN_INPUT = 2
-LN_INFO = 3
-
-
-def qso2str(qso, pos, cnt) -> tuple[str, str]:
-    d = adif_date2iso(qso["QSO_DATE"])
-    t = adif_time2iso(qso["TIME_ON"])
-
-    opt_info = ''
-    for i, f in (
-            ('.', 'RST_RCVD'),
-            (',', 'RST_SENT'),
-            ('\'', 'NAME'),
-            ('f', 'FREQ'),
-            ('p', 'TX_PWR'),
-            ('*', 'QSL_RCVD'),
-            ('#', 'COMMENT'),
-    ):
-        if f in qso:
-            val = qso[f]
-            if f == 'FREQ':
-                val = f'{float(val) * 1000:0.3f}'.rstrip('0').rstrip('.')
-            if f in ('FREQ', 'TX_PWR'):
-                opt_info += f'| {val} {i} '
-            else:
-                opt_info += f'| {i} {val} '
+from .hamcc import CassiopeiaConsole
 
-    event_info = ''
-    if 'CONTEST_ID' in qso and qso["CONTEST_ID"]:
-        event_info = (f'[ $ {qso["CONTEST_ID"]} | -N {qso.get("STX", qso["STX_STRING"])} | '
-                      f'% {qso.get("SRX", qso.get("SRX_STRING", ""))} ]')
-    elif 'MY_SIG' in qso:
-        event_info = f'[ $ {qso["MY_SIG"]} | -N {qso["MY_SIG_INFO"]} | % {qso.get("SIG_INFO", "")} ]'
 
-    loc = ''
-    if 'GRIDSQUARE' in qso:
-        loc = f'{qso["QTH"]} ({qso["GRIDSQUARE"]})' if 'QTH' in qso else qso["GRIDSQUARE"]
+def qso_iterator(qso_stream: TextIO) -> Iterator:
+    line = qso_stream.readline().strip()
+    if line:
+        yield line
 
-    my_loc = ''
-    if 'MY_GRIDSQUARE' in qso:
-        my_loc = f'{qso["MY_CITY"]} ({qso["MY_GRIDSQUARE"]})' if 'MY_CITY' in qso else qso["MY_GRIDSQUARE"]
+    while line:
+        line = qso_stream.readline().strip()
+        if line:
+            yield line
 
-    line1 = (f'[ {"*" if pos == -1 else pos + 1}/{"-" if cnt == 0 else cnt} ] '
-             f'[ -c {qso["STATION_CALLSIGN"]} | -l {my_loc} | -n {qso.get("MY_NAME", "")} ] {event_info}')
-    line2 = (f'[ {d} d | {t} t | {qso["BAND"] if qso["BAND"] else "Band"} | {qso["MODE"] if qso["MODE"] else "Mode"} | '
-             f'{qso["CALL"] if qso["CALL"] else "Call"} | @ {loc} {opt_info}]')
 
-    return line1, line2
+def process_qsos(qsos: list[list[str]] | TextIO, file: str,
+                 own_call: str, own_loc: str, own_name: str, append: bool = False,  # noqa: C901
+                 contest_id: str = '', qso_number: int = 1):
+    """Process a list of text input from stdin or commandline as it was typed in console"""
 
-
-def read_adi(file: str) -> tuple[dict[str, str], dict[str, tuple[str, str]]]:
-    last_qso = {}
-    worked_calls = {}
-
-    doc = adi.load(file)
-    for r in doc['RECORDS']:
-        if all(f in r for f in ('CALL', 'QSO_DATE', 'TIME_ON')):
-            last_qso = r
-            worked_calls[r['CALL']] = (r['QSO_DATE'], r['TIME_ON'])
-    return last_qso, worked_calls
-
-
-def command_console(stdscr: window, file, own_call, own_loc, own_name, append=False,  # noqa: C901
-                    contest_id='', qso_number=1, records: list = []):
     adi_f = None
     try:
         fmode = 'a' if append else 'w'
         fexists = os.path.isfile(file)
 
         last_qso = {}
-        worked_calls = []
         if fexists and append:
-            logger.info('Loading last QSO and worked before...')
-            last_qso, worked_calls = read_adi(file)
+            logger.info('Loading last QSO...')
+            doc = adi.load(file)
+            last_qso = doc['RECORDS'][-1] if doc['RECORDS'] else {}
 
         adi_f = open(file, fmode)
 
@@ -115,118 +61,47 @@ def command_console(stdscr: window, file, own_call, own_loc, own_name, append=Fa
             adi_f.flush()
             logger.info('...done')
 
-        if records:
-            last_qso = records[-1]
-
-        cc = CassiopeiaConsole(own_call, own_loc, own_name, contest_id, qso_number, last_qso, worked_calls)
-        if records:
-            logger.info('Loading QSOs...')
-            for r in records:
-                cc.append_qso(r)
-            logger.info(f'...done {len(cc.qsos)} QSOs')
-
-        # Clear screen
-        stdscr.clear()
-        ln1, ln2 = qso2str(cc.current_qso, cc.edit_pos, 0)
-        stdscr.addstr(LN_MYDATA, 0, ln1)
-        stdscr.addstr(LN_QSODATA, 0, ln2)
-
-        fname = '...' + adi_f.name[-40:] if len(adi_f.name) > 40 else adi_f.name
-        last_qso_str = (f'. Last QSO: {last_qso["CALL"]} '
-                        f'worked on {adif_date2iso(last_qso["QSO_DATE"])} '
-                        f'at {adif_time2iso(last_qso["TIME_ON"])}') if last_qso and 'CALL' in last_qso else ''
-        stdscr.addstr(LN_INFO, 0, f'{"Appending to" if append else "Overwriting"} "{fname}"{last_qso_str}')
-        stdscr.addstr(LN_INPUT, 0, PROMPT)
-
-        stdscr.refresh()
-        stdscr.nodelay(True)
-
-        try:
-            logger.info('Entering main loop...')
-            while True:
-                py, px = stdscr.getyx()
-                ln1, ln2 = qso2str(cc.current_qso, cc.edit_pos, len(cc.qsos))
-                stdscr.addstr(LN_MYDATA, 0, ln1)
-                stdscr.clrtoeol()
-                stdscr.addstr(LN_QSODATA, 0, ln2)
-                stdscr.clrtoeol()
-                stdscr.addstr(py, px, '')
-
-                while True:
-                    try:
-                        c = stdscr.getkey()
-                        break
-                    except error:
-                        time.sleep(.01)
+        cc = CassiopeiaConsole(own_call, own_loc, own_name, contest_id, qso_number, last_qso)
+        qsos = qsos if type(qsos) is list else qso_iterator(qsos)
+        for qso in qsos:
+            if type(qso) is str:
+                qso = qso.strip().split(' ')
+                logger.info(f'Processing QSO: {qso}')
+                for chunk in qso:
+                    for char in chunk:
+                        cc.append_char(char)
+                    res = cc.append_char(' ')
+                    if res:
+                        if res.startswith('Warning:'):
+                            logger.warning(f'{res} for "{chunk}"')
+                        elif res.startswith('Error:'):
+                            logger.error(f'{res} for "{chunk}"')
+            else:
+                logger.info(f'Processing QSO: {qso}')
+                for val in qso:
+                    res = cc.evaluate(val)
+                    if res:
+                        if res.startswith('Warning:'):
+                            logger.warning(f'{res} for "{val}"')
+                        elif res.startswith('Error:'):
+                            logger.error(f'{res} for "{val}"')
+
+            res = cc.finalize_qso()
+            if res:
+                if res.startswith('Warning:'):
+                    logger.warning(res)
+                elif res.startswith('Error:'):
+                    logger.error(res)
+                else:
+                    logger.info(res)
 
-                if c == 'KEY_UP':
-                    cc.load_prev()
-                    stdscr.addstr(LN_INFO, 0, '')
-                    stdscr.clrtoeol()
-                    stdscr.addstr(LN_INPUT, 0, PROMPT)
-                    stdscr.clrtoeol()
-                elif c == 'KEY_DOWN':
-                    cc.load_next()
-                    stdscr.addstr(LN_INFO, 0, '')
-                    stdscr.clrtoeol()
-                    stdscr.addstr(LN_INPUT, 0, PROMPT)
-                    stdscr.clrtoeol()
-                elif c == 'KEY_DC':
-                    res = cc.del_selected()
-                    if res >= 0:
-                        stdscr.addstr(LN_INFO, 0, f'Deleted QSO #{res + 1}')
-                    else:
-                        stdscr.addstr(LN_INFO, 0, '')
-                    stdscr.clrtoeol()
-                    stdscr.addstr(LN_INPUT, 0, PROMPT)
-                    stdscr.clrtoeol()
-                elif len(c) > 1 or c in '\r\t':
-                    continue
-                elif c == '\n':  # Flush QSO to stack
-                    res = cc.append_char(c)
-                    stdscr.addstr(LN_INFO, 0, res)
-                    stdscr.clrtoeol()
-                    stdscr.addstr(LN_INPUT, 0, PROMPT)
-                    stdscr.clrtoeol()
-                elif c == '!':  # Write QSOs to disk
-                    cc.append_char('\n')
-                    i = 0
-                    msg = ''
-                    while cc.has_qsos():
-                        adi_f.write('\n\n' + adi.dumps({'RECORDS': [cc.pop_qso()]}))
-                        adi_f.flush()
-                        i += 1
-                        msg = f'{i} QSO(s) written to disk'
-                    stdscr.addstr(LN_INFO, 0, msg)
-                    stdscr.clrtoeol()
-                    stdscr.addstr(LN_INPUT, 0, PROMPT)
-                    stdscr.clrtoeol()
-                else:  # Concat sequence
-                    res = cc.append_char(c)
-                    stdscr.addstr(LN_INFO, 0, res)
-                    stdscr.clrtoeol()
-                    if c in ('~', '?'):
-                        stdscr.addstr(LN_INPUT, 0, PROMPT)
-                        stdscr.clrtoeol()
-                    else:
-                        stdscr.addstr(py, px, '')
-                        if c == '\b':  # TODO: Is there a better way?
-                            if res == '\b':
-                                stdscr.addstr(c)
-                                stdscr.clrtoeol()
-                        else:
-                            stdscr.addstr(c)
-        except KeyboardInterrupt:
-            logger.info('Received keyboard interrupt')
-        finally:
-            logger.info(f'Saving {len(cc.qsos)} QSOs...')
             while cc.has_qsos():
+                logger.info(f'Saving {len(cc.qsos)} QSO(s)...')
                 adi_f.write('\n\n' + adi.dumps({'RECORDS': [cc.pop_qso()]}))
                 adi_f.flush()
-            logger.info('...done')
-    except Exception as exc:  # Print exception info due to curses wrapper removes traceback
-        print(f'{type(exc).__name__}: {exc}', file=sys.stderr)
-        logger.exception(exc)
+        logger.info('...done')
+    except KeyboardInterrupt:
+        logger.info('Received keyboard interrupt')
     finally:
         if adi_f:
             adi_f.close()
@@ -235,8 +110,6 @@ def command_console(stdscr: window, file, own_call, own_loc, own_name, append=Fa
 
 def main():
     import argparse
-    from datetime import datetime
-    logger.info('Starting...')
 
     parser = argparse.ArgumentParser(description='Log Ham Radio QSOs via console',
                                      formatter_class=argparse.RawDescriptionHelpFormatter,
@@ -248,6 +121,10 @@ def main():
     parser.add_argument('file', metavar='ADIF_FILE', nargs='?',
                         default=os.path.expanduser('~/hamcc_log.adi'),
                         help='the file to store the QSOs')
+    parser.add_argument('-q', '--qso', dest='qso', metavar='VALUE', nargs='+', action='append',
+                        help='a QSO string to import instead of running the console (argument can be used repeatedly per QSO)')
+    parser.add_argument('--stdin', dest='stdin', action='store_true',
+                        help='read QSO strings from STDIN instead of running the console')
     parser.add_argument('-c', '--call', dest='own_call', default='',
                         help='your callsign')
     parser.add_argument('-l', '--locator', dest='own_loc', default='',
@@ -272,24 +149,36 @@ def main():
         logger.setLevel(args.log_level)
         logging.getLogger('hamcc').setLevel(args.log_level)
 
-    if os.name == 'nt':
-        os.system("mode con cols=120 lines=25")
-
-    records = []
-    if args.load_qsos:
-        if os.path.isfile(args.file):
-            bak_date = datetime.now().strftime('%Y-%m-%d_%H.%M.%S')
-            phead, ptail = os.path.split(args.file)
-            bak_file = os.path.join(phead, f'{bak_date}_{ptail}')
-            logger.info(f'Creating backup "{bak_file}" from "{args.file}"...')
-            os.rename(args.file, bak_file)
-            doc = adi.load(bak_file)
-            records = doc['RECORDS']
-
-    wrapper(command_console, args.file, args.own_call, args.own_loc, args.own_name,
-            not args.overwrite, args.event, args.exchange, records)
-
-    logger.info('Stopped')
+    if args.qso or args.stdin:
+        stderr_handler = logging.StreamHandler()
+        stderr_handler.setFormatter(logging.Formatter(LOG_FMT))
+        stderr_handler.setLevel(args.log_level)
+        logger.addHandler(stderr_handler)
+        logging.getLogger('hamcc').addHandler(stderr_handler)
+
+        qsos =  sys.stdin if args.stdin else args.qso
+        process_qsos(qsos, args.file, args.own_call, args.own_loc, args.own_name,
+                     not args.overwrite, args.event, args.exchange)
+    else:
+        from datetime import datetime
+        from ._console_ import run_console
+        logger.info('Starting console...')
+
+        records = []
+        if args.load_qsos:
+            if os.path.isfile(args.file):
+                bak_date = datetime.now().strftime('%Y-%m-%d_%H.%M.%S')
+                phead, ptail = os.path.split(args.file)
+                bak_file = os.path.join(phead, f'{bak_date}_{ptail}')
+                logger.info(f'Creating backup "{bak_file}" from "{args.file}"...')
+                os.rename(args.file, bak_file)
+                doc = adi.load(bak_file)
+                records = doc['RECORDS']
+
+        run_console(args.file, args.own_call, args.own_loc, args.own_name,
+                    args.overwrite, args.event, args.exchange, records)
+
+        logger.info('Stopped console')
 
 
 if __name__ == '__main__':