test.py

import aprslib
import asyncio
import signal
import logging
from aioax25 import kiss as kiss
from aioax25.aprs import APRSInterface
from aioax25 import interface

import sys

LOG = logging.getLogger(__name__)


LOG_LEVELS = {
    "CRITICAL": logging.CRITICAL,
    "ERROR": logging.ERROR,
    "WARNING": logging.WARNING,
    "INFO": logging.INFO,
    "DEBUG": logging.DEBUG,
}


def setup_logging(loglevel):
    log_level = LOG_LEVELS[loglevel]
    LOG.setLevel(log_level)
    log_format = "[%(asctime)s][%(threadName)-12s][%(levelname)-5.5s] [%(pathname)s.%(funcName)s:%(lineno)d]" " %(message)s"
    date_format = "%m/%d/%Y %I:%M:%S %p"
    log_formatter = logging.Formatter(fmt=log_format, datefmt=date_format)
    sh = logging.StreamHandler(sys.stdout)
    sh.setFormatter(log_formatter)
    LOG.addHandler(sh)

    kiss_logger = logging.getLogger("kiss.classes")
    kiss_logger.setLevel(log_level)
    kiss_logger.addHandler(sh)

    ax_logger = logging.getLogger("aioax25.frame")
    ax_logger.setLevel(log_level)
    ax_logger.addHandler(sh)


def p(interface, frame, match):
    LOG.debug("Got an APRS frame")
    LOG.debug(frame.header)
    LOG.debug(frame.payload)
    payload = str(frame.payload.decode())
    msg = "{}:{}".format(str(frame.header), payload)

    aprs_msg = aprslib.parse(msg)
    LOG.debug(aprs_msg)


def main():

    loop = asyncio.get_event_loop()

    setup_logging("DEBUG")
    kissdev = kiss.TCPKISSDevice("192.168.1.7", 8001, log=LOG,
                                 loop=loop)
    kissdev.open()

    kissport0 = kissdev[0]

    ax25int = interface.AX25Interface(
        kissport=kissport0,
        loop=loop,
        log=LOG
    )

    aprsint = APRSInterface(
        ax25int=ax25int,
        mycall='WB4BOR',
        log=LOG
    )

    aprsint.bind(callback=p, callsign=".*", regex=True)

    def _shutdown():
        LOG.debug("shutdown signal")
        kissdev.close()
        loop.call_later(5, loop.stop)

    for sig in (signal.SIGINT, signal.SIGTERM, signal.SIGQUIT):
        loop.add_signal_handler(sig, _shutdown)

    LOG.debug("Main loop")
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        pass


if __name__ == "__main__":
    main()