From d291330e3001f9245b473a1fa7899d66d2b3937c Mon Sep 17 00:00:00 2001
From: "Robin.Mueller" <robin.mueller.m@gmail.com>
Date: Sun, 8 Mar 2020 21:09:43 +0100
Subject: [PATCH] Preparations for GUIWq
---
OBSW_TmTcClient.py | 46 ++++++++-----
comIF/OBSW_ComInterface.py | 3 +
comIF/OBSW_Ethernet_ComIF.py | 1 -
comIF/OBSW_Serial_ComIF.py | 16 +++--
sendreceive/OBSW_CommandSenderReceiver.py | 27 ++++++--
.../OBSW_SingleCommandSenderReceiver.py | 16 +++--
sendreceive/OBSW_TmListener.py | 69 +++++++++++++++++++
tc/OBSW_TcPacket.py | 4 +-
test/OBSW_PusServiceTest.py | 2 +-
test/OBSW_UnitTest.py | 9 +--
utility/OBSW_ArgParser.py | 4 ++
utility/OBSW_TmTcPrinter.py | 6 +-
12 files changed, 159 insertions(+), 44 deletions(-)
create mode 100644 sendreceive/OBSW_TmListener.py
diff --git a/OBSW_TmTcClient.py b/OBSW_TmTcClient.py
index 4e8d08b..95dc9de 100644
--- a/OBSW_TmTcClient.py
+++ b/OBSW_TmTcClient.py
@@ -1,4 +1,4 @@
-#!/usr/bin/python3.7
+#!/usr/bin/python3.8
# -*- coding: utf-8 -*-
"""
@file
@@ -57,28 +57,29 @@ import atexit
import queue
import unittest
import logging
-from multiprocessing import Process
from test import OBSW_PusServiceTest
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
from sendreceive.OBSW_SingleCommandSenderReceiver import SingleCommandSenderReceiver
from sendreceive.OBSW_SequentialSenderReceiver import SequentialCommandSenderReceiver
+from sendreceive.OBSW_TmListener import TmListener
from utility.OBSW_ArgParser import parseInputArguments
-from utility.OBSW_TmTcPrinter import TmtcPrinter
+from utility.OBSW_TmTcPrinter import TmTcPrinter, TmTcPrinterT
from comIF.OBSW_Ethernet_ComIF import EthernetComIF
from comIF.OBSW_Serial_ComIF import SerialComIF
from gui.OBSW_TmtcGUI import TmTcGUI
from gui.OBSW_BackendTest import TmTcBackend
from utility.OBSW_ExitHandler import keyboardInterruptHandler
+from comIF.OBSW_ComInterface import ComIF_T
def main():
args = parseInputArguments()
setGlobals(args)
- tmtcPrinter = TmtcPrinter(g.displayMode, g.printToFile, True)
+ tmtcPrinter = TmTcPrinter(g.displayMode, g.printToFile, True)
+ tmListener = 0
communicationInterface = 0
if g.modeId == "GUIMode":
backend = TmTcBackend()
@@ -92,28 +93,30 @@ def main():
else:
communicationInterface = setCommunicationInterface(tmtcPrinter)
atexit.register(keyboardInterruptHandler, comInterface=communicationInterface)
+ tmListener = TmListener(communicationInterface)
+ tmListener.start()
if g.modeId == "ListenerMode":
- Receiver = CommandSenderReceiver(communicationInterface, tmtcPrinter, g.tmTimeout, g.tcSendTimeoutFactor,
- g.printToFile)
- Receiver.comInterface.performListenerMode()
-
+ pass
elif g.modeId == "SingleCommandMode":
pusPacketTuple = commandPreparation()
- SenderAndReceiver = SingleCommandSenderReceiver(communicationInterface, tmtcPrinter, pusPacketTuple,
- g.tmTimeout, g.tcSendTimeoutFactor, g.printToFile)
+ SenderAndReceiver = SingleCommandSenderReceiver(
+ comInterface=communicationInterface, tmtcPrinter=tmtcPrinter, pusPacketTuple=pusPacketTuple,
+ tmTimeout=g.tmTimeout, tcTimeoutFactor=g.tcSendTimeoutFactor, doPrintToFile=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 = SequentialCommandSenderReceiver(
+ comInterface=communicationInterface, tmtcPrinter=tmtcPrinter, tmTimeout=g.tmTimeout,
+ tcQueue=serviceTestSelect(g.service, serviceQueue), tcTimeoutFactor=g.tcSendTimeoutFactor,
+ doPrintToFile=g.printToFile)
SenderAndReceiver.sendQueueTcAndReceiveTmSequentially()
elif g.modeId == "SoftwareTestMode":
allTcQueue = createTotalTcQueue()
- SenderAndReceiver = SequentialCommandSenderReceiver(communicationInterface, tmtcPrinter, g.tmTimeout,
- allTcQueue, g.tcSendTimeoutFactor, g.printToFile)
+ SenderAndReceiver = SequentialCommandSenderReceiver(
+ comInterface=communicationInterface, tmtcPrinter=tmtcPrinter, tmTimeout=g.tmTimeout,
+ tcQueue=allTcQueue, tcTimeoutFactor=g.tcSendTimeoutFactor, doPrintToFile=g.printToFile)
SenderAndReceiver.sendQueueTcAndReceiveTmSequentially()
elif g.modeId == "OBSWUnitTest":
@@ -133,14 +136,19 @@ 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)
- command.print()
+ # command.print()
# file = bytearray([1, 2, 3, 4, 5])
# command = PUSTelecommand(service=23, subservice=1, SSC=21, data=file)
# command.packCommandTuple()
return command.packCommandTuple()
-def setCommunicationInterface(tmtcPrinter):
+def setCommunicationInterface(tmtcPrinter: TmTcPrinterT) -> ComIF_T:
+ """
+ Return the desired communication interface object
+ :param tmtcPrinter: TmTcPrinter object.
+ :return: CommunicationInterface object
+ """
try:
if g.comIF == 0 and g.modeId != "OBSWUnitTest":
communicationInterface = EthernetComIF(tmtcPrinter, g.tmTimeout, g.tcSendTimeoutFactor,
@@ -156,7 +164,7 @@ def setCommunicationInterface(tmtcPrinter):
except (IOError, OSError):
print("Error setting up communication interface")
logging.exception("Error: ")
- return exit()
+ exit()
if __name__ == "__main__":
diff --git a/comIF/OBSW_ComInterface.py b/comIF/OBSW_ComInterface.py
index 050bcc3..1572b78 100644
--- a/comIF/OBSW_ComInterface.py
+++ b/comIF/OBSW_ComInterface.py
@@ -8,6 +8,9 @@ Description: Generic Communication Interface. Defines the syntax of the communic
@author: R. Mueller
"""
from abc import abstractmethod
+from typing import TypeVar
+
+ComIF_T = TypeVar('ComIF_T', bound='CommunicationInterface')
class CommunicationInterface:
diff --git a/comIF/OBSW_Ethernet_ComIF.py b/comIF/OBSW_Ethernet_ComIF.py
index de7db0a..a80948d 100644
--- a/comIF/OBSW_Ethernet_ComIF.py
+++ b/comIF/OBSW_Ethernet_ComIF.py
@@ -7,7 +7,6 @@ Description: Ethernet Communication Interface
@author: R. Mueller
"""
import select
-import logging
from comIF.OBSW_ComInterface import CommunicationInterface
from tm.OBSW_TmPacket import PUSTelemetryFactory
diff --git a/comIF/OBSW_Serial_ComIF.py b/comIF/OBSW_Serial_ComIF.py
index 4729b41..424c3ea 100644
--- a/comIF/OBSW_Serial_ComIF.py
+++ b/comIF/OBSW_Serial_ComIF.py
@@ -89,17 +89,19 @@ class SerialComIF(CommunicationInterface):
self.numberOfPackets = self.numberOfPackets + 1
return endIndex
- def performListenerMode(self):
- print("Listening for packages ...")
- while True:
- self.pollInterface()
+ # This is basically a permanent loop.. and shouldnt be here. TmListener Class will do this.
+ # def performListenerMode(self):
+ # print("Listening for packages ...")
+ # while True:
+ # self.pollInterface()
def receiveTelemetryAndStoreIntoQueue(self, tmQueue):
packet = self.pollInterface()
tmQueue.put(packet)
def receiveTelemetryAndStoreTuple(self, tmTupleQueue):
- packet = self.pollInterface()
- tmInfo = packet.packTmInformation()
- tmTupleQueue.put(packet, tmInfo)
+ packets = self.pollInterface()
+ for packet in packets:
+ tmInfo = packet.packTmInformation()
+ tmTupleQueue.put(packet, tmInfo)
diff --git a/sendreceive/OBSW_CommandSenderReceiver.py b/sendreceive/OBSW_CommandSenderReceiver.py
index 4288b6d..edbd0ab 100644
--- a/sendreceive/OBSW_CommandSenderReceiver.py
+++ b/sendreceive/OBSW_CommandSenderReceiver.py
@@ -8,22 +8,41 @@ Set up the UDP client as specified in the header comment and use the unit testin
A separate thread is used to listen for replies and send a new telecommand
if the first reply has not been received.
-This is still experimental.
@author: R. Mueller
"""
import time
from config import OBSW_Config as g
+from comIF.OBSW_ComInterface import CommunicationInterface, ComIF_T
+from utility.OBSW_TmTcPrinter import TmTcPrinterT, TmTcPrinter
# Generic TMTC SendReceive class which is implemented
# by specific implementations (e.g. SingleCommandSenderReceiver)
class CommandSenderReceiver:
- def __init__(self, comInterface, tmtcPrinter, tmTimeout, tcSendTimeoutFactor, doPrintToFile):
+ """
+ This is the generic CommandSenderReceiver object. All TMTC objects inherit this object
+ """
+ def __init__(self, comInterface: ComIF_T, tmtcPrinter: TmTcPrinterT, tmTimeout: int,
+ tcSendTimeoutFactor: int, doPrintToFile: bool):
+ """
+ :param comInterface: CommunicationInterface object. Instantiate the desired one and pass it here
+ :param tmtcPrinter: TmTcPrinter object. Instantiate it and pass it here.
+ :param tmTimeout: TM Timeout. After each sent telecommand, listen for TMs for this time period
+ :param tcSendTimeoutFactor: If TM is not received, resend TC after tmTimeout * tcSendTimeoutFactor
+ :param doPrintToFile: Specify whether output is also printed to a file
+ """
self.tmTimeout = tmTimeout
- self.comInterface = comInterface
- self.tmtcPrinter = tmtcPrinter
+ if isinstance(comInterface, CommunicationInterface):
+ self.comInterface = comInterface
+ else:
+ raise TypeError("Invalid communication interface type!")
+
+ if isinstance(tmtcPrinter, TmTcPrinter):
+ self.tmtcPrinter = tmtcPrinter
+ else:
+ raise TypeError("Invalid TMTC Printer type!")
self.replyReceived = False
self.tmReady = False
diff --git a/sendreceive/OBSW_SingleCommandSenderReceiver.py b/sendreceive/OBSW_SingleCommandSenderReceiver.py
index d686b05..9d97759 100644
--- a/sendreceive/OBSW_SingleCommandSenderReceiver.py
+++ b/sendreceive/OBSW_SingleCommandSenderReceiver.py
@@ -1,5 +1,4 @@
-#!/usr/bin/python3.7
-# -*- coding: utf-8 -*-
+#!/usr/bin/python3.8
"""
@file
OBSW_Config.py
@@ -9,14 +8,23 @@
Used to send single tcs and listen for replies after that
"""
from sendreceive.OBSW_CommandSenderReceiver import CommandSenderReceiver
+from comIF.OBSW_ComInterface import ComIF_T
import threading
import time
# Specific implementation of CommandSenderReceiver to send a single telecommand
class SingleCommandSenderReceiver(CommandSenderReceiver):
- def __init__(self, comInterface, displayMode, pusPacketTuple, tmTimeout, tcTimeoutFactor, doPrintToFile):
- super().__init__(comInterface, displayMode, tmTimeout, tcTimeoutFactor, doPrintToFile)
+ """ This object can be used by instantiating it and calling sendSingleTcAndReceiveTm() """
+ def __init__(self, comInterface: ComIF_T, tmtcPrinter, pusPacketTuple, tmTimeout, tcTimeoutFactor, doPrintToFile):
+ """
+ :param comInterface: CommunicationInterface object, passed on to CommandSenderReceiver
+ :param tmtcPrinter: TmTcPrinter object, passed on to CommandSenderReceiver
+ :param tmTimeout: TM Timeout. After each sent telecommand, listen for TMs for this time period
+ :param tcTimeoutFactor: If TM is not received, resend TC after tmTimeout * tcSendTimeoutFactor
+ :param doPrintToFile: Specify whether output is also printed to a file
+ """
+ super().__init__(comInterface, tmtcPrinter, tmTimeout, tcTimeoutFactor, doPrintToFile)
self.pusPacketTuple = pusPacketTuple
try:
(self.pusPacketInfo, self.pusPacket) = self.pusPacketTuple
diff --git a/sendreceive/OBSW_TmListener.py b/sendreceive/OBSW_TmListener.py
new file mode 100644
index 0000000..18e64dd
--- /dev/null
+++ b/sendreceive/OBSW_TmListener.py
@@ -0,0 +1,69 @@
+"""
+@file
+ OBSW_TmListener.py
+@date
+ 01.11.2019
+@brief
+ Separate class to listen to telecommands.
+ This will enable to run the listener in a separate thread later.
+ This Listener will propably have some kind of mode. In default configuration, it will just listen for packets
+"""
+import time
+import threading
+
+from comIF.OBSW_ComInterface import ComIF_T
+import config.OBSW_Config as g
+
+
+class TmListener:
+ def __init__(self, comInterface: ComIF_T):
+ self.tmTimeout = g.tmTimeout
+ self.comInterface = comInterface
+ # this will be the default mode (listener mode)
+ self.mode = 0
+ self.active = True
+ self.listenerThread = threading.Thread(target=self.performOperation())
+ self.modeChangeEvent = threading.Event()
+
+ def start(self):
+ self.listenerThread.start()
+
+ def performOperation(self):
+ while self.active:
+ if self.mode == 0:
+ self.defaultOperation()
+
+ def defaultOperation(self):
+ while not self.modeChangeEvent.is_set():
+ self.comInterface.pollInterface()
+
+ def checkForOneTelemetrySequence(self):
+ tmReady = self.comInterface.dataAvailable(g.tmTimeout * g.tcSendTimeoutFactor)
+ if tmReady is False:
+ return False
+ else:
+ self.comInterface.receiveTelemetry()
+ start_time = time.time()
+ elapsed_time = 0
+ while elapsed_time < self.tmTimeout:
+ tmReady = self.comInterface.dataAvailable(1.0)
+ if tmReady:
+ self.comInterface.receiveTelemetry()
+ elapsed_time = time.time() - start_time
+ # the timeout value can be set by special TC queue entries if packet handling takes longer,
+ # but it is reset here to the global value
+ if self.tmTimeout is not g.tmTimeout:
+ self.tmTimeout = g.tmTimeout
+ return True
+
+ # checks for replies. if no reply is received, send telecommand again
+ def checkForFirstReply(self):
+ success = self.checkForOneTelemetrySequence()
+ if success:
+ self.replyReceived = True
+ else:
+ if len(self.pusPacket) == 0:
+ print("Command Sender Receiver: No command has been sent yet")
+ else:
+ print("Command Sender Receiver: Sending command again")
+ self.comInterface.sendTelecommand(self.pusPacket, self.pusPacketInfo)
\ No newline at end of file
diff --git a/tc/OBSW_TcPacket.py b/tc/OBSW_TcPacket.py
index 123f03c..02ae70d 100644
--- a/tc/OBSW_TcPacket.py
+++ b/tc/OBSW_TcPacket.py
@@ -32,9 +32,9 @@ class PUSTelecommand:
try:
length = 4 + len(self.data) + 1
return length
- except TypeError as e:
+ except TypeError:
print("OBSW_TcPacket: Invalid type of data")
- logging.exception("Error: ")
+ logging.exception("Error")
exit()
def pack(self):
diff --git a/test/OBSW_PusServiceTest.py b/test/OBSW_PusServiceTest.py
index 5299a10..34d28fc 100644
--- a/test/OBSW_PusServiceTest.py
+++ b/test/OBSW_PusServiceTest.py
@@ -115,4 +115,4 @@ class TestService17(TestService):
@classmethod
def tearDownClass(cls):
print("Testing Service 17 finished")
- super().tearDownClass()
\ No newline at end of file
+ super().tearDownClass()
diff --git a/test/OBSW_UnitTest.py b/test/OBSW_UnitTest.py
index be89bd2..0101102 100644
--- a/test/OBSW_UnitTest.py
+++ b/test/OBSW_UnitTest.py
@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
"""
Program: OBSW_UnitTest.py
Date: 01.11.2019
@@ -12,7 +11,9 @@ TestService is the template method, analyseTmInfo and analyseTcInfo are implemen
specific services or devices by setting up assertionDict entries which count received packets
based on subservice or entry values. There is a default implementation provided for analyseTcInfo.
The log files generated by individual service test or the software test mode when providing
-the -p flag are very useful for extending the unit tester
+the -p flag are very useful for extending the unit tester.
+To check the dictionary keys of the telecommand and telemetry information queue entries,
+go look up the packTmInformation() and packTcInformation() implementations in the TM and TC folder.
Example Service 17:
TC[17,1] and TC[17,128] are sent, TM[1,1] and TM[1,7] (TC verification) are expected twice, TM[17,2] (ping reply)
@@ -31,7 +32,7 @@ import queue
from tc.OBSW_TcPacker import packDummyDeviceTestInto
from abc import abstractmethod
from sendreceive.OBSW_MultipleCommandsSenderReceiver import MultipleCommandSenderReceiver
-from utility.OBSW_TmTcPrinter import TmtcPrinter
+from utility.OBSW_TmTcPrinter import TmTcPrinter
from comIF.OBSW_Ethernet_ComIF import EthernetComIF
from comIF.OBSW_Serial_ComIF import SerialComIF
from config import OBSW_Config as g
@@ -61,7 +62,7 @@ class TestService(unittest.TestCase):
cls.testQueue = queue.Queue()
cls.tcTimeoutFactor = g.tcSendTimeoutFactor
cls.printFile = g.printToFile
- cls.tmtcPrinter = TmtcPrinter(cls.displayMode, cls.printFile, True)
+ cls.tmtcPrinter = TmTcPrinter(cls.displayMode, cls.printFile, True)
if g.comIF == 0:
cls.communicationInterface = EthernetComIF(cls.tmtcPrinter, g.tmTimeout, g.tcSendTimeoutFactor,
g.sockSend, g.sockReceive, g.sendAddress)
diff --git a/utility/OBSW_ArgParser.py b/utility/OBSW_ArgParser.py
index 80c4850..4c7f519 100644
--- a/utility/OBSW_ArgParser.py
+++ b/utility/OBSW_ArgParser.py
@@ -13,6 +13,10 @@ import sys
def parseInputArguments():
+ """
+ Parses all input arguments
+ :return: Input arguments
+ """
argParser = argparse.ArgumentParser(description="TMTC Client Command Line Interface")
argParser.add_argument('-m', '--mode', type=int, help='Target Mode. Default is 1(Listener Mode), '
diff --git a/utility/OBSW_TmTcPrinter.py b/utility/OBSW_TmTcPrinter.py
index 489c20e..e7970ce 100644
--- a/utility/OBSW_TmTcPrinter.py
+++ b/utility/OBSW_TmTcPrinter.py
@@ -9,11 +9,13 @@
Class that performs all printing functionalities
"""
import os
-
+from typing import TypeVar
from config import OBSW_Config as g
+TmTcPrinterT = TypeVar('TmTcPrinterT', bound='TmTcPrinter')
+
-class TmtcPrinter:
+class TmTcPrinter:
def __init__(self, displayMode="long", doPrintToFile=False, printTc=True):
self.printBuffer = ""
# global print buffer which will be useful to print something to file
--
GitLab