diff --git a/vna_system/core/processors/configs/bscan_config.json b/vna_system/core/processors/configs/bscan_config.json index b0a8735..6db4813 100644 --- a/vna_system/core/processors/configs/bscan_config.json +++ b/vna_system/core/processors/configs/bscan_config.json @@ -4,7 +4,7 @@ "s11_norm_enabled": false, "subtract_mean_ascan": false, "hardcoded_notch_enabled": false, - "axis": "real", + "axis": "imag", "cut": 0.0, "max": 3.0, "gain": 1.5, diff --git a/vna_system/core/processors/implementations/bscan_processor.py b/vna_system/core/processors/implementations/bscan_processor.py index 2111f28..7ae50ec 100644 --- a/vna_system/core/processors/implementations/bscan_processor.py +++ b/vna_system/core/processors/implementations/bscan_processor.py @@ -512,7 +512,6 @@ class BScanProcessor(BaseProcessor): y_coords: list[float] = [] z_values: list[float] = [] - z_values_square = np.zeros((len(history[0]["distance_data"]),len(history)),dtype=float) time_series = [record["time_domain_data"] for record in history] adjusted_time_series = self._apply_mean_ascan_subtraction(time_series) @@ -525,7 +524,6 @@ class BScanProcessor(BaseProcessor): normalized_ampls = np.array(amps) / np.max(np.array(amps)[depth_mask]) else: normalized_ampls = np.array(amps) - z_values_square[:,sweep_index-1] = normalized_ampls for d, a in zip(depths, normalized_ampls, strict=False): x_coords.append(sweep_index) diff --git a/vna_system/core/processors/websocket_handler.py b/vna_system/core/processors/websocket_handler.py index 14f0955..6f9514e 100644 --- a/vna_system/core/processors/websocket_handler.py +++ b/vna_system/core/processors/websocket_handler.py @@ -71,6 +71,7 @@ class ProcessorWebSocketHandler: await websocket.accept() self.active_connections.add(websocket) + self._send_locks[websocket] = asyncio.Lock() logger.info("WebSocket connected", total_connections=len(self.active_connections)) try: @@ -87,8 +88,22 @@ class ProcessorWebSocketHandler: """Remove a connection and log the updated count.""" if websocket in self.active_connections: self.active_connections.remove(websocket) + self._send_locks.pop(websocket, None) logger.info("WebSocket disconnected", total_connections=len(self.active_connections)) + async def _send(self, websocket: WebSocket, message_str: str) -> None: + """ + Send text on a connection, serialized against any other send on it. + + All outbound traffic must go through here; see `_send_locks`. + """ + lock = self._send_locks.get(websocket) + if lock is None: + # Connection already torn down. + return + async with lock: + await websocket.send_text(message_str) + # --------------------------------------------------------------------- # # Inbound messages # --------------------------------------------------------------------- # @@ -96,7 +111,7 @@ class ProcessorWebSocketHandler: """Parse and route an inbound client message.""" # Handle ping/pong messages first (they are not JSON) if data == "ping": - await websocket.send_text("pong") + await self._send(websocket, "pong") return elif data == "pong": # Just acknowledge, no response needed @@ -189,7 +204,7 @@ class ProcessorWebSocketHandler: for r in history ], } - await websocket.send_text(json.dumps(response)) + await self._send(websocket, json.dumps(response)) except Exception as exc: # noqa: BLE001 logger.error("Error getting history") await self._send_error(websocket, f"Ошибка получения истории: {exc}") @@ -270,7 +285,7 @@ class ProcessorWebSocketHandler: try: response = await asyncio.to_thread(self.processor_manager.build_processor_state, processor_id) - await websocket.send_text(json.dumps(response)) + await self._send(websocket, json.dumps(response)) except Exception as exc: # noqa: BLE001 logger.error("Error getting processor state", processor_id=processor_id, error=repr(exc)) await self._send_error(websocket, f"Ошибка получения состояния процессора: {exc}") @@ -371,7 +386,7 @@ class ProcessorWebSocketHandler: if raw_data: payload["raw_data"] = raw_data - await websocket.send_text(json.dumps(payload)) + await self._send(websocket, json.dumps(payload)) except Exception as exc: # noqa: BLE001 logger.error("Error sending error message", error=repr(exc)) @@ -432,7 +447,7 @@ class ProcessorWebSocketHandler: disconnected: list[WebSocket] = [] for websocket in list(self.active_connections): # snapshot try: - await websocket.send_text(message_str) + await self._send(websocket, message_str) except Exception as exc: # noqa: BLE001 logger.error("Broadcast to client failed; marking for disconnect", error=repr(exc)) disconnected.append(websocket) diff --git a/vna_system/main.py b/vna_system/main.py index 676b364..f42cb48 100644 --- a/vna_system/main.py +++ b/vna_system/main.py @@ -103,6 +103,11 @@ def main() -> None: # (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, + # The client drives its own ping/pong (see web_ui websocket.js), and the + # library's keepalive task races with our large result frames on the same + # transport, tripping an assertion inside websockets' drain handling. + ws_ping_interval=None, + ws_ping_timeout=None, )