diff --git a/OBSW_TmTcClient.py b/OBSW_TmTcClient.py
index a930fc4db8c476c493a2b444ed0d1fc60fcc5624..522d6d93230b559b9922aa1628a6642e40499904 100644
--- a/OBSW_TmTcClient.py
+++ b/OBSW_TmTcClient.py
@@ -157,13 +157,15 @@ def setCommunicationInterface(tmtcPrinter: TmTcPrinterT) -> ComIF_T:
communicationInterface = None
try:
if g.comIF == 0 and g.modeId != g.modeList.UnitTest:
- communicationInterface = EthernetComIF(tmtcPrinter, g.tmTimeout, g.tcSendTimeoutFactor,
- g.sockSend, g.sockReceive, g.sendAddress)
+ communicationInterface = EthernetComIF(
+ tmtcPrinter=tmtcPrinter, tmTimeout=g.tmTimeout, tcTimeoutFactor=g.tcSendTimeoutFactor,
+ sendAddress=g.sendAddress, receiveAddress=g.recAddress)
elif g.modeId != g.modeList.UnitTest:
comPort = g.comPort
baudRate = 115200
g.serialTimeout = 0.05
- communicationInterface = SerialComIF(tmtcPrinter, comPort, baudRate, g.serialTimeout)
+ communicationInterface = SerialComIF(tmtcPrinter=tmtcPrinter, comPort=comPort, baudRate=baudRate,
+ serialTimeout=g.serialTimeout)
else:
print("Could not set up communication interface !")
exit()
diff --git a/config/OBSW_Config.py b/config/OBSW_Config.py
index 3060081c817e31d0674de14e9489728d952beb7b..d4f1ca66dae854425a44f04958b09f084be1bb58 100644
--- a/config/OBSW_Config.py
+++ b/config/OBSW_Config.py
@@ -6,15 +6,26 @@
@brief
Global settings for UDP client
"""
-import socket
import enum
from typing import Tuple, TypeVar
-# Contains IP address as string and port as int
+"""
+Global type definitions
+"""
ethernetAddressT = Tuple[str, int]
socketType = TypeVar['socketType']
+# Mode options, set by args parser
+class modeList(enum.Enum):
+ GUIMode = 0
+ ListenerMode = 1
+ SingleCommandMode = 2
+ ServiceTestMode = 3
+ SoftwareTestMode = 4
+ UnitTest = 5
+
+
"""
Mission/Device specific information.
"""
@@ -34,16 +45,6 @@ All global variables, set in main program with arg parser
"""
-# Mode options, set by args parser
-class modeList(enum.Enum):
- GUIMode = 0
- ListenerMode = 1
- SingleCommandMode = 2
- ServiceTestMode = 3
- SoftwareTestMode = 4
- UnitTest = 5
-
-
# General Settings
scriptMode = 1
modeId = 0
@@ -72,7 +73,7 @@ printRawTmData = False
# noinspection PyUnusedLocal
def setGlobals(args):
global recAddress, sendAddress, scriptMode, modeId, service, displayMode, comIF, comPort, serialTimeout
- global tmTimeout, tcSendTimeoutFactor, sockSend, sockReceive, printToFile, printHkData, printRawTmData
+ global tmTimeout, tcSendTimeoutFactor, printToFile, printHkData, printRawTmData
if args.mode == 0:
print("GUI mode not implemented yet !")
if args.shortDisplayMode:
@@ -100,8 +101,8 @@ def setGlobals(args):
recAddress = recAddressToSet
sendAddress = sendAddressToSet
comIF = args.comIF
- sockReceive = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
- sockSend = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+ # sockReceive = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+ # sockSend = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
modeId = modeId
comPort = args.COM
printHkData = args.hk
@@ -110,9 +111,3 @@ def setGlobals(args):
displayMode = displayMode
service = service
printToFile = args.printFile
-
-
-def setGlobalsUnitTest():
- global sockReceive
- sockReceive = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-
diff --git a/sendreceive/OBSW_CommandSenderReceiver.py b/sendreceive/OBSW_CommandSenderReceiver.py
index e50981c3df412b45f021908eb6c25386fe639646..8695b8bccc9e1db8a436b54dc3945191f0c1c889 100644
--- a/sendreceive/OBSW_CommandSenderReceiver.py
+++ b/sendreceive/OBSW_CommandSenderReceiver.py
@@ -34,17 +34,17 @@ class CommandSenderReceiver:
self.tmTimeout = tmTimeout
if isinstance(comInterface, CommunicationInterface):
- self.comInterface = comInterface
+ self._comInterface = comInterface
else:
raise TypeError("Invalid communication interface type!")
if isinstance(tmtcPrinter, TmTcPrinter):
- self.tmtcPrinter = tmtcPrinter
+ self._tmtcPrinter = tmtcPrinter
else:
raise TypeError("Invalid TMTC Printer type!")
if isinstance(tmListener, TmListener):
- self.tmListener = tmListener
+ self._tmListener = tmListener
else:
raise TypeError("Invalid TM Listener type!")
@@ -61,8 +61,8 @@ class CommandSenderReceiver:
self.queueEntryIsTelecommand = True
# needed to store last actual tc packet from queue
- self.lastTc = []
- self.lastTcInfo = []
+ self.lastTc = bytearray()
+ self.lastTcInfo = dict()
# ignore periodic packets for timeout when checking a sequence
self.isPacketToIgnore = False
@@ -71,9 +71,9 @@ class CommandSenderReceiver:
Checks for replies. If no reply is received, send telecommand again in checkForTimeout()
:return: None
"""
- if self.tmListener.replyEvent.is_set():
+ if self._tmListener.replyEvent.is_set():
self.replyReceived = True
- self.tmListener.replyEvent.clear()
+ self._tmListener.replyEvent.clear()
else:
if len(self.pusPacket) == 0:
print("Command Sender Receiver: No command has been sent yet")
@@ -90,11 +90,11 @@ class CommandSenderReceiver:
time.sleep(waitTime)
elif self.pusPacketInfo == "print":
printString = self.pusPacket
- self.tmtcPrinter.printString(printString)
+ self._tmtcPrinter.printString(printString)
elif self.pusPacketInfo == "export":
exportName = self.pusPacket
if self.doPrintToFile:
- self.tmtcPrinter.printToFile(exportName, True)
+ self._tmtcPrinter.printToFile(exportName, True)
elif self.pusPacketInfo == "timeout":
self.tmTimeout = self.pusPacket
else:
@@ -114,7 +114,7 @@ class CommandSenderReceiver:
self.elapsed_time = time.time() - self.start_time
if self.elapsed_time > self.tmTimeout * self.tcSendTimeoutFactor:
print("Command Sender Receiver: Timeout, sending TC again !")
- self.comInterface.sendTelecommand(self.lastTc, self.lastTcInfo)
+ self._comInterface.sendTelecommand(self.lastTc, self.lastTcInfo)
self.timeoutCounter = self.timeoutCounter + 1
self.start_time = time.time()
time.sleep(1)
diff --git a/sendreceive/OBSW_MultipleCommandsSenderReceiver.py b/sendreceive/OBSW_MultipleCommandsSenderReceiver.py
index fc3e866c21041e3ae9848fccef1acf63027cfe67..f9243b3123fc8b4b3303c5845319590eb049ee85 100644
--- a/sendreceive/OBSW_MultipleCommandsSenderReceiver.py
+++ b/sendreceive/OBSW_MultipleCommandsSenderReceiver.py
@@ -55,11 +55,11 @@ class MultipleCommandSenderReceiver(SequentialCommandSenderReceiver):
try:
self.sendAllQueue()
self.handleLastRepliesListening(self.tmTimeout/2.0)
- self.tmListener.modeOpFinished.set()
- self.tmInfoQueue = self.tmListener.retrieveTmInfoQueue()
+ self._tmListener.modeOpFinished.set()
+ self.tmInfoQueue = self._tmListener.retrieveTmInfoQueue()
# self.handleTcResending() Turned off for now, not needed
if self.doPrintToFile:
- self.tmtcPrinter.printToFile()
+ self._tmtcPrinter.printToFile()
except (KeyboardInterrupt, SystemExit):
print("Keyboard Interrupt or System Exit detected")
exit()
@@ -80,7 +80,7 @@ class MultipleCommandSenderReceiver(SequentialCommandSenderReceiver):
self.checkQueueEntry(self.tcQueue.pop())
if self.queueEntryIsTelecommand:
self.tcInfoQueue.append(self.pusPacketInfo)
- self.comInterface.sendTelecommand(self.pusPacket, self.pusPacketInfo)
+ self._comInterface.sendTelecommand(self.pusPacket, self.pusPacketInfo)
self.handleWaiting()
def handleWaiting(self):
@@ -92,8 +92,8 @@ class MultipleCommandSenderReceiver(SequentialCommandSenderReceiver):
time.sleep(self.waitTime)
def retrieveTmInfoQueue(self):
- if self.tmListener.replyEvent.is_set():
- return self.tmListener.tmInfoQueue
+ if self._tmListener.replyEvent.is_set():
+ return self._tmListener.tmInfoQueue
else:
print("Multiple Command SenderReceiver: Configuration error, reply event not set in TM listener")
diff --git a/sendreceive/OBSW_SequentialSenderReceiver.py b/sendreceive/OBSW_SequentialSenderReceiver.py
index c7c7e432357988473500e18fb0112eccc3c2c43f..a39966eeff11654cb1ba7fbaaf62994baa636661 100644
--- a/sendreceive/OBSW_SequentialSenderReceiver.py
+++ b/sendreceive/OBSW_SequentialSenderReceiver.py
@@ -43,8 +43,8 @@ class SequentialCommandSenderReceiver(CommandSenderReceiver):
self.abortFlag = False
def sendQueueTcAndReceiveTmSequentially(self):
- self.tmListener.modeId = g.modeList.ServiceTestMode
- self.tmListener.modeChangeEvent.set()
+ self._tmListener.modeId = g.modeList.ServiceTestMode
+ self._tmListener.modeChangeEvent.set()
self.sendAndReceiveFirstPacket()
# this flag is set in the separate thread !
try:
@@ -62,16 +62,16 @@ class SequentialCommandSenderReceiver(CommandSenderReceiver):
self.start_time = time.time()
self.checkForTimeout()
print("Sequential SenderReceiver: All replies received!")
- self.tmListener.modeOpFinished.set()
+ self._tmListener.modeOpFinished.set()
if self.doPrintToFile:
print("Sequential SenderReceiver: Exporting output to log file.")
- self.tmtcPrinter.printToFile()
+ self._tmtcPrinter.printToFile()
print("Listening for packages ...")
def performNextTcSend(self):
- if self.tmListener.replyEvent.is_set():
+ if self._tmListener.replyEvent.is_set():
self.replyReceived = True
- self.tmListener.replyEvent.clear()
+ self._tmListener.replyEvent.clear()
# this flag is set in the separate receiver thread too
if self.replyReceived:
self.sendNextTelecommand()
@@ -83,7 +83,7 @@ class SequentialCommandSenderReceiver(CommandSenderReceiver):
def sendAndReceiveFirstPacket(self):
self.checkQueueEntry(self.tcQueue.pop())
if self.queueEntryIsTelecommand:
- self.comInterface.sendTelecommand(self.pusPacket, self.pusPacketInfo)
+ self._comInterface.sendTelecommand(self.pusPacket, self.pusPacketInfo)
else:
self.sendAndReceiveFirstPacket()
@@ -91,7 +91,7 @@ class SequentialCommandSenderReceiver(CommandSenderReceiver):
self.checkQueueEntry(self.tcQueue.pop())
if self.queueEntryIsTelecommand:
self.start_time = time.time()
- self.comInterface.sendTelecommand(self.pusPacket, self.pusPacketInfo)
+ self._comInterface.sendTelecommand(self.pusPacket, self.pusPacketInfo)
elif self.tcQueue.__len__() == 0:
# Special case: Last queue entry is not a Telecommand
self.allRepliesReceived = True
diff --git a/sendreceive/OBSW_SingleCommandSenderReceiver.py b/sendreceive/OBSW_SingleCommandSenderReceiver.py
index 0c670dbfb4a6f48c123c707774f6f7b85e31e5dc..8280333ac196c33893b8aaaa2b5866ed5af3a9d0 100644
--- a/sendreceive/OBSW_SingleCommandSenderReceiver.py
+++ b/sendreceive/OBSW_SingleCommandSenderReceiver.py
@@ -39,14 +39,14 @@ class SingleCommandSenderReceiver(CommandSenderReceiver):
exit()
def sendSingleTcAndReceiveTm(self):
- self.tmListener.modeId = g.modeList.SingleCommandMode
- self.tmListener.modeChangeEvent.set()
- self.comInterface.sendTelecommand(self.pusPacket, self.pusPacketInfo)
+ self._tmListener.modeId = g.modeList.SingleCommandMode
+ self._tmListener.modeChangeEvent.set()
+ self._comInterface.sendTelecommand(self.pusPacket, self.pusPacketInfo)
while not self.replyReceived:
# wait until reply is received
super().checkForFirstReply()
if self.replyReceived:
print("Single Command SenderReceiver: Reply received")
- self.tmListener.modeOpFinished.set()
+ self._tmListener.modeOpFinished.set()
print("Listening for packages ...")
diff --git a/tc/OBSW_TcService200.py b/tc/OBSW_TcService200.py
index b37942a41a8d90407681bb55262cfe689dae9d59..ee4d20c1f047c67653cdceeef3028628a8e9e4c7 100644
--- a/tc/OBSW_TcService200.py
+++ b/tc/OBSW_TcService200.py
@@ -13,7 +13,6 @@ Run script with -s <Service or Device> -m 3 with optional -p parameter for file
"""
from tc.OBSW_TcPacket import PUSTelecommand
from tc.OBSW_TcPacker import tcQueueT
-from typing import Deque
import struct
diff --git a/test/OBSW_PusServiceTest.py b/test/OBSW_PusServiceTest.py
index 3e9310d230fada96f365f1639603ce758f116cb1..0583ce939cb0c33cf8a55cd66222ef1ae8bc6db7 100644
--- a/test/OBSW_PusServiceTest.py
+++ b/test/OBSW_PusServiceTest.py
@@ -1,5 +1,6 @@
import unittest
-from test.OBSW_UnitTest import TestService, TmInfoQueueT
+from test.OBSW_UnitTest import TestService
+from tm.OBSW_PusTm import pusTmInfoQueueT
from tc.OBSW_TcPacker import packService17TestInto, packService5TestInto, packService2TestInto
import config.OBSW_Config as g
@@ -20,7 +21,7 @@ class TestService2(TestService):
self.miscExpected = 4
super().performGenericAssertionTest(assertionDict)
- def analyseTmInfo(self, tmInfoQueue: TmInfoQueueT, assertionDict: dict):
+ def analyseTmInfo(self, tmInfoQueue: pusTmInfoQueueT, assertionDict: dict):
while not tmInfoQueue.__len__() == 0:
currentTmInfo = tmInfoQueue.pop()
# Tc verification scanning is generic and has been moved to the superclass
diff --git a/test/OBSW_UnitTest.py b/test/OBSW_UnitTest.py
index 1bc09609fbf088e590b397408b830ca0d8714452..06cfb9745b218bd4d27f6de54cb7d455958e0657 100644
--- a/test/OBSW_UnitTest.py
+++ b/test/OBSW_UnitTest.py
@@ -30,7 +30,6 @@ from collections import deque
from typing import Deque
from tc.OBSW_TcPacker import packDummyDeviceTestInto
-from tm.OBSW_PusTm import pusPacketInfoT
from tm.OBSW_TmService1 import pusPacketInfoService1T
from sendreceive.OBSW_TmListener import TmListener
from abc import abstractmethod
@@ -40,7 +39,6 @@ from comIF.OBSW_Ethernet_ComIF import EthernetComIF
from comIF.OBSW_Serial_ComIF import SerialComIF
from config import OBSW_Config as g
-TmInfoQueueT = Deque[pusPacketInfoT]
TmInfoQueueService1T = Deque[pusPacketInfoService1T]
@@ -75,12 +73,15 @@ class TestService(unittest.TestCase):
cls.tmtcPrinter = TmTcPrinter(cls._displayMode, cls.printFile, True)
cls.communicationInterface = None
if g.comIF == 0:
- cls.communicationInterface = EthernetComIF(cls.tmtcPrinter, g.tmTimeout, g.tcSendTimeoutFactor,
- g.sockSend, g.sockReceive, g.sendAddress)
+ cls.communicationInterface = EthernetComIF(
+ tmtcPrinter=cls.tmtcPrinter,tmTimeout=g.tmTimeout, tcTimeoutFactor=g.tcSendTimeoutFactor,
+ sendAddress=g.sendAddress, receiveAddress=g.recAddress)
else:
baudRate = 115200
- cls.communicationInterface = SerialComIF(cls.tmtcPrinter, g.comPort, baudRate, g.serialTimeout)
- cls.tmListener = TmListener(cls.communicationInterface, g.tmTimeout, cls.tcTimeoutFactor)
+ cls.communicationInterface = SerialComIF(
+ tmtcPrinter=cls.tmtcPrinter, comPort=g.comPort, baudRate=baudRate, serialTimeout=g.serialTimeout)
+ cls.tmListener = TmListener(
+ comInterface=cls.communicationInterface, tmTimeout=g.tmTimeout, tcTimeoutFactor=cls.tcTimeoutFactor)
# connectToBoard()
# This function should be called in each individual test to send the actual telecommands