OBSW_TmTcClient.py

#!/usr/bin/python3.7
# -*- coding: utf-8 -*-
"""
@file
    OBSW_TmTcClient.py
@date
    01.11.2019
@brief
    This client was developed by KSat for the SOURCE project to test the on-board software.
    It can be used to to send and receive TMTC packages.
@manual
    Manual installation of crcmod might be needed
        1. Install pip if it is not installed yet
        2. Install crcmod and all other reqiored packages:
           Command: pip install crcmod
           or use IDE (interpreter settings -> pip in PyCharm)

    The script can be used by specifying command line parameters. Please run this script with the -h flag
    or without any command line parameters to display options. GUI is work-in-progress
    It might be necessary to set board or PC IP address if using ethernet communication.
    Default values should work normally though. Use higher timeout value (-t Parameter) for STM32

    Example command to test service 17,
    assuming no set client IP (set manually to PC IP Address if necessary) and default board IP 169.254.1.38:
    OBSW_TmTcClient.py -m 3 -s 17
    Example to run Unit Test:
    OBSW_TmTcClient.py -m 5
    Example to test service 17 with HK output and serial communication:
    OBSW_TmTcClient.py -m 3 -s 17 --hk -c 1
    Get command line help:
    OBSW_TmTcClient.py -h

    There are four different Modes:
        0. GUI Mode: Experimental mode, also called if no input parameter are specified
        1. Listener Mode: Only Listen for incoming TM packets
        2. SingleCommandMode: Send Single Command repeatedly until answer is received,
            only listen after that
        3. ServiceTestMode: Send all Telecommands belonging to a certain service
            and scan for replies for each telecommand. Listen after that
        4. SoftwareTestMode: Send all services and perform reply scanning like mode 3.
            Listen after that
        5. Unit Test Mode: Performs a unit test which returns a simple OK or NOT OK. This mode
            has the capability to send TCs in bursts, where applicable

    If there are problems receiving packets with Ethernet Communication,
    use the tool Wireshark to track ethernet communication
    for UDP echo packets (requests and response).
    If the packets appear, there might be a problematic firewall setting.
    Please ensure that python.exe UDP packets are not blocked in advanced firewall settings
    and create a rule to allow packets from port 2008.

@author:
    S. Gaisser, J. Meier, R. Mueller
"""

import atexit
import signal
import queue
import unittest

from test import OBSW_UnitTest
from config import OBSW_Config as g
from config.OBSW_Config import setGlobals
from tc.OBSW_TcPacker import PUSTelecommand, createTotalTcQueue, serviceTestSelect
from sendreceive.OBSW_CommandSenderReceiver import CommandSenderReceiver, connectToBoard
from sendreceive.OBSW_SingleCommandSenderReceiver import SingleCommandSenderReceiver
from sendreceive.OBSW_SequentialSenderReceiver import SequentialCommandSenderReceiver
from utility.OBSW_ArgParser import parseInputArguments
from utility.OBSW_TmTcPrinter import TmtcPrinter
from comIF.OBSW_Ethernet_ComIF import EthernetComIF
from comIF.OBSW_Serial_ComIF import SerialComIF


def main():
    args = parseInputArguments()
    setGlobals(args)
    tmtcPrinter = TmtcPrinter(g.displayMode, g.printToFile, True)
    communicationInterface = setCommunicationInterface(tmtcPrinter)
    atexit.register(communicationInterface.keyboardInterruptHandler)
    if g.modeId == "ListenerMode":
        Receiver = CommandSenderReceiver(communicationInterface, tmtcPrinter, g.tmTimeout, g.tcSendTimeoutFactor,
                                         g.printToFile)
        Receiver.comInterface.performListenerMode()

    elif g.modeId == "SingleCommandMode":
        pusPacketTuple = commandPreparation()
        SenderAndReceiver = SingleCommandSenderReceiver(communicationInterface, tmtcPrinter, pusPacketTuple,
                                                        g.tmTimeout, g.tcSendTimeoutFactor, g.printToFile)
        SenderAndReceiver.sendSingleTcAndReceiveTm()

    elif g.modeId == "ServiceTestMode":
        serviceQueue = queue.Queue()
        SenderAndReceiver = SequentialCommandSenderReceiver(communicationInterface, tmtcPrinter, g.tmTimeout,
                                                            serviceTestSelect(g.service, serviceQueue),
                                                            g.tcSendTimeoutFactor, g.printToFile)
        SenderAndReceiver.sendQueueTcAndReceiveTmSequentially()

    elif g.modeId == "SoftwareTestMode":
        allTcQueue = createTotalTcQueue()
        SenderAndReceiver = SequentialCommandSenderReceiver(communicationInterface, tmtcPrinter, g.tmTimeout,
                                                            allTcQueue, g.tcSendTimeoutFactor, g.printToFile)
        SenderAndReceiver.sendQueueTcAndReceiveTmSequentially()

    elif g.modeId == "OBSWUnitTest":
        if g.comIF == 1:
            communicationInterface.serial.close()
        # Set up test suite and run it with runner
        # Verbosity specifies detail level
        # noinspection PyTypeChecker
        suite = unittest.TestLoader().loadTestsFromModule(OBSW_UnitTest)
        unittest.TextTestRunner(verbosity=2).run(suite)

    else:
        print("Unknown Mode, Configuration error !")
        exit()


# Prepare command for single command testing
def commandPreparation():
    # Direct command which triggers an additional step reply and one completion reply
    # Single Command Testing
    #command = PUSTelecommand(service=17, subservice=1, SSC=21)
    file = bytearray([1, 2, 3, 4, 5])
    command = PUSTelecommand(service=23, subservice=1, SSC=21, data=file)
    command.packCommandTuple()
    return command.packCommandTuple()


def setUpSocket():
    try:
        g.sockReceive.bind(g.recAddress)
        g.sockReceive.setblocking(False)
    except OSError:
        print("Error setting up sockets.")
        print("Socket Receive Address: " + str(g.recAddress))
        exit()
    except TypeError:
        print("Invalid Receive Address")
        exit()


def setCommunicationInterface(tmtcPrinter):
    if g.comIF == 0:
        setUpSocket()
        connectToBoard()
        communicationInterface = EthernetComIF(tmtcPrinter, g.tmTimeout, g.tcSendTimeoutFactor,
                                               g.sockSend, g.sockReceive, g.sendAddress)
    else:
        comPort = g.comPort
        baudRate = 115200
        g.serialTimeout = 0.05
        communicationInterface = SerialComIF(tmtcPrinter, comPort, baudRate, g.serialTimeout)
    return communicationInterface


class GracefulKiller:
    kill_now = False

    def __init__(self):
        signal.signal(signal.SIGINT, self.exit_gracefully)
        signal.signal(signal.SIGTERM, self.exit_gracefully)

    def exit_gracefully(self):
        self.kill_now = True
        print("I was killed")


if __name__ == "__main__":
    main()