use std::time::{SystemTime, UNIX_EPOCH};
use crate::protocol::{EEG_EVENT_ID, EEG_SCALING_FACTOR, NUM_EEG_CHANNELS, PACKET_SIZE, SYNC_BYTE};
use crate::types::{ChecksumStats, EegPacket, Mw75Event};
fn now_secs() -> f64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("system clock is before Unix epoch")
.as_secs_f64()
}
pub fn validate_checksum(packet: &[u8]) -> (bool, u16, u16) {
if packet.len() < PACKET_SIZE {
return (false, 0, 0);
}
let calculated: u16 = packet[..61].iter().map(|&b| b as u16).sum::<u16>();
let received: u16 = packet[61] as u16 | ((packet[62] as u16) << 8);
(calculated == received, calculated, received)
}
pub fn parse_eeg_packet(packet: &[u8]) -> Option<EegPacket> {
if packet.len() != PACKET_SIZE || packet[0] != SYNC_BYTE {
return None;
}
let (is_valid, _calc, _recv) = validate_checksum(packet);
if !is_valid {
return None;
}
let event_id = packet[1];
let counter = packet[3];
let timestamp = now_secs();
let ref_value = f32::from_le_bytes([packet[4], packet[5], packet[6], packet[7]]);
let drl = f32::from_le_bytes([packet[8], packet[9], packet[10], packet[11]]);
let mut channels = Vec::with_capacity(NUM_EEG_CHANNELS);
for ch in 0..NUM_EEG_CHANNELS {
let offset = 12 + ch * 4;
if offset + 4 <= packet.len() {
let raw = f32::from_le_bytes([
packet[offset],
packet[offset + 1],
packet[offset + 2],
packet[offset + 3],
]);
channels.push(raw * EEG_SCALING_FACTOR);
}
}
let feature_status = if packet.len() > 60 { packet[60] } else { 0 };
Some(EegPacket {
timestamp,
event_id,
counter,
ref_value,
drl,
channels,
feature_status,
checksum_valid: true,
})
}
pub struct PacketProcessor {
buffer: Vec<u8>,
pub stats: ChecksumStats,
pub verbose: bool,
}
impl PacketProcessor {
pub fn new(verbose: bool) -> Self {
Self {
buffer: Vec::with_capacity(PACKET_SIZE * 10),
stats: ChecksumStats::default(),
verbose,
}
}
pub fn process_data(&mut self, data: &[u8]) -> Vec<Mw75Event> {
self.buffer.extend_from_slice(data);
let mut events = Vec::new();
let mut i = 0;
while i < self.buffer.len() {
if self.buffer[i] == SYNC_BYTE {
if i + PACKET_SIZE <= self.buffer.len() {
let packet = &self.buffer[i..i + PACKET_SIZE];
let (is_valid, calc, recv) = validate_checksum(packet);
self.stats.total_packets += 1;
if !is_valid {
self.stats.invalid_packets += 1;
if self.verbose {
log::warn!(
"Checksum mismatch: calc=0x{:04x} recv=0x{:04x} (event={}, counter={})",
calc, recv, packet[1], packet[3]
);
}
i += 1;
continue;
}
self.stats.valid_packets += 1;
if packet[1] == EEG_EVENT_ID {
if let Some(eeg) = parse_eeg_packet(packet) {
events.push(Mw75Event::Eeg(eeg));
}
} else {
events.push(Mw75Event::OtherEvent {
event_id: packet[1],
counter: packet[3],
raw: packet.to_vec(),
});
}
i += PACKET_SIZE;
} else {
break;
}
} else {
i += 1;
}
}
if i > 0 {
self.buffer.drain(..i);
}
let max_buf = PACKET_SIZE * 10;
if self.buffer.len() > max_buf {
if let Some(pos) = self.buffer.iter().rposition(|&b| b == SYNC_BYTE) {
self.buffer.drain(..pos);
log::debug!("Buffer overflow — recovered sync at {pos}");
} else {
self.buffer.clear();
log::warn!("Buffer overflow — no sync byte found, cleared");
}
}
events
}
pub fn buffered_len(&self) -> usize {
self.buffer.len()
}
pub fn get_stats(&self) -> ChecksumStats {
self.stats.clone()
}
pub fn reset(&mut self) {
self.buffer.clear();
self.stats = ChecksumStats::default();
}
}
#[cfg(test)]
mod tests {
use super::*;
fn make_packet(event_id: u8, counter: u8) -> Vec<u8> {
let mut pkt = vec![0u8; PACKET_SIZE];
pkt[0] = SYNC_BYTE;
pkt[1] = event_id;
pkt[2] = 58; pkt[3] = counter;
fix_checksum(&mut pkt);
pkt
}
fn make_packet_with_channels(counter: u8, raw_values: &[f32; 12]) -> Vec<u8> {
let mut pkt = vec![0u8; PACKET_SIZE];
pkt[0] = SYNC_BYTE;
pkt[1] = EEG_EVENT_ID;
pkt[2] = 58;
pkt[3] = counter;
pkt[4..8].copy_from_slice(&42.0_f32.to_le_bytes());
pkt[8..12].copy_from_slice(&(-7.5_f32).to_le_bytes());
for (i, &val) in raw_values.iter().enumerate() {
let off = 12 + i * 4;
pkt[off..off + 4].copy_from_slice(&val.to_le_bytes());
}
pkt[60] = 0x01; fix_checksum(&mut pkt);
pkt
}
fn fix_checksum(pkt: &mut [u8]) {
let sum: u16 = pkt[..61].iter().map(|&b| b as u16).sum::<u16>();
pkt[61] = (sum & 0xFF) as u8;
pkt[62] = (sum >> 8) as u8;
}
#[test]
fn checksum_valid_packet() {
let pkt = make_packet(EEG_EVENT_ID, 0);
let (valid, calc, recv) = validate_checksum(&pkt);
assert!(valid);
assert_eq!(calc, recv);
}
#[test]
fn checksum_invalid_corrupted_byte() {
let mut pkt = make_packet(EEG_EVENT_ID, 0);
pkt[62] ^= 0xFF; let (valid, _, _) = validate_checksum(&pkt);
assert!(!valid);
}
#[test]
fn checksum_invalid_corrupted_payload() {
let mut pkt = make_packet(EEG_EVENT_ID, 10);
pkt[30] = 0xFF; let (valid, _, _) = validate_checksum(&pkt);
assert!(!valid);
}
#[test]
fn checksum_too_short() {
let (valid, calc, recv) = validate_checksum(&[0xAA, 0x00]);
assert!(!valid);
assert_eq!(calc, 0);
assert_eq!(recv, 0);
}
#[test]
fn checksum_empty() {
let (valid, _, _) = validate_checksum(&[]);
assert!(!valid);
}
#[test]
fn checksum_exact_minimum_length() {
let pkt = make_packet(EEG_EVENT_ID, 0);
assert_eq!(pkt.len(), 63);
let (valid, _, _) = validate_checksum(&pkt);
assert!(valid);
}
#[test]
fn checksum_longer_than_packet_still_valid() {
let mut pkt = make_packet(EEG_EVENT_ID, 0);
pkt.extend_from_slice(&[0xFF, 0xFF, 0xFF]);
let (valid, _, _) = validate_checksum(&pkt);
assert!(valid);
}
#[test]
fn parse_basic_eeg_packet() {
let pkt = make_packet(EEG_EVENT_ID, 42);
let eeg = parse_eeg_packet(&pkt).expect("should parse");
assert_eq!(eeg.event_id, EEG_EVENT_ID);
assert_eq!(eeg.counter, 42);
assert_eq!(eeg.channels.len(), NUM_EEG_CHANNELS);
assert!(eeg.checksum_valid);
assert!(eeg.timestamp > 0.0);
}
#[test]
fn parse_rejects_wrong_sync_byte() {
let mut pkt = make_packet(EEG_EVENT_ID, 0);
pkt[0] = 0xBB; fix_checksum(&mut pkt);
assert!(parse_eeg_packet(&pkt).is_none());
}
#[test]
fn parse_rejects_short_packet() {
assert!(parse_eeg_packet(&[0xAA]).is_none());
assert!(parse_eeg_packet(&[0xAA; 62]).is_none());
}
#[test]
fn parse_rejects_invalid_checksum() {
let mut pkt = make_packet(EEG_EVENT_ID, 0);
pkt[61] = 0; pkt[62] = 0;
assert!(parse_eeg_packet(&pkt).is_none());
}
#[test]
fn parse_channel_values_scaled() {
let raw = [1000.0_f32; 12];
let pkt = make_packet_with_channels(0, &raw);
let eeg = parse_eeg_packet(&pkt).unwrap();
for &ch in &eeg.channels {
assert!((ch - 23.842).abs() < 0.01, "Expected ~23.842, got {ch}");
}
}
#[test]
fn parse_negative_channel_values() {
let raw = [-5000.0_f32; 12];
let pkt = make_packet_with_channels(0, &raw);
let eeg = parse_eeg_packet(&pkt).unwrap();
for &ch in &eeg.channels {
let expected = -5000.0 * EEG_SCALING_FACTOR;
assert!(
(ch - expected).abs() < 0.1,
"Expected ~{expected}, got {ch}"
);
}
}
#[test]
fn parse_ref_and_drl() {
let raw = [0.0_f32; 12];
let pkt = make_packet_with_channels(0, &raw);
let eeg = parse_eeg_packet(&pkt).unwrap();
assert!((eeg.ref_value - 42.0).abs() < 0.001);
assert!((eeg.drl - (-7.5)).abs() < 0.001);
}
#[test]
fn parse_feature_status() {
let raw = [0.0_f32; 12];
let pkt = make_packet_with_channels(0, &raw);
let eeg = parse_eeg_packet(&pkt).unwrap();
assert_eq!(eeg.feature_status, 0x01);
}
#[test]
fn parse_all_counter_values() {
for c in 0..=255u8 {
let pkt = make_packet(EEG_EVENT_ID, c);
let eeg = parse_eeg_packet(&pkt).unwrap();
assert_eq!(eeg.counter, c);
}
}
#[test]
fn processor_basic_single_packet() {
let mut proc = PacketProcessor::new(false);
let pkt = make_packet(EEG_EVENT_ID, 1);
let events = proc.process_data(&pkt);
assert_eq!(events.len(), 1);
assert!(matches!(&events[0], Mw75Event::Eeg(e) if e.counter == 1));
assert_eq!(proc.stats.valid_packets, 1);
assert_eq!(proc.stats.total_packets, 1);
assert_eq!(proc.stats.invalid_packets, 0);
}
#[test]
fn processor_multiple_packets_in_one_call() {
let mut proc = PacketProcessor::new(false);
let mut data = Vec::new();
for i in 0..5 {
data.extend_from_slice(&make_packet(EEG_EVENT_ID, i));
}
let events = proc.process_data(&data);
assert_eq!(events.len(), 5);
assert_eq!(proc.stats.valid_packets, 5);
}
#[test]
fn processor_split_delivery_across_two_calls() {
let mut proc = PacketProcessor::new(false);
let pkt = make_packet(EEG_EVENT_ID, 1);
let events1 = proc.process_data(&pkt[..30]);
assert!(events1.is_empty());
assert_eq!(proc.buffered_len(), 30);
let events2 = proc.process_data(&pkt[30..]);
assert_eq!(events2.len(), 1);
assert_eq!(proc.buffered_len(), 0);
}
#[test]
fn processor_split_at_every_byte() {
let mut proc = PacketProcessor::new(false);
let pkt = make_packet(EEG_EVENT_ID, 99);
let mut total_events = 0;
for &byte in &pkt {
let events = proc.process_data(&[byte]);
total_events += events.len();
}
assert_eq!(total_events, 1);
}
#[test]
fn processor_garbage_prefix_skipped() {
let mut proc = PacketProcessor::new(false);
let pkt = make_packet(EEG_EVENT_ID, 5);
let mut data = vec![0x01, 0x02, 0x03, 0x04, 0x05];
data.extend_from_slice(&pkt);
let events = proc.process_data(&data);
assert_eq!(events.len(), 1);
assert!(matches!(&events[0], Mw75Event::Eeg(e) if e.counter == 5));
}
#[test]
fn processor_garbage_between_packets() {
let mut proc = PacketProcessor::new(false);
let mut data = Vec::new();
data.extend_from_slice(&make_packet(EEG_EVENT_ID, 1));
data.extend_from_slice(&[0x01, 0x02, 0x03]); data.extend_from_slice(&make_packet(EEG_EVENT_ID, 2));
let events = proc.process_data(&data);
assert_eq!(events.len(), 2);
}
#[test]
fn processor_other_event_type() {
let mut proc = PacketProcessor::new(false);
let pkt = make_packet(100, 7); let events = proc.process_data(&pkt);
assert_eq!(events.len(), 1);
assert!(matches!(
&events[0],
Mw75Event::OtherEvent {
event_id: 100,
counter: 7,
..
}
));
}
#[test]
fn processor_invalid_checksum_skips_and_counts() {
let mut proc = PacketProcessor::new(false);
let mut pkt = make_packet(EEG_EVENT_ID, 1);
pkt[30] = 0xFF;
let events = proc.process_data(&pkt);
assert!(events.is_empty());
assert!(proc.stats.invalid_packets > 0);
}
#[test]
fn processor_invalid_then_valid() {
let mut proc = PacketProcessor::new(false);
let mut bad = make_packet(EEG_EVENT_ID, 1);
bad[30] = 0xFF;
let good = make_packet(EEG_EVENT_ID, 2);
let mut data = Vec::new();
data.extend_from_slice(&bad);
data.extend_from_slice(&good);
let events = proc.process_data(&data);
assert!(!events.is_empty());
assert!(proc.stats.valid_packets >= 1);
}
#[test]
fn processor_sync_byte_in_payload() {
let mut pkt = vec![0u8; PACKET_SIZE];
pkt[0] = SYNC_BYTE;
pkt[1] = EEG_EVENT_ID;
pkt[2] = 58;
pkt[3] = 10;
pkt[15] = 0xAA;
pkt[20] = 0xAA;
pkt[40] = 0xAA;
fix_checksum(&mut pkt);
let mut proc = PacketProcessor::new(false);
let events = proc.process_data(&pkt);
assert_eq!(events.len(), 1);
assert!(matches!(&events[0], Mw75Event::Eeg(e) if e.counter == 10));
}
#[test]
fn processor_reset_clears_state() {
let mut proc = PacketProcessor::new(false);
let pkt = make_packet(EEG_EVENT_ID, 1);
proc.process_data(&pkt);
assert_eq!(proc.stats.valid_packets, 1);
proc.reset();
assert_eq!(proc.stats.valid_packets, 0);
assert_eq!(proc.stats.total_packets, 0);
assert_eq!(proc.buffered_len(), 0);
}
#[test]
fn processor_partial_packet_retained() {
let mut proc = PacketProcessor::new(false);
let pkt = make_packet(EEG_EVENT_ID, 1);
let events = proc.process_data(&pkt[..40]);
assert!(events.is_empty());
assert_eq!(proc.buffered_len(), 40);
let events = proc.process_data(&pkt[40..]);
assert_eq!(events.len(), 1);
assert_eq!(proc.buffered_len(), 0);
}
#[test]
fn processor_buffer_overflow_protection() {
let mut proc = PacketProcessor::new(false);
let garbage = vec![0x01; PACKET_SIZE * 15];
let events = proc.process_data(&garbage);
assert!(events.is_empty());
assert!(proc.buffered_len() < PACKET_SIZE * 11);
}
#[test]
fn processor_stats_error_rate() {
let mut proc = PacketProcessor::new(false);
for i in 0..3 {
let pkt = make_packet(EEG_EVENT_ID, i);
proc.process_data(&pkt);
}
assert_eq!(proc.stats.valid_packets, 3);
assert_eq!(proc.stats.total_packets, 3);
assert!((proc.stats.error_rate() - 0.0).abs() < f64::EPSILON);
}
#[test]
fn processor_two_packets_in_64_byte_chunk() {
let mut proc = PacketProcessor::new(false);
let pkt1 = make_packet(EEG_EVENT_ID, 1);
let pkt2 = make_packet(EEG_EVENT_ID, 2);
let mut combined = Vec::new();
combined.extend_from_slice(&pkt1);
combined.extend_from_slice(&pkt2);
let events1 = proc.process_data(&combined[..64]);
assert_eq!(events1.len(), 1);
let events2 = proc.process_data(&combined[64..]);
assert_eq!(events2.len(), 1);
}
#[test]
fn processor_empty_input() {
let mut proc = PacketProcessor::new(false);
let events = proc.process_data(&[]);
assert!(events.is_empty());
assert_eq!(proc.buffered_len(), 0);
}
#[test]
fn processor_verbose_mode() {
let mut proc = PacketProcessor::new(true);
assert!(proc.verbose);
let mut bad = make_packet(EEG_EVENT_ID, 0);
bad[50] = 0xFF;
proc.process_data(&bad);
assert!(proc.stats.invalid_packets > 0);
}
#[test]
fn stats_default() {
let stats = ChecksumStats::default();
assert_eq!(stats.valid_packets, 0);
assert_eq!(stats.invalid_packets, 0);
assert_eq!(stats.total_packets, 0);
assert_eq!(stats.error_rate(), 0.0);
}
#[test]
fn stats_error_rate_calculation() {
let stats = ChecksumStats {
valid_packets: 90,
invalid_packets: 10,
total_packets: 100,
};
assert!((stats.error_rate() - 10.0).abs() < 0.01);
}
#[test]
fn stats_error_rate_zero_packets() {
let stats = ChecksumStats::default();
assert_eq!(stats.error_rate(), 0.0); }
#[test]
fn stats_error_rate_all_invalid() {
let stats = ChecksumStats {
valid_packets: 0,
invalid_packets: 50,
total_packets: 50,
};
assert!((stats.error_rate() - 100.0).abs() < 0.01);
}
}