{"record":{"id":"05f8bc140c0dad24","repo":"ruvnet/RuView","slug":"insufficient-data-for-csi","errorCode":null,"errorMessage":"Insufficient data for CSI","messagePattern":"Insufficient data for CSI","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"plans/phase2-architecture/hardware-integration.md","lineNumber":375,"sourceCode":"        return {\n            'timestamp': header[0],\n            'channel': header[1],\n            'rate': header[2],\n            'rssi': header[3] - 256 if header[3] > 127 else header[3],\n            'noise': header[4] - 256 if header[4] > 127 else header[4],\n            'antenna_config': header[5],\n            'csi_length': header[6],\n            'mac_address': header[7]\n        }\n    \n    @staticmethod\n    def parse_csi_data(data, header):\n        \"\"\"Parse CSI complex values\"\"\"\n        csi_start = AtherosCSIFormat.HEADER_SIZE\n        csi_length = header['csi_length']\n        \n        if len(data) < csi_start + csi_length:\n            raise ValueError(\"Insufficient data for CSI\")\n        \n        # Atheros format: 10-bit values packed\n        # [real_0|imag_0|real_1|imag_1|...]\n        csi_raw = data[csi_start:csi_start + csi_length]\n        \n        # Unpack 10-bit values\n        num_values = csi_length * 8 // 10\n        csi_complex = np.zeros(num_values // 2, dtype=complex)\n        \n        bit_offset = 0\n        for i in range(0, num_values, 2):\n            # Extract 10-bit real and imaginary parts\n            real = AtherosCSIFormat._extract_10bit(csi_raw, bit_offset)\n            imag = AtherosCSIFormat._extract_10bit(csi_raw, bit_offset + 10)\n            \n            # Convert to signed values\n            real = real - 512 if real > 511 else real\n            imag = imag - 512 if imag > 511 else imag","sourceCodeStart":357,"sourceCodeEnd":393,"githubUrl":"https://github.com/ruvnet/RuView/blob/4685618388a5e49fad5b3005806f3bdd6a7c25c3/plans/phase2-architecture/hardware-integration.md#L357-L393","documentation":"ValueError from parse_csi_data when the buffer is shorter than HEADER_SIZE + header['csi_length'], i.e. the header parsed fine and declares a CSI payload size that the remaining bytes do not cover. The guard protects the slice data[csi_start:csi_start + csi_length] and the 10-bit unpack loop that follows.","triggerScenarios":"Calling parse_csi_data with only the header bytes; a truncated capture where csi_length bytes were not yet read; a corrupted csi_length field (e.g. bit-flip) making the declared payload larger than the actual packet; off-by-N desync after a dropped packet.","commonSituations":"Splitting header parsing and CSI parsing across two reads without accumulating; firmware/driver version changing the CSI payload size while the parser assumes the old layout; noisy RF captures occasionally producing garbage headers.","solutions":["Accumulate data until len(data) >= HEADER_SIZE + header['csi_length'] before calling parse_csi_data","Sanity-check header['csi_length'] (0 < csi_length <= max expected for the antenna config / channel width) and discard the packet as corrupt if it fails","If lengths are consistently wrong after a firmware update, update the parser's expected layout","Treat this as a drop-and-continue error in the capture loop; log the declared vs available lengths for diagnosis"],"exampleFix":"# before\ncsi = AtherosCSIFormat.parse_csi_data(buf, header)  # ValueError when payload truncated\n\n# after\nneed = AtherosCSIFormat.HEADER_SIZE + header['csi_length']\nif len(buf) < need:\n    buf += read_more()\ncsi = AtherosCSIFormat.parse_csi_data(buf, header)","handlingStrategy":"validation","validationCode":"need = AtherosCSIFormat.HEADER_SIZE + header['csi_length']\nif len(buf) < need:\n    buf += read_more()\nif not (0 < header['csi_length'] <= MAX_EXPECTED_CSI_LEN):\n    drop_packet()  # corrupt header; parsing would fail or yield garbage","typeGuard":"def has_full_csi(buf: bytes, header: dict) -> bool:\n    return len(buf) >= AtherosCSIFormat.HEADER_SIZE + header['csi_length']","tryCatchPattern":"try:\n    csi = AtherosCSIFormat.parse_csi_data(buf, header)\nexcept ValueError as e:\n    if 'Insufficient data for CSI' in str(e):\n        logging.warning('truncated CSI: declared %d, have %d', header['csi_length'], len(buf))\n        continue  # drop, stay in sync via framing\n    raise","preventionTips":["Header and CSI must come from the same accumulated buffer, not separate reads","Bound csi_length with a plausibility check before trusting it","After firmware changes, re-validate expected CSI payload sizes for the new build"],"tags":["python","csi","atheros","parsing","validation","data-truncation"],"backgroundTag":null,"analyzedSha":"4685618388a5e49fad5b3005806f3bdd6a7c25c3","analyzedAt":"2026-08-16T06:09:40.886Z","schemaVersion":2},"datasetVersion":"2026-08-16T08:17:34.114Z"}