Unified TimeSeriesStore with pluggable backends, global rewrite of timed event storage

[leshy]

Summary

This is a first pass on memory, cleans up the way we deal with timed events in dimos
Creates a TimeSeriesStore[T] abstraction for timestamped data with multiple backend implementations.

Backend implementations are very simple and TBH I don't expect to use psql but sqlite, implemented them more as examples.

TimeSeriesStore[T] Base Class (dimos/memory/timeseries/base.py)

• Generic abstract base — backends implement _save, _load, _delete, _iter_items, _find_closest_timestamp, _count, _last_timestamp, _find_before, _find_after

• API built on top: save(), find_closest(), find_before(), find_after(), prune_old(), slice_by_time(), iterate(), iterate_realtime(), stream(), consume_stream()

• T bound to Timestamped — timestamps come from .ts attribute

Backend Implementations

Backend	File	Use Case
InMemoryStore	inmemory.py	Live streaming, buffers, testing. O(log n) via SortedKeyList
PickleDirStore	pickledir.py	Simple file-based, one pickle per timestamp
SqliteStore	sqlite.py	Single-file DB with indexed queries
PostgresStore	postgres.py	Multi-user, implements Resource lifecycle
LegacyPickleStore	legacy.py	Backward compat with old TimedSensorReplay recordings

Unified the way we treat storage of timed items

Replaced TimestampedCollection

TimestampedCollection and its subclass TimestampedBufferCollection were used for:

• Transform buffers in TF service (TBuffer)
• Message timestamp alignment (align_timestamped)
• Quality filtering (image sharpness windows)

All now use InMemoryStore directly. TimestampedBufferCollection is a thin wrapper adding auto-prune on insert. TimestampedCollection class deleted entirely.

Replaced Pickle Sensor Replay System with standard time series store interface

future recording and reply can go via sqlite for example

Replaced Transform service in-memory transform storage

they all depend on the same base, storing transforms in postgres (not that we want to) is a one line change

Other changes

• Removed generic Embedding type — all embedding models return the same type

Usage

from dimos.memory.timeseries import InMemoryStore, SqliteStore, PostgresStore

# In-memory buffer
store = InMemoryStore()
# or
# SQLite persistent storage
# store = SqliteStore("recordings/lidar")
# or
# Postgres storage
# store = PostgresStore("my_lidar_table")

store.save(data)  # data.ts used as timestamp
latest = store.last()
closest = store.find_closest(target_ts, tolerance=0.5)

for item in store.iterate_realtime(speed=2.0):
    process(item)

# Observable streaming
store.stream(speed=1.0, seek=10.0).subscribe(on_next=process)

[greptile-apps]

_{42 files reviewed, no comments}

_{Edit Code Review Agent Settings | Greptile}

[paul-nechifor]

Personally, I don't like the API. It's similar to what we do with LCM() and others where we don't encode the lifecycle into the API. The API is simple, but doesn't give you enough control.

Sessions

store = SqliteStore("recordings/lidar")
store.save(data)

Without a session, API calls race to create self._conn. You could add a lock. But it just complicates the code to always check the lock and check that self._conn exists. If you have this:

store = SqliteStore("recordings/lidar")
session = store.session()
with session:
    session.save(data)

Then SqliteStoreSession has self._conn as non-None and you never have to check it's there.

Lack of owner

Most systems separate the database from the sessions. This makes it easy to close the database which closes all the sessions.

store = SqliteStore("recordings/lidar")
session = store.session()
store.close()

store knows about all its sessions and can close them. Additionally, without it we can't know if we don't have two SqliteStore objects using the same sqlite file.

Low cohesion in TimeSeriesStore

TimeSeriesStore is a database, a session, and a query set.

Most of the methods belong to the query set. (QuerySet is a Django term, others use Manager or Collection for the object which does the querying.) But the code forces people to extend TimeSeriesStore to get what they need instead of TimeSeriesStore using what it needs.

Example. _save is abstract, but save is provided, even if not wanted.

class TimeSeriesStore(Generic[T], ABC):
    @abstractmethod
    def _save(self, timestamp: float, data: T) -> None: ..

    def save(self, *data: T) -> None:
        for item in data:
            self._save(item.ts, item)

class MyStore(TimeSeriesStore[T]):
    def _save(self, timestamp: float, data: T) -> None: ..

I suggest this:

class Session(Protocol[T]):
    def save(self, timestamp: float, data: T) -> None: ...

class MySession(Session[T]):
    def save(self, timestamp: float, data: T) -> None:
        # implementation
        ...

class TimeSeriesStore(Generic[T])
    def __init__(self, session: Session[T]):
        self._session = session

    def save(self, *data: T) -> None:
        for item in data:
            self._session.save(data)

TimeSeriesStore uses a Session, it doesn't have to force people to inherit from it. It doesn't mix its namespace with whatever inherits from it.

Verbosity

If the concern is that this is too verbose, you can always have helper methods. But if the API mixes concerns too much that's not easy to untangle. Example helper method:

with store_session("sqlite", "/path/to/sensors.db") as store:
    store.save(...)

[leshy]

Personally, I don't like the API. It's similar to what we do with LCM() and others where we don't encode the lifecycle into the API. The API is simple, but doesn't give you enough control.

greptile is saying 5/5 paul is saying 0/5, this is a classic good cop bad cop situation

sounds good! will externalize the sessions

[leshy]

We decided to merge this large thing to not dangle, Pauls feedback on store init/shutdown API is good, so I will rewrite to include external session handling in a follow up