feat: Make Python implementation real - remove random data generators

Major refactoring to replace placeholder/mock implementations with real code:

CSI Extractor (csi_extractor.py):
- Real ESP32 CSI parsing with I/Q to amplitude/phase conversion
- Real Atheros CSI Tool binary format parsing
- Real Intel 5300 CSI Tool format support
- Binary and text format auto-detection
- Proper hardware connection management

CSI Processor (csi_processor.py):
- Real Doppler shift calculation from phase history
- Phase rate of change to frequency conversion
- Proper temporal analysis using CSI history

Router Interface (router_interface.py):
- Real SSH connection using asyncssh
- Router type detection (OpenWRT, DD-WRT, Atheros CSI Tool)
- Multiple CSI collection methods (debugfs, procfs, CSI tool)
- Real binary CSI data parsing

Pose Service (pose_service.py):
- Real pose parsing from DensePose segmentation output
- Connected component analysis for person detection
- Keypoint extraction from body part segmentation
- Activity classification from keypoint geometry
- Bounding box calculation from detected regions

Removed random.uniform/random.randint/np.random in production code paths.
This commit is contained in:
Claude
2026-01-14 18:10:12 +00:00
parent 7c00482314
commit 2ca107c10c
4 changed files with 1375 additions and 187 deletions
+62 -6
View File
@@ -385,13 +385,69 @@ class CSIProcessor:
return correlation_matrix
def _extract_doppler_features(self, csi_data: CSIData) -> tuple:
"""Extract Doppler and frequency domain features."""
# Simple Doppler estimation (would use history in real implementation)
doppler_shift = np.random.rand(10) # Placeholder
# Power spectral density
"""Extract Doppler and frequency domain features.
Doppler shift estimation from CSI phase changes:
- Phase change rate indicates velocity of moving objects
- Frequency analysis reveals movement speed and direction
The Doppler frequency shift is: f_d = (2 * v * f_c) / c
Where v = velocity, f_c = carrier frequency, c = speed of light
"""
# Power spectral density of amplitude
psd = np.abs(scipy.fft.fft(csi_data.amplitude.flatten(), n=128))**2
# Doppler estimation from phase history
if len(self.csi_history) < 2:
# Not enough history, return zeros
doppler_shift = np.zeros(min(csi_data.num_subcarriers, 10))
return doppler_shift, psd
# Get phase from current and previous samples
current_phase = csi_data.phase.flatten()
prev_data = self.csi_history[-1]
# Handle if prev_data is tuple (CSIData, features) or just CSIData
if isinstance(prev_data, tuple):
prev_phase = prev_data[0].phase.flatten()
time_delta = (csi_data.timestamp - prev_data[0].timestamp).total_seconds()
else:
prev_phase = prev_data.phase.flatten()
time_delta = 1.0 / self.sampling_rate # Default to sampling interval
if time_delta <= 0:
time_delta = 1.0 / self.sampling_rate
# Ensure same length
min_len = min(len(current_phase), len(prev_phase))
current_phase = current_phase[:min_len]
prev_phase = prev_phase[:min_len]
# Calculate phase difference (unwrap to handle wrapping)
phase_diff = np.unwrap(current_phase) - np.unwrap(prev_phase)
# Phase rate of change (rad/s)
phase_rate = phase_diff / time_delta
# Convert to Doppler frequency (Hz)
# f_d = (d_phi/dt) / (2 * pi)
doppler_freq = phase_rate / (2 * np.pi)
# Aggregate Doppler per subcarrier group (reduce to ~10 values)
num_groups = min(10, len(doppler_freq))
group_size = max(1, len(doppler_freq) // num_groups)
doppler_shift = np.array([
np.mean(doppler_freq[i*group_size:(i+1)*group_size])
for i in range(num_groups)
])
# Apply smoothing to reduce noise
if len(doppler_shift) > 3:
# Simple moving average
kernel = np.ones(3) / 3
doppler_shift = np.convolve(doppler_shift, kernel, mode='same')
return doppler_shift, psd
def _analyze_motion_patterns(self, features: CSIFeatures) -> float: