diff --git a/vna_system/core/processors/base_processor.py b/vna_system/core/processors/base_processor.py index 06d064e..1d8efc1 100644 --- a/vna_system/core/processors/base_processor.py +++ b/vna_system/core/processors/base_processor.py @@ -894,7 +894,13 @@ class BaseProcessor: "timestamp": entry.get("timestamp"), }) - logger.info("History imported", processor_id=self.processor_id, records=len(history_data)) + self._trim_history() + logger.info( + "History imported", + processor_id=self.processor_id, + records=len(history_data), + kept=len(self._sweep_history), + ) @staticmethod def _points_to_list(points: Any) -> list[tuple[float, float]]: diff --git a/vna_system/core/processors/configs/bscan_config.json b/vna_system/core/processors/configs/bscan_config.json index 6fe3cc7..b0a8735 100644 --- a/vna_system/core/processors/configs/bscan_config.json +++ b/vna_system/core/processors/configs/bscan_config.json @@ -4,9 +4,9 @@ "s11_norm_enabled": false, "subtract_mean_ascan": false, "hardcoded_notch_enabled": false, - "axis": "abs", + "axis": "real", "cut": 0.0, - "max": 12.0, + "max": 3.0, "gain": 1.5, "start_freq": 1730.0, "stop_freq": 6000.0, diff --git a/vna_system/core/processors/implementations/bscan_processor.py b/vna_system/core/processors/implementations/bscan_processor.py index c802f56..2111f28 100644 --- a/vna_system/core/processors/implementations/bscan_processor.py +++ b/vna_system/core/processors/implementations/bscan_processor.py @@ -60,7 +60,7 @@ class BScanProcessor(BaseProcessor): super().__init__("bscan", config_dir) # Increase history size for multi-sweep plotting - self._max_history = 50 + self._max_history = 1000 # Local plot history (separate from sweep history maintained by BaseProcessor) self._plot_history: list[dict[str, Any]] = [] @@ -320,6 +320,45 @@ class BScanProcessor(BaseProcessor): # Processing # ------------------------------------------------------------------------- + def add_sweep_data( + self, + sweep_data: Any, + calibrated_data: Any, + vna_config: Any, + reference_data: Any = None, + reference_info: Any = None, + raw_reference_data: Any = None, + calibration_standards: dict | None = None, + ) -> ProcessedResult | None: + """ + Add the latest sweep and process it incrementally. + + Overridden to avoid BaseProcessor's default of calling `self.recalculate()`, + which for BScanProcessor rebuilds the *entire* plot history from scratch. + Doing that on every incoming sweep is O(history_size) per sweep, so with a + large `_max_history` a live stream turns into O(n^2) work and floods the + websocket with full-heatmap payloads on every sweep. Here we only process + the new sweep and append it to the existing plot history; a full rebuild + is still done by `recalculate()` when the user changes config or loads history. + """ + with self._lock: + self._sweep_history.append( + { + "sweep_data": sweep_data, + "calibrated_data": calibrated_data, + "vna_config": self._snapshot_vna_config(vna_config), + "reference_data": reference_data, + "reference_info": reference_info, + "raw_reference_data": raw_reference_data, + "calibration_standards": calibration_standards, + "timestamp": datetime.now().timestamp(), + } + ) + self._trim_history() + latest_vna_config = self._sweep_history[-1]["vna_config"] + + return self._process_data(sweep_data, calibrated_data, latest_vna_config) + def process_sweep( self, sweep_data: SweepData, diff --git a/vna_system/core/processors/websocket_handler.py b/vna_system/core/processors/websocket_handler.py index a176cbc..14f0955 100644 --- a/vna_system/core/processors/websocket_handler.py +++ b/vna_system/core/processors/websocket_handler.py @@ -38,6 +38,16 @@ class ProcessorWebSocketHandler: self.active_connections: set[WebSocket] = set() + # Starlette's WebSocket.send() is not safe to call concurrently from + # multiple tasks on the same connection (it corrupts internal ASGI + # protocol state, raising AssertionError and killing the connection). + # Broadcasts triggered by unrelated processor results (scheduled from + # worker threads) can otherwise race with a direct reply to a client + # command, especially once a payload is large enough to take a while + # to transmit (e.g. a big B-scan history). Serialize all sends per + # connection with a lock to eliminate that race. + self._send_locks: dict[WebSocket, asyncio.Lock] = {} + # Main FastAPI/uvicorn event loop handle (set on first connection). self._loop: asyncio.AbstractEventLoop | None = None @@ -138,10 +148,16 @@ class ProcessorWebSocketHandler: return try: - result = self.processor_manager.recalculate_processor(processor_id, config_updates) - if result: - await websocket.send_text(json.dumps(self._result_to_message(processor_id, result))) - else: + # Note: recalculate_processor() already broadcasts the result to every + # connected client (including this one) via the manager's result + # callbacks, so we must NOT also send it here directly — two + # concurrent writes to the same connection (the broadcast task and + # this handler) can race on the ASGI transport and corrupt/close it, + # especially with large payloads that take a while to transmit. + result = await asyncio.to_thread( + self.processor_manager.recalculate_processor, processor_id, config_updates + ) + if not result: await self._send_error(websocket, f"Нет результата от процессора {processor_id}") except Exception as exc: # noqa: BLE001 logger.error("Recalculation failed") @@ -197,10 +213,14 @@ class ProcessorWebSocketHandler: return try: - result = self.processor_manager.load_processor_history(processor_id, history_data, config) - if result: - await websocket.send_text(json.dumps(self._result_to_message(processor_id, result))) - else: + # See note in _handle_recalculate: load_processor_history() already + # broadcasts the result to all clients via result callbacks, so no + # explicit send here (avoids racing two concurrent writes on the + # same connection, which reliably crashes it for large histories). + result = await asyncio.to_thread( + self.processor_manager.load_processor_history, processor_id, history_data, config + ) + if not result: await self._send_error(websocket, f"Нет результата от процессора {processor_id} после загрузки истории") except Exception as exc: # noqa: BLE001 logger.error("History load failed", processor_id=processor_id, error=repr(exc)) @@ -224,10 +244,12 @@ class ProcessorWebSocketHandler: return try: - result = self.processor_manager.append_processor_history(processor_id, history_data) - if result: - await websocket.send_text(json.dumps(self._result_to_message(processor_id, result))) - else: + # See note in _handle_recalculate: append_processor_history() already + # broadcasts the result to all clients via result callbacks. + result = await asyncio.to_thread( + self.processor_manager.append_processor_history, processor_id, history_data + ) + if not result: await self._send_error(websocket, f"Нет результата от процессора {processor_id} после дополнения истории") except Exception as exc: # noqa: BLE001 logger.error("History append failed", processor_id=processor_id, error=repr(exc)) @@ -247,7 +269,7 @@ class ProcessorWebSocketHandler: return try: - response = self.processor_manager.build_processor_state(processor_id) + response = await asyncio.to_thread(self.processor_manager.build_processor_state, processor_id) await websocket.send_text(json.dumps(response)) except Exception as exc: # noqa: BLE001 logger.error("Error getting processor state", processor_id=processor_id, error=repr(exc)) @@ -293,7 +315,7 @@ class ProcessorWebSocketHandler: return # Recalculate and send updated result to all clients - result = processor.recalculate() + result = await asyncio.to_thread(processor.recalculate) if result: # Broadcast to all connected clients message_str = json.dumps(self._result_to_message(processor_id, result)) diff --git a/vna_system/main.py b/vna_system/main.py index fba59d7..676b364 100644 --- a/vna_system/main.py +++ b/vna_system/main.py @@ -99,6 +99,10 @@ def main() -> None: port=port, log_level="info", reload=False, + # Default 16MB is too small for uploading large B-scan history files + # (hundreds of sweeps x 1000 points each easily exceeds it); the + # connection is silently reset with no application-level error. + ws_max_size=256 * 1024 * 1024, )