Deprecate single-channel only methods, unify new methods

Add deprecation for purely single-channel only legacy methods, place new
methods as reusable base methods, which can potentially be implemented
in the future for single-channel as well.

Signed-off-by: Pierre R. Mai <pmai@pmsf.de>

Author: pmai

osi3trace/osi_trace.py

@@ -7,6 +7,7 @@
 import struct
 
 from abc import ABC, abstractmethod
+from typing_extensions import deprecated
 
 from mcap_protobuf.decoder import DecoderFactory
 from mcap.reader import make_reader
@@ -139,40 +140,38 @@ def __iter__(self):
     def close(self):
         return self.reader.close()
 
+    @deprecated("This is a legacy interface only supported for single-channel traces, which will be removed in future versions.")
     def retrieve_offsets(self, limit=None):
         if isinstance(self.reader, OSITraceSingle):
             return self.reader.retrieve_offsets(limit)
         raise NotImplementedError("Offsets are only supported for single-channel traces.")
 
+    @deprecated("This is a legacy interface only supported for single-channel traces, which will be removed in future versions.")
     def retrieve_message(self, index=None, skip=False):
         if isinstance(self.reader, OSITraceSingle):
             return self.reader.retrieve_message(index, skip)
         raise NotImplementedError("Index-based message retrieval is only supported for single-channel traces.")
 
+    @deprecated("This is a legacy interface only supported for single-channel traces, which will be removed in future versions.")
     def get_message_by_index(self, index):
         if isinstance(self.reader, OSITraceSingle):
             return self.reader.get_message_by_index(index)
         raise NotImplementedError("Index-based message retrieval is only supported for single-channel traces.")
 
+    @deprecated("This is a legacy interface only supported for single-channel traces, which will be removed in future versions.")
     def get_messages_in_index_range(self, begin, end):
         if isinstance(self.reader, OSITraceSingle):
             return self.reader.get_messages_in_index_range(begin, end)
         raise NotImplementedError("Index-based message retrieval is only supported for single-channel traces.")
 
     def get_available_topics(self):
-        if isinstance(self.reader, OSITraceMulti):
-            return self.reader.get_available_topics()
-        raise NotImplementedError("Getting available topics is only supported for multi-channel traces.")
+        return self.reader.get_available_topics()
 
     def get_file_metadata(self):
-        if isinstance(self.reader, OSITraceMulti):
-            return self.reader.get_file_metadata()
-        raise NotImplementedError("Getting file metadata is only supported for multi-channel traces.")
+        return self.reader.get_file_metadata()
 
     def get_channel_metadata(self):
-        if isinstance(self.reader, OSITraceMulti):
-            return self.reader.get_channel_metadata()
-        raise NotImplementedError("Getting channel metadata is only supported for multi-channel traces.")
+        return self.reader.get_channel_metadata()
 
 
 class ReaderBase(ABC):
@@ -190,6 +189,18 @@ def __iter__(self):
     def close(self):
         pass
 
+    @abstractmethod
+    def get_available_topics(self):
+        pass
+
+    @abstractmethod
+    def get_file_metadata(self):
+        pass
+
+    @abstractmethod
+    def get_channel_metadata(self):
+        pass
+
 
 class OSITraceSingle(ReaderBase):
     """OSI single-channel trace reader"""
@@ -203,21 +214,16 @@ def __init__(self, path=None, type_name="SensorView", cache_messages=False):
         self.message_cache = {} if cache_messages else None
         self._header_length = 4
         if path:
-            self.from_file(path, type_name, cache_messages)
-
-    def from_file(self, path, type_name="SensorView", cache_messages=False):
-        """Import a trace from a file"""
-        self.type = OSITrace.map_message_type(type_name)
+            self.type = OSITrace.map_message_type(type_name)
 
-        if path.suffix.lower() in [".lzma", ".xz"]:
-            self.file = lzma.open(path, "rb")
-        else:
-            self.file = open(path, "rb")
-
-        self.read_complete = False
-        self.current_index = 0
-        self.message_offsets = [0]
-        self.message_cache = {} if cache_messages else None
+            if path.suffix.lower() in [".lzma", ".xz"]:
+                self.file = lzma.open(path, "rb")
+            else:
+                self.file = open(path, "rb")
+            self.read_complete = False
+            self.current_index = 0
+            self.message_offsets = [0]
+            self.message_cache = {} if cache_messages else None
 
     def retrieve_offsets(self, limit=None):
         """Retrieve the offsets of the messages from the file."""
@@ -337,6 +343,15 @@ def close(self):
         self.read_limit = None
         self.type = None
 
+    def get_available_topics(self):
+        raise NotImplementedError("Getting available topics is only supported for multi-channel traces.")
+
+    def get_file_metadata(self):
+        raise NotImplementedError("Getting file metadata is only supported for multi-channel traces.")
+
+    def get_channel_metadata(self):
+        raise NotImplementedError("Getting channel metadata is only supported for multi-channel traces.")
+
 
 class OSITraceMulti(ReaderBase):
     """OSI multi-channel trace reader"""
@@ -387,7 +402,7 @@ def get_channel_metadata(self):
             if channel.topic == self.topic:
                 return channel.metadata
         return None
-    
+
     def get_message_type(self):
         for channel in self._summary.channels.values():
             if channel.topic == self.topic:
@@ -397,7 +412,7 @@ def get_message_type(self):
                 else:
                     raise ValueError(f"Schema '{schema.name}' is not an 'osi3.' schema.")
         return None
-    
+
     def _channel_is_of_type(self, channel, type_name):
         schema = self._summary.schemas[channel.schema_id]  
         return type_name is None or schema.name == f"osi3.{type_name}"

poetry.lock

@@ -12,6 +12,7 @@ readme = "README.md"
 dynamic = ["version"]
 requires-python = ">=3.10"
 dependencies = [
+    "typing-extensions~=4.9",
     "protobuf>=6.30.2",
     "mcap>=1.2.2",
     "mcap-protobuf-support>=0.5.3"

pyproject.toml

@@ -1,6 +1,7 @@
 import os
 import tempfile
 import unittest
+import warnings
 
 from osi3trace.osi_trace import OSITrace
 from osi3.osi_sensorview_pb2 import SensorView
@@ -30,7 +31,9 @@ def test_osi_trace_sv(self):
                     self.assertIsInstance(message, SensorView)
                     f.write(str(message))
 
-            self.assertEqual(len(trace.retrieve_offsets()), 10)
+            with warnings.catch_warnings():
+                warnings.simplefilter("ignore")
+                self.assertEqual(len(trace.retrieve_offsets()), 10)
             trace.close()
 
             self.assertTrue(os.path.exists(path_output))
@@ -47,7 +50,9 @@ def test_osi_trace_svc(self):
                     self.assertIsInstance(message, SensorViewConfiguration)
                     f.write(str(message))
 
-            self.assertEqual(len(trace.retrieve_offsets()), 1)
+            with warnings.catch_warnings():
+                warnings.simplefilter("ignore")
+                self.assertEqual(len(trace.retrieve_offsets()), 1)
             trace.close()
 
             self.assertTrue(os.path.exists(path_output))
@@ -64,7 +69,9 @@ def test_osi_trace_gt(self):
                     self.assertIsInstance(message, GroundTruth)
                     f.write(str(message))
 
-            self.assertEqual(len(trace.retrieve_offsets()), 10)
+            with warnings.catch_warnings():
+                warnings.simplefilter("ignore")
+                self.assertEqual(len(trace.retrieve_offsets()), 10)
             trace.close()
 
             self.assertTrue(os.path.exists(path_output))
@@ -81,7 +88,9 @@ def test_osi_trace_hvd(self):
                     self.assertIsInstance(message, HostVehicleData)
                     f.write(str(message))
 
-            self.assertEqual(len(trace.retrieve_offsets()), 10)
+            with warnings.catch_warnings():
+                warnings.simplefilter("ignore")
+                self.assertEqual(len(trace.retrieve_offsets()), 10)
             trace.close()
 
             self.assertTrue(os.path.exists(path_output))
@@ -98,7 +107,9 @@ def test_osi_trace_sd(self):
                     self.assertIsInstance(message, SensorData)
                     f.write(str(message))
 
-            self.assertEqual(len(trace.retrieve_offsets()), 10)
+            with warnings.catch_warnings():
+                warnings.simplefilter("ignore")
+                self.assertEqual(len(trace.retrieve_offsets()), 10)
             trace.close()
 
             self.assertTrue(os.path.exists(path_output))
@@ -115,7 +126,9 @@ def test_osi_trace_tc(self):
                     self.assertIsInstance(message, TrafficCommand)
                     f.write(str(message))
 
-            self.assertEqual(len(trace.retrieve_offsets()), 10)
+            with warnings.catch_warnings():
+                warnings.simplefilter("ignore")
+                self.assertEqual(len(trace.retrieve_offsets()), 10)
             trace.close()
 
             self.assertTrue(os.path.exists(path_output))
@@ -132,7 +145,9 @@ def test_osi_trace_tcu(self):
                     self.assertIsInstance(message, TrafficCommandUpdate)
                     f.write(str(message))
 
-            self.assertEqual(len(trace.retrieve_offsets()), 10)
+            with warnings.catch_warnings():
+                warnings.simplefilter("ignore")
+                self.assertEqual(len(trace.retrieve_offsets()), 10)
             trace.close()
 
             self.assertTrue(os.path.exists(path_output))
@@ -149,7 +164,9 @@ def test_osi_trace_tu(self):
                     self.assertIsInstance(message, TrafficUpdate)
                     f.write(str(message))
 
-            self.assertEqual(len(trace.retrieve_offsets()), 10)
+            with warnings.catch_warnings():
+                warnings.simplefilter("ignore")
+                self.assertEqual(len(trace.retrieve_offsets()), 10)
             trace.close()
 
             self.assertTrue(os.path.exists(path_output))
@@ -166,7 +183,9 @@ def test_osi_trace_mr(self):
                     self.assertIsInstance(message, MotionRequest)
                     f.write(str(message))
 
-            self.assertEqual(len(trace.retrieve_offsets()), 10)
+            with warnings.catch_warnings():
+                warnings.simplefilter("ignore")
+                self.assertEqual(len(trace.retrieve_offsets()), 10)
             trace.close()
 
             self.assertTrue(os.path.exists(path_output))
@@ -183,7 +202,9 @@ def test_osi_trace_su(self):
                     self.assertIsInstance(message, StreamingUpdate)
                     f.write(str(message))
 
-            self.assertEqual(len(trace.retrieve_offsets()), 10)
+            with warnings.catch_warnings():
+                warnings.simplefilter("ignore")
+                self.assertEqual(len(trace.retrieve_offsets()), 10)
             trace.close()
 
             self.assertTrue(os.path.exists(path_output))
@@ -195,8 +216,10 @@ def test_osi_trace_offsets_robustness(self):
 
             trace = OSITrace(path_input)
             # Test whether the function can handle be run multiple times safely
-            offsets = trace.retrieve_offsets(None)
-            offsets2 = trace.retrieve_offsets(None)
+            with warnings.catch_warnings():
+                warnings.simplefilter("ignore")
+                offsets = trace.retrieve_offsets(None)
+                offsets2 = trace.retrieve_offsets(None)
             trace.close()
 
             self.assertEqual(len(offsets), 10)