Spaces:
Running
Running
File size: 2,803 Bytes
47ffedc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | """
Sensor Buffer for Rolling Window
=================================
Maintains last 60 samples for LSTM/CNN inference.
"""
import numpy as np
from collections import deque
class SensorBuffer:
"""
Rolling buffer that stores last N sensor readings.
Usage:
buffer = SensorBuffer(window_size=60, n_features=40)
# Add new sample
buffer.add_sample(features_scaled)
# Check if ready
if buffer.is_ready():
sequence = buffer.get_sequence()
# Use sequence for LSTM/CNN
"""
def __init__(self, window_size=60, n_features=40):
"""
Initialize buffer.
Args:
window_size (int): Number of samples to store (default: 60)
n_features (int): Number of features per sample (default: 40)
"""
self.window_size = window_size
self.n_features = n_features
# Use deque for efficient FIFO operations
self.buffer = deque(maxlen=window_size)
#print(f"✅ SensorBuffer initialized: window={window_size}, features={n_features}")
def add_sample(self, features):
"""
Add new sample to buffer.
Args:
features (np.array): (40,) array of scaled features
"""
# Convert to 1D array if needed
if features.ndim == 2:
features = features.flatten()
# Add to buffer (automatically removes oldest if full)
self.buffer.append(features)
def get_sequence(self):
"""
Get sequence for LSTM/CNN (60 x 40 array).
Returns:
np.array: (window_size, n_features) array
"""
# If buffer not full yet, pad with zeros
if len(self.buffer) < self.window_size:
# Pad beginning with zeros
padding_needed = self.window_size - len(self.buffer)
padding = [np.zeros(self.n_features)] * padding_needed
return np.array(padding + list(self.buffer))
# Buffer is full - return as numpy array
return np.array(list(self.buffer))
def is_ready(self):
"""
Check if buffer has enough samples for reliable prediction.
Returns:
bool: True if buffer has at least window_size samples
"""
return len(self.buffer) >= self.window_size
def size(self):
"""Current number of samples in buffer"""
return len(self.buffer)
def clear(self):
"""Clear all samples from buffer"""
self.buffer.clear()
def __len__(self):
return len(self.buffer)
def __repr__(self):
return f"SensorBuffer(size={len(self.buffer)}/{self.window_size}, features={self.n_features})"
|