{"record":{"id":"93984e38d229891c","repo":"ruvnet/RuView","slug":"pipeline-processing-failed-e","errorCode":null,"errorMessage":"Pipeline processing failed: {e}","messagePattern":"Pipeline processing failed: (.+?)","errorType":"exception","errorClass":"CSIProcessingError","httpStatus":null,"severity":"error","filePath":"archive/v1/src/core/csi_processor.py","lineNumber":265,"sourceCode":"            self._total_processed += 1\n            \n            # Preprocess the data\n            preprocessed_data = self.preprocess_csi_data(csi_data)\n            \n            # Extract features\n            features = self.extract_features(preprocessed_data)\n            \n            # Detect human presence\n            detection_result = self.detect_human_presence(features)\n            \n            # Add to history\n            self.add_to_history(csi_data)\n            \n            return detection_result\n            \n        except Exception as e:\n            self._processing_errors += 1\n            raise CSIProcessingError(f\"Pipeline processing failed: {e}\")\n    \n    def add_to_history(self, csi_data: CSIData) -> None:\n        \"\"\"Add CSI data to processing history.\n\n        Args:\n            csi_data: CSI data to add to history\n        \"\"\"\n        self.csi_history.append(csi_data)\n        # Cache mean phase for fast Doppler extraction\n        if csi_data.phase.ndim == 2:\n            self._phase_cache.append(np.mean(csi_data.phase, axis=0))\n        else:\n            self._phase_cache.append(csi_data.phase.flatten())\n    \n    def clear_history(self) -> None:\n        \"\"\"Clear the CSI data history.\"\"\"\n        self.csi_history.clear()\n        self._phase_cache.clear()","sourceCodeStart":247,"sourceCodeEnd":283,"githubUrl":"https://github.com/ruvnet/RuView/blob/4685618388a5e49fad5b3005806f3bdd6a7c25c3/archive/v1/src/core/csi_processor.py#L247-L283","documentation":"CSIProcessingError raised by the async CSIProcessor.process_csi_data, the end-to-end pipeline (preprocess -> extract -> detect -> add_to_history). Any exception from any stage is re-wrapped under 'Pipeline processing failed'; the handler also increments the internal _processing_errors counter, so repeated failures are visible via processor stats. The nested original message (possibly double-wrapped, e.g. 'Failed to preprocess CSI data: ...') identifies the failing stage.","triggerScenarios":"Awaiting processor.process_csi_data(csi_data) with data that fails any single stage: undersized batches (< window_size), shape mismatches, NaNs, or invalid internal state. Because it catches Exception, programming errors in custom config or data classes also surface here.","commonSituations":"Streaming loops pushing every ESP32 packet immediately, including the first undersized ones; network packet loss producing ragged final windows; switching hardware profiles mid-stream changing subcarrier counts; concurrency where multiple coroutines share one processor and corrupt history state.","solutions":["Unwrap the message chain: the text nests stage errors — fix the innermost cause (undersized batch, shape, NaN).","Buffer at the source: only call process_csi_data when len(amplitude) >= window_size.","Validate CSIData shape/NaN before entering the pipeline (see validationCode) and drop bad batches with a logged warning.","Give each concurrent consumer its own CSIProcessor; check processor._processing_errors in monitoring to detect chronic bad input."],"exampleFix":"# before\nresult = await processor.process_csi_data(csi_data)  # first tiny packet raises\n\n# after\nif csi_data.amplitude.shape[0] >= processor.window_size:\n    result = await processor.process_csi_data(csi_data)\nelse:\n    buffer.append(csi_data)  # wait for more samples","handlingStrategy":"try-catch","validationCode":"def pipeline_ready(processor, csi_data) -> bool:\n    return (\n        csi_data.amplitude.ndim == 2\n        and csi_data.amplitude.shape == csi_data.phase.shape\n        and csi_data.amplitude.shape[0] >= processor.window_size\n        and bool(np.isfinite(csi_data.amplitude).all())\n    )\n\nif pipeline_ready(processor, csi_data):\n    result = await processor.process_csi_data(csi_data)","typeGuard":null,"tryCatchPattern":"try:\n    result = await processor.process_csi_data(csi_data)\nexcept CSIProcessingError as e:\n    logger.warning('pipeline failure on batch %s: %s',\n                   getattr(csi_data, 'timestamp', '?'), e)\n    continue  # drop batch; monitor processor._processing_errors rate\n# abort (re-raise) when the error rate exceeds a threshold","preventionTips":["Gate the stream: only await process_csi_data once the buffer holds >= window_size samples.","Drop corrupted batches at ingestion with a logged warning instead of feeding them to the pipeline.","Track _processing_errors / total ratio as a health metric; alert on sustained bad-input rates.","Use one CSIProcessor per coroutine; never share across tasks."],"tags":["csi-processing","runtime","async","pipeline","wrapper-exception"],"backgroundTag":null,"analyzedSha":"4685618388a5e49fad5b3005806f3bdd6a7c25c3","analyzedAt":"2026-08-16T06:09:40.886Z","schemaVersion":2},"datasetVersion":"2026-08-16T08:17:34.114Z"}