diff --git a/OBSW_TmTcClient.py b/OBSW_TmTcClient.py
index c202aef75bfb098e6f7fb15866b2e6ed396a5d6f..7313cf076b5f9c4b7a71d582c1c004879b6f5ea3 100644
--- a/OBSW_TmTcClient.py
+++ b/OBSW_TmTcClient.py
@@ -51,9 +51,8 @@
@author:
S. Gaisser, J. Meier, R. Mueller
"""
-
import atexit
-import queue
+from collections import deque
import unittest
import logging
@@ -80,7 +79,7 @@ def main():
tmtcPrinter = TmTcPrinter(g.displayMode, g.printToFile, True)
tmListener = 0
communicationInterface = 0
- if g.modeId == "GUIMode":
+ if g.modeId == g.modeList.GUIMode:
backend = TmTcBackend()
backend.start()
GUI = TmTcGUI()
@@ -95,9 +94,9 @@ def main():
tmListener = TmListener(
comInterface=communicationInterface, tmTimeout=g.tmTimeout, tcTimeoutFactor=g.tcSendTimeoutFactor)
tmListener.start()
- if g.modeId == "ListenerMode":
+ if g.modeId == g.modeList.ListenerMode:
print("Listening for packages...")
- elif g.modeId == "SingleCommandMode":
+ elif g.modeId == g.modeList.SingleCommandMode:
pusPacketTuple = commandPreparation()
SenderAndReceiver = SingleCommandSenderReceiver(
comInterface=communicationInterface, tmtcPrinter=tmtcPrinter, tmListener=tmListener,
@@ -105,15 +104,15 @@ def main():
doPrintToFile=g.printToFile)
SenderAndReceiver.sendSingleTcAndReceiveTm()
- elif g.modeId == "ServiceTestMode":
- serviceQueue = queue.Queue()
+ elif g.modeId == g.modeList.ServiceTestMode:
+ serviceQueue = deque()
SenderAndReceiver = SequentialCommandSenderReceiver(
comInterface=communicationInterface, tmtcPrinter=tmtcPrinter, tmTimeout=g.tmTimeout,
tcQueue=serviceTestSelect(g.service, serviceQueue), tcTimeoutFactor=g.tcSendTimeoutFactor,
doPrintToFile=g.printToFile, tmListener=tmListener)
SenderAndReceiver.sendQueueTcAndReceiveTmSequentially()
- elif g.modeId == "SoftwareTestMode":
+ elif g.modeId == g.modeList.SoftwareTestMode:
allTcQueue = createTotalTcQueue()
SenderAndReceiver = SequentialCommandSenderReceiver(
comInterface=communicationInterface, tmtcPrinter=tmtcPrinter, tmTimeout=g.tmTimeout,
@@ -121,7 +120,7 @@ def main():
tmListener=tmListener)
SenderAndReceiver.sendQueueTcAndReceiveTmSequentially()
- elif g.modeId == "OBSWUnitTest":
+ elif g.modeId == g.modeList.UnitTest:
# Set up test suite and run it with runner
# Verbosity specifies detail level
# noinspection PyTypeChecker
diff --git a/comIF/OBSW_ComInterface.py b/comIF/OBSW_ComInterface.py
index 718f4d7edd1e18aaeba9dd49b45a0ee04f797589..5c971409cb82d0c0e3b1f4b8c207517c6971106a 100644
--- a/comIF/OBSW_ComInterface.py
+++ b/comIF/OBSW_ComInterface.py
@@ -8,9 +8,12 @@ Description: Generic Communication Interface. Defines the syntax of the communic
@author: R. Mueller
"""
from abc import abstractmethod
-from typing import TypeVar, Tuple
+from typing import TypeVar, Tuple, Union, List
+from tm.OBSW_PusTm import pusPacketT, pusPacketInfoT
ComIF_T = TypeVar('ComIF_T', bound='CommunicationInterface')
+pusTupleT = Tuple[pusPacketT, pusPacketInfoT]
+packetListT = List[pusPacketT]
class CommunicationInterface:
@@ -38,7 +41,7 @@ class CommunicationInterface:
return packetList
@abstractmethod
- def pollInterface(self, parameters) -> Tuple[bool, list]:
+ def pollInterface(self, parameters) -> Tuple[bool, packetListT]:
"""
Poll the interface and return a list of received packets
:param parameters:
@@ -61,7 +64,7 @@ class CommunicationInterface:
pass
# Receive Telemetry and store a tuple consisting of TM information and TM packet into queue
- def receiveTelemetryAndStoreTuple(self, tmTupleQueue):
+ def receiveTelemetryAndStoreTuple(self, tmTupleQueue) -> Union[None, pusTupleT]:
pass
diff --git a/comIF/OBSW_Serial_ComIF.py b/comIF/OBSW_Serial_ComIF.py
index cf9e4f4e5ef1cff84f55a3b3d0855484f5a4fb2e..5f234eecc866500af2661aa4a18a18294e308095 100644
--- a/comIF/OBSW_Serial_ComIF.py
+++ b/comIF/OBSW_Serial_ComIF.py
@@ -8,8 +8,8 @@
import time
import serial
import logging
-from typing import Tuple
-from comIF.OBSW_ComInterface import CommunicationInterface
+from typing import Tuple, Union
+from comIF.OBSW_ComInterface import CommunicationInterface, pusTupleT, packetListT
from tm.OBSW_TmPacket import PUSTelemetryFactory
@@ -25,17 +25,19 @@ class SerialComIF(CommunicationInterface):
exit()
self.data = bytearray()
self.numberOfPackets = 0
- # self.serial.open()
def sendTelecommand(self, tcPacket, tcPacketInfo=""):
self.tmtcPrinter.printTelecommand(tcPacket, tcPacketInfo)
self.serial.write(tcPacket)
- def receiveTelemetry(self, parameters=0):
+ def receiveTelemetry(self, parameters=0) -> list:
(packetReceived, packetList) = self.pollInterface()
- return packetList
+ if packetReceived:
+ return packetList
+ else:
+ return []
- def pollInterface(self, parameter=0) -> Tuple[bool, list]:
+ def pollInterface(self, parameter=0) -> Tuple[bool, packetListT]:
if self.dataAvailable():
pusDataList, numberOfPackets = self.pollPusPackets()
packetList = []
@@ -97,19 +99,19 @@ class SerialComIF(CommunicationInterface):
self.numberOfPackets = self.numberOfPackets + 1
return endIndex
- # 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)
+ (packetReceived, pusPackets) = self.pollInterface()
+ if packetReceived:
+ for packet in pusPackets:
+ tmQueue.put(packet)
- def receiveTelemetryAndStoreTuple(self, tmTupleQueue):
- packets = self.pollInterface()
- for packet in packets:
- tmInfo = packet.packTmInformation()
- tmTupleQueue.put(packet, tmInfo)
+ def receiveTelemetryAndStoreTuple(self, tmTupleQueue) -> Union[None, pusTupleT]:
+ (packetReceived, pusPackets) = self.pollInterface()
+ if packetReceived:
+ for packet in pusPackets:
+ tmInfo = packet.packTmInformation()
+ tmTupleQueue.put(packet, tmInfo)
+ return tmTupleQueue
+ else:
+ return None
diff --git a/config/OBSW_Config.py b/config/OBSW_Config.py
index 8de036d322454362e52520c5cdb7ee4b22590a68..a6f3246bc77f7f1346a43b1cb1b7fe0745aa8f33 100644
--- a/config/OBSW_Config.py
+++ b/config/OBSW_Config.py
@@ -7,6 +7,7 @@
Global settings for UDP client
"""
import socket
+import enum
"""
@@ -27,15 +28,16 @@ GPS1_SID = bytearray([0x00, 0x00, 0x2f, 0x00])
All global variables, set in main program with arg parser
"""
+
# Mode options, set by args parser
-modeList = {
- 0: "GUIMode",
- 1: "ListenerMode",
- 2: "SingleCommandMode",
- 3: "ServiceTestMode",
- 4: "SoftwareTestMode",
- 5: "OBSWUnitTest"
-}
+class modeList(enum.Enum):
+ GUIMode = 0
+ ListenerMode = 1
+ SingleCommandMode = 2
+ ServiceTestMode = 3
+ SoftwareTestMode = 4
+ UnitTest = 5
+
# General Settings
scriptMode = 1
diff --git a/sendreceive/OBSW_MultipleCommandsSenderReceiver.py b/sendreceive/OBSW_MultipleCommandsSenderReceiver.py
index 1c8becbc68e2fd3c0be8b9adf06f6bc674def1f8..2008cbe2289c44d376a1ebe47ccd61ce7180ba85 100644
--- a/sendreceive/OBSW_MultipleCommandsSenderReceiver.py
+++ b/sendreceive/OBSW_MultipleCommandsSenderReceiver.py
@@ -6,15 +6,16 @@
@brief
Used to send multiple TCs as bursts and listen for replies simultaneously. Used by UnitTester
"""
+import queue
+import time
+import threading
+from typing import Union
+
from config import OBSW_Config as g
from sendreceive.OBSW_SequentialSenderReceiver import SequentialCommandSenderReceiver
from comIF.OBSW_ComInterface import ComIF_T
from utility.OBSW_TmTcPrinter import TmTcPrinterT
from sendreceive.OBSW_TmListener import TmListenerT
-from typing import Union
-import queue
-import time
-import threading
class MultipleCommandSenderReceiver(SequentialCommandSenderReceiver):
@@ -57,11 +58,13 @@ class MultipleCommandSenderReceiver(SequentialCommandSenderReceiver):
self.sendAllQueue()
while not self.allRepliesReceived:
time.sleep(0.5)
+ self.tmListener.retrieveTmInfoQueue()
# self.handleTcResending() Turned off for now, not needed
if self.doPrintToFile:
self.tmtcPrinter.printToFile()
except (KeyboardInterrupt, SystemExit):
- super().handleInterrupt(receiverThread)
+ print("Keyboard Interrupt or System Exit detected")
+ exit()
return self.tcInfoQueue, self.tmInfoQueue
def handleTcResending(self):
@@ -93,9 +96,20 @@ class MultipleCommandSenderReceiver(SequentialCommandSenderReceiver):
else:
time.sleep(self.waitTime)
+ def retrieveTmInfoQueue(self):
+ if self.tmListener.replyEvent.is_set():
+ self.tmInfoQueue = self.tmListener.tmInfoQueue
+ else:
+ print("Multiple Command SenderReceiver: Configuration error, reply event not set in TM listener")
+
def checkForMultipleReplies(self):
super().checkForMultipleReplies()
+ # def checkForMultipleReplies(self):
+ # while not self.allRepliesReceived and self.run_event.is_set():
+ # # listen for duration timeoutInSeconds for replies
+ # self.handleReplyListening()
+
def handleReplyListening(self):
tmReady = self.comInterface.dataAvailable(2.0)
if tmReady:
diff --git a/sendreceive/OBSW_TmListener.py b/sendreceive/OBSW_TmListener.py
index 2f47c904036a6de779944fef89045142f654f4e3..7f2821c1556c32652c5006f1b8333a687ac42e8e 100644
--- a/sendreceive/OBSW_TmListener.py
+++ b/sendreceive/OBSW_TmListener.py
@@ -10,6 +10,7 @@
"""
import time
import threading
+import queue
from typing import TypeVar
from comIF.OBSW_ComInterface import ComIF_T
@@ -25,7 +26,7 @@ class TmListener:
and changing the mode to do special mode operations. The mode operation ends as soon
the modeOpFinished Event is set() !
"""
- def __init__(self, comInterface: ComIF_T, tmTimeout: int, tcTimeoutFactor: int):
+ def __init__(self, comInterface: ComIF_T, tmTimeout: float, tcTimeoutFactor: float):
self.tmTimeout = tmTimeout
self.tcTimeoutFactor = tcTimeoutFactor
self.comInterface = comInterface
@@ -45,6 +46,8 @@ class TmListener:
# self.terminationEvent = threading.Event()
# This Event is set and cleared by the listener to inform the sender objects if a reply has been received
self.replyEvent = threading.Event()
+ # Will be filled for the Unit Test
+ self.tmInfoQueue = queue.Queue()
def start(self):
self.listenerThread.start()
@@ -73,23 +76,26 @@ class TmListener:
:return:
"""
# Listener Mode
- if self.modeId == g.modeList[1]:
+ if self.modeId == g.modeList.ListenerMode:
pass
# Single Command Mode
- elif self.modeId == g.modeList[2]:
+ elif self.modeId == g.modeList.SingleCommandMode:
# Listen for one reply sequence.
if self.checkForOneTelemetrySequence():
# Set reply event, will be cleared by checkForFirstReply()
self.replyEvent.set()
# Sequential Command Mode
- elif self.modeId == g.modeList[3]:
+ elif self.modeId == g.modeList.ServiceTestMode or self.modeId == g.modeList.SoftwareTestMode:
if self.checkForOneTelemetrySequence():
print("TM Listener: Reply sequence received!")
self.replyEvent.set()
+ elif self.modeId == g.modeList.UnitTest:
+ currentPackets = self.comInterface.receiveTelemetryAndStoreTuple()
def checkForOneTelemetrySequence(self) -> bool:
"""
Receive all telemetry for a specified time period.
+ This function prints the telemetry sequence but does not return it.
:return:
"""
tmReady = self.comInterface.dataAvailable(self.tmTimeout * self.tcTimeoutFactor)
diff --git a/tc/OBSW_TcPacker.py b/tc/OBSW_TcPacker.py
index e7f10a492bc6c5e41e714bb442567136b97957ce..cc8dd5090ade2f240dcee9f33c8dab88ef9508e4 100644
--- a/tc/OBSW_TcPacker.py
+++ b/tc/OBSW_TcPacker.py
@@ -20,7 +20,7 @@ from tc.OBSW_TcService8 import packService8TestInto
from tc.OBSW_TcService200 import packModeData, packService200TestInto
import config.OBSW_Config as g
from datetime import datetime
-import queue
+from collections import deque
def serviceTestSelect(service, serviceQueue):
diff --git a/tm/OBSW_PusTm.py b/tm/OBSW_PusTm.py
index 703fe8ec6a2f8e41155259a1b28932167a0b8763..591ff3bbb88b47508843a94697888335628063c4 100644
--- a/tm/OBSW_PusTm.py
+++ b/tm/OBSW_PusTm.py
@@ -1,4 +1,8 @@
from tm.OBSW_PusPacket import OBSWPusPacket
+from typing import TypeVar, Dict
+
+pusPacketT = TypeVar('pusPacketT', bound='PUSTelemetry')
+pusPacketInfoT = Dict[str, any]
class PUSTelemetry(OBSWPusPacket):
@@ -22,7 +26,7 @@ class PUSTelemetry(OBSWPusPacket):
def printTelemetryColumnHeaders(self, array):
super().printPusPacketHeaderColumnHeaders(array)
- def packTmInformation(self):
+ def packTmInformation(self) -> pusPacketInfoT:
tmInformation = {
"service": self.getService(),
"subservice": self.getSubservice(),