diff --git a/obsw_tmtc_client.py b/obsw_tmtc_client.py
index b2b06c2ac01addd2015ae52f2148e27825392f01..167165ad7178b8dff5ec5e29471eec50f32771db 100644
--- a/obsw_tmtc_client.py
+++ b/obsw_tmtc_client.py
@@ -1,7 +1,7 @@
#!/usr/bin/python3.8
# -*- coding: utf-8 -*-
-
import atexit
+import time
import unittest
import logging
import sys
@@ -19,14 +19,15 @@ from sendreceive.obsw_tm_listener import TmListener
from utility.obsw_args_parser import parse_input_arguments
from utility.obsw_tmtc_printer import TmTcPrinter
from utility.obsw_exit_handler import keyboard_interrupt_handler
+from utility.obsw_logger import set_tmtc_logger, get_logger
from comIF.obsw_com_config import set_communication_interface
from gui.obsw_tmtc_gui import TmTcGUI
from gui.obsw_backend_test import TmTcBackend
-from utility.obsw_logger import set_tmtc_logger
-from utility.obsw_logger import get_logger
+
+logger = get_logger()
def main():
@@ -35,7 +36,7 @@ def main():
It can be used to to send and receive TMTC packets and TMTC sequences.
@manual
- Manual installation of crcmod and pyserial might be needed
+ Manual installation of crcmod and pyserial for serial communication might be needed
1. Install pip if it is not installed yet
2. Install crcmod and all other required packages:
Command: python3.8 -m pip install crcmod
@@ -76,77 +77,20 @@ def main():
If the packets appear, there might be a problematic firewall setting.
Please ensure that python.exe UDP packets are not blocked in advanced firewall settings
and create a rule to allow packets from port 2008.
- TODO: The whole code below could ba packed into a class too.
- If a backend/frontend architecture is to be implemented, this will make things easier
- We propably instantiate all sender/receiver objects (or maybe instantiate them once
- we need them) on object construction. The main could spawn a process for the frontend
- (e.g. GUI or web interface) and a process for the backend (the sender/receiver classes,
- passing values to a SQL database, loading the MIB database...)
"""
set_tmtc_logger()
- logger = get_logger()
logger.info("Starting TMTC Client")
+
logger.info("Parsing input arguments")
args = parse_input_arguments()
+
logger.info("Setting global variables")
set_globals(args)
- 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()
- gui.start()
- backend.join()
- gui.join()
- logger.info("Both processes have closed")
- sys.exit()
- else:
- communication_interface = set_communication_interface(tmtc_printer)
- atexit.register(keyboard_interrupt_handler, comInterface=communication_interface)
- tm_listener = TmListener(
- com_interface=communication_interface, tm_timeout=g.G_TM_TIMEOUT,
- tc_timeout_factor=g.G_TC_SEND_TIMEOUT_FACTOR)
- tm_listener.start()
- if g.G_MODE_ID == g.ModeList.ListenerMode:
- logger.info("Listening for packages...")
- elif g.G_MODE_ID == g.ModeList.SingleCommandMode:
- pus_packet_tuple = command_preparation()
- sender_and_receiver = SingleCommandSenderReceiver(
- com_interface=communication_interface, tmtc_printer=tmtc_printer, tm_listener=tm_listener,
- pus_packet_tuple=pus_packet_tuple)
- logger.info("Performing single command operation")
- sender_and_receiver.send_single_tc_and_receive_tm()
-
- elif g.G_MODE_ID == g.ModeList.ServiceTestMode:
- service_queue = deque()
- logger.info("Performing service command operation")
- sender_and_receiver = SequentialCommandSenderReceiver(
- com_interface=communication_interface, tmtc_printer=tmtc_printer, tm_listener=tm_listener,
- tc_queue=service_test_select(g.G_SERVICE, service_queue))
- sender_and_receiver.send_queue_tc_and_receive_tm_sequentially()
-
- elif g.G_MODE_ID == g.ModeList.SoftwareTestMode:
- all_tc_queue = create_total_tc_queue()
- logger.info("Performing multjple service commands operation")
- sender_and_receiver = SequentialCommandSenderReceiver(
- com_interface=communication_interface, tmtc_printer=tmtc_printer,
- tc_queue=all_tc_queue, tm_listener=tm_listener)
- sender_and_receiver.send_queue_tc_and_receive_tm_sequentially()
-
- elif g.G_MODE_ID == g.ModeList.UnitTest:
- # Set up test suite and run it with runner. Verbosity specifies detail level
- g.G_TM_LISTENER = tm_listener
- g.G_COM_INTERFACE = communication_interface
- g.G_TMTC_PRINTER = tmtc_printer
- logger.info("Performing module tests")
- # noinspection PyTypeChecker
- suite = unittest.TestLoader().loadTestsFromModule(obsw_pus_service_test)
- unittest.TextTestRunner(verbosity=2).run(suite)
-
- else:
- logging.error("Unknown Mode, Configuration error !")
- sys.exit()
+ logger.info("Starting TMTC Handler")
+ tmtc_handler = TmTcHandler()
+ tmtc_handler.perform_operation()
+
# At some later point, the program will run permanently and be able to take commands.
# For now we put a permanent loop here so the program
# doesn't exit automatically (TM Listener is daemonic)
@@ -165,5 +109,80 @@ def command_preparation():
return command.pack_command_tuple()
+class TmTcHandler:
+ def __init__(self):
+ self.mode = g.G_MODE_ID
+ self.comIF = g.G_COM_IF
+ # This flag could be used later to command the TMTC Client with a front-end
+ self.command_received = True
+
+ def perform_operation(self):
+ while True:
+ if self.command_received:
+ self.handle_action()
+ self.command_received = False
+ logger.info("TMTC Client in idle mode")
+ time.sleep(5)
+
+
+ def handle_action(self):
+ tmtc_printer = TmTcPrinter(g.G_DISPLAY_MODE, g.G_PRINT_TO_FILE, True)
+ if self.mode == g.ModeList.GUIMode:
+ backend = TmTcBackend()
+ backend.start()
+ gui = TmTcGUI()
+ gui.start()
+ backend.join()
+ gui.join()
+ logger.info("Both processes have closed")
+ sys.exit()
+ else:
+ communication_interface = set_communication_interface(tmtc_printer)
+ atexit.register(keyboard_interrupt_handler, comInterface=communication_interface)
+ tm_listener = TmListener(
+ com_interface=communication_interface, tm_timeout=g.G_TM_TIMEOUT,
+ tc_timeout_factor=g.G_TC_SEND_TIMEOUT_FACTOR)
+ tm_listener.start()
+ if self.mode == g.ModeList.ListenerMode:
+ logger.info("Listening for packages...")
+ elif self.mode == g.ModeList.SingleCommandMode:
+ pus_packet_tuple = command_preparation()
+ sender_and_receiver = SingleCommandSenderReceiver(
+ com_interface=communication_interface, tmtc_printer=tmtc_printer,
+ tm_listener=tm_listener)
+ logger.info("Performing single command operation")
+ sender_and_receiver.send_single_tc_and_receive_tm(pus_packet_tuple=pus_packet_tuple)
+
+ elif self.mode== g.ModeList.ServiceTestMode:
+ service_queue = deque()
+ logger.info("Performing service command operation")
+ sender_and_receiver = SequentialCommandSenderReceiver(
+ com_interface=communication_interface, tmtc_printer=tmtc_printer,
+ tm_listener=tm_listener,
+ tc_queue=service_test_select(g.G_SERVICE, service_queue))
+ sender_and_receiver.send_queue_tc_and_receive_tm_sequentially()
+
+ elif self.mode == g.ModeList.SoftwareTestMode:
+ all_tc_queue = create_total_tc_queue()
+ logger.info("Performing multjple service commands operation")
+ sender_and_receiver = SequentialCommandSenderReceiver(
+ com_interface=communication_interface, tmtc_printer=tmtc_printer,
+ tc_queue=all_tc_queue, tm_listener=tm_listener)
+ sender_and_receiver.send_queue_tc_and_receive_tm_sequentially()
+
+ elif self.mode == g.ModeList.UnitTest:
+ # Set up test suite and run it with runner. Verbosity specifies detail level
+ g.G_TM_LISTENER = tm_listener
+ g.G_COM_INTERFACE = communication_interface
+ g.G_TMTC_PRINTER = tmtc_printer
+ logger.info("Performing module tests")
+ # noinspection PyTypeChecker
+ suite = unittest.TestLoader().loadTestsFromModule(obsw_pus_service_test)
+ unittest.TextTestRunner(verbosity=2).run(suite)
+
+ else:
+ logging.error("Unknown Mode, Configuration error !")
+ sys.exit()
+
if __name__ == "__main__":
main()
diff --git a/sendreceive/obsw_command_sender_receiver.py b/sendreceive/obsw_command_sender_receiver.py
index c09a01fe96bfffe752396425ecf595661f5ba3f2..07cdb1ed2738e40266099dbeaae19737377cfa35 100644
--- a/sendreceive/obsw_command_sender_receiver.py
+++ b/sendreceive/obsw_command_sender_receiver.py
@@ -43,27 +43,34 @@ class CommandSenderReceiver:
if isinstance(com_interface, CommunicationInterface):
self._com_interface = com_interface
else:
- raise TypeError("Invalid communication interface service_type!")
+ logger.error("CommandSenderReceiver: Invalid communication interface!")
+ raise TypeError("CommandSenderReceiver: Invalid communication interface!")
if isinstance(tmtc_printer, TmTcPrinter):
self._tmtc_printer = tmtc_printer
else:
- raise TypeError("Invalid TMTC Printer service_type!")
+ logger.error("CommandSenderReceiver: Invalid TMTC printer!")
+ raise TypeError("CommandSenderReceiver: Invalid TMTC printer!")
if isinstance(tm_listener, TmListener):
self._tm_listener = tm_listener
else:
- raise TypeError("Invalid TM Listener service_type!")
- self._reply_received = False
+ logger.error("CommandSenderReceiver: Invalid TM listener!")
+ raise TypeError("Invalid TM Listener!")
self._start_time = 0
self._elapsed_time = 0
self._timeout_counter = 0
- # needed to store last actual tc wiretapping_packet from queue
+ # needed to store last actual TC packet from queue
self._last_tc = bytearray()
self._last_tc_info = dict()
+ # this flag can be used to notify when the operation is finished
+ self._operation_pending = False
+ # This flag can be used to notify when a reply was received.
+ self._reply_received = False
+
def set_tm_timeout(self, tm_timeout: float = g.G_TM_TIMEOUT):
"""
Set the TM timeout. Usually, the global value set by the args parser is set,
@@ -89,6 +96,7 @@ class CommandSenderReceiver:
"""
if self._tm_listener.replyEvent.is_set():
self._reply_received = True
+ self._operation_pending = False
self._tm_listener.replyEvent.clear()
else:
self._check_for_timeout()
@@ -129,8 +137,8 @@ class CommandSenderReceiver:
if self._start_time == 0:
self._start_time = time.time()
if self._timeout_counter == 5:
- logger.info("Command Sender Receiver: No response from command !")
- sys.exit()
+ logger.info("CommandSenderReceiver: No response from command !")
+ self._operation_pending = False
if self._start_time != 0:
self._elapsed_time = time.time() - self._start_time
if self._elapsed_time >= self._tm_timeout * self._tc_send_timeout_factor:
diff --git a/sendreceive/obsw_sequential_sender_receiver.py b/sendreceive/obsw_sequential_sender_receiver.py
index bf8cf18c58fe03faab95bdb3ad2acfa2eb5ba32a..952b54aab6a4e00ac67c38bd21b292e5e66f9592 100644
--- a/sendreceive/obsw_sequential_sender_receiver.py
+++ b/sendreceive/obsw_sequential_sender_receiver.py
@@ -9,7 +9,7 @@ import time
import config.obsw_config as g
from sendreceive.obsw_command_sender_receiver import CommandSenderReceiver
-from sendreceive.obsw_tm_listener import TmListenerT
+from sendreceive.obsw_tm_listener import TmListener
from comIF.obsw_com_interface import CommunicationInterface
from utility.obsw_tmtc_printer import TmTcPrinter
from tc.obsw_pus_tc_base import TcQueueT
@@ -20,7 +20,7 @@ class SequentialCommandSenderReceiver(CommandSenderReceiver):
Specific implementation of CommandSenderReceiver to send multiple telecommands in sequence
"""
def __init__(self, com_interface: CommunicationInterface, tmtc_printer: TmTcPrinter,
- tm_listener: TmListenerT, tc_queue: TcQueueT):
+ tm_listener: TmListener, tc_queue: TcQueueT):
"""
:param com_interface: CommunicationInterface object, passed on to CommandSenderReceiver
:param tm_listener: TmListener object which runs in the background and receives
diff --git a/sendreceive/obsw_single_command_sender_receiver.py b/sendreceive/obsw_single_command_sender_receiver.py
index 677e3893b17323d3420b11eadf12ee947b79ee0b..1cdb00449d36b6f7b73b3535db5a9f9c2ad2e2cd 100644
--- a/sendreceive/obsw_single_command_sender_receiver.py
+++ b/sendreceive/obsw_single_command_sender_receiver.py
@@ -7,22 +7,29 @@
@brief
Used to send single tcs and listen for replies after that
"""
-import logging
from sendreceive.obsw_command_sender_receiver import CommandSenderReceiver
-from sendreceive.obsw_tm_listener import TmListenerT
+from sendreceive.obsw_tm_listener import TmListener
+
from comIF.obsw_com_interface import CommunicationInterface
+
from utility.obsw_tmtc_printer import TmTcPrinter
+from utility.obsw_logger import get_logger
+
+from tc.obsw_pus_tc_base import PusTcTupleT
import config.obsw_config as g
+
+logger = get_logger()
+
class SingleCommandSenderReceiver(CommandSenderReceiver):
"""
Specific implementation of CommandSenderReceiver to send a single telecommand
This object can be used by instantiating it and calling sendSingleTcAndReceiveTm()
"""
def __init__(self, com_interface: CommunicationInterface, tmtc_printer: TmTcPrinter,
- tm_listener: TmListenerT, pus_packet_tuple: tuple):
+ tm_listener: TmListener):
"""
:param com_interface: CommunicationInterface object, passed on to CommandSenderReceiver
:param tm_listener: TmListener object which runs in the background and receives all TM
@@ -30,31 +37,30 @@ class SingleCommandSenderReceiver(CommandSenderReceiver):
"""
super().__init__(com_interface=com_interface, tm_listener=tm_listener,
tmtc_printer=tmtc_printer)
- self.faulty_input = False
- try:
- self.pus_packet, self.pus_packet_info = pus_packet_tuple
- except TypeError:
- print("Invalid queue entry detected")
- logging.exception("Error")
- self.faulty_input = True
- def send_single_tc_and_receive_tm(self):
+
+ def send_single_tc_and_receive_tm(self, pus_packet_tuple: PusTcTupleT):
"""
Send a single telecommand passed to the class and wait for replies
:return:
"""
- if not self.faulty_input:
- self._tm_listener.mode_id = g.ModeList.SingleCommandMode
- self._tm_listener.mode_change_event.set()
- self._tmtc_printer.print_telecommand(self.pus_packet, self.pus_packet_info)
- self._com_interface.send_telecommand(self.pus_packet, self.pus_packet_info)
- self._last_tc = self.pus_packet
- self._last_tc_info = self.pus_packet_info
- while not self._reply_received:
- # wait until reply is received
- super()._check_for_first_reply()
- if self._reply_received:
- self._tm_listener.mode_op_finished.set()
- self.print_tm_queue(self._tm_listener.retrieve_tm_packet_queue())
- print("Single Command SenderReceiver: Reply received")
- print("Listening for packages ...")
+ try:
+ pus_packet, pus_packet_info = pus_packet_tuple
+ except TypeError:
+ logger.error("SingleCommandSenderReceiver: Invalid command input")
+ return
+ self._operation_pending = True
+ self._tm_listener.mode_id = g.ModeList.SingleCommandMode
+ self._tm_listener.mode_change_event.set()
+ self._tmtc_printer.print_telecommand(pus_packet, pus_packet_info)
+ self._com_interface.send_telecommand(pus_packet, pus_packet_info)
+ self._last_tc = pus_packet
+ self._last_tc_info = pus_packet_info
+ while self._operation_pending:
+ # wait until reply is received
+ super()._check_for_first_reply()
+ if self._reply_received:
+ self._tm_listener.mode_op_finished.set()
+ self.print_tm_queue(self._tm_listener.retrieve_tm_packet_queue())
+ logger.info("SingleCommandSenderReceiver: Reply received")
+ logger.info("Listening for packages ...")
diff --git a/tm/obsw_pus_tm_base.py b/tm/obsw_pus_tm_base.py
index b18784c7e8bc85ee062d79e2cf6e5b0b35526c83..a35eb6877082f73968ba0aa491f0e8379fb0c6df 100644
--- a/tm/obsw_pus_tm_base.py
+++ b/tm/obsw_pus_tm_base.py
@@ -186,6 +186,9 @@ class PusTelemetryPacked(PusTelemetry):
self.pack_subcounter = 0
self.time = 0
+ def pack(self) -> bytearray:
+
+
# pylint: disable=too-many-instance-attributes
diff --git a/tm/obsw_pus_tm_factory.py b/tm/obsw_pus_tm_factory.py
index b77ff9682bb65e64dea062fd0019de085de245f5..547b2a42aef206d5df522f6567a9e576c7408013 100644
--- a/tm/obsw_pus_tm_factory.py
+++ b/tm/obsw_pus_tm_factory.py
@@ -1,10 +1,4 @@
# -*- coding: utf-8 -*-
-"""
-Program: obsw_pus_tm_factory.py
-Date: 01.11.2019
-Description: Deserialize TM byte header_list into PUS TM Class
-Author: R.Mueller, S. Gaisser
-"""
from typing import Deque, List, Dict, Tuple
from tm.obsw_pus_tm_base import PusTelemetry, PusTmInfoT
from tm.obsw_tm_service_1 import Service1TM
@@ -21,8 +15,11 @@ PusTmTupleQueueT = Deque[PusTmTupleT]
class PusTelemetryFactory(object):
+ """
+ Deserialize TM bytearrays into PUS TM Classes
+ """
@staticmethod
- def create(raw_tm_packet: bytes) -> PusTelemetry:
+ def create(raw_tm_packet: bytearray) -> PusTelemetry:
service_type = raw_tm_packet[7]
if service_type == 1:
return Service1TM(raw_tm_packet)
@@ -43,7 +40,7 @@ class PusTelemetryFactory(object):
class Service2TM(PusTelemetry):
- def __init__(self, byte_array: bytes):
+ def __init__(self, byte_array: bytearray):
super().__init__(byte_array)
self.specify_packet_info("Raw Commanding Reply")
diff --git a/utility/obsw_args_parser.py b/utility/obsw_args_parser.py
index 33803a47ea1201805c3f1d4d7d5b160ecd5785f7..f0d3161761a85fe2686d3de20d532cfca52bb987 100644
--- a/utility/obsw_args_parser.py
+++ b/utility/obsw_args_parser.py
@@ -1,16 +1,6 @@
-#!/usr/bin/python3.7
-"""
-@file
- obsw_args_parser.py
-@date
- 11.01.2020
-@brief
- Reads all input arguments or requests them from user.
-"""
-
+#!/usr/bin/python3.8
import argparse
import sys
-import logging
from utility.obsw_logger import get_logger
diff --git a/utility/obsw_logger.py b/utility/obsw_logger.py
index 96e9511452e4fd44dd7201c60d34ae0e529ce229..ab343805763b9d830eabdf316c915b12cef242b4 100644
--- a/utility/obsw_logger.py
+++ b/utility/obsw_logger.py
@@ -5,12 +5,18 @@ import config.obsw_config as g
class InfoFilter(logging.Filter):
+ """
+ Filter object, which is used so that only INFO and DEBUG messages are printed to stdout.
+ """
def filter(self, rec):
if rec.levelno == logging.INFO or rec.levelno == logging.DEBUG:
return rec.levelno
def set_tmtc_logger() -> logging.Logger:
+ """
+ Sets the logger object which will be used globally.
+ """
logger = logging.getLogger(g.G_TMTC_LOGGER_NAME)
logger.setLevel(level=logging.DEBUG)
generic_format = logging.Formatter(