tcpip-to-serial-agent.py

#!/usr/bin/env python3
import argparse
import enum
import threading
import signal
import os
import time

from agent.log import get_console_logger
from agent.tcp_handler import TcpHandler
from agent.conf import JSON_CONF_NAME, DEFAULT_BAUDRATE, DEFAULT_MAX_DEQUE_SLOTS
from agent.json import check_json_file
from agent.serial_handler import SerialHandler, determine_com_port, find_com_port_from_hint
from agent.buffer import init_reception_buffer, set_max_slots
from agent.exit_handler import service_shutdown, ServiceExit, self_destruct_signalled


LOGGER = get_console_logger()


class ConfigKeys(enum.IntEnum):
    MAX_BUF_SLOTS = 0
    BAUDRATE = 1
    SERIAL_PORT = 2


def main():
    print('-- Starting TCP/IP to Serial Agent for the iOBC --')
    error_setup()
    try:
        start_application()
    except ServiceExit:
        LOGGER.info('Closing TCP/IP to Serial Agent')


def start_application():
    cfg_dict = args_setup()
    set_max_slots(max_slots=cfg_dict[ConfigKeys.MAX_BUF_SLOTS])
    init_reception_buffer()

    ser_handler = SerialHandler(
        serial_port=cfg_dict[ConfigKeys.SERIAL_PORT], baud_rate=cfg_dict[ConfigKeys.BAUDRATE]
    )
    tcp_handler = TcpHandler()

    serial_polling_thread = None
    tcp_thread = None
    try:
        serial_polling_thread = threading.Thread(target=ser_handler.run)
        tcp_thread = threading.Thread(target=tcp_handler.run)
        serial_polling_thread.start()
        tcp_thread.start()
        # Keep the main thread running, otherwise signals are ignored.
        while True:
            if self_destruct_signalled():
                raise ServiceExit
            time.sleep(0.5)
    except ServiceExit:
        # Graceful exit
        ser_handler.stop()
        tcp_handler.stop()
        serial_polling_thread.join()
        tcp_thread.join()
    finally:
        LOGGER.warning('TCP/IP to Serial Agent exited unexpectedly')


def error_setup():
    # atexit.register(keyboard_interrupt_handler)
    # Register the signal handlers
    signal.signal(signal.SIGTERM, service_shutdown)
    signal.signal(signal.SIGINT, service_shutdown)


def args_setup() -> dict:
    cfg_dict = dict()
    arg_parser = argparse.ArgumentParser(description='TCP/IP to Serial Bridge for the iOBC')
    arg_parser.add_argument('-p', '--port', dest='port', type=str, help='Serial port')
    arg_parser.add_argument('-b', '--baudrate', dest='baud', type=int, help='Serial baudrate')
    arg_parser.add_argument('--hint', dest='hint', type=str, help='Hint for serial port name')
    arg_parser.add_argument(
        '--buf_slots', dest='buf_slots', type=int, help='Buffer slots for TM reception queue'
    )
    arg_parser.add_argument(
        '-d', '--delete', dest='del_conf', action='store_true', help='Delete configuration JSON'
    )
    args = arg_parser.parse_args()

    check_json_file(JSON_CONF_NAME)
    serial_port = ''
    if args.del_conf:
        LOGGER.info(f'Deleting JSON configuration file {JSON_CONF_NAME}')
        os.remove(JSON_CONF_NAME)
    if args.port is not None:
        LOGGER.info(f'Serial port {args.port} was specified')
        serial_port = args.port
    else:
        com_port_found = False
        if args.hint is not None:
            com_port_found, serial_port = find_com_port_from_hint(hint=args.hint)
        if not com_port_found:
            serial_port = determine_com_port(json_cfg_path=JSON_CONF_NAME)
    if args.baud is not None:
        baudrate = args.baud
    else:
        baudrate = DEFAULT_BAUDRATE
    if args.buf_slots is not None:
        max_buf_slots = args.buf_slots
    else:
        max_buf_slots = DEFAULT_MAX_DEQUE_SLOTS

    cfg_dict[ConfigKeys.SERIAL_PORT] = serial_port
    cfg_dict[ConfigKeys.MAX_BUF_SLOTS] = max_buf_slots
    cfg_dict[ConfigKeys.BAUDRATE] = baudrate
    return cfg_dict


if __name__ == "__main__":
    main()