32 lines
911 B
Python
32 lines
911 B
Python
from __future__ import annotations
|
|
|
|
from collections import defaultdict
|
|
from collections.abc import Iterable
|
|
from dataclasses import dataclass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FeatureVector:
|
|
sensor_id: str
|
|
values: dict[str, float]
|
|
label: str | None = None
|
|
|
|
|
|
class FeatureStore:
|
|
def __init__(self) -> None:
|
|
self._vectors: dict[str, list[FeatureVector]] = defaultdict(list)
|
|
|
|
def add(self, vector: FeatureVector) -> None:
|
|
self._vectors[vector.sensor_id].append(vector)
|
|
|
|
def add_batch(self, vectors: Iterable[FeatureVector]) -> None:
|
|
for vector in vectors:
|
|
self.add(vector)
|
|
|
|
def latest(self, sensor_id: str) -> FeatureVector | None:
|
|
series = self._vectors.get(sensor_id)
|
|
return series[-1] if series else None
|
|
|
|
def all(self) -> list[FeatureVector]:
|
|
return [vector for vectors in self._vectors.values() for vector in vectors]
|