Add mcap support to osi3trace

[thomassedlmayer]

This PR adds multi-channel trace file support to the existing OSITrace reader class while keeping it backwards compatible in terms of the existing interface/features for single-channel trace files.

The OSITrace class now acts as dispatcher/wrapper for the underlying reader class (OSITraceMulti or OSITraceSingle) which is created depending on the input file type. The functionality of the old OSITrace reader was moved to the OSITraceSingle class (unchanged) and the existing methods of OSITrace are forwarded to the corresponding reader's method depending on feature support.

Note on feature support:
The old single-channel trace reader provides index-based message retrieval while the mcap library does only provides log_time-based message retrieval. This means that existing methods (retrieve_offset, get_message_by_index, etc.) are not supported when reading from an mcap file. On the other hand, metadata features are only supported for mcap. This leaves a kind of unpleasant non-unified interface depending on the underlying file format. The only unified interface for reading both types of file formats is the stateful iterator functionality and the restart and close methods (see ReaderBase class).

In the future we might want to unify the features of the single- and multi-channel trace file readers and think about a better interface.

[TimmRuppert]

Last week, we had an extensive discussion about maintaining 100 % backward compatibility versus introducing a clean new interface through an intentional breaking change. The current version of this PR doesn’t provide the former, while the latter might take too much time to implement (or to agree on an interface).

At that time, I mentioned using a factory or some form of delegation pattern. Below is a sketch of the factory approach as a basis for discussion. This should preserve full backward compatibility while still allowing delegation across two subclasses.

from typing import Optional
from pathlib import Path

class OSITrace:
    """Factory base class that returns appropriate subclass instances based on file format."""

    def __new__(cls, path: Path = None, type_name: Optional[str] = "SensorView",
                cache_messages: Optional[bool] = False, *args, **kwargs):

        if cls is OSITrace:  # only run factory when called directly
            if path.suffix.lower() == ".osi":
                return super().__new__(SingleOSITrace)
            elif path.suffix.lower() == ".mcap":
                return super().__new__(McapOSITrace)
            else:
                raise ValueError(f"Unsupported trace format detected for: {path}")
        return super().__new__(cls)

    def __init__(self, path=None, type_name="SensorView", cache_messages=False, *args, **kwargs):
        # Base class init does nothing
        pass


class SingleOSITrace(OSITrace):  
    def __init__(self, path=None, type_name="SensorView", cache_messages=False, *args, **kwargs):
        print(f"Initialized SingleOSITrace with path: {path}")
        self.path = path
        self.type_name = type_name
        self.cache_messages = cache_messages
        self.member_only_in_single = "I am unique to SingleOSITrace"


class McapOSITrace(OSITrace):  
    def __init__(self, path=None, type_name="SensorView", cache_messages=False, *args, **kwargs):
        print(f"Initialized McapOSITrace with path: {path}")
        self.path = path
        self.type_name = type_name
        self.start_time = kwargs.get("start_time", None) # unique to McapOSITrace


if __name__ == "__main__":
    print("Creating .osi trace:")
    trace1 = OSITrace(path=Path("example.osi"))

    print("\nCreating .mcap trace:")
    trace2 = OSITrace(path=Path("example.mcap"), start_time=123)

    print("\nChecking types:")
    print(f"trace1 type: {type(trace1).__name__}")
    print(f"trace2 type: {type(trace2).__name__}")
    print(f"trace1 isinstance OSITrace: {isinstance(trace1, OSITrace)}")
    print(f"trace2 isinstance OSITrace: {isinstance(trace2, OSITrace)}")

    print("\nAccessing some unique values:")
    print(f"trace1 member_only_in_single: {trace1.member_only_in_single}")
    print(f"trace2 start_time: {trace2.start_time}")

    print("\nCreating unsupported trace format:")
    trace3 = OSITrace(path=Path("example.txt"))  # raises ValueError

Example Output:

Creating .osi trace:
Initialized SingleOSITrace with path: example.osi

Creating .mcap trace:
Initialized McapOSITrace with path: example.mcap

Checking types:
trace1 type: SingleOSITrace
trace2 type: McapOSITrace
trace1 isinstance OSITrace: True
trace2 isinstance OSITrace: True

Accessing some unique values:
trace1 member_only_in_single: I am unique to SingleOSITrace
trace2 start_time: 123

Creating unsupported trace format:
Traceback (most recent call last):
  File "/home/timm/Downloads/test_factory.py", line 59, in <module>
    trace3 = OSITrace(path=Path("example.txt"))  # raises ValueError
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/timm/Downloads/test_factory.py", line 16, in __new__
    raise ValueError(f"Unsupported trace format detected for: {path}")
ValueError: Unsupported trace format detected for: example.txt

[pmai]

I would propose to take a look of integrating either the factory approach or exposing more of the existing interface and potentially also supporting more of the existing interface for the multi-trace file reader.

[thomassedlmayer]

I tried out the factory approach and it worked fine in general and looks cleaner than my original approach when combined with an abstract class for auto completion features. But I did not find a way to allow initializing an empty OSITrace (ositrace_instance = OSITrace()) and then calling ositrace_instance.from_file(...) with all the arguments filled. This was possible with the current OSITrace implementation and I guess we "want" to keep it like this. @TimmRuppert If you have any idea on how to solve this, let me know.

For now I continued with the proxy approach and tried to expose the OSITrace legacy attributes by overwriting the attribute getter/setter and forward the calls to the reader in case it's an OSITraceSingle instance. This looks a bit ugly and this also comes with some potential issues I guess but it's the closest I've come to keep the old OSITrace behaviour.

[TimmRuppert]

I tried out the factory approach and it worked fine in general and looks cleaner than my original approach when combined with an abstract class for auto completion features. But I did not find a way to allow initializing an empty OSITrace (ositrace_instance = OSITrace()) and then calling ositrace_instance.from_file(...) with all the arguments filled. This was possible with the current OSITrace implementation and I guess we "want" to keep it like this. @TimmRuppert If you have any idea on how to solve this, let me know.

Although I don’t think this is a great practice for new code, we could apply morphism here. In this case, it seems appropriate since we’re working with an existing codebase and need to expose a specific set of “public” functions/members. Python allows overwriting self.__class__, but you’ll need to call __init__ (and possibly other python dunder functions if used) manually to properly initialize class members. I asked AI to extend my earlier example to include a from_file function, which takes care of handling the morphism. Seems to work fine but in all honesty I have only worked with morphism in python once.

from typing import Optional
from pathlib import Path

class OSITrace:
    """
    Factory class that returns the correct trace handler (.osi or .mcap)
    and supports deferred initialization via the .from_file() method.
    """
    def __new__(cls, path: Optional[Path] = None, *args, **kwargs):
        # If this is a call to a subclass (e.g., SingleOSITrace()), don't use the factory.
        if cls is not OSITrace:
            return super().__new__(cls)

        # If a path is provided, act as a factory immediately.
        if path:
            if not isinstance(path, Path): path = Path(path) # Ensure path is a Path object

            if path.suffix.lower() == ".osi":
                # Return an instance of the specific subclass
                return super().__new__(SingleOSITrace)
            elif path.suffix.lower() == ".mcap":
                return super().__new__(McapOSITrace)
            else:
                raise ValueError(f"Unsupported trace format detected for: {path}")
        
        # If no path is provided, create a base OSITrace instance
        # that is waiting for the from_file() call.
        return super().__new__(cls)

    def __init__(self, path: Optional[Path] = None, *args, **kwargs):
        """
        The __init__ of the base class is called after __new__.
        For subclasses, their own __init__ will handle initialization.
        For a base instance (path=None), we don't need to do anything here yet.
        """
        pass

    def from_file(self, path: Path, type_name: str = "SensorView", 
                  cache_messages: bool = False, *args, **kwargs):
        """
        Loads data from a file, morphing the current base instance into the
        correct subclass based on the file extension.
        """
        if not isinstance(path, Path): path = Path(path)
        
        target_cls = None
        if path.suffix.lower() == ".osi":
            target_cls = SingleOSITrace
        elif path.suffix.lower() == ".mcap":
            target_cls = McapOSITrace
        else:
            raise ValueError(f"Unsupported trace format detected for: {path}")

        # === The Magic ===
        # 1. Change the class of the current instance to the target class.
        self.__class__ = target_cls
        
        # 2. Call the __init__ of the new class to properly initialize the instance.
        target_cls.__init__(self, path=path, type_name=type_name, 
                            cache_messages=cache_messages, *args, **kwargs)

class SingleOSITrace(OSITrace):
    def __init__(self, path=None, type_name="SensorView", cache_messages=False, *args, **kwargs):
        print(f"-> Initialized SingleOSITrace with path: {path}")
        self.path = path
        self.type_name = type_name
        self.cache_messages = cache_messages
        self.member_only_in_single = "I am unique to SingleOSITrace"

class McapOSITrace(OSITrace):
    def __init__(self, path=None, type_name="SensorView", cache_messages=False, *args, **kwargs):
        print(f"-> Initialized McapOSITrace with path: {path}")
        self.path = path
        self.type_name = type_name
        # Example of subclass-specific argument
        self.start_time = kwargs.get("start_time", None)

if __name__ == "__main__":
    print("--- Scenario 1: Initialization with path (Factory behavior) ---")
    
    print("\nCreating .osi trace directly:")
    trace1 = OSITrace(path=Path("example.osi"))
    print(f"trace1 type: {type(trace1).__name__}")
    print(f"trace1 member: {trace1.member_only_in_single}")

    print("\nCreating .mcap trace directly:")
    trace2 = OSITrace(path=Path("example.mcap"), start_time=12345)
    print(f"trace2 type: {type(trace2).__name__}")
    print(f"trace2 start_time: {trace2.start_time}")
    
    print("\n" + "="*60 + "\n")

    print("--- Scenario 2: Deferred initialization (Added after your comment) ---")
    
    print("\nCreating empty instance first:")
    trace3 = OSITrace()
    print(f"Initial trace3 type: {type(trace3).__name__}")
    
    print("\nCalling from_file() with .osi path:")
    trace3.from_file(path=Path("another.osi"))
    print(f"Morphed trace3 type: {type(trace3).__name__}")
    print(f"Morphed trace3 member: {trace3.member_only_in_single}")
    print(f"Is it still an OSITrace? {isinstance(trace3, OSITrace)}")

Output

--- Scenario 1: Initialization with path (Factory behavior) ---

Creating .osi trace directly:
-> Initialized SingleOSITrace with path: example.osi
trace1 type: SingleOSITrace
trace1 member: I am unique to SingleOSITrace

Creating .mcap trace directly:
-> Initialized McapOSITrace with path: example.mcap
trace2 type: McapOSITrace
trace2 start_time: 12345

============================================================

--- Scenario 2: Deferred initialization (Backwards compatibility) ---

Creating empty instance first:
Initial trace3 type: OSITrace

Calling from_file() with .osi path:
-> Initialized SingleOSITrace with path: another.osi
Morphed trace3 type: SingleOSITrace
Morphed trace3 member: I am unique to SingleOSITrace
Is it still an OSITrace? True