Merge pull request #55 from robamu-org/eive

Various improvements and fixes

CHANGELOG.md

@@ -15,6 +15,12 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/).
 - Auto-Complete feature for service and op-code selection using the `prompt-toolkit`
   packaged
 
+### Fixed
+
+- Added missing super constructor call for HkReplyUnpacked
+- Extended Op Code options functionality and actually use it. Allows to set custom timeout
+  or and enter listener mode for certain op codes
+
 ## [v1.12.0]
 
 ### API Changes

src/tmtccmd/__init__.py

@@ -1,8 +1,8 @@
 VERSION_NAME = "tmtccmd"
 VERSION_MAJOR = 1
-VERSION_MINOR = 12
+VERSION_MINOR = 13
 VERSION_REVISION = 0
 
 # I think this needs to be in string representation to be parsed so we can't
 # use a formatted string here.
-__version__ = "1.12.0"
+__version__ = "1.13.0"

src/tmtccmd/config/__init__.py

@@ -1,4 +1,8 @@
-from .globals import add_op_code_entry, add_service_op_code_entry
+from .globals import (
+    add_op_code_entry,
+    add_service_op_code_entry,
+    generate_op_code_options,
+)
 from .definitions import (
     QueueCommands,
     ServiceOpCodeDictT,

src/tmtccmd/config/args.py

@@ -6,7 +6,12 @@
 
 from prompt_toolkit.completion import WordCompleter
 import prompt_toolkit
-from tmtccmd.config.definitions import CoreModeList, ServiceOpCodeDictT, OpCodeEntryT
+from tmtccmd.config.definitions import (
+    CoreModeList,
+    ServiceOpCodeDictT,
+    OpCodeEntryT,
+    OpCodeDictKeys,
+)
 from tmtccmd.utility.logger import get_console_logger
 
 
@@ -121,7 +126,8 @@ def add_generic_arguments(arg_parser: argparse.ArgumentParser):
         "-l",
         "--listener",
         help="Determine whether the listener mode will be active after performing the operation",
-        action="store_false",
+        action="store_true",
+        default=None,
     )
     arg_parser.add_argument(
         "-t",
@@ -262,6 +268,32 @@ def handle_unspecified_args(args) -> None:
         args.op_code = prompt_op_code(
             service_op_code_dict=service_op_code_dict, service=current_service
         )
+    op_code_value = service_op_code_dict[args.service][1]
+    op_code_options = op_code_value[args.op_code][1]
+    if op_code_options is not None and isinstance(op_code_options, dict):
+        if op_code_options.get(OpCodeDictKeys.ENTER_LISTENER_MODE):
+            if args.listener is None:
+                LOGGER.info(
+                    "Detected op code configuration: Enter listener mode after command"
+                )
+                args.listener = True
+            else:
+                LOGGER.warning(
+                    "Detected op code listerner mode configuration but is overriden by CLI argument"
+                )
+        timeout = op_code_options.get(OpCodeDictKeys.TIMEOUT)
+        if timeout is not None:
+            if args.tm_timeout is None:
+                LOGGER.info(
+                    f"Detected op code configuration: Set custom timeout {timeout}"
+                )
+                args.tm_timeout = timeout
+            else:
+                LOGGER.warning(
+                    "Detected op code timeout configuration but is overriden by CLI argument"
+                )
+    if args.listener is None:
+        args.listener = False
 
 
 def handle_empty_args(args) -> None:
@@ -319,7 +351,7 @@ def build_service_word_completer(
 
 def prompt_op_code(service_op_code_dict: ServiceOpCodeDictT, service: str) -> str:
     op_code_adjustment = 24
-    info_adjustment = 40
+    info_adjustment = 56
     horz_line_num = op_code_adjustment + info_adjustment + 3
     horiz_line = horz_line_num * "-"
     op_code_info_str = "Operation Code".ljust(op_code_adjustment)

src/tmtccmd/config/definitions.py

@@ -46,6 +46,7 @@ class CoreGlobalIds(enum.IntEnum):
 
 class OpCodeDictKeys(enum.IntEnum):
     TIMEOUT = CoreGlobalIds.TM_TIMEOUT
+    ENTER_LISTENER_MODE = CoreGlobalIds.USE_LISTENER_AFTER_OP
 
 
 # Service Op Code Dictionary Types
@@ -77,6 +78,7 @@ def __init__(self):
 
 class HkReplyUnpacked(DataReplyUnpacked):
     def __init__(self):
+        super().__init__()
         # Validity buffer
         self.validity_buffer = bytearray()
         # Number of variables contained in HK set

src/tmtccmd/config/globals.py

@@ -1,7 +1,7 @@
 import argparse
 import collections.abc
 import pprint
-from typing import Union, List, Dict
+from typing import Union, List, Dict, Optional
 
 from tmtccmd.utility.logger import get_console_logger
 from tmtccmd.utility.conf_util import check_args_in_dict, print_core_globals
@@ -338,3 +338,12 @@ def add_service_op_code_entry(
     op_code_entry: OpCodeEntryT,
 ):
     srv_op_code_dict.update({name: (info, op_code_entry)})
+
+
+def generate_op_code_options(
+    enter_listener_mode: bool = False, custom_timeout: Optional[float] = None
+):
+    op_code_opts = dict()
+    op_code_opts.update({OpCodeDictKeys.ENTER_LISTENER_MODE: enter_listener_mode})
+    if custom_timeout is not None:
+        op_code_opts.update({OpCodeDictKeys.TIMEOUT: custom_timeout})

src/tmtccmd/core/backend.py

@@ -4,7 +4,7 @@
 from threading import Thread
 from abc import abstractmethod
 from collections import deque
-from typing import Union
+from typing import Union, cast
 
 from tmtccmd.config.definitions import CoreServiceList, CoreModeList
 from tmtccmd.tm.definitions import TmTypes
@@ -63,13 +63,13 @@ def __init__(
         self.__apid = 0
 
         # This flag could be used later to command the TMTC Client with a front-end
-        self.one_shot_operation = True
+        self.one_shot_operation = False
 
         self.__com_if = com_if
         self.__tmtc_printer = tmtc_printer
         self.__tm_listener = tm_listener
         if tm_handler.get_type() == TmTypes.CCSDS_SPACE_PACKETS:
-            self.__tm_handler: CcsdsTmHandler = tm_handler
+            self.__tm_handler: CcsdsTmHandler = cast(CcsdsTmHandler, tm_handler)
             for apid_queue_len_tuple in self.__tm_handler.get_apid_queue_len_list():
                 self.__tm_listener.subscribe_ccsds_tm_handler(
                     apid_queue_len_tuple[0], apid_queue_len_tuple[1]
@@ -141,6 +141,7 @@ def prepare_tmtc_handler_start(
         com_if: CommunicationInterface,
         tmtc_printer: TmTcPrinter,
         tm_listener: TmListener,
+        tm_handler: TmHandler,
         init_mode: int,
         init_service: Union[str, int] = CoreServiceList.SERVICE_17.value,
         init_opcode: str = "0",
@@ -154,6 +155,7 @@ def prepare_tmtc_handler_start(
             init_mode=init_mode,
             init_service=init_service,
             init_opcode=init_opcode,
+            tm_handler=tm_handler,
         )
         tmtc_task = Process(target=TmTcHandler.start_handler, args=(tmtc_handler,))
         return tmtc_task
@@ -273,7 +275,7 @@ def __handle_action(self):
                 print(error)
                 LOGGER.error("Custom mode handling module not provided!")
 
-    def __core_operation(self, one_shot):
+    def __core_operation(self, one_shot: bool):
         if self.mode == CoreModeList.LISTENER_MODE:
             one_shot = False
         if not one_shot:

src/tmtccmd/runner.py

@@ -294,6 +294,6 @@ def get_default_tmtc_backend(
     )
     tmtc_backend.set_current_apid(apid=apid)
     tmtc_backend.set_one_shot_or_loop_handling(
-        get_global(CoreGlobalIds.USE_LISTENER_AFTER_OP)
+        not get_global(CoreGlobalIds.USE_LISTENER_AFTER_OP)
     )
     return tmtc_backend