Revert everything to 3.8 compatible

src/pyrtma/data_logger/__init__.py

@@ -1,11 +1,3 @@
-import sys
-
-if sys.version_info.major < 3:
-    raise ImportError("DataLogger requires python 3.10 or greater")
-
-if sys.version_info.minor < 10:
-    raise ImportError("DataLogger requires python 3.10 or greater")
-
 from .data_logger import DataLogger
 
 

src/pyrtma/data_logger/__main__.py

@@ -37,7 +37,17 @@ def main():
 
     args = parser.parse_args()
 
-    log_level = logging.getLevelNamesMapping().get(args.log_level) or logging.INFO
+    if args.log_level == "DEBUG":
+        level = logging.DEBUG
+    elif args.log_level == "INFO":
+        level = logging.INFO
+    elif args.log_level == "WARN":
+        level = logging.WARN
+    elif args.log_level == "ERROR":
+        level = logging.ERROR
+    else:
+        print("Unknown log level. Using INFO instead")
+        level = logging.INFO
 
     if args.defs_file:
         base = pathlib.Path(args.defs_file).absolute().parent
@@ -46,7 +56,7 @@ def main():
         sys.path.insert(0, (str(base.absolute())))
         importlib.import_module(fname)
 
-    d = DataLogger(args.mm_ip, log_level=log_level)
+    d = DataLogger(args.mm_ip, log_level=level)
 
     try:
         d.run()

src/pyrtma/data_logger/cli.py

@@ -81,40 +81,39 @@ def main():
         while True:
             cmd = input("(data_log)>> ")
 
-            match (cmd):
-                case "add-collection":
-                    add_collection(mod)
-                case "add-data-set":
-                    add_data_set(mod)
-                case "rm-collection":
-                    rm_collection(mod)
-                case "rm-data-set":
-                    rm_data_set(mod)
-                case "pause":
-                    mod.send_signal(cd.MT_DATA_LOGGER_PAUSE)
-                case "resume":
-                    mod.send_signal(cd.MT_DATA_LOGGER_RESUME)
-                case "reset":
-                    mod.send_signal(cd.MT_DATA_LOGGER_RESET)
-                case "start":
-                    mod.send_signal(cd.MT_DATA_LOGGER_START)
-                case "stop":
-                    mod.send_signal(cd.MT_DATA_LOGGER_STOP)
-                case "get-status":
-                    mod.send_signal(cd.MT_DATA_LOGGER_STATUS_REQUEST)
-                case "get-config":
-                    mod.send_signal(cd.MT_DATA_COLLECTION_CONFIG_REQUEST)
-                case "get-metadata":
-                    mod.send_signal(cd.MT_DATA_LOGGER_METADATA_REQUEST)
-                case "update-metadata":
-                    update_metadata(mod, metadata)
-                case "help":
-                    help()
-                case "exit":
-                    break
-                case _:
-                    print(f"Unknown command: {cmd}")
-                    help()
+            if cmd == "add-collection":
+                add_collection(mod)
+            elif cmd == "add-data-set":
+                add_data_set(mod)
+            elif cmd == "rm-collection":
+                rm_collection(mod)
+            elif cmd == "rm-data-set":
+                rm_data_set(mod)
+            elif cmd == "pause":
+                mod.send_signal(cd.MT_DATA_LOGGER_PAUSE)
+            elif cmd == "resume":
+                mod.send_signal(cd.MT_DATA_LOGGER_RESUME)
+            elif cmd == "reset":
+                mod.send_signal(cd.MT_DATA_LOGGER_RESET)
+            elif cmd == "start":
+                mod.send_signal(cd.MT_DATA_LOGGER_START)
+            elif cmd == "stop":
+                mod.send_signal(cd.MT_DATA_LOGGER_STOP)
+            elif cmd == "get-status":
+                mod.send_signal(cd.MT_DATA_LOGGER_STATUS_REQUEST)
+            elif cmd == "get-config":
+                mod.send_signal(cd.MT_DATA_COLLECTION_CONFIG_REQUEST)
+            elif cmd == "get-metadata":
+                mod.send_signal(cd.MT_DATA_LOGGER_METADATA_REQUEST)
+            elif cmd == "update-metadata":
+                update_metadata(mod, metadata)
+            elif cmd == "help":
+                help()
+            elif cmd == "exit":
+                break
+            else:
+                print(f"Unknown command: {cmd}")
+                help()
 
             msg = mod.read_message(0.200)
             if msg:

src/pyrtma/data_logger/data_formatter.py

@@ -1,5 +1,5 @@
 import pyrtma
-from typing import Type, Optional, ClassVar, IO, Any
+from typing import Type, Optional, ClassVar, IO, Any, Union, List
 from abc import ABC
 from .exceptions import InvalidFormatter, DataFormatterKeyError
 
@@ -17,19 +17,19 @@ def __init__(self, fd: IO[Any], **kwargs):
         if hdr := self.format_header():
             self.fd.write(hdr)
 
-    def format_header(self) -> Optional[bytes | str]:
+    def format_header(self) -> Optional[Union[bytes, str]]:
         return None
 
-    def format_footer(self) -> Optional[bytes | str]:
+    def format_footer(self) -> Optional[Union[bytes, str]]:
         return None
 
-    def format_message(self, msg: pyrtma.Message) -> Optional[str | bytes]:
+    def format_message(self, msg: pyrtma.Message) -> Optional[Union[str, bytes]]:
         return None
 
-    def write(self, wbuf: list[pyrtma.Message]):
+    def write(self, wbuf: List[pyrtma.Message]):
         self.fd.writelines(self.format_message(msg) for msg in wbuf)
 
-    def finalize(self, wbuf: list[pyrtma.Message]):
+    def finalize(self, wbuf: List[pyrtma.Message]):
         # Write the last bit of data
         self.write(wbuf)
 

src/pyrtma/data_logger/data_logger.py

@@ -300,43 +300,42 @@ def run(self):
                     continue
 
                 try:
-                    match (msg.data):
-                        case cd.MT_DATA_LOGGER_START:
-                            self.start_logging()
-                        case cd.MT_DATA_LOGGER_STOP:
-                            self.stop_logging()
-                        case cd.MT_DATA_LOGGER_PAUSE:
-                            self.pause_logging()
-                        case cd.MT_DATA_LOGGER_RESUME:
-                            self.resume_logging()
-                        case cd.MT_ADD_DATA_COLLECTION:
-                            self.add_data_collection(
-                                cast(cd.MDF_ADD_DATA_COLLECTION, msg.data)
-                            )
-                        case cd.MT_ADD_DATA_SET:
-                            self.add_data_set(cast(cd.MDF_ADD_DATA_SET, msg.data))
-                        case cd.MT_REMOVE_DATA_COLLECTION:
-                            self.rm_data_collection()
-                        case cd.MT_REMOVE_DATA_SET:
-                            self.rm_data_set(cast(cd.MDF_REMOVE_DATA_SET, msg.data))
-                        case cd.MT_DATA_LOGGER_RESET:
-                            self.reset()
-                        case cd.MT_DATA_LOGGER_STATUS_REQUEST:
-                            self.send_status()
-                        case cd.MT_DATA_COLLECTION_CONFIG_REQUEST:
-                            self.send_config()
-                        case cd.MT_DATA_LOGGER_METADATA_UPDATE:
-                            self.update_metadata(
-                                cast(cd.MDF_DATA_LOGGER_METADATA_UPDATE, msg.data).json
-                            )
-                        case cd.MT_DATA_LOGGER_METADATA_REQUEST:
-                            meta = cd.MDF_DATA_LOGGER_METADATA()
-                            meta.json = self.metadata.to_json()
-                            self.mod.send_message(meta)
-                        case cd.MT_EXIT:
-                            if msg.header.dest_mod_id == self.mod.module_id:
-                                self._running = False
-                                self.mod.info("Received EXIT request. Closing...")
+                    if msg.type_id == cd.MT_DATA_LOGGER_START:
+                        self.start_logging()
+                    elif msg.type_id == cd.MT_DATA_LOGGER_STOP:
+                        self.stop_logging()
+                    elif msg.type_id == cd.MT_DATA_LOGGER_PAUSE:
+                        self.pause_logging()
+                    elif msg.type_id == cd.MT_DATA_LOGGER_RESUME:
+                        self.resume_logging()
+                    elif msg.type_id == cd.MT_ADD_DATA_COLLECTION:
+                        self.add_data_collection(
+                            cast(cd.MDF_ADD_DATA_COLLECTION, msg.data)
+                        )
+                    elif msg.type_id == cd.MT_ADD_DATA_SET:
+                        self.add_data_set(cast(cd.MDF_ADD_DATA_SET, msg.data))
+                    elif msg.type_id == cd.MT_REMOVE_DATA_COLLECTION:
+                        self.rm_data_collection()
+                    elif msg.type_id == cd.MT_REMOVE_DATA_SET:
+                        self.rm_data_set(cast(cd.MDF_REMOVE_DATA_SET, msg.data))
+                    elif msg.type_id == cd.MT_DATA_LOGGER_RESET:
+                        self.reset()
+                    elif msg.type_id == cd.MT_DATA_LOGGER_STATUS_REQUEST:
+                        self.send_status()
+                    elif msg.type_id == cd.MT_DATA_COLLECTION_CONFIG_REQUEST:
+                        self.send_config()
+                    elif msg.type_id == cd.MT_DATA_LOGGER_METADATA_UPDATE:
+                        self.update_metadata(
+                            cast(cd.MDF_DATA_LOGGER_METADATA_UPDATE, msg.data).json
+                        )
+                    elif msg.type_id == cd.MT_DATA_LOGGER_METADATA_REQUEST:
+                        meta = cd.MDF_DATA_LOGGER_METADATA()
+                        meta.json = self.metadata.to_json()
+                        self.mod.send_message(meta)
+                    elif msg.type_id == cd.MT_EXIT:
+                        if msg.header.dest_mod_id == self.mod.module_id:
+                            self._running = False
+                            self.mod.info("Received EXIT request. Closing...")
                 except DataLoggerError as e:
                     self.mod.error(e.args[0])
                     self.send_error(f"{e.__class__.__name__}:{e.args[0]}")

src/pyrtma/data_logger/data_set.py

@@ -5,7 +5,7 @@
 
 from ..message import Message
 from ..core_defs import ALL_MESSAGE_TYPES
-from typing import Type, Optional, IO, Any
+from typing import Type, Optional, IO, Any, List
 
 from .metadata import LoggingMetadata
 from .data_formatter import DataFormatter
@@ -25,7 +25,7 @@ def __init__(
         file_name_fmt: str,
         formatter_cls: Type[DataFormatter],
         subdivide_interval: int,
-        msg_types: list[int],
+        msg_types: List[int],
         metadata: LoggingMetadata,
     ):
         self.logger = logging.getLogger("data_logger").getChild(
@@ -53,8 +53,8 @@ def __init__(
         else:
             self.formatter = formatter_cls(io.StringIO())
 
-        self.rbuf: list[Message] = []
-        self.wbuf: list[Message] = []
+        self.rbuf: List[Message] = []
+        self.wbuf: List[Message] = []
 
         self.fd: Optional[IO[Any]] = None
         self.file_path = pathlib.Path()

src/pyrtma/data_logger/metadata.py

@@ -4,11 +4,12 @@
 import os
 
 from .exceptions import MissingMetadata, InvalidMetadata
+from typing import Union
 
 
 class LoggingMetadata:
     def __init__(self):
-        self._metadata: dict[str, str | float | int] = {}
+        self._metadata: dict[str, Union[str, float, int]] = {}
 
     def update(self, json_str: str):
         try:

src/pyrtma/manager.py

@@ -931,7 +931,17 @@ def main():
     else:
         ip_addr = ""  # socket.INADDR_ANY
 
-    level = logging.getLevelNamesMapping().get(args.log_level) or logging.INFO
+    if args.log_level == "DEBUG":
+        level = logging.DEBUG
+    elif args.log_level == "INFO":
+        level = logging.INFO
+    elif args.log_level == "WARN":
+        level = logging.WARN
+    elif args.log_level == "ERROR":
+        level = logging.ERROR
+    else:
+        print("Unknown log level. Using INFO instead")
+        level = logging.INFO
 
     with disable_message_validation():
         msg_mgr = MessageManager(