From 73f5b0bc7fdf90453c95738fcfb0a9124a89e08d Mon Sep 17 00:00:00 2001
From: "Robin.Mueller" <robin.mueller.m@gmail.com>
Date: Thu, 3 Dec 2020 23:35:03 +0100
Subject: [PATCH] cleaning up
---
core/tmtc_backend.py | 45 ++++++++++++++++-------------
pus_tc/tmtcc_tc_service2_raw_cmd.py | 2 +-
2 files changed, 26 insertions(+), 21 deletions(-)
diff --git a/core/tmtc_backend.py b/core/tmtc_backend.py
index e06d41d..05eb615 100644
--- a/core/tmtc_backend.py
+++ b/core/tmtc_backend.py
@@ -6,11 +6,6 @@ from multiprocessing import Process
from collections import deque
from typing import Tuple, Union
-from config import tmtcc_config as g
-from config.tmtcc_definitions import ModeList
-from config.tmtcc_user_code import command_preparation_hook
-from pus_tc.tmtcc_tc_packer_hook import create_total_tc_queue
-
from tmtc_core.tmtc_core_definitions import ComInterfaces
from tmtc_core.utility.obsw_logger import get_logger
from tmtc_core.pus_tc.tmtcc_pus_tc_base import PusTcInfo
@@ -20,15 +15,16 @@ from tmtc_core.sendreceive.obsw_tm_listener import TmListener
from tmtc_core.comIF.obsw_com_interface import CommunicationInterface
from tmtc_core.utility.obsw_tmtc_printer import TmTcPrinter
from tmtc_core.utility.obsw_exit_handler import keyboard_interrupt_handler
-
from tmtc_core.pus_tc.tmtcc_pus_tc_packer import ServiceQueuePacker
-from test.obsw_pus_service_test import run_selected_pus_tests
+
from config.tmtcc_com_config import set_communication_interface
+from config.tmtcc_definitions import ModeList
+from pus_tc.tmtcc_tc_packer_hook import create_total_tc_queue
+from test.obsw_pus_service_test import run_selected_pus_tests
from utility.tmtcc_binary_uploader import BinaryFileUploader
LOGGER = get_logger()
-
class TmTcHandler:
"""
This is the primary class which handles TMTC reception. This can be seen as the backend
@@ -67,7 +63,7 @@ class TmTcHandler:
@staticmethod
def prepare_tmtc_handler_start(
init_com_if: ComInterfaces = ComInterfaces.Dummy,
- init_mode: ModeList = g.ModeList.ListenerMode):
+ init_mode: ModeList = ModeList.ListenerMode):
tmtc_handler = TmTcHandler(init_com_if, init_mode)
tmtc_task = Process(target=TmTcHandler.start_handler, args=(tmtc_handler, ))
return tmtc_task
@@ -81,6 +77,7 @@ class TmTcHandler:
executed_handler.start()
def initialize(self):
+ from config import tmtcc_config as g
"""
Perform initialization steps which might be necessary after class construction.
This has to be called at some point before using the class!
@@ -122,17 +119,17 @@ class TmTcHandler:
"""
Command handling.
"""
- if self.mode == g.ModeList.PromptMode:
+ if self.mode == ModeList.PromptMode:
self.prompt_mode()
- if self.mode == g.ModeList.ListenerMode:
+ if self.mode == ModeList.ListenerMode:
if self.tm_listener.reply_event():
LOGGER.info("TmTcHandler: Packets received.")
self.tmtc_printer.print_telemetry_queue(self.tm_listener.retrieve_tm_packet_queue())
self.tm_listener.clear_tm_packet_queue()
self.tm_listener.clear_reply_event()
- elif self.mode == g.ModeList.SingleCommandMode:
+ elif self.mode == ModeList.SingleCommandMode:
if self.single_command_package is None:
pus_packet_tuple = command_preparation()
else:
@@ -143,9 +140,10 @@ class TmTcHandler:
tm_listener=self.tm_listener)
LOGGER.info("Performing single command operation..")
sender_and_receiver.send_single_tc_and_receive_tm(pus_packet_tuple=pus_packet_tuple)
- self.mode = g.ModeList.PromptMode
+ self.mode = ModeList.PromptMode
- elif self.mode == g.ModeList.ServiceTestMode:
+ elif self.mode == ModeList.ServiceTestMode:
+ from config import tmtcc_config as g
service_queue = deque()
service_queue_packer = ServiceQueuePacker()
op_code = g.G_OP_CODE
@@ -170,15 +168,16 @@ class TmTcHandler:
LOGGER.info("SequentialSenderReceiver: Exporting output to log file.")
self.tmtc_printer.print_file_buffer_list_to_file("log/tmtc_log.txt", True)
- elif self.mode == g.ModeList.BinaryUploadMode:
+ elif self.mode == ModeList.BinaryUploadMode:
# Upload binary, prompt user for input, in the end prompt for new mode and enter that
# mode
file_uploader = BinaryFileUploader(self.communication_interface, self.tmtc_printer,
self.tm_listener)
file_uploader.perform_file_upload()
- self.mode = g.ModeList.ListenerMode
+ self.mode = ModeList.ListenerMode
- elif self.mode == g.ModeList.UnitTest:
+ elif self.mode == ModeList.UnitTest:
+ from config import tmtcc_config as g
# Set up test suite and run it with runner. Verbosity specifies detail level
g.G_TM_LISTENER = self.tm_listener
g.G_COM_INTERFACE = self.communication_interface
@@ -191,15 +190,15 @@ class TmTcHandler:
sys.exit()
def __core_operation(self, one_shot):
- if self.mode == g.ModeList.ListenerMode:
+ if self.mode == ModeList.ListenerMode:
one_shot = False
if not one_shot:
while True:
self.__handle_action()
- if self.mode == g.ModeList.Idle:
+ if self.mode == ModeList.Idle:
LOGGER.info("TMTC Client in idle mode")
time.sleep(5)
- elif self.mode == g.ModeList.ListenerMode:
+ elif self.mode == ModeList.ListenerMode:
time.sleep(1)
else:
self.__handle_action()
@@ -238,4 +237,10 @@ def command_preparation() -> Tuple[bytearray, Union[None, PusTcInfo]]:
Prepare command for single command testing
:return:
"""
+ try:
+ from config.tmtcc_user_code import command_preparation_hook
+ except ImportError as e:
+ print(e)
+ LOGGER.error("Hook function for command application not implemented!")
+ sys.exit(1)
return command_preparation_hook()
diff --git a/pus_tc/tmtcc_tc_service2_raw_cmd.py b/pus_tc/tmtcc_tc_service2_raw_cmd.py
index deee9e6..d40f21e 100644
--- a/pus_tc/tmtcc_tc_service2_raw_cmd.py
+++ b/pus_tc/tmtcc_tc_service2_raw_cmd.py
@@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
"""
@file tmtcc_tc_service2_raw_cmd.py
-@brief PUS Service 2: Device Access, Native low-level commanding
+@brief PUS Service 2: Device Access, native low-level commanding
@author R. Mueller
@date 01.11.2019
"""
--
GitLab