From 4c1b8501bc739a8c6f87ed539bbef8f781464a5b Mon Sep 17 00:00:00 2001
From: "Robin.Mueller" <robin.mueller.m@gmail.com>
Date: Mon, 6 Apr 2020 13:59:27 +0200
Subject: [PATCH] various bugfixes and improvements
---
comIF/OBSW_Ethernet_ComIF.py | 2 +-
config/OBSW_Config.py | 82 +++++++++----------
gui/OBSW_TmtcGUI.py | 2 +-
obsw_tmtc_client.py | 54 ++++++------
.../OBSW_MultipleCommandsSenderReceiver.py | 17 ++--
sendreceive/OBSW_SequentialSenderReceiver.py | 2 +-
sendreceive/obsw_command_sender_receiver.py | 2 +-
.../obsw_single_command_sender_receiver.py | 2 +-
sendreceive/obsw_tm_listener.py | 4 +-
tc/OBSW_TcPacker.py | 6 +-
tc/OBSW_TcPacket.py | 2 +-
tc/OBSW_TcService2.py | 2 +-
tc/OBSW_TcService200.py | 4 +-
test/OBSW_PusServiceTest.py | 14 ++--
test/OBSW_UnitTest.py | 18 ++--
tm/OBSW_PusTm.py | 2 +-
tm/OBSW_TmPacket.py | 4 +-
utility/OBSW_ArgParser.py | 71 +++++++++-------
utility/obsw_tmtc_printer.py | 6 +-
19 files changed, 157 insertions(+), 139 deletions(-)
diff --git a/comIF/OBSW_Ethernet_ComIF.py b/comIF/OBSW_Ethernet_ComIF.py
index d4de40a..010c0ab 100644
--- a/comIF/OBSW_Ethernet_ComIF.py
+++ b/comIF/OBSW_Ethernet_ComIF.py
@@ -89,7 +89,7 @@ class EthernetComIF(CommunicationInterface):
self.sockReceive.setblocking(False)
except OSError:
print("Socket already set-up.")
- # print("Socket Receive Address: " + str(g.recAddress))
+ # print("Socket Receive Address: " + str(g.G_REC_ADDRESS))
# logging.exception("Error: ")
# exit()
except TypeError:
diff --git a/config/OBSW_Config.py b/config/OBSW_Config.py
index 622fba0..bf38c78 100644
--- a/config/OBSW_Config.py
+++ b/config/OBSW_Config.py
@@ -51,28 +51,28 @@ All global variables, set in main program with arg parser
# General Settings
-scriptMode = 1
-modeId = 0
-service = 17
-displayMode = "long"
+G_SCRIPT_MODE = 1
+G_MODE_ID = 0
+G_SERVICE = 17
+G_DISPLAY_MODE = "long"
-comIF = 0
+G_COM_IF = 0
# COM Port for serial communication
-comPort = 'COM0'
-serial_timeout = 0.01
+G_COM_PORT = 'COM0'
+G_SERIAL_TIMEOUT = 0.01
# Time related
-tmTimeout = 10
-tcSendTimeoutFactor = 2.0
+G_TM_TIMEOUT = 10
+G_TC_SEND_TIMEOUT_FACTOR = 2.0
# Ethernet connection settings
-recAddress = 0
-sendAddress = (0, 0)
+G_REC_ADDRESS = 0
+G_SEND_ADDRESS = (0, 0)
# Print Settings
-G_PRINT_TO_FILE = False
+G_PRINT_TO_FILE = True
G_PRINT_HK_DATA = False
-printRawTmData = False
+G_PRINT_RAW_TM = False
"""
These objects are set for the Unit Test, no better solution found yet
@@ -84,15 +84,15 @@ tmtcPrinter = None
# noinspection PyUnusedLocal
def setGlobals(args):
- global recAddress, sendAddress, scriptMode, modeId, service, displayMode, comIF, comPort
- global serial_timeout, tmTimeout, tcSendTimeoutFactor, G_PRINT_TO_FILE
- global G_PRINT_HK_DATA, printRawTmData
+ global G_REC_ADDRESS, G_SEND_ADDRESS, G_SCRIPT_MODE, G_MODE_ID, G_SERVICE, G_DISPLAY_MODE
+ global G_COM_IF, G_COM_PORT, G_SERIAL_TIMEOUT, G_TM_TIMEOUT, G_TC_SEND_TIMEOUT_FACTOR, G_PRINT_TO_FILE
+ global G_PRINT_HK_DATA, G_PRINT_RAW_TM
if args.mode == 0:
print("GUI mode not implemented yet !")
if args.shortDisplayMode:
- displayMode = "short"
+ G_DISPLAY_MODE = "short"
else:
- displayMode = "long"
+ G_DISPLAY_MODE = "long"
# Board IP address and ethernet port IP address can be passed optionally
# by passing line parameter In PyCharm: Set parameters in run configuration
# Add IP address of Ethernet port. Use command ipconfig in windows console or ifconfig in Linux.
@@ -103,41 +103,41 @@ def setGlobals(args):
sendAddressToSet = (args.boardIP, portSend)
if 0 <= args.mode <= 5:
if args.mode == 0:
- modeId = ModeList.GUIMode
+ G_MODE_ID = ModeList.GUIMode
elif args.mode == 1:
- modeId = ModeList.ListenerMode
+ G_MODE_ID = ModeList.ListenerMode
elif args.mode == 2:
- modeId = ModeList.SingleCommandMode
+ G_MODE_ID = ModeList.SingleCommandMode
elif args.mode == 3:
- modeId = ModeList.ServiceTestMode
+ G_MODE_ID = ModeList.ServiceTestMode
elif args.mode == 4:
- modeId = ModeList.SoftwareTestMode
+ G_MODE_ID = ModeList.SoftwareTestMode
elif args.mode == 5:
- modeId = ModeList.UnitTest
+ G_MODE_ID = ModeList.UnitTest
else:
print("Invalid mode argument, setting default mode (1)")
- modeId = ModeList[1]
+ G_MODE_ID = ModeList[1]
if args.comIF == 0:
- comIF = ComIF.Ethernet
+ G_COM_IF = ComIF.Ethernet
elif args.comIF == 1:
- comIF = ComIF.Serial
+ G_COM_IF = ComIF.Serial
else:
print("Invalid Communication Interface, setting serial (1)")
- comIF = ComIF.Serial
+ G_COM_IF = ComIF.Serial
- service = str(args.service)
- if service.isdigit():
- service = int(args.service)
+ G_SERVICE = str(args.service)
+ if G_SERVICE.isdigit():
+ G_SERVICE = int(args.service)
else:
- service = args.service
- recAddress = recAddressToSet
- sendAddress = sendAddressToSet
- modeId = modeId
- comPort = args.COM
- G_PRINT_HK_DATA = args.hk
- tmTimeout = args.tmTimeout
- printRawTmData = args.rawDataPrint
- displayMode = displayMode
- service = service
+ G_SERVICE = args.service
+ G_REC_ADDRESS = recAddressToSet
+ G_SEND_ADDRESS = sendAddressToSet
+ G_MODE_ID = G_MODE_ID
+ G_PRINT_HK_DATA = args.print_hk
G_PRINT_TO_FILE = args.printFile
+ G_PRINT_RAW_TM = args.rawDataPrint
+ G_SERVICE = G_SERVICE
+ G_COM_PORT = args.com_port
+ G_TM_TIMEOUT = args.tm_timeout
+
diff --git a/gui/OBSW_TmtcGUI.py b/gui/OBSW_TmtcGUI.py
index b6684bb..5a55c6b 100644
--- a/gui/OBSW_TmtcGUI.py
+++ b/gui/OBSW_TmtcGUI.py
@@ -27,7 +27,7 @@ import time
# include a really nice source badge and make it large !
# plan this on paper first...
# Step 1: Huge Mission Badge in Tkinter window because that is cool.
-# Step 2: Simple buttons to run service test around the badge.
+# Step 2: Simple buttons to run G_SERVICE test around the badge.
class TmTcGUI(Process):
def __init__(self):
super(TmTcGUI, self).__init__()
diff --git a/obsw_tmtc_client.py b/obsw_tmtc_client.py
index b87a27f..7ec3c63 100644
--- a/obsw_tmtc_client.py
+++ b/obsw_tmtc_client.py
@@ -20,13 +20,13 @@ 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,
+Example command to test G_SERVICE 17,
assuming no set client IP (set manually to PC IP Address if necessary)
and default board IP 169.254.1.38:
obsw_tmtc_client.py -m 3 -s 17
Example to run Unit Test:
obsw_tmtc_client.py -m 5
-Example to test service 17 with HK output and serial communication:
+Example to test G_SERVICE 17 with HK output and serial communication:
obsw_tmtc_client.py -m 3 -s 17 --hk -c 1
Get command line help:
obsw_tmtc_client.py -h
@@ -36,7 +36,7 @@ There are four different Modes:
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
+ 3. ServiceTestMode: Send all Telecommands belonging to a certain G_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
@@ -87,8 +87,8 @@ def main():
"""
args = parseInputArguments()
setGlobals(args)
- tmtc_printer = TmTcPrinter(g.displayMode, g.G_PRINT_TO_FILE, True)
- if g.modeId == g.ModeList.GUIMode:
+ tmtc_printer = TmTcPrinter(g.G_DISPLAY_MODE, g.G_PRINT_TO_FILE, True)
+ if g.G_MODE_ID == g.ModeList.GUIMode:
backend = TmTcBackend()
backend.start()
gui = TmTcGUI()
@@ -101,35 +101,35 @@ def main():
communication_interface = set_communication_interface(tmtc_printer)
atexit.register(keyboardInterruptHandler, comInterface=communication_interface)
tm_listener = TmListener(
- comInterface=communication_interface, tmTimeout=g.tmTimeout,
- tcTimeoutFactor=g.tcSendTimeoutFactor)
+ comInterface=communication_interface, tmTimeout=g.G_TM_TIMEOUT,
+ tcTimeoutFactor=g.G_TC_SEND_TIMEOUT_FACTOR)
tm_listener.start()
- if g.modeId == g.ModeList.ListenerMode:
+ if g.G_MODE_ID == g.ModeList.ListenerMode:
print("Listening for packages...")
- elif g.modeId == g.ModeList.SingleCommandMode:
+ elif g.G_MODE_ID == g.ModeList.SingleCommandMode:
pus_packet_tuple = command_preparation()
sender_and_receiver = SingleCommandSenderReceiver(
comInterface=communication_interface, tmtcPrinter=tmtc_printer, tmListener=tm_listener,
- pusPacketTuple=pus_packet_tuple, tmTimeout=g.tmTimeout,
- tcTimeoutFactor=g.tcSendTimeoutFactor, doPrintToFile=g.G_PRINT_TO_FILE)
+ pusPacketTuple=pus_packet_tuple, tmTimeout=g.G_TM_TIMEOUT,
+ tcTimeoutFactor=g.G_TC_SEND_TIMEOUT_FACTOR, doPrintToFile=g.G_PRINT_TO_FILE)
sender_and_receiver.send_single_tc_and_receive_tm()
- elif g.modeId == g.ModeList.ServiceTestMode:
+ elif g.G_MODE_ID == g.ModeList.ServiceTestMode:
service_queue = deque()
sender_and_receiver = SequentialCommandSenderReceiver(
- comInterface=communication_interface, tmtcPrinter=tmtc_printer, tmTimeout=g.tmTimeout,
- tcQueue=serviceTestSelect(g.service, service_queue),
- tcTimeoutFactor=g.tcSendTimeoutFactor, tmListener=tm_listener)
+ comInterface=communication_interface, tmtcPrinter=tmtc_printer, tmTimeout=g.G_TM_TIMEOUT,
+ tcQueue=serviceTestSelect(g.G_SERVICE, service_queue),
+ tcTimeoutFactor=g.G_TC_SEND_TIMEOUT_FACTOR, tmListener=tm_listener)
sender_and_receiver.sendQueueTcAndReceiveTmSequentially()
- elif g.modeId == g.ModeList.SoftwareTestMode:
+ elif g.G_MODE_ID == g.ModeList.SoftwareTestMode:
all_tc_queue = createTotalTcQueue()
sender_and_receiver = SequentialCommandSenderReceiver(
- comInterface=communication_interface, tmtcPrinter=tmtc_printer, tmTimeout=g.tmTimeout,
- tcQueue=all_tc_queue, tcTimeoutFactor=g.tcSendTimeoutFactor, tmListener=tm_listener)
+ comInterface=communication_interface, tmtcPrinter=tmtc_printer, tmTimeout=g.G_TM_TIMEOUT,
+ tcQueue=all_tc_queue, tcTimeoutFactor=g.G_TC_SEND_TIMEOUT_FACTOR, tmListener=tm_listener)
sender_and_receiver.sendQueueTcAndReceiveTmSequentially()
- elif g.modeId == g.ModeList.UnitTest:
+ elif g.G_MODE_ID == g.ModeList.UnitTest:
# Set up test suite and run it with runner. Verbosity specifies detail level
g.tmListener = tm_listener
g.comInterface = communication_interface
@@ -158,7 +158,7 @@ def command_preparation():
command = PUSTelecommand(service=17, subservice=1, SSC=21)
# command.print()
# file = bytearray([1, 2, 3, 4, 5])
- # command = PUSTelecommand(service=23, subservice=1, SSC=21, data=file)
+ # command = PUSTelecommand(G_SERVICE=23, subservice=1, SSC=21, data=file)
# command.packCommandTuple()
return command.packCommandTuple()
@@ -170,18 +170,18 @@ def set_communication_interface(tmtc_printer: TmTcPrinterT) -> ComIF_T:
:return: CommunicationInterface object
"""
try:
- if g.comIF == g.ComIF.Ethernet:
+ if g.G_COM_IF == g.ComIF.Ethernet:
communication_interface = EthernetComIF(
- tmtcPrinter=tmtc_printer, tmTimeout=g.tmTimeout,
- tcTimeoutFactor=g.tcSendTimeoutFactor, sendAddress=g.sendAddress,
- receiveAddress=g.recAddress)
+ tmtcPrinter=tmtc_printer, tmTimeout=g.G_TM_TIMEOUT,
+ tcTimeoutFactor=g.G_TC_SEND_TIMEOUT_FACTOR, sendAddress=g.G_SEND_ADDRESS,
+ receiveAddress=g.G_REC_ADDRESS)
else:
- com_port = g.comPort
+ com_port = g.G_COM_PORT
baud_rate = 115200
- g.serial_timeout = 0.05
+ g.G_SERIAL_TIMEOUT = 0.05
communication_interface = SerialComIF(
tmtcPrinter=tmtc_printer, comPort=com_port, baudRate=baud_rate,
- serialTimeout=g.serial_timeout)
+ serialTimeout=g.G_SERIAL_TIMEOUT)
# else:
# pass
return communication_interface
diff --git a/sendreceive/OBSW_MultipleCommandsSenderReceiver.py b/sendreceive/OBSW_MultipleCommandsSenderReceiver.py
index 9447a27..2b86b4e 100644
--- a/sendreceive/OBSW_MultipleCommandsSenderReceiver.py
+++ b/sendreceive/OBSW_MultipleCommandsSenderReceiver.py
@@ -22,9 +22,9 @@ class MultipleCommandSenderReceiver(SequentialCommandSenderReceiver):
Difference to seqential sender: This class can send TCs in bursts. Wait intervals can be specified with
wait time between the send bursts. This is generally done in the separate test classes in UnitTest
"""
- def __init__(self, comInterface: ComIF_T, tmtcPrinter: TmTcPrinterT, tcQueue: Deque, tmListener: TmListenerT,
- tmTimeout: float, waitIntervals: list, waitTime: Union[float, list], printTm: bool,
- tcTimeoutFactor: float, doPrintToFile: bool):
+ def __init__(self, comInterface: ComIF_T, tmtcPrinter: TmTcPrinterT, tcQueue: Deque,
+ tmListener: TmListenerT, tmTimeout: float, waitIntervals: list,
+ waitTime: Union[float, list], printTm: bool, tcTimeoutFactor: float):
"""
TCs are sent in burst when applicable. Wait intervals can be specified by supplying
respective arguments
@@ -57,10 +57,11 @@ class MultipleCommandSenderReceiver(SequentialCommandSenderReceiver):
self._tm_listener.modeChangeEvent.set()
# TC info queue is set in this function
self.sendAllQueue()
- self.handleLastRepliesListening(self.tm_timeout / 2.0)
+ self.handleLastRepliesListening(self.tm_timeout / 1.5)
self.tmInfoQueue = self.retrieveListenerTmInfoQueue()
self._tm_listener.modeOpFinished.set()
if g.G_PRINT_TO_FILE:
+ print("blub")
self._tmtc_printer.print_to_file()
except (KeyboardInterrupt, SystemExit):
print("Keyboard Interrupt or System Exit detected")
@@ -80,9 +81,11 @@ class MultipleCommandSenderReceiver(SequentialCommandSenderReceiver):
self.sendAndPrintTc()
def sendAndPrintTc(self):
- if self.check_queue_entry(self._tcQueue.pop()):
- self.tcInfoQueue.append(self.pusPacketInfo)
- self._com_interface.sendTelecommand(self.pusPacket, self.pusPacketInfo)
+ tc_queue_tuple = self._tcQueue.pop()
+ if self.check_queue_entry(tc_queue_tuple):
+ pus_packet, pus_packet_info = tc_queue_tuple
+ self.tcInfoQueue.append(pus_packet_info)
+ self._com_interface.sendTelecommand(pus_packet, pus_packet_info)
self.handleWaiting()
def handleWaiting(self):
diff --git a/sendreceive/OBSW_SequentialSenderReceiver.py b/sendreceive/OBSW_SequentialSenderReceiver.py
index 4c2ae27..ef73d6b 100644
--- a/sendreceive/OBSW_SequentialSenderReceiver.py
+++ b/sendreceive/OBSW_SequentialSenderReceiver.py
@@ -29,7 +29,7 @@ class SequentialCommandSenderReceiver(CommandSenderReceiver):
:param tmListener: TmListener object which runs in the background and receives all Telemetry
: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 tcTimeoutFactor: If TM is not received, resend TC after G_TM_TIMEOUT * G_TC_SEND_TIMEOUT_FACTOR
:param doPrintToFile: Specify whether output is also printed to a file
"""
super().__init__(
diff --git a/sendreceive/obsw_command_sender_receiver.py b/sendreceive/obsw_command_sender_receiver.py
index 453caaf..36fcb7a 100644
--- a/sendreceive/obsw_command_sender_receiver.py
+++ b/sendreceive/obsw_command_sender_receiver.py
@@ -38,7 +38,7 @@ class CommandSenderReceiver:
: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
+ resend TC after G_TM_TIMEOUT * G_TC_SEND_TIMEOUT_FACTOR
"""
self.tm_timeout = tmTimeout
self.tc_send_timeout_factor = tcSendTimeoutFactor
diff --git a/sendreceive/obsw_single_command_sender_receiver.py b/sendreceive/obsw_single_command_sender_receiver.py
index 27be3c1..238f0d3 100644
--- a/sendreceive/obsw_single_command_sender_receiver.py
+++ b/sendreceive/obsw_single_command_sender_receiver.py
@@ -31,7 +31,7 @@ class SingleCommandSenderReceiver(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
+ resend TC after G_TM_TIMEOUT * G_TC_SEND_TIMEOUT_FACTOR
:param doPrintToFile: Specify whether output is also printed to a file
"""
super().__init__(comInterface=comInterface, tmListener=tmListener, tmtcPrinter=tmtcPrinter,
diff --git a/sendreceive/obsw_tm_listener.py b/sendreceive/obsw_tm_listener.py
index 0df22c0..0155d7f 100644
--- a/sendreceive/obsw_tm_listener.py
+++ b/sendreceive/obsw_tm_listener.py
@@ -123,6 +123,6 @@ class TmListener:
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
+ if self.tmTimeout is not g.G_TM_TIMEOUT:
+ self.tmTimeout = g.G_TM_TIMEOUT
return True
diff --git a/tc/OBSW_TcPacker.py b/tc/OBSW_TcPacker.py
index fbab63c..2bd4765 100644
--- a/tc/OBSW_TcPacker.py
+++ b/tc/OBSW_TcPacker.py
@@ -2,11 +2,11 @@
"""
Program: OBSW_UnitTest.py
Date: 01.11.2019
-Description: Packs the TC queue for specific service or device testing
+Description: Packs the TC queue for specific G_SERVICE or device testing
Manual:
-Contains a service select factory which returns specific service or device tests.
-Also contains an all service and all device tc queue packer.
+Contains a G_SERVICE select factory which returns specific G_SERVICE or device tests.
+Also contains an all G_SERVICE and all device tc queue packer.
Run script with -s <Service or Device> -m 3 with optional -p parameter for file output inside the log folder
@author: R. Mueller
diff --git a/tc/OBSW_TcPacket.py b/tc/OBSW_TcPacket.py
index fad67cb..9bd5e21 100644
--- a/tc/OBSW_TcPacket.py
+++ b/tc/OBSW_TcPacket.py
@@ -70,7 +70,7 @@ class PUSTelecommand:
def packInformation(self) -> pusTcInfoT:
tcInformation = {
- "service": self.service,
+ "G_SERVICE": self.service,
"subservice": self.subservice,
"ssc": self.ssc,
"packetId": self.packetId,
diff --git a/tc/OBSW_TcService2.py b/tc/OBSW_TcService2.py
index f0ca154..4b8d629 100644
--- a/tc/OBSW_TcService2.py
+++ b/tc/OBSW_TcService2.py
@@ -17,7 +17,7 @@ def packService2TestInto(tcQueue: Deque, calledExternally: bool = False) -> Dequ
if calledExternally is False:
tcQueue.appendleft(("print", "Testing Service 2"))
objectId = bytearray([0x44, 0x00, 0xAF, 0xFE]) # dummy device
- # don't forget to set object mode raw (service 200) before calling this !
+ # don't forget to set object mode raw (G_SERVICE 200) before calling this !
# Set Raw Mode
tcQueue.appendleft(("print", "\r\nTesting Service 2: Setting Raw Mode"))
modeData = packModeData(objectId, 3, 0)
diff --git a/tc/OBSW_TcService200.py b/tc/OBSW_TcService200.py
index 3abc8e9..31105db 100644
--- a/tc/OBSW_TcService200.py
+++ b/tc/OBSW_TcService200.py
@@ -5,8 +5,8 @@ Date: 01.11.2019
Description: PUS Custom Service 200: Mode Commanding
Manual:
-Contains a service select factory which returns specific service or device tests.
-Also contains an all service and all device tc queue packer.
+Contains a G_SERVICE select factory which returns specific G_SERVICE or device tests.
+Also contains an all G_SERVICE and all device tc queue packer.
Run script with -s <Service or Device> -m 3 with optional -p parameter for file output inside the log folder
@author: R. Mueller
diff --git a/test/OBSW_PusServiceTest.py b/test/OBSW_PusServiceTest.py
index b73404f..6770b15 100644
--- a/test/OBSW_PusServiceTest.py
+++ b/test/OBSW_PusServiceTest.py
@@ -12,7 +12,7 @@ class TestService2(TestService):
print("Testing Service 2")
# all commands must be sent sequentially, not as a burst
cls.waitIntervals = [1, 2, 3, 4]
- cls.waitTime = [2, 2, 2, 2]
+ cls.waitTime = [2, 2, 2, 3]
packService2TestInto(cls.testQueue)
def test_Service2(self):
@@ -25,20 +25,20 @@ class TestService2(TestService):
while not tmInfoQueue.__len__() == 0:
currentTmInfo = tmInfoQueue.pop()
# Tc verification scanning is generic and has been moved to the superclass
- if currentTmInfo["service"] == 1:
+ if currentTmInfo["G_SERVICE"] == 1:
super().scanForRespectiveTc(currentTmInfo)
# Here, the desired event Id or RID can be specified
- if currentTmInfo["service"] == 5:
+ if currentTmInfo["G_SERVICE"] == 5:
# mode change
if currentTmInfo["EventID"] == 7401 and currentTmInfo["RID"] == 0x4400affe:
self.eventCounter = self.eventCounter + 1
- if currentTmInfo["service"] == 200 and currentTmInfo["subservice"] == 6:
+ if currentTmInfo["G_SERVICE"] == 200 and currentTmInfo["subservice"] == 6:
# mode change confirmation
self.miscCounter = self.miscCounter + 1
# wiretapping messages
- elif currentTmInfo["service"] == 2 and currentTmInfo["subservice"] == 130:
+ elif currentTmInfo["G_SERVICE"] == 2 and currentTmInfo["subservice"] == 130:
self.miscCounter = self.miscCounter + 1
- elif currentTmInfo["service"] == 2 and currentTmInfo["subservice"] == 131:
+ elif currentTmInfo["G_SERVICE"] == 2 and currentTmInfo["subservice"] == 131:
self.miscCounter = self.miscCounter + 1
if currentTmInfo["valid"] == 0:
self.valid = False
@@ -96,7 +96,7 @@ class TestService17(TestService):
print("Testing Service 17")
cls.waitIntervals = [2]
cls.waitTime = 2
- cls.tmTimeout = g.tmTimeout
+ cls.tmTimeout = g.G_TM_TIMEOUT
packService17TestInto(cls.testQueue)
def test_Service17(self):
diff --git a/test/OBSW_UnitTest.py b/test/OBSW_UnitTest.py
index 800559c..40e8aab 100644
--- a/test/OBSW_UnitTest.py
+++ b/test/OBSW_UnitTest.py
@@ -13,7 +13,7 @@ For Developers:
TestService is the template method, analyseTmInfo and analyseTcInfo are implemented for
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 log files generated by individual G_SERVICE test or the software test mode when providing
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.
@@ -45,7 +45,7 @@ TmInfoQueueService1T = Deque[pusPacketInfoService1T]
class TestService(unittest.TestCase):
"""
- Generic service test class.
+ Generic G_SERVICE test class.
See documentation: https://docs.python.org/3/library/unittest.html#unittest.TestCase.setUpClass
"""
@classmethod
@@ -67,8 +67,8 @@ class TestService(unittest.TestCase):
cls.testQueue = deque()
# Extremely ugly solution so that we can use the Python Unit Test Framework
# default timeout for receiving TM, set in subclass manually
- cls.tmTimeout = g.tmTimeout
- cls.tcTimeoutFactor = g.tcSendTimeoutFactor
+ cls.tmTimeout = g.G_TM_TIMEOUT
+ cls.tcTimeoutFactor = g.G_TC_SEND_TIMEOUT_FACTOR
cls.printFile = g.G_PRINT_TO_FILE
cls.tmtcPrinter = g.tmtcPrinter
cls.tmListener = g.tmListener
@@ -84,7 +84,7 @@ class TestService(unittest.TestCase):
comInterface=self.communicationInterface, tmtcPrinter=self.tmtcPrinter,
tmListener=self.tmListener, tcQueue=self.testQueue, tmTimeout=self.tmTimeout,
waitIntervals=self.waitIntervals, waitTime=self.waitTime, printTm=self.printTm,
- tcTimeoutFactor=self.tcTimeoutFactor, doPrintToFile=self.printFile)
+ tcTimeoutFactor=self.tcTimeoutFactor)
(tcInfoQueue, tmInfoQueue) = UnitTester.sendTcQueueAndReturnInfo()
assertionDict = self.analyseTmTcInfo(tmInfoQueue, tcInfoQueue)
return assertionDict
@@ -134,7 +134,7 @@ class TestService(unittest.TestCase):
# For commands with multiple steps, update this value manually !
self.tcVerifyStepCounter = self.tcVerifyCounter + 1
self.tcSscArray.append(currentTcInfo["ssc"])
- self.tcServiceArray.append(currentTcInfo["service"])
+ self.tcServiceArray.append(currentTcInfo["G_SERVICE"])
self.tcSubserviceArray.append(currentTcInfo["subservice"])
# must be implemented by child test !
@@ -178,13 +178,13 @@ class TestService(unittest.TestCase):
while not tmInfoQueue.__len__() == 0:
currentTmInfo = tmInfoQueue.pop()
# Tc verification scanning is generic and has been moved to the superclass
- if currentTmInfo["service"] == 1:
+ if currentTmInfo["G_SERVICE"] == 1:
self.scanForRespectiveTc(currentTmInfo)
# Here, the desired event Id or RID can be specified
- if currentTmInfo["service"] == 5:
+ if currentTmInfo["G_SERVICE"] == 5:
if currentTmInfo["EventID"] == 8200 and currentTmInfo["RID"] == 0x51001700:
self.eventCounter = self.eventCounter + 1
- if currentTmInfo["service"] == 17:
+ if currentTmInfo["G_SERVICE"] == 17:
self.miscCounter = self.miscCounter + 1
if currentTmInfo["valid"] == 0:
self.valid = False
diff --git a/tm/OBSW_PusTm.py b/tm/OBSW_PusTm.py
index 5cc6c57..cc82f7d 100644
--- a/tm/OBSW_PusTm.py
+++ b/tm/OBSW_PusTm.py
@@ -34,7 +34,7 @@ class PUSTelemetry(OBSWPusPacket):
def packTmInformation(self) -> pusTmInfoT:
tmInformation = {
- "service": self.getService(),
+ "G_SERVICE": self.getService(),
"subservice": self.getSubservice(),
"ssc": self.getSSC(),
"data": self.byteArrayData,
diff --git a/tm/OBSW_TmPacket.py b/tm/OBSW_TmPacket.py
index a263d11..e05f52e 100644
--- a/tm/OBSW_TmPacket.py
+++ b/tm/OBSW_TmPacket.py
@@ -15,7 +15,7 @@ import struct
def PUSTelemetryFactory(rawPacket):
servicetype = rawPacket[7]
- # extract service type from data
+ # extract G_SERVICE type from data
if servicetype == 1:
return Service1TM(rawPacket)
elif servicetype == 2:
@@ -31,7 +31,7 @@ def PUSTelemetryFactory(rawPacket):
elif servicetype == 200:
return Service200TM(rawPacket)
else:
- print("The service " + str(servicetype) + " is not implemented in Telemetry Factory")
+ print("The G_SERVICE " + str(servicetype) + " is not implemented in Telemetry Factory")
return PUSTelemetry(rawPacket)
diff --git a/utility/OBSW_ArgParser.py b/utility/OBSW_ArgParser.py
index dba2a69..24c7700 100644
--- a/utility/OBSW_ArgParser.py
+++ b/utility/OBSW_ArgParser.py
@@ -19,37 +19,51 @@ def parseInputArguments():
"""
argParser = argparse.ArgumentParser(description="TMTC Client Command Line Interface")
- argParser.add_argument('-m', '--mode', type=int, help='Target Mode. Default is 1(Listener Mode), '
- '0: GUI Mode, 1:Listener Mode, '
- '2: Single Command Mode, 3: Service Test Mode, '
- '4: Software Test Mode, 5: Unit Test Mode ', default=0)
- argParser.add_argument('-c', '--comIF', type=int, help='Communication Interface. 0 for Ethernet, 1 for Serial',
- default=0)
+ argParser.add_argument(
+ '-m', '--mode', type=int, help='Target Mode. Default is 1(Listener Mode), '
+ '0: GUI Mode, 1:Listener Mode, 2: Single Command Mode, 3: Service Test Mode, '
+ '4: Software Test Mode, 5: Unit Test Mode ', default=0)
+ argParser.add_argument(
+ '-c', '--comIF', type=int, help='Communication Interface. 0 for Ethernet, 1 for Serial',
+ default=0)
argParser.add_argument('--clientIP', help='Client(Computer) IP. Default:\'\'', default='')
- argParser.add_argument('--boardIP', help='Board IP. Default: 169.254.1.38', default='169.254.1.38')
+ argParser.add_argument(
+ '--boardIP', help='Board IP. Default: 169.254.1.38', default='169.254.1.38')
argParser.add_argument('-s', '--service', help='Service to test. Default: 17', default=17)
- argParser.add_argument('-t', '--tmTimeout', type=float, help='TM Timeout when listening to verification sequence.'
- ' Default: 12, 6(Serial)', default=6.0)
- argParser.add_argument('-p', '--printFile', help='Supply -p to print output to file. Default: False',
- action='store_true')
+ argParser.add_argument('-t', '--tm_timeout', type=float, help='TM Timeout when listening to '
+ 'verification sequence. Default: 12, 6(Serial)', default=6.0)
+ argParser.add_argument('--np', dest='printFile', help='Supply --np to suppress print '
+ 'output to file.', action='store_false')
argParser.add_argument('-o', '--tcTimeoutFactor', type=float,
help='TC Timeout Factor. Multiplied with TM Timeout, '
'TC sent again after this time period. Default: 3.5', default=3.5)
- argParser.add_argument('-r', '--rawDataPrint', help='Supply -r to print all raw data directly', action='store_true')
- argParser.add_argument('-d', '--shortDisplayMode', help='Supply -d to print short output', action='store_true')
- argParser.add_argument('-k', '--hk', help='Supply -k or --hk to print HK data', action='store_true')
- argParser.add_argument('--COM', help='COM Port for serial communication')
+ argParser.add_argument('-r', '--rawDataPrint', help='Supply -r to print all raw data directly',
+ action='store_true')
+ argParser.add_argument('-d', '--shortDisplayMode', help='Supply -d to print short output',
+ action='store_true')
+ argParser.add_argument('--hk', dest='print_hk', help='Supply -k or --hk to print HK data',
+ action='store_true')
+ argParser.add_argument('--COM', dest="com_port", help='COM Port for serial communication')
if len(sys.argv) == 1:
print("No Input Arguments specified.")
argParser.print_help()
- args = argParser.parse_args()
- assignEmptyArgs(args)
+ args, unknown = argParser.parse_known_args()
print(args)
+ handle_args(args, unknown)
return args
-def assignEmptyArgs(args):
+def handle_args(args, unknown: list):
+ """
+ Handles the parsed arguments.
+ :param args: Namespace objects
+ (see https://docs.python.org/dev/library/argparse.html#argparse.Namespace)
+ :param unknown: List of unknown parameters.
+ :return:
+ """
+ if len(unknown) > 0:
+ print("Unknown arguments detected: " + str(unknown))
if len(sys.argv) > 1:
handleUnspecifiedArgs(args)
if len(sys.argv) == 1:
@@ -57,10 +71,11 @@ def assignEmptyArgs(args):
def handleUnspecifiedArgs(args):
- if args.comIF == 1 and args.tmTimeout is None:
+ if args.comIF == 1 and args.tm_timeout is None:
args.tmTimeout = 6.0
- if args.comIF == 1 and args.COM is None:
- args.COM = input("Serial Commuinication specified without COM port. Please enter COM Port: ")
+ if args.comIF == 1 and args.com_port is None:
+ args.com_port = input("Serial Commuinication specified without COM port. "
+ "Please enter COM Port: ")
if args.mode is None:
print("No mode specified with -m Parameter.")
print("Possible Modes: ")
@@ -72,7 +87,7 @@ def handleUnspecifiedArgs(args):
args.mode = input("Please enter Mode: ")
if args.mode == 1 and args.service is None:
args.service = input("No Service specified for Service Mode. "
- "Please enter PUS service number: ")
+ "Please enter PUS G_SERVICE number: ")
def handleEmptyArgs(args):
@@ -82,15 +97,15 @@ def handleEmptyArgs(args):
except TypeError:
pass
if printHk in ('y', 'yes', 1):
- args.hk = True
+ args.print_hk = True
else:
- args.hk = False
- printToLog = input("Export service test output to log files ? (y/n or yes/no)")
+ args.print_hk = False
+ printToLog = input("Export G_SERVICE test output to log files ? (y/n or yes/no)")
try:
printToLog = printToLog.lower()
except TypeError:
pass
- if printToLog == 'y' or printToLog == 'yes' or printHk == 1:
- args.printFile = True
- else:
+ if printToLog == 'n' or printToLog == 'no' or printToLog == 0:
args.printFile = False
+ else:
+ args.printFile = True
diff --git a/utility/obsw_tmtc_printer.py b/utility/obsw_tmtc_printer.py
index d3822bc..3c65c1a 100644
--- a/utility/obsw_tmtc_printer.py
+++ b/utility/obsw_tmtc_printer.py
@@ -55,7 +55,7 @@ class TmTcPrinter:
if packet.getService() == 3 and \
(packet.getSubservice() == 10 or packet.getSubservice() == 12):
self.__handle_hk_definition_print(packet)
- if g.printRawTmData:
+ if g.G_PRINT_RAW_TM:
self.print_buffer = "TM Data:" + "\n" + self.return_data_string(packet.data)
print(self.print_buffer)
self.add_print_buffer_to_file_buffer()
@@ -275,7 +275,7 @@ class TmTcPrinter:
:param tc_packet_info:
:return:
"""
- self.print_buffer = "Sent TC[" + str(tc_packet_info["service"]) + "," + \
+ self.print_buffer = "Sent TC[" + str(tc_packet_info["G_SERVICE"]) + "," + \
str(tc_packet_info["subservice"]) + "] " + " with SSC " + \
str(tc_packet_info["ssc"])
print(self.print_buffer)
@@ -288,7 +288,7 @@ class TmTcPrinter:
:return:
"""
try:
- self.print_buffer = "Telecommand TC[" + str(tc_packet_info["service"]) + "," + \
+ self.print_buffer = "Telecommand TC[" + str(tc_packet_info["G_SERVICE"]) + "," + \
str(tc_packet_info["subservice"]) + "] with SSC " + \
str(tc_packet_info["ssc"]) + " sent with data " + \
self.return_data_string(tc_packet_info["data"])
--
GitLab