Refactor continuous data handling to use a dictionary structure for improved access by index and stream name

Author: anjaldoshi

src/open_ephys/analysis/README.md

@@ -65,9 +65,9 @@ Recording Index: 0
 
 ## Loading continuous data
 
-Continuous data for each recording is accessed via the `.continuous` property of each `Recording` object. This returns a list of continuous data, grouped by processor/sub-processor. For example, if you have two data streams merged into a single Record Node, each data stream will be associated with a different processor ID. If you're recording Neuropixels data, each probe's data stream will be stored in a separate sub-processor, which must be loaded individually.
+Continuous data for each recording is accessed via the `.continuous` property of each `Recording` object. This now returns a dictionary of continuous data grouped by processor/sub-processor. Each stream is stored twice in the dictionary: once under its zero-based index and once under its stream name. For example, if you have two data streams merged into a single Record Node, each data stream will be associated with a different processor ID. If you're recording Neuropixels data, each probe's data stream will be stored in a separate sub-processor, which must be loaded individually.
 
-Continuous data for individual data streams can be accessed by index (e.g., `continuous[0]`), or by stream name (e.g., `continuous["example_data"]`). If there are multiple streams with the same name, the source processor ID will be appended to the stream name so they can be distinguished (e.g., `continuous["example_data_100"]`).
+Continuous data for individual data streams can be accessed by index (e.g., `continuous[0]`), or by stream name (e.g., `continuous["example_data"]`). If there are multiple streams with the same name, the source processor ID will be appended to the stream name so they can be distinguished (e.g., `continuous["example_data_100"]`). Iterating over the dictionary yields the continuous objects in index order, and `continuous.keys()` lists both the integer indices and stream names that can be used for lookup.
 
 Each `continuous` object has four properties:
 

src/open_ephys/analysis/formats/BinaryRecording.py

@@ -37,7 +37,7 @@
     RecordingFormat,
     Spikes,
     SpikeMetadata,
-    create_continuous_named_tuple
+    create_continuous_dict
 )
 from open_ephys.analysis.utils import alphanum_key
 
@@ -295,7 +295,7 @@ def load_continuous(self):
                     names[idx2] = name2 + "_" + source_processor_ids[idx2]
                     break
 
-        self._continuous = create_continuous_named_tuple(names, values)
+        self._continuous = create_continuous_dict(names, values)
 
     def load_spikes(self):
         self._spikes = []

src/open_ephys/analysis/formats/NwbRecording.py

@@ -36,7 +36,7 @@
     SpikeMetadata,
     RecordingFormat,
     Recording,
-    create_continuous_named_tuple
+    create_continuous_dict
 )
 
 
@@ -178,7 +178,7 @@ def load_continuous(self):
                     names[idx2] = name2 + "_" + source_processor_ids[idx2]
                     break
 
-        self._continuous = create_continuous_named_tuple(names, values)
+        self._continuous = create_continuous_dict(names, values)
 
     def load_spikes(self):
 

src/open_ephys/analysis/formats/OpenEphysRecording.py

@@ -40,7 +40,7 @@
     RecordingFormat,
     Spikes,
     SpikeMetadata,
-    create_continuous_named_tuple
+    create_continuous_dict
 )
 
 
@@ -286,7 +286,8 @@ def load_continuous(self):
         continuous_files, stream_indexes, unique_stream_indexes, stream_info = (
             self.find_continuous_files()
         )
-        self._continuous = []
+        values = []
+        names = []
 
         for stream_index in unique_stream_indexes:
 
@@ -295,11 +296,15 @@ def load_continuous(self):
                 if stream_indexes[ind] == stream_index:
                     files_for_stream.append(os.path.join(self.directory, filename))
 
-            self._continuous.append(
+            names.append(stream_info[stream_index]["stream_name"])
+
+            values.append(
                 OpenEphysContinuous(
                     stream_info[stream_index], files_for_stream, self.recording_index
                 )
             )
+        
+        self._continuous = create_continuous_dict(names, values)
 
     def load_spikes(self):
 

src/open_ephys/analysis/recording.py

@@ -25,7 +25,7 @@
 
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from collections import namedtuple
+from collections.abc import Sequence
 from enum import StrEnum
 import warnings
 import numpy as np
@@ -51,51 +51,54 @@ class ContinuousMetadata:
     bit_volts: list[float]
 
 
-def create_continuous_named_tuple(names, values):
-    """
-    Create a named tuple from the given names and values.
-    """
-    NT = namedtuple("DynamicTuple", names)
-    
-    class ContinuousWrapper:
-        """
-        Allow the .continuous attribute to be accessed as a dictionary.
-        """
-        def __init__(self, nt, names):
-            self._nt = nt
-            self._names = names
-            self._index = {name: i for i, name in enumerate(names)}
-
-        def __getitem__(self, key):
-            if isinstance(key, str):
-                return getattr(self._nt, key)
-            return self._nt[key]
-
-        def __getattr__(self, attr):
-            return getattr(self._nt, attr)
-
-        def __len__(self):
-            return len(self._nt)
-
-        def __iter__(self):
-            return iter(self._nt)
-        
-        def keys(self):
-            """Return available field names (like dict.keys())."""
-            return list(self._names)
-
-        def items(self):
-            """Return (name, value) pairs."""
-            return [(name, getattr(self._nt, name)) for name in self._names]
-
-        def values(self):
-            """Return values (like dict.values())."""
-            return list(self._nt)
-
-        def __repr__(self):
-            return repr(self._nt)
-        
-    return ContinuousWrapper(NT(*values), names)
+class ContinuousDict(dict):
+    """Dictionary access to continuous streams by numeric index or string name."""
+
+    def __init__(self, names: Sequence[str], values: Sequence["Continuous"]):
+        if len(names) != len(values):
+            raise ValueError("`names` and `values` must have the same length.")
+
+        super().__init__()
+        self._names = list(names)
+        self._values = list(values)
+
+        for idx, (name, value) in enumerate(zip(self._names, self._values)):
+            super().__setitem__(idx, value)
+            super().__setitem__(name, value)
+
+    def __getitem__(self, key):
+        if isinstance(key, (int, np.integer)):
+            key = int(key)
+        return super().__getitem__(key)
+
+    def __iter__(self):
+        return iter(self._values)
+
+    def __len__(self):
+        return len(self._values)
+
+    def keys(self):
+        return list(range(len(self._values))) + list(self._names)
+
+    def items(self):
+        return [(idx, value) for idx, value in enumerate(self._values)] + [
+            (name, value) for name, value in zip(self._names, self._values)
+        ]
+
+    def values(self):
+        return list(self._values)
+
+    def __repr__(self):
+        entries = ", ".join(
+            f"{name!r}: {value!r}" for name, value in zip(self._names, self._values)
+        )
+        return f"ContinuousDict({{{entries}}})"
+
+
+def create_continuous_dict(names, values):
+    """Return continuous data as a dictionary keyed by index and stream name."""
+
+    return ContinuousDict(names, values)
 class Continuous(ABC):
     metadata: ContinuousMetadata
     samples: np.ndarray
@@ -132,52 +135,6 @@ def get_samples(
         """
         pass
 
-        def create_continuous_named_tuple(names, values):
-            """
-            Create a named tuple from the given names and values.
-            """
-            NT = namedtuple("DynamicTuple", names)
-            
-            class ContinuousWrapper:
-                """
-                Allow the .continuous attribute to be accessed as a dictionary.
-                """
-                def __init__(self, nt, names):
-                    self._nt = nt
-                    self._names = names
-                    self._index = {name: i for i, name in enumerate(names)}
-
-                def __getitem__(self, key):
-                    if isinstance(key, str):
-                        return getattr(self._nt, key)
-                    return self._nt[key]
-
-                def __getattr__(self, attr):
-                    return getattr(self._nt, attr)
-
-                def __len__(self):
-                    return len(self._nt)
-
-                def __iter__(self):
-                    return iter(self._nt)
-                
-                def keys(self):
-                    """Return available field names (like dict.keys())."""
-                    return list(self._names)
-
-                def items(self):
-                    """Return (name, value) pairs."""
-                    return [(name, getattr(self._nt, name)) for name in self._names]
-
-                def values(self):
-                    """Return values (like dict.values())."""
-                    return list(self._nt)
-
-                def __repr__(self):
-                    return repr(self._nt)
-                
-            return ContinuousWrapper(NT(*values), names)
-
 class Spikes(ABC):
     metadata: SpikeMetadata
     waveforms: np.ndarray | None
@@ -242,8 +199,10 @@ class Recording(ABC):
     """
 
     @property
-    def continuous(self) -> list[Continuous] | None:
-        """Returns a list of Continuous objects"""
+    def continuous(self) -> ContinuousDict | None:
+        """Returns a ContinuousDict containing Continuous objects
+            which can be accessed by index or stream name.
+        """
         if self._continuous is None:
             self.load_continuous()
         return self._continuous