Incremental buffer for streaming signal updates in research notebooks.
| 127 | |
| 128 | @dataclass |
| 129 | class SignalStreamBuffer: |
| 130 | """Incremental buffer for streaming signal updates in research notebooks.""" |
| 131 | |
| 132 | _frames: list[pl.DataFrame] |
| 133 | |
| 134 | def __init__(self) -> None: |
| 135 | self._frames = [] |
| 136 | |
| 137 | def append( |
| 138 | self, |
| 139 | timestamps: Sequence[str], |
| 140 | signal: Sequence[float], |
| 141 | side: Sequence[float] | None = None, |
| 142 | symbol: str | None = None, |
| 143 | ) -> None: |
| 144 | self._frames.append(to_polars_signal_frame(timestamps, signal, side=side, symbol=symbol)) |
| 145 | |
| 146 | def frame(self) -> pl.DataFrame: |
| 147 | if not self._frames: |
| 148 | return pl.DataFrame({"ts": [], "signal": []}) |
| 149 | return pl.concat(self._frames, how="vertical") |
| 150 | |
| 151 | def clear(self) -> None: |
| 152 | self._frames.clear() |
| 153 | |
| 154 | |
| 155 | def to_pandas(df: pl.DataFrame): # type: ignore[no-untyped-def] |
nothing calls this directly
no outgoing calls
no test coverage detected