diff --git a/OBSW_TmTcClient.py b/OBSW_TmTcClient.py
index 4e8d08b3b292edced6b9f8075ee8727a8580c939..95dc9de4889455d673178fcc8c88a4de35d58565 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 050bcc3b486a4b1dd81d4996ef07091bfa448b31..1572b784ac58499cde7e0ca9277f01c16531ffb0 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 de7db0a5ad4793bf1c080d30798bcda4db54604d..a80948d2ce13a593d0fa3135b725e25a0f1d46fe 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 4729b41eee005c3ff52e65afbb9f1463ad5c10c5..424c3ead41894250854dc5c4e0896ca30b395803 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 4288b6d7d949a488fda98272359a61c5e684d67a..edbd0ab81be3e30b607fc2ea5a3a71ca7e23c4e4 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 d686b05b8ea6a494799daf9167d535854ae3f739..9d977590d03a4f6b308aca7fb9d5c2f524180c19 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 0000000000000000000000000000000000000000..18e64dd7965d2d3e6bf2db8a305e143e94e40b1d
--- /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 123f03c1eeb04768320b09024199e85d47078f9a..02ae70d4b84bcefae6f28966258547cd1542d949 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 5299a10d3916c45a87405c823289cc23ac46fe0c..34d28fcbae0b45fbaba862c3e0331953f39f45b0 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 be89bd250f115bf156ef5d3f67d78f7dcf211452..01011029d21e133f63c73c082ac7db16e9d8b545 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 80c485064fea75d292c0238adf22d3e6e6f03a7f..4c7f519a6f2a5c29072f3ca6605c31fbedacaf23 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 489c20ef59f6898261ffb7c4e1ce4282d229982a..e7970cef35b3981dfaed4e248c7e7f24f5b621f6 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