diff --git a/archive/v1/src/hardware/csi_extractor.py b/archive/v1/src/hardware/csi_extractor.py index edb43325..db56f8c8 100644 --- a/archive/v1/src/hardware/csi_extractor.py +++ b/archive/v1/src/hardware/csi_extractor.py @@ -143,13 +143,28 @@ class ESP32BinaryParser: 12 4 Sequence number (LE u32) 16 1 RSSI (i8) 17 1 Noise floor (i8) - 18 2 Reserved + 18 1 PPDU type (ADR-110): 0=HT/legacy, 1=HE-SU, 2=HE-MU, + 3=HE-TB, 0xFF=unknown. Pre-ADR-110 firmware sends 0. + 19 1 Flags (ADR-110): bit 0 = bw40, bit 2 = STBC, + bit 3 = LDPC, bit 4 = 802.15.4 sync valid. 20 N*2 I/Q pairs (n_antennas * n_subcarriers * 2 bytes, signed i8) """ MAGIC = 0xC5110001 HEADER_SIZE = 20 - HEADER_FMT = ' CSIData: """Parse an ADR-018 binary frame into CSIData. @@ -168,8 +183,8 @@ class ESP32BinaryParser: f"Frame too short: need {self.HEADER_SIZE} bytes, got {len(raw_data)}" ) - magic, node_id, n_antennas, n_subcarriers, freq_mhz, sequence, rssi_u8, noise_u8 = \ - struct.unpack_from(self.HEADER_FMT, raw_data, 0) + magic, node_id, n_antennas, n_subcarriers, freq_mhz, sequence, rssi_u8, noise_u8, \ + ppdu_byte, flags_byte = struct.unpack_from(self.HEADER_FMT, raw_data, 0) if magic != self.MAGIC: raise CSIParseError( @@ -226,6 +241,17 @@ class ESP32BinaryParser: 'rssi_dbm': rssi, 'noise_floor_dbm': noise_floor, 'channel_freq_mhz': freq_mhz, + # ADR-110 extension — zeros from pre-ADR-110 firmware land here as + # 'ht_legacy' + all-flags-false. New consumers can branch on + # ppdu_type / he_capable for HE-LTF-aware DSP. + 'ppdu_type': self._PPDU_NAMES.get(ppdu_byte, 'unknown'), + 'ppdu_type_raw': ppdu_byte, + 'he_capable': ppdu_byte in (1, 2, 3), + 'bw40': bool(flags_byte & 0x01), + 'stbc': bool(flags_byte & 0x04), + 'ldpc': bool(flags_byte & 0x08), + 'ieee802154_sync_valid': bool(flags_byte & 0x10), + 'adr018_flags_raw': flags_byte, } ) diff --git a/archive/v1/tests/unit/test_esp32_binary_parser.py b/archive/v1/tests/unit/test_esp32_binary_parser.py index 9f8f4e7b..64681a3a 100644 --- a/archive/v1/tests/unit/test_esp32_binary_parser.py +++ b/archive/v1/tests/unit/test_esp32_binary_parser.py @@ -23,7 +23,10 @@ from hardware.csi_extractor import ( # ADR-018 constants MAGIC = 0xC5110001 -HEADER_FMT = ' bytes: """Build an ADR-018 binary frame for testing.""" if iq_pairs is None: @@ -54,6 +59,8 @@ def build_binary_frame( sequence, rssi_u8, noise_u8, + ppdu_byte, + flags_byte, ) iq_data = b'' @@ -63,6 +70,52 @@ def build_binary_frame( return header + iq_data +class TestAdr110ByteEncoding: + """ADR-110: byte 18 = PPDU type, byte 19 = flags.""" + + def setup_method(self): + self.parser = ESP32BinaryParser() + + def test_pre_adr110_zeros_decode_as_ht_legacy(self): + """Pre-ADR-110 firmware sends zeros → must surface as HT/legacy + no flags.""" + frame = build_binary_frame() # ppdu_byte=0, flags_byte=0 default + csi = self.parser.parse(frame) + assert csi.metadata['ppdu_type'] == 'ht_legacy' + assert csi.metadata['ppdu_type_raw'] == 0 + assert csi.metadata['he_capable'] is False + assert csi.metadata['bw40'] is False + assert csi.metadata['stbc'] is False + assert csi.metadata['ldpc'] is False + assert csi.metadata['ieee802154_sync_valid'] is False + + def test_he_su_decodes(self): + frame = build_binary_frame(ppdu_byte=1) + csi = self.parser.parse(frame) + assert csi.metadata['ppdu_type'] == 'he_su' + assert csi.metadata['he_capable'] is True + + def test_he_mu_and_he_tb_decode(self): + for byte, expected in [(2, 'he_mu'), (3, 'he_tb')]: + csi = self.parser.parse(build_binary_frame(ppdu_byte=byte)) + assert csi.metadata['ppdu_type'] == expected + assert csi.metadata['he_capable'] is True + + def test_unknown_ppdu_byte(self): + csi = self.parser.parse(build_binary_frame(ppdu_byte=0xFF)) + assert csi.metadata['ppdu_type'] == 'unknown' + assert csi.metadata['ppdu_type_raw'] == 0xFF + assert csi.metadata['he_capable'] is False + + def test_all_flags_set_round_trip(self): + # bw40 (0x01) + STBC (0x04) + LDPC (0x08) + 15.4-sync (0x10) = 0x1D + csi = self.parser.parse(build_binary_frame(ppdu_byte=1, flags_byte=0x1D)) + assert csi.metadata['bw40'] is True + assert csi.metadata['stbc'] is True + assert csi.metadata['ldpc'] is True + assert csi.metadata['ieee802154_sync_valid'] is True + assert csi.metadata['adr018_flags_raw'] == 0x1D + + class TestESP32BinaryParser: """Tests for ESP32BinaryParser."""