optimization fix

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]]:

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,

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,

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)
+            # Note: recalculate_processor() already broadcasts the result to every
-            if result:
+            # connected client (including this one) via the manager's result
-                await websocket.send_text(json.dumps(self._result_to_message(processor_id, result)))
+            # callbacks, so we must NOT also send it here directly — two
-            else:
+            # 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)
+            # See note in _handle_recalculate: load_processor_history() already
-            if result:
+            # broadcasts the result to all clients via result callbacks, so no
-                await websocket.send_text(json.dumps(self._result_to_message(processor_id, result)))
+            # explicit send here (avoids racing two concurrent writes on the
-            else:
+            # 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)
+            # See note in _handle_recalculate: append_processor_history() already
-            if result:
+            # broadcasts the result to all clients via result callbacks.
-                await websocket.send_text(json.dumps(self._result_to_message(processor_id, result)))
+            result = await asyncio.to_thread(
-            else:
+                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))

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,
     )