From 2fdc0becc3ff25eb77c0cedf2b13c4033a2043d2 Mon Sep 17 00:00:00 2001 From: ljiang-slac Date: Fri, 10 Jul 2026 13:12:50 -0700 Subject: [PATCH 1/5] updated the way to calculate time window --- src/components/MultiPVPlot.jsx | 31 +++++++++++++++++++++---- start-frontend.sh | 42 ++++++++++++++++++++++++++++++++++ 2 files changed, 68 insertions(+), 5 deletions(-) create mode 100755 start-frontend.sh diff --git a/src/components/MultiPVPlot.jsx b/src/components/MultiPVPlot.jsx index b8ae177..b2578f6 100644 --- a/src/components/MultiPVPlot.jsx +++ b/src/components/MultiPVPlot.jsx @@ -44,7 +44,7 @@ export default function MultiPVPlot({ plotId, pvNames }) { alert('No data to export'); return; } - + const data = buffer.getData(); const csv = ['Timestamp,Value'] @@ -199,13 +199,34 @@ export default function MultiPVPlot({ plotId, pvNames }) { let xMin = null; let xMax = null; - + + // use the last point data time stamp to replace current time if (timeSyncEnabled) { - const now = new Date(); - xMax = now; - xMin = new Date(now.getTime() - globalTimeWindow * 1000); + // Anchor the window's right edge to the newest DATA timestamp, + // not the front-end wall clock (they can differ if clocks aren't synced). + let latest = null; + pvNames.forEach((pvName) => { + const buffer = buffersRef.current[pvName]; + const ts = buffer ? buffer.getLatestTimestamp() : null; + if (ts && (!latest || ts > latest)) { + latest = ts; + } + }); + + // Use latest data time as the right edge; fall back to now if no data yet + xMax = latest || new Date(); + xMin = new Date(xMax.getTime() - globalTimeWindow * 1000); } + + + + //if (timeSyncEnabled) { + // const now = new Date(); + // xMax = now; + // xMin = new Date(now.getTime() - globalTimeWindow * 1000); + //} + const traces = pvNames.map((pvName) => { const buffer = buffersRef.current[pvName]; let data = buffer ? buffer.getData() : { x: [], y: [] }; diff --git a/start-frontend.sh b/start-frontend.sh new file mode 100755 index 0000000..1d2c61e --- /dev/null +++ b/start-frontend.sh @@ -0,0 +1,42 @@ +#!/usr/bin/env bash +# +# start-frontend.sh +# Starts the EPICS PV Plotter frontend (Vite dev server). +# Can be run manually for testing, or invoked by systemd. +# + +set -e # 出错即退出 + +# ------------------------------------------------------------ +# 1. 加载 nvm（关键：让 node/npm 可用） +# ------------------------------------------------------------ +export NVM_DIR="/home/b_bluesky/.nvm" +# shellcheck disable=SC1091 +[ -s "$NVM_DIR/nvm.sh" ] && source "$NVM_DIR/nvm.sh" + +# 使用指定的 node 版本（和你 which node 一致） +nvm use v20.11.1 >/dev/null 2>&1 || true + +# 兜底：直接把 node bin 目录加进 PATH +export PATH="/home/b_bluesky/.nvm/versions/node/v20.11.1/bin:$PATH" + +# ------------------------------------------------------------ +# 2. 进入项目目录 +# ------------------------------------------------------------ +cd /home/b_bluesky/Documents/epics-pv-plotter + +# ------------------------------------------------------------ +# 3. 打印环境信息（方便调试） +# ------------------------------------------------------------ +echo "==========================================" +echo " Starting EPICS PV Plotter Frontend" +echo " node: $(which node) ($(node -v))" +echo " npm : $(which npm) ($(npm -v))" +echo " cwd : $(pwd)" +echo "==========================================" + +# ------------------------------------------------------------ +# 4. 启动 Vite dev server +# exec 让 npm 进程替换当前 shell，便于 systemd 正确管理进程 +# ------------------------------------------------------------ +exec npm run dev -- --host 0.0.0.0 --port 5173 From 6c0b2c6cbfb81fe35fd53bf365181e9edeef9af7 Mon Sep 17 00:00:00 2001 From: ljiang-slac Date: Mon, 13 Jul 2026 13:46:14 -0700 Subject: [PATCH 2/5] added health endpoint for health header request --- epics_fastapi_gateway.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/epics_fastapi_gateway.py b/epics_fastapi_gateway.py index 865a97d..82a3228 100644 --- a/epics_fastapi_gateway.py +++ b/epics_fastapi_gateway.py @@ -20,6 +20,7 @@ from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse +from fastapi import Response import uvicorn from caproto.asyncio.client import Context @@ -313,6 +314,17 @@ async def root(): """ return html +@app.head("/health") +async def health_head(): + return Response( + + status_code=200, + headers={ + "X-Status": "healthy", + "X-Active-Connections": str(len(active_subscriptions)), + }, + + ) @app.get("/health") async def health_check(): From d941cdabf27b9cf09dcb502712a38a67c1d527e8 Mon Sep 17 00:00:00 2001 From: ljiang-slac Date: Mon, 20 Jul 2026 12:16:31 -0700 Subject: [PATCH 3/5] added a new global connection pool --- src/components/MultiPVPlot.jsx | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/components/MultiPVPlot.jsx b/src/components/MultiPVPlot.jsx index b2578f6..e07361a 100644 --- a/src/components/MultiPVPlot.jsx +++ b/src/components/MultiPVPlot.jsx @@ -201,6 +201,7 @@ export default function MultiPVPlot({ plotId, pvNames }) { let xMax = null; // use the last point data time stamp to replace current time + if (timeSyncEnabled) { // Anchor the window's right edge to the newest DATA timestamp, // not the front-end wall clock (they can differ if clocks aren't synced). @@ -217,10 +218,10 @@ export default function MultiPVPlot({ plotId, pvNames }) { xMax = latest || new Date(); xMin = new Date(xMax.getTime() - globalTimeWindow * 1000); } + - - + //if (timeSyncEnabled) { // const now = new Date(); // xMax = now; From b4d349451c976c52f404f3dc487893ee74d74409 Mon Sep 17 00:00:00 2001 From: ljiang-slac Date: Mon, 20 Jul 2026 12:17:38 -0700 Subject: [PATCH 4/5] added global PV connection pool --- epics-ws-gateway-caproto-sec.py | 4 + epics_fastapi_sse_gateway.py | 235 +++++++++++++++++++++++++++++++ src/services/PVConnectionPool.js | 64 +++++++++ src/services/WebSocketManager.js | 12 +- src/stores/usePlotStore.js | 130 +++++++++-------- src/utils/constants.js | 2 +- tests/commands.txt | 1 + tests/test_fastapi_client.py | 17 +++ tests/test_iterable_object.py | 101 +++++++++++++ tests/test_prototype.js | 51 +++++++ tests/test_sse_client.py | 11 ++ tests/test_subscription.py | 83 +++++++++++ tests/test_ws_client.py | 8 +- tutorial/uvicorn.py | 2 +- tutorial/websocket_manager.txt | 101 +++++++++++++ 15 files changed, 755 insertions(+), 67 deletions(-) create mode 100644 epics_fastapi_sse_gateway.py create mode 100644 src/services/PVConnectionPool.js create mode 100644 tests/commands.txt create mode 100644 tests/test_fastapi_client.py create mode 100644 tests/test_iterable_object.py create mode 100644 tests/test_prototype.js create mode 100644 tests/test_sse_client.py create mode 100644 tests/test_subscription.py create mode 100644 tutorial/websocket_manager.txt diff --git a/epics-ws-gateway-caproto-sec.py b/epics-ws-gateway-caproto-sec.py index 1833fff..fa6f56a 100644 --- a/epics-ws-gateway-caproto-sec.py +++ b/epics-ws-gateway-caproto-sec.py @@ -210,12 +210,16 @@ def signal_handler(): # Register signal handlers # Signal Terminate : command, Signal Interrupt : keyborad # call the handler when the loop gets signal + # Signal Interrupt, Ctrl + C + # Signal Terminate kill for sig in (signal.SIGTERM, signal.SIGINT): + loop.add_signal_handler(sig, signal_handler) # Start WebSocket server try: async with websockets.serve(handle_client, HOST, PORT): + # let stop event wait until get signal to stop await stop_event.wait() finally: # Clean up all active subscriptions diff --git a/epics_fastapi_sse_gateway.py b/epics_fastapi_sse_gateway.py new file mode 100644 index 0000000..c540383 --- /dev/null +++ b/epics_fastapi_sse_gateway.py @@ -0,0 +1,235 @@ +#!/usr/bin/env python3 +""" +EPICS Channel Access SSE Gateway using FastAPI + caproto + +- GET /sse?pv=PV:NAME streams PV updates via Server-Sent Events (SSE) +- GET /health +- GET /api/subscriptions + +Run: + python epics_fastapi_sse_gateway.py +or: + uvicorn epics_fastapi_sse_gateway:app --host 0.0.0.0 --port 8001 +""" + +import asyncio +import json +import logging +from contextlib import asynccontextmanager +from datetime import datetime +from typing import Dict, Optional, AsyncIterator + +import numpy as np +from caproto.asyncio.client import Context +from fastapi import FastAPI, Query +from fastapi import Response +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import StreamingResponse +import uvicorn + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + datefmt="%H:%M:%S", +) +log = logging.getLogger("epics_sse_gateway") + +CA_CONTEXT: Optional[Context] = None + +# Track active SSE subscriptions +# key: client_id, value: caproto subscription +active_subscriptions: Dict[str, object] = {} + + +def extract_value(data): + # Similar to your WebSocket gateway helper + if hasattr(data, "__iter__") and not isinstance(data, (str, bytes)): + try: + result = [] + for item in data: + if isinstance(item, bytes): + result.append(item.decode("utf-8", errors="replace").rstrip("\x00")) + elif isinstance(item, (np.number, int, float)): + result.append(float(item)) + else: + result.append(item) + return result[0] if len(result) == 1 else result + except Exception: + pass + + if isinstance(data, bytes): + return data.decode("utf-8", errors="replace").rstrip("\x00") + if isinstance(data, str): + return data.rstrip("\x00") + if hasattr(data, "tolist"): + val = data.tolist() + if isinstance(val, list) and len(val) == 1: + v = val[0] + return float(v) if isinstance(v, np.number) else v + return val + if isinstance(data, (int, float, np.number)): + return float(data) + + try: + return float(data) + except (ValueError, TypeError): + return str(data) + + +def sse_event(data: dict, event: str = "update", event_id: Optional[str] = None) -> str: + """ + Format one SSE message. + SSE format: + id: \n + event: \n + data: \n + \n + """ + payload = json.dumps(data, ensure_ascii=False) + lines = [] + if event_id is not None: + lines.append(f"id: {event_id}") + if event: + lines.append(f"event: {event}") + lines.append(f"data: {payload}") + return "\n".join(lines) + "\n\n" + + +@asynccontextmanager +async def lifespan(app: FastAPI) -> AsyncIterator[None]: + global CA_CONTEXT + log.info("Starting EPICS SSE Gateway...") + CA_CONTEXT = Context() + yield + log.info("Shutting down... clearing subscriptions") + for cid, sub in list(active_subscriptions.items()): + try: + sub.clear() + except Exception as e: + log.warning(f"Cleanup error for {cid}: {e}") + active_subscriptions.clear() + log.info("Cleanup complete") + + +app = FastAPI( + title="EPICS SSE Gateway", + version="1.0.0", + description="Stream EPICS CA PV updates via Server-Sent Events (SSE)", + lifespan=lifespan, +) + +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], # tighten in production + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + + +@app.head("/health") +async def health_head(): + return Response( + status_code=200, + headers={ + "X-Status": "healthy", + "X-Active-Subscriptions": str(len(active_subscriptions)), + }, + ) + + +@app.get("/health") +async def health(): + return { + "status": "healthy", + "active_subscriptions": len(active_subscriptions), + "timestamp": datetime.now().isoformat(), + } + + +@app.get("/api/subscriptions") +async def subscriptions(): + return {"count": len(active_subscriptions), "subscriptions": list(active_subscriptions.keys())} + + +@app.get("/sse") +async def sse(pv: str = Query(..., description="EPICS PV name to monitor via SSE")): + """ + SSE endpoint: + GET /sse?pv=PV:NAME + + Client example (browser): + const es = new EventSource("http://host:8001/sse?pv=TEST:PV"); + es.onmessage = (e) => console.log(e.data); + """ + if CA_CONTEXT is None: + # should not happen if lifespan ran + raise RuntimeError("CA context not initialized") + + # A simple client id (no direct access to client IP like WebSocket object) + client_id = f"sse:{pv}:{datetime.now().timestamp()}" + log.info(f"SSE subscribe: {client_id}") + + async def event_generator(): + subscription = None + try: + epics_pv, = await CA_CONTEXT.get_pvs(pv) + + # Send an initial value as a "snapshot" event + initial = await epics_pv.read(data_type="time") + snapshot = { + "pv": pv, + "value": extract_value(initial.data), + "timestamp": initial.metadata.timestamp, + "status": initial.metadata.status, + "severity": initial.metadata.severity, + "type": "snapshot", + } + yield sse_event(snapshot, event="snapshot", event_id="0") + + # Subscribe for updates + subscription = epics_pv.subscribe(data_type="time") + active_subscriptions[client_id] = subscription + + event_count = 0 + async for ev in subscription: + event_count += 1 + msg = { + "pv": pv, + "value": extract_value(ev.data), + "timestamp": ev.metadata.timestamp, + "status": ev.metadata.status, + "severity": ev.metadata.severity, + "type": "update", + } + yield sse_event(msg, event="update", event_id=str(event_count)) + + except asyncio.CancelledError: + # Happens when client disconnects and StreamingResponse cancels the generator + raise + except Exception as e: + err = {"pv": pv, "error": str(e), "type": "error", "timestamp": datetime.now().isoformat()} + yield sse_event(err, event="error") + finally: + # cleanup subscription + if subscription is not None: + try: + subscription.clear() + except Exception: + pass + active_subscriptions.pop(client_id, None) + log.info(f"SSE cleaned up: {client_id}") + + return StreamingResponse( + event_generator(), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + # If behind nginx, you may also want: "X-Accel-Buffering": "no" + }, + ) + + +if __name__ == "__main__": + uvicorn.run(app, host="0.0.0.0", port=8001, log_level="info") diff --git a/src/services/PVConnectionPool.js b/src/services/PVConnectionPool.js new file mode 100644 index 0000000..659d7ca --- /dev/null +++ b/src/services/PVConnectionPool.js @@ -0,0 +1,64 @@ +import { PVWebSocket } from "./WebSocketManager"; + +class PVConnectionPool { + constructor() { + // pvName -> { ws: PVWebSocket, subscribers: Set } + this.pool = new Map(); + } + + subscribe(pvName, callbacks) { + let entry = this.pool.get(pvName); + + if (!entry) { + const subscribers = new Set(); + + // Create ONE shared PVWebSocket for this PV + const ws = new PVWebSocket( + pvName, + (value, timestamp) => { + // Fan-out to all subscribers of this PV + subscribers.forEach((s) => s.onData?.(value, timestamp)); + }, + (error) => { + subscribers.forEach((s) => s.onError?.(error)); + }, + () => { + subscribers.forEach((s) => s.onConnect?.()); + } + ); + + ws.connect(); + entry = { ws, subscribers }; + this.pool.set(pvName, entry); + } + + // Register this subscriber (one per plot per PV) + const subscriber = { + onData: callbacks.onData, + onError: callbacks.onError, + onConnect: callbacks.onConnect, + }; + + entry.subscribers.add(subscriber); + + // Return an unsubscribe function + return () => { + const e = this.pool.get(pvName); + if (!e) return; + + e.subscribers.delete(subscriber); + + // If no more subscribers, close the shared connection + if (e.subscribers.size === 0) { + e.ws.disconnect(); + this.pool.delete(pvName); + } + }; + } + + getActiveConnectionCount() { + return this.pool.size; + } +} + +export const pvConnectionPool = new PVConnectionPool(); diff --git a/src/services/WebSocketManager.js b/src/services/WebSocketManager.js index 895d3a1..4aee699 100644 --- a/src/services/WebSocketManager.js +++ b/src/services/WebSocketManager.js @@ -17,22 +17,26 @@ export class PVWebSocket { this.reconnectTimer = null; this.messageCount = 0; // Track number of messages received } - + //called when the WebSocket connection is opened connect() { try { + //WS_CONFIG.BASE_URL saved in utils/constant.js IP address + Port Number + //encodeURIComponent: convert special characters into UTF-8 + percent-encoding const wsUrl = `${WS_CONFIG.BASE_URL}/ws?pv=${encodeURIComponent(this.pvName)}`; console.log(`🔗 Connecting to: ${wsUrl}`); this.ws = new WebSocket(wsUrl); - // + //Fire when the WebSocket handshake completes and the connection is open this.ws.onopen = () => { console.log(`✅ Connected: ${this.pvName}`); this.reconnectAttempts = 0; this.messageCount = 0; + //call onConnect callbackfunction/Notify the caller if it is passed if (this.onConnect) this.onConnect(); }; - // + + //WebSocket Message Event Handler this.ws.onmessage = (event) => { try { const data = JSON.parse(event.data); @@ -59,7 +63,7 @@ export class PVWebSocket { if (this.onError) this.onError('Data parse error'); } }; - // + //WebSocket Error Handler this.ws.onerror = (error) => { // Only log error if not initial connection attempt if (this.reconnectAttempts > 0) { diff --git a/src/stores/usePlotStore.js b/src/stores/usePlotStore.js index ebf87e3..58cf3b1 100644 --- a/src/stores/usePlotStore.js +++ b/src/stores/usePlotStore.js @@ -1,7 +1,7 @@ // src/stores/usePlotStore.js -import { create } from 'zustand'; -import { persist, createJSONStorage } from 'zustand/middleware'; -import { PLOT_CONFIG } from '../utils/constants'; +import { create } from "zustand"; +import { persist, createJSONStorage } from "zustand/middleware"; +import { PLOT_CONFIG } from "../utils/constants"; let nextPlotId = 1; @@ -9,7 +9,7 @@ export const usePlotStore = create( persist( (set, get) => ({ plots: [], - + // Time synchronization settings timeSyncEnabled: true, globalTimeWindow: 60, @@ -21,21 +21,19 @@ export const usePlotStore = create( set((state) => ({ latestValues: { ...state.latestValues, - [pvName]: { value, timestamp } - } + [pvName]: { value, timestamp }, + }, })); }, - - // Toggle time synchronization toggleTimeSync: () => { set((state) => ({ - timeSyncEnabled: !state.timeSyncEnabled + timeSyncEnabled: !state.timeSyncEnabled, })); - console.log(`Time sync: ${get().timeSyncEnabled ? 'ON' : 'OFF'}`); + console.log(`Time sync: ${get().timeSyncEnabled ? "ON" : "OFF"}`); }, - + // Set global time window setTimeWindow: (seconds) => { set({ globalTimeWindow: seconds }); @@ -45,45 +43,50 @@ export const usePlotStore = create( // Add plot with automatic layout calculation addPlot: (pvNames, width, height) => { const plots = get().plots; - + const newPlotWidth = PLOT_CONFIG.DEFAULT_WIDTH * width; const newPlotHeight = PLOT_CONFIG.DEFAULT_HEIGHT * height; - + // Find the best position for the new plot const position = findBestPosition(plots, newPlotWidth, newPlotHeight); - + const newPlot = { id: nextPlotId++, pvNames: Array.isArray(pvNames) ? pvNames : [pvNames], x: position.x, y: position.y, w: newPlotWidth, - h: newPlotHeight + h: newPlotHeight, }; - + set({ plots: [...plots, newPlot] }); - console.log(`Plot added at (${newPlot.x}, ${newPlot.y}), size: ${newPlot.w}x${newPlot.h}`, newPlot); + console.log( + `Plot added at (${newPlot.x}, ${newPlot.y}), size: ${newPlot.w}x${newPlot.h}`, + newPlot, + ); }, - // Remove Plot + // Remove Plot removePlot: (plotId) => { set((state) => ({ - plots: state.plots.filter((plot) => plot.id !== plotId) + plots: state.plots.filter((plot) => plot.id !== plotId), })); console.log(`Plot removed: ${plotId}`); }, removePVFromPlot: (plotId, pvName) => { set((state) => ({ - plots: state.plots.map((plot) => { - if (plot.id === plotId) { - const updatedPVs = plot.pvNames.filter((pv) => pv !== pvName); - return updatedPVs.length > 0 - ? { ...plot, pvNames: updatedPVs } - : null; - } - return plot; - }).filter(Boolean) + plots: state.plots + .map((plot) => { + if (plot.id === plotId) { + const updatedPVs = plot.pvNames.filter((pv) => pv !== pvName); + return updatedPVs.length > 0 + ? { ...plot, pvNames: updatedPVs } + : null; + } + return plot; + }) + .filter(Boolean), })); console.log(`PV removed: ${pvName} from plot ${plotId}`); }, @@ -91,50 +94,53 @@ export const usePlotStore = create( updateLayout: (newLayout) => { set((state) => ({ plots: state.plots.map((plot) => { - const layoutItem = newLayout.find((item) => item.i === plot.id.toString()); + const layoutItem = newLayout.find( + (item) => item.i === plot.id.toString(), + ); if (layoutItem) { return { ...plot, x: layoutItem.x, y: layoutItem.y, w: layoutItem.w, - h: layoutItem.h + h: layoutItem.h, }; } return plot; - }) + }), })); }, clearAll: () => { set({ plots: [] }); nextPlotId = 1; - console.log('All plots cleared'); - } + console.log("All plots cleared"); + }, }), { - name: 'epics-plot-storage', + name: "epics-plot-storage", storage: createJSONStorage(() => localStorage), - + onRehydrateStorage: () => (state) => { if (state) { - const maxId = state.plots.reduce((max, plot) => - Math.max(max, plot.id), 0 + const maxId = state.plots.reduce( + (max, plot) => Math.max(max, plot.id), + 0, ); nextPlotId = maxId + 1; - + console.log(`State rehydrated: ${state.plots.length} plots restored`); console.log(`Next plot ID will be: ${nextPlotId}`); } }, - - partialize: (state) => ({ + + partialize: (state) => ({ plots: state.plots, timeSyncEnabled: state.timeSyncEnabled, - globalTimeWindow: state.globalTimeWindow - }) - } - ) + globalTimeWindow: state.globalTimeWindow, + }), + }, + ), ); // Helper function to find the best position for a new plot @@ -144,34 +150,40 @@ function findBestPosition(existingPlots, width, height) { } const gridCols = PLOT_CONFIG.GRID_COLS; - + // Try to place in rows, starting from y=0 let currentRow = 0; - + while (true) { // Try each column position in this row for (let col = 0; col <= gridCols - width; col++) { const candidate = { x: col, y: currentRow }; - + // Check if this position conflicts with any existing plot - const hasConflict = existingPlots.some(plot => + const hasConflict = existingPlots.some((plot) => rectanglesOverlap( - candidate.x, candidate.y, width, height, - plot.x, plot.y, plot.w, plot.h - ) + candidate.x, + candidate.y, + width, + height, + plot.x, + plot.y, + plot.w, + plot.h, + ), ); - + if (!hasConflict) { return candidate; } } - + // Move to next row currentRow += PLOT_CONFIG.DEFAULT_HEIGHT; - + // Safety check: don't go beyond reasonable rows if (currentRow > 100) { - console.warn('Could not find position, placing at end'); + console.warn("Could not find position, placing at end"); return { x: 0, y: currentRow }; } } @@ -180,9 +192,9 @@ function findBestPosition(existingPlots, width, height) { // Helper function to check if two rectangles overlap function rectanglesOverlap(x1, y1, w1, h1, x2, y2, w2, h2) { return !( - x1 + w1 <= x2 || // rect1 is left of rect2 - x2 + w2 <= x1 || // rect2 is left of rect1 - y1 + h1 <= y2 || // rect1 is above rect2 - y2 + h2 <= y1 // rect2 is above rect1 + x1 + w1 <= x2 || // rect1 is left of rect2 + x2 + w2 <= x1 || // rect2 is left of rect1 + y1 + h1 <= y2 || // rect1 is above rect2 + y2 + h2 <= y1 // rect2 is above rect1 ); } diff --git a/src/utils/constants.js b/src/utils/constants.js index b87574f..ceabe05 100644 --- a/src/utils/constants.js +++ b/src/utils/constants.js @@ -20,7 +20,7 @@ export const WS_CONFIG = { // ================================================================ export const PLOT_CONFIG = { // Data buffer settings - MAX_POINTS: 6000, // Maximum number of data points stored per PV + MAX_POINTS: 1000, // Maximum number of data points stored per PV UPDATE_INTERVAL: 100, // Plot update interval (milliseconds) // Grid layout settings diff --git a/tests/commands.txt b/tests/commands.txt new file mode 100644 index 0000000..4cda37f --- /dev/null +++ b/tests/commands.txt @@ -0,0 +1 @@ +websocat 'ws://localhost:8000/ws?pv=SPEAR:BeamCurrAvg' diff --git a/tests/test_fastapi_client.py b/tests/test_fastapi_client.py new file mode 100644 index 0000000..afabbe0 --- /dev/null +++ b/tests/test_fastapi_client.py @@ -0,0 +1,17 @@ +#!/usr/bin/env python3 +import asyncio +import websockets +from urllib.parse import quote + +async def test(): + pv = "BL22:SCAN:MASTER:ADC1" + uri = f"ws://192.168.22.4:8000/ws?pv={quote(pv)}" + print(f"Connecting to {uri}...") + + async with websockets.connect(uri) as websocket: + print("✅ Connected!") + while True: + msg = await websocket.recv() + print("Received:", msg) + +asyncio.run(test()) diff --git a/tests/test_iterable_object.py b/tests/test_iterable_object.py new file mode 100644 index 0000000..3f43fc4 --- /dev/null +++ b/tests/test_iterable_object.py @@ -0,0 +1,101 @@ +#!/usr/bin/env python3 +import asyncio + +class MyAsyncIterable: + def __init__(self, values, name="sub"): + self.values = list(values) + self.name = name + + def __aiter__(self): + print(f"{self.name}.__aiter__() called") + return MyAsyncIterator(self.values, name=f"{self.name}.it") + +class MyAsyncIterator: + def __init__(self, values, name="it"): + self.values = list(values) + self.i = 0 + self.name = name + + def __aiter__(self): + return self + + async def __anext__(self): + if self.i >= len(self.values): + raise StopAsyncIteration + v = self.values[self.i] + print(f"{self.name}.__anext__() -> {v}") + self.i += 1 + await asyncio.sleep(0) + return v + + +async def demo_skip_bug(): + print("\n=== demo_skip_bug (mixing async for + anext on same iterator) ===") + sub = MyAsyncIterable([10, 11, 12, 13, 14, 15], name="sub_skip") + it = aiter(sub) + + print("\n-- WRONG: async for event in it + anext(it) inside loop --") + async for event in it: + print("async for got:", event) + + try: + extra = await anext(it) + print("manual anext got:", extra) + except StopAsyncIteration: + print("manual anext: StopAsyncIteration") + break + + +async def demo_equivalence_manual_vs_asyncfor(): + print("\n=== demo_equivalence (manual loop vs async for) ===") + + # Manual loop: aiter(obj) + repeatedly anext(it) + obj = MyAsyncIterable([1, 2, 3, 4], name="obj_manual") + print("\n-- manual: it = aiter(obj); while True: x = await anext(it) --") + it = aiter(obj) + manual = [] + while True: + try: + x = await anext(it) + except StopAsyncIteration: + break + manual.append(x) + print("manual collected:", manual) + + # async for: async for x in obj + obj = MyAsyncIterable([1, 2, 3, 4], name="obj_asyncfor") + print("\n-- async for x in obj --") + asyncfor_vals = [] + async for x in obj: + asyncfor_vals.append(x) + print("async for collected:", asyncfor_vals) + + print("\nSame result?", manual == asyncfor_vals) + + +async def demo_skip_visible(): + print("\n=== demo_skip_visible (events are skipped for the async for variable) ===") + sub = MyAsyncIterable([10, 11, 12, 13, 14, 15], name="sub_visible") + it = aiter(sub) + + print("\n-- WRONG pattern: discard one item via anext(it) each iteration --") + processed = [] + async for event in it: + # Consume and discard the next item (bug) + try: + await anext(it) + except StopAsyncIteration: + pass + processed.append(event) + + print("processed (from async for variable only):", processed) + print("expected if no bug:", [10, 11, 12, 13, 14, 15]) + + +async def main(): + await demo_skip_bug() + await demo_equivalence_manual_vs_asyncfor() + await demo_skip_visible() + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tests/test_prototype.js b/tests/test_prototype.js new file mode 100644 index 0000000..2a647be --- /dev/null +++ b/tests/test_prototype.js @@ -0,0 +1,51 @@ +class ProtoMethodDemo { + constructor(name) { + this.name = name; + + // Instance function property: each instance gets its own function object + this.instanceHello = function () { + return `instanceHello from ${this.name}`; + }; + } + + // Prototype method: shared by all instances via ProtoMethodDemo.prototype + protoHello() { + return `protoHello from ${this.name}`; + } +} + +const a = new ProtoMethodDemo("A"); +const b = new ProtoMethodDemo("B"); + +console.log("=== Where do methods live? ==="); +console.log("a.hasOwnProperty('protoHello'):", a.hasOwnProperty("protoHello")); // false +console.log("a.hasOwnProperty('instanceHello'):", a.hasOwnProperty("instanceHello")); // true +console.log("protoHello is from prototype:", + a.protoHello === ProtoMethodDemo.prototype.protoHello); // true + +console.log(a.protoHello === ProtoMethodDemo.prototype.protoHello); // true +console.log(Object.getPrototypeOf(a) === ProtoMethodDemo.prototype); // true + + + +console.log("\n=== Are methods shared? ==="); +console.log("a.protoHello === b.protoHello:", a.protoHello === b.protoHello); // true (shared) +console.log("a.instanceHello === b.instanceHello:", a.instanceHello === b.instanceHello); // false (per instance) + +console.log("\n=== Prototype linkage ==="); +console.log("Object.getPrototypeOf(a) === ProtoMethodDemo.prototype:", + Object.getPrototypeOf(a) === ProtoMethodDemo.prototype); // true + +console.log("\n=== Property lookup order (instance first, then prototype) ==="); +console.log("a.protoHello():", a.protoHello()); // uses prototype method +console.log("a.instanceHello():", a.instanceHello()); // uses instance property + +// Override: define a property on instance with same name as prototype method +a.protoHello = function () { + return "OVERRIDDEN protoHello on instance A"; +}; + +console.log("\n=== After overriding a.protoHello on the instance ==="); +console.log("a.hasOwnProperty('protoHello'):", a.hasOwnProperty("protoHello")); // true now +console.log("a.protoHello():", a.protoHello()); // instance version +console.log("b.protoHello():", b.protoHello()); // still prototype version diff --git a/tests/test_sse_client.py b/tests/test_sse_client.py new file mode 100644 index 0000000..f0068e8 --- /dev/null +++ b/tests/test_sse_client.py @@ -0,0 +1,11 @@ +import httpx + +pv = "SPEAR:BeamCurrAvg" +url = f"http://localhost:8001/sse?pv={pv}" + +with httpx.stream("GET", url, timeout=None) as r: + r.raise_for_status() + for line in r.iter_lines(): + if not line: + continue + print(line) diff --git a/tests/test_subscription.py b/tests/test_subscription.py new file mode 100644 index 0000000..7af025b --- /dev/null +++ b/tests/test_subscription.py @@ -0,0 +1,83 @@ +#!/usr/bin/env python3 +import asyncio +from caproto.asyncio.client import Context + +PV = "SPEAR:BeamCurrAvg" # Change this to a PV that updates periodically + +''' +async for equals to + + + + +_it = aiter(subscription) +while True: + try: + event = await anext(_it) # to get the next element + except StopAsyncIteration: + break + + +''' + + +''' + +async def main(): + ctx = Context() + pv, = await ctx.get_pvs(PV) + + subscription = pv.subscribe(data_type="time") + print("subscription type:", type(subscription)) + print("has __aiter__:", hasattr(subscription, "__aiter__")) + + n = 0 + async for event in subscription: + n += 1 + print("\n=== EVENT", n, "===") + print("event type:", type(event)) + print("dir(event):", [a for a in dir(event) if not a.startswith("_")]) + + print("has data:", hasattr(event, "data")) + print("has metadata:", hasattr(event, "metadata")) + + print("event.data type:", type(event.data)) + print("event.data repr:", repr(event.data)) + + md = event.metadata + print("metadata type:", type(md)) + print("dir(metadata):", [a for a in dir(md) if not a.startswith("_")]) + + for k in ["timestamp", "status", "severity"]: + if hasattr(md, k): + print(f"metadata.{k} =", getattr(md, k)) + + if n >= 1: + break + + await subscription.clear() + await ctx.disconnect() + +''' +async def main(): + ctx = Context() + pv, = await ctx.get_pvs(PV) + + subscription = pv.subscribe(data_type="time") + print("subscription type:", type(subscription)) + print("has __aiter__:", hasattr(subscription, "__aiter__")) + print("has __anext__:", hasattr(subscription, "__anext__")) + + async_iter = aiter(subscription) + + event1 = await anext(async_iter) + print("event1:", event1) + + event2 = await anext(async_iter) + print("event2:", event2) + + + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tests/test_ws_client.py b/tests/test_ws_client.py index b8e177d..612ad5a 100644 --- a/tests/test_ws_client.py +++ b/tests/test_ws_client.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 import asyncio import websockets - +from urllib.parse import quote async def test(): # get Uniform Resource Identifier (URI) @@ -31,7 +31,11 @@ async def test(): - Fragment (#section1) is NOT sent to server (browser-only) ''' - uri = "ws://192.168.22.4:8082?pv=BL22:SCAN:MASTER:ADC1" + pv = "BL22:SCAN:MASTER:ADC1" + uri = f"ws://192.168.22.4:8000/ws?pv={quote(pv)}" + + + #uri = "ws://192.168.22.4:8082?pv=BL22:SRS570_AMP1:NAME" print(f"Connecting to {uri}...") diff --git a/tutorial/uvicorn.py b/tutorial/uvicorn.py index a3936be..244e126 100644 --- a/tutorial/uvicorn.py +++ b/tutorial/uvicorn.py @@ -9,7 +9,7 @@ async def serve(): # 2. 🔥 call app.startup() await app.startup() # all the code before yield execution in lifespan - # CA_CONTEXT 已创建 + # CA_CONTEXT created # 3. list to the connection await server.start_listening() diff --git a/tutorial/websocket_manager.txt b/tutorial/websocket_manager.txt new file mode 100644 index 0000000..7a7851d --- /dev/null +++ b/tutorial/websocket_manager.txt @@ -0,0 +1,101 @@ +## WebSocket Data Format (Server → Client/FastAPI Gateway) + +The gateway streams PV updates as JSON messages over WebSocket. + +Example payload: + +```json +{ + "pv": "SPEAR:BeamCurrAvg", + "value": 305.12, + "timestamp": 1782321020.304764, + "status": 0, + "severity": 0 +} +``` + +Field descriptions: +- `pv`: PV name +- `value`: PV value (scalar or array, JSON-serializable) +- `timestamp`: EPICS timestamp (seconds since Unix epoch) +- `status`: EPICS alarm status +- `severity`: EPICS alarm severity + +--- + +## Frontend WebSocket Connection Management (`PVWebSocket`) + +Raw WebSocket usage in the browser requires handling several concerns: +- Building the URL correctly +- Handling `open` / `message` / `error` / `close` events +- Parsing JSON safely +- Reconnecting when the connection drops +- Cleaning up timers and connections when the component unmounts + +To encapsulate this logic, the frontend uses the `PVWebSocket` class. + +### Responsibilities + +`PVWebSocket` does four main things: +1. **Connect** to the gateway endpoint: `/ws?pv=...` +2. **Receive and parse messages**, then forward values to the UI via callbacks +3. **Handle errors** consistently +4. **Reconnect automatically** if the connection closes unexpectedly + +### Constructor (callback-based design) + +```js +new PVWebSocket(pvName, onData, onError, onConnect) +``` + +Parameters: +- `pvName`: PV to subscribe to +- `onData(value, timestamp)`: called on every PV update +- `onError(message)`: called on error conditions +- `onConnect()`: called when the WebSocket connection is opened + +This callback-based design keeps connection mechanics inside the manager, while the UI decides how to render or store incoming data. + +Internal state initialized by the constructor: +- `ws`: the underlying browser `WebSocket` object +- `reconnectAttempts` / `reconnectTimer`: reconnection control +- `messageCount`: debug counter to confirm updates are arriving + +--- + +## Connection Lifecycle + +### `connect()`: URL construction and event handlers + +`connect()` builds a URL like: + +``` +ws://HOST/ws?pv= +``` + +Then it creates the WebSocket and attaches handlers: + +- `onopen`: reset counters and call `onConnect()` +- `onmessage`: + - parse JSON + - if the payload contains `{ "error": ... }`, call `onError(...)` + - otherwise increment `messageCount` and call `onData(data.value, data.timestamp)` +- `onerror`: report connection errors (optionally reduced logging noise on first attempt) +- `onclose`: trigger reconnection logic + +### `attemptReconnect()`: reconnection strategy + +When the socket closes: +- stop after `MAX_RECONNECT_ATTEMPTS` +- otherwise wait `RECONNECT_DELAY` milliseconds and call `connect()` again + +This improves resilience during transient network issues or gateway restarts. + +### `disconnect()`: cleanup (React lifecycle) + +When the UI no longer needs the PV (e.g., component unmount), call `disconnect()` to: +- cancel any scheduled reconnection timer +- close the WebSocket +- release resources + +--- From 5a9534699171def0f7bd1922465af839f06cec87 Mon Sep 17 00:00:00 2001 From: ljiang-slac Date: Tue, 21 Jul 2026 06:18:10 -0700 Subject: [PATCH 5/5] updated the websocket manager --- src/components/MultiPVPlot.jsx | 301 ++++++------- .../MultiPVPlot_singlePVConnection.jsx | 418 ++++++++++++++++++ src/services/PVConnectionPool.js | 9 + src/services/WebSocketManager.js | 95 ++-- src/services/WebSocketManager_oldwq.js | 119 +++++ 5 files changed, 734 insertions(+), 208 deletions(-) create mode 100644 src/components/MultiPVPlot_singlePVConnection.jsx create mode 100644 src/services/WebSocketManager_oldwq.js diff --git a/src/components/MultiPVPlot.jsx b/src/components/MultiPVPlot.jsx index e07361a..cd3eef9 100644 --- a/src/components/MultiPVPlot.jsx +++ b/src/components/MultiPVPlot.jsx @@ -1,15 +1,16 @@ // src/components/MultiPVPlot.jsx -import { useState, useEffect, useRef } from 'react'; -import Plot from 'react-plotly.js'; -import { X, Wifi, WifiOff, AlertCircle, Download } from 'lucide-react'; -import { PVWebSocket } from '../services/WebSocketManager'; -import { DataBuffer } from '../services/DataBuffer'; -import { PLOT_CONFIG, PLOT_LAYOUT_TEMPLATE, getPVColor } from '../utils/constants'; -import { usePlotStore } from '../stores/usePlotStore'; -import './MultiPVPlot.css'; +import { useState, useEffect, useRef } from "react"; +import Plot from "react-plotly.js"; +import { X, Wifi, WifiOff, AlertCircle, Download } from "lucide-react"; + +import { DataBuffer } from "../services/DataBuffer"; +import { pvConnectionPool } from "../services/PVConnectionPool"; + +import { PLOT_CONFIG, PLOT_LAYOUT_TEMPLATE, getPVColor } from "../utils/constants"; +import { usePlotStore } from "../stores/usePlotStore"; +import "./MultiPVPlot.css"; export default function MultiPVPlot({ plotId, pvNames }) { - // Toggle to show/hide the per-PV tags (status icon, point count, remove button). const SHOW_PV_TAGS = false; const [plotData, setPlotData] = useState([]); @@ -17,8 +18,9 @@ export default function MultiPVPlot({ plotId, pvNames }) { const [yAxisRange, setYAxisRange] = useState(null); const [xAxisRange, setXAxisRange] = useState(null); const [revision, setRevision] = useState(0); - const buffersRef = useRef({}); - const websocketsRef = useRef({}); + + const buffersRef = useRef({}); // pvName -> DataBuffer + const unsubscribersRef = useRef({}); // pvName -> unsubscribe() const updateTimerRef = useRef(null); const { @@ -26,39 +28,33 @@ export default function MultiPVPlot({ plotId, pvNames }) { removePVFromPlot, timeSyncEnabled, globalTimeWindow, - updateLatestValue + updateLatestValue, } = usePlotStore(); - // Returns a stable, cross-plot-consistent color for each PV name - const getTraceColor = (pvName) => { - return getPVColor(pvName); - }; + const getTraceColor = (pvName) => getPVColor(pvName); - // ============================================================ - // Export data function for a single PV - // ============================================================ const exportData = (pvName) => { const buffer = buffersRef.current[pvName]; if (!buffer || buffer.getPointCount() === 0) { - alert('No data to export'); + alert("No data to export"); return; } - + const data = buffer.getData(); - const csv = ['Timestamp,Value'] - .concat(data.x.map((time, i) => - `${time.toISOString()},${data.y[i]}` - )) - .join('\n'); + const csv = ["Timestamp,Value"] + .concat( + data.x.map((time, i) => `${time.toISOString()},${data.y[i]}`) + ) + .join("\n"); - const blob = new Blob([csv], { type: 'text/csv;charset=utf-8;' }); + const blob = new Blob([csv], { type: "text/csv;charset=utf-8;" }); const url = URL.createObjectURL(blob); - const link = document.createElement('a'); + const link = document.createElement("a"); link.href = url; - const timestamp = new Date().toISOString().replace(/[:.]/g, '-'); + const timestamp = new Date().toISOString().replace(/[:.]/g, "-"); link.download = `${pvName}_${timestamp}.csv`; document.body.appendChild(link); @@ -66,56 +62,54 @@ export default function MultiPVPlot({ plotId, pvNames }) { document.body.removeChild(link); URL.revokeObjectURL(url); - console.log(`✅ Exported ${data.x.length} data points for ${pvName}`); + console.log(`Exported ${data.x.length} data points for ${pvName}`); }; - // ============================================================ - // Export all data for multi-PV plots - // ============================================================ const exportAllData = () => { if (pvNames.length === 1) { exportData(pvNames[0]); return; } - const allData = pvNames.map(pvName => { + const allData = pvNames.map((pvName) => { const buffer = buffersRef.current[pvName]; return buffer ? buffer.getData() : { x: [], y: [] }; }); - if (allData.every(d => d.x.length === 0)) { - alert('No data to export'); + if (allData.every((d) => d.x.length === 0)) { + alert("No data to export"); return; } - const header = ['Timestamp'].concat(pvNames.map(pv => `${pv}_Value`)); + const header = ["Timestamp"].concat(pvNames.map((pv) => `${pv}_Value`)); + const allTimestamps = new Set(); - allData.forEach(data => { - data.x.forEach(time => allTimestamps.add(time.getTime())); + allData.forEach((data) => { + data.x.forEach((time) => allTimestamps.add(time.getTime())); }); const sortedTimestamps = Array.from(allTimestamps).sort(); - const rows = sortedTimestamps.map(timestamp => { - const row = [new Date(timestamp).toISOString()]; + const rows = sortedTimestamps.map((ts) => { + const row = [new Date(ts).toISOString()]; pvNames.forEach((pvName, idx) => { const data = allData[idx]; - const timeIndex = data.x.findIndex(t => t.getTime() === timestamp); - row.push(timeIndex >= 0 ? data.y[timeIndex] : ''); + const timeIndex = data.x.findIndex((t) => t.getTime() === ts); + row.push(timeIndex >= 0 ? data.y[timeIndex] : ""); }); - return row.join(','); + return row.join(","); }); - const csv = [header.join(',')].concat(rows).join('\n'); + const csv = [header.join(",")].concat(rows).join("\n"); - const blob = new Blob([csv], { type: 'text/csv;charset=utf-8;' }); + const blob = new Blob([csv], { type: "text/csv;charset=utf-8;" }); const url = URL.createObjectURL(blob); - const link = document.createElement('a'); + const link = document.createElement("a"); link.href = url; - const timestamp = new Date().toISOString().replace(/[:.]/g, '-'); + const timestamp = new Date().toISOString().replace(/[:.]/g, "-"); link.download = `multi-pv_${timestamp}.csv`; document.body.appendChild(link); @@ -123,137 +117,114 @@ export default function MultiPVPlot({ plotId, pvNames }) { document.body.removeChild(link); URL.revokeObjectURL(url); - console.log(`✅ Exported data for ${pvNames.length} PVs`); + console.log(`Exported data for ${pvNames.length} PVs`); }; - // ============================================================ - // Effect 1: Manage WebSocket connections + buffers - // Depends ONLY on pvNames. - // ============================================================ + // Effect 1: subscribe/unsubscribe PVs for this plot useEffect(() => { - console.log(`🔧 Initializing connections for PVs:`, pvNames); + console.log("Sync subscriptions:", pvNames); + // Subscribe new PVs pvNames.forEach((pvName) => { if (!buffersRef.current[pvName]) { buffersRef.current[pvName] = new DataBuffer(PLOT_CONFIG.MAX_POINTS); + setConnectionStatus((prev) => ({ ...prev, [pvName]: "connecting" })); - const ws = new PVWebSocket( - pvName, - (value, timestamp) => { - buffersRef.current[pvName].addPoint(value, timestamp); + const unsubscribe = pvConnectionPool.subscribe(pvName, { + onData: (value, timestamp) => { + const buf = buffersRef.current[pvName]; + if (!buf) return; + buf.addPoint(value, timestamp); updateLatestValue(pvName, value, timestamp); }, - (error) => { + onError: (error) => { console.error(`Error for ${pvName}:`, error); - setConnectionStatus((prev) => ({ ...prev, [pvName]: 'error' })); + setConnectionStatus((prev) => ({ ...prev, [pvName]: "error" })); }, - () => { - setConnectionStatus((prev) => ({ ...prev, [pvName]: 'connected' })); - } - ); + onConnect: () => { + setConnectionStatus((prev) => ({ ...prev, [pvName]: "connected" })); + }, + }); - ws.connect(); - websocketsRef.current[pvName] = ws; - setConnectionStatus((prev) => ({ ...prev, [pvName]: 'connecting' })); + unsubscribersRef.current[pvName] = unsubscribe; } }); + // Unsubscribe PVs removed from this plot Object.keys(buffersRef.current).forEach((pvName) => { if (!pvNames.includes(pvName)) { - websocketsRef.current[pvName]?.disconnect(); - delete websocketsRef.current[pvName]; + unsubscribersRef.current[pvName]?.(); + delete unsubscribersRef.current[pvName]; + delete buffersRef.current[pvName]; + setConnectionStatus((prev) => { - const newStatus = { ...prev }; - delete newStatus[pvName]; - return newStatus; + const next = { ...prev }; + delete next[pvName]; + return next; }); } }); - // No cleanup here: pvNames changes are handled above (connect new PVs / disconnect removed PVs). - // Cleanup happens on unmount. // eslint-disable-next-line react-hooks/exhaustive-deps }, [pvNames]); - - // Disconnect everything on unmount only + + /////////////////////////////////////////////////////////////////////// + // Unsubscribe everything on unmount useEffect(() => { return () => { - console.log('🛑 Disconnecting all websockets (unmount)'); - Object.values(websocketsRef.current).forEach((ws) => ws.disconnect()); - websocketsRef.current = {}; + console.log("Unsubscribing all PVs (unmount)"); + Object.keys(unsubscribersRef.current).forEach((pvName) => { + unsubscribersRef.current[pvName]?.(); + }); + unsubscribersRef.current = {}; buffersRef.current = {}; }; // eslint-disable-next-line react-hooks/exhaustive-deps }, []); - // ============================================================ - // Effect 2: Periodic plot redraw timer - // Depends on pvNames + time settings. - // ============================================================ + // Effect 2: periodic plot redraw useEffect(() => { let updateCount = 0; updateTimerRef.current = setInterval(() => { - updateCount++; + updateCount += 1; let xMin = null; let xMax = null; - - // use the last point data time stamp to replace current time - + if (timeSyncEnabled) { - // Anchor the window's right edge to the newest DATA timestamp, - // not the front-end wall clock (they can differ if clocks aren't synced). let latest = null; pvNames.forEach((pvName) => { const buffer = buffersRef.current[pvName]; const ts = buffer ? buffer.getLatestTimestamp() : null; - if (ts && (!latest || ts > latest)) { - latest = ts; - } + if (ts && (!latest || ts > latest)) latest = ts; }); - - // Use latest data time as the right edge; fall back to now if no data yet + xMax = latest || new Date(); xMin = new Date(xMax.getTime() - globalTimeWindow * 1000); } - - - - - //if (timeSyncEnabled) { - // const now = new Date(); - // xMax = now; - // xMin = new Date(now.getTime() - globalTimeWindow * 1000); - //} const traces = pvNames.map((pvName) => { const buffer = buffersRef.current[pvName]; let data = buffer ? buffer.getData() : { x: [], y: [] }; if (timeSyncEnabled && xMin && xMax && data.x.length > 0) { - const filteredIndices = []; - data.x.forEach((time, idx) => { - if (time >= xMin && time <= xMax) { - filteredIndices.push(idx); - } + const idxs = []; + data.x.forEach((t, i) => { + if (t >= xMin && t <= xMax) idxs.push(i); }); - - data = { - x: filteredIndices.map(i => data.x[i]), - y: filteredIndices.map(i => data.y[i]) - }; + data = { x: idxs.map((i) => data.x[i]), y: idxs.map((i) => data.y[i]) }; } return { x: data.x, y: data.y, - type: 'scatter', - mode: 'lines+markers', + type: "scatter", + mode: "lines+markers", name: pvName, line: { width: 2, color: getTraceColor(pvName) }, - marker: { size: 4 } + marker: { size: 4 }, }; }); @@ -267,7 +238,7 @@ export default function MultiPVPlot({ plotId, pvNames }) { const max = Math.max(...allValues); const range = max - min; const padding = range * 0.2 || 0.0001; - setYAxisRange([min - padding * 0.8, max + padding * 3.]); + setYAxisRange([min - padding * 0.8, max + padding * 3.0]); } if (timeSyncEnabled && xMin && xMax) { @@ -281,97 +252,83 @@ export default function MultiPVPlot({ plotId, pvNames }) { const buffer = buffersRef.current[pvName]; return sum + (buffer ? buffer.getPointCount() : 0); }, 0); - console.log(`🔄 Plot update #${updateCount}: ${totalPoints} total points across ${pvNames.length} PV(s)`); - - if (timeSyncEnabled) { - console.log(`🕐 Time window: ${xMin?.toLocaleTimeString()} - ${xMax?.toLocaleTimeString()}`); - } + console.log( + `Plot update #${updateCount}: ${totalPoints} total points across ${pvNames.length} PV(s)` + ); } setPlotData(traces); - setRevision(prev => prev + 1); + setRevision((prev) => prev + 1); }, PLOT_CONFIG.UPDATE_INTERVAL); - console.log(`✅ Plot update timer started (interval: ${PLOT_CONFIG.UPDATE_INTERVAL}ms)`); + console.log(`Plot update timer started (${PLOT_CONFIG.UPDATE_INTERVAL}ms)`); return () => { if (updateTimerRef.current) { clearInterval(updateTimerRef.current); - console.log(`✅ Plot update timer stopped`); + console.log("Plot update timer stopped"); } }; // eslint-disable-next-line react-hooks/exhaustive-deps }, [pvNames, timeSyncEnabled, globalTimeWindow]); - // ============================================================ - // Connection status icon helper (used by pv-tags when enabled) - // ============================================================ const getStatusIcon = (status) => { switch (status) { - case 'connected': + case "connected": return ; - case 'error': + case "error": return ; default: return ; } }; - // ============================================================ - // Plot layout - // ============================================================ const plotLayout = { ...PLOT_LAYOUT_TEMPLATE, - // title removed intentionally datarevision: revision, showlegend: true, - yaxis: { ...PLOT_LAYOUT_TEMPLATE.yaxis, autorange: yAxisRange ? false : true, range: yAxisRange, - exponentformat: 'e', - tickformat: '.2e' + exponentformat: "e", + tickformat: ".2e", }, xaxis: { ...PLOT_LAYOUT_TEMPLATE.xaxis, - type: 'date', - tickformat: '%H:%M:%S', + type: "date", + tickformat: "%H:%M:%S", autorange: xAxisRange ? false : true, - range: xAxisRange + range: xAxisRange, }, - margin: { l: 85, r: 30, t: 10, b: 50 } + margin: { l: 85, r: 30, t: 10, b: 50 }, }; return (
- - {/* - Per-PV tags. Controlled by SHOW_PV_TAGS (top of component). - Hidden by default; logic preserved and re-enabled by SHOW_PV_TAGS = true. - */}
- {SHOW_PV_TAGS && pvNames.map((pvName) => ( -
- {getStatusIcon(connectionStatus[pvName])} - {pvName} - {buffersRef.current[pvName] && ( - - ({buffersRef.current[pvName].getPointCount()}) - - )} - {pvNames.length > 1 && ( - - )} -
- ))} + {SHOW_PV_TAGS && + pvNames.map((pvName) => ( +
+ {getStatusIcon(connectionStatus[pvName])} + {pvName} + {buffersRef.current[pvName] && ( + + ({buffersRef.current[pvName].getPointCount()}) + + )} + {pvNames.length > 1 && ( + + )} +
+ ))}
@@ -401,15 +358,15 @@ export default function MultiPVPlot({ plotId, pvNames }) { responsive: true, displayModeBar: false, displaylogo: false, - modeBarButtonsToRemove: ['lasso2d', 'select2d'], + modeBarButtonsToRemove: ["lasso2d", "select2d"], toImageButtonOptions: { - format: 'png', - filename: pvNames.join('_'), + format: "png", + filename: pvNames.join("_"), height: 600, - width: 1000 - } + width: 1000, + }, }} - style={{ width: '100%', height: '100%' }} + style={{ width: "100%", height: "100%" }} useResizeHandler={true} />
diff --git a/src/components/MultiPVPlot_singlePVConnection.jsx b/src/components/MultiPVPlot_singlePVConnection.jsx new file mode 100644 index 0000000..e07361a --- /dev/null +++ b/src/components/MultiPVPlot_singlePVConnection.jsx @@ -0,0 +1,418 @@ +// src/components/MultiPVPlot.jsx +import { useState, useEffect, useRef } from 'react'; +import Plot from 'react-plotly.js'; +import { X, Wifi, WifiOff, AlertCircle, Download } from 'lucide-react'; +import { PVWebSocket } from '../services/WebSocketManager'; +import { DataBuffer } from '../services/DataBuffer'; +import { PLOT_CONFIG, PLOT_LAYOUT_TEMPLATE, getPVColor } from '../utils/constants'; +import { usePlotStore } from '../stores/usePlotStore'; +import './MultiPVPlot.css'; + +export default function MultiPVPlot({ plotId, pvNames }) { + // Toggle to show/hide the per-PV tags (status icon, point count, remove button). + const SHOW_PV_TAGS = false; + + const [plotData, setPlotData] = useState([]); + const [connectionStatus, setConnectionStatus] = useState({}); + const [yAxisRange, setYAxisRange] = useState(null); + const [xAxisRange, setXAxisRange] = useState(null); + const [revision, setRevision] = useState(0); + const buffersRef = useRef({}); + const websocketsRef = useRef({}); + const updateTimerRef = useRef(null); + + const { + removePlot, + removePVFromPlot, + timeSyncEnabled, + globalTimeWindow, + updateLatestValue + } = usePlotStore(); + + // Returns a stable, cross-plot-consistent color for each PV name + const getTraceColor = (pvName) => { + return getPVColor(pvName); + }; + + // ============================================================ + // Export data function for a single PV + // ============================================================ + const exportData = (pvName) => { + const buffer = buffersRef.current[pvName]; + + if (!buffer || buffer.getPointCount() === 0) { + alert('No data to export'); + return; + } + + const data = buffer.getData(); + + const csv = ['Timestamp,Value'] + .concat(data.x.map((time, i) => + `${time.toISOString()},${data.y[i]}` + )) + .join('\n'); + + const blob = new Blob([csv], { type: 'text/csv;charset=utf-8;' }); + const url = URL.createObjectURL(blob); + const link = document.createElement('a'); + link.href = url; + + const timestamp = new Date().toISOString().replace(/[:.]/g, '-'); + link.download = `${pvName}_${timestamp}.csv`; + + document.body.appendChild(link); + link.click(); + document.body.removeChild(link); + URL.revokeObjectURL(url); + + console.log(`✅ Exported ${data.x.length} data points for ${pvName}`); + }; + + // ============================================================ + // Export all data for multi-PV plots + // ============================================================ + const exportAllData = () => { + if (pvNames.length === 1) { + exportData(pvNames[0]); + return; + } + + const allData = pvNames.map(pvName => { + const buffer = buffersRef.current[pvName]; + return buffer ? buffer.getData() : { x: [], y: [] }; + }); + + if (allData.every(d => d.x.length === 0)) { + alert('No data to export'); + return; + } + + const header = ['Timestamp'].concat(pvNames.map(pv => `${pv}_Value`)); + const allTimestamps = new Set(); + allData.forEach(data => { + data.x.forEach(time => allTimestamps.add(time.getTime())); + }); + + const sortedTimestamps = Array.from(allTimestamps).sort(); + + const rows = sortedTimestamps.map(timestamp => { + const row = [new Date(timestamp).toISOString()]; + + pvNames.forEach((pvName, idx) => { + const data = allData[idx]; + const timeIndex = data.x.findIndex(t => t.getTime() === timestamp); + row.push(timeIndex >= 0 ? data.y[timeIndex] : ''); + }); + + return row.join(','); + }); + + const csv = [header.join(',')].concat(rows).join('\n'); + + const blob = new Blob([csv], { type: 'text/csv;charset=utf-8;' }); + const url = URL.createObjectURL(blob); + const link = document.createElement('a'); + link.href = url; + + const timestamp = new Date().toISOString().replace(/[:.]/g, '-'); + link.download = `multi-pv_${timestamp}.csv`; + + document.body.appendChild(link); + link.click(); + document.body.removeChild(link); + URL.revokeObjectURL(url); + + console.log(`✅ Exported data for ${pvNames.length} PVs`); + }; + + // ============================================================ + // Effect 1: Manage WebSocket connections + buffers + // Depends ONLY on pvNames. + // ============================================================ + useEffect(() => { + console.log(`🔧 Initializing connections for PVs:`, pvNames); + + pvNames.forEach((pvName) => { + if (!buffersRef.current[pvName]) { + buffersRef.current[pvName] = new DataBuffer(PLOT_CONFIG.MAX_POINTS); + + const ws = new PVWebSocket( + pvName, + (value, timestamp) => { + buffersRef.current[pvName].addPoint(value, timestamp); + updateLatestValue(pvName, value, timestamp); + }, + (error) => { + console.error(`Error for ${pvName}:`, error); + setConnectionStatus((prev) => ({ ...prev, [pvName]: 'error' })); + }, + () => { + setConnectionStatus((prev) => ({ ...prev, [pvName]: 'connected' })); + } + ); + + ws.connect(); + websocketsRef.current[pvName] = ws; + setConnectionStatus((prev) => ({ ...prev, [pvName]: 'connecting' })); + } + }); + + Object.keys(buffersRef.current).forEach((pvName) => { + if (!pvNames.includes(pvName)) { + websocketsRef.current[pvName]?.disconnect(); + delete websocketsRef.current[pvName]; + delete buffersRef.current[pvName]; + setConnectionStatus((prev) => { + const newStatus = { ...prev }; + delete newStatus[pvName]; + return newStatus; + }); + } + }); + + // No cleanup here: pvNames changes are handled above (connect new PVs / disconnect removed PVs). + // Cleanup happens on unmount. + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [pvNames]); + + // Disconnect everything on unmount only + useEffect(() => { + return () => { + console.log('🛑 Disconnecting all websockets (unmount)'); + Object.values(websocketsRef.current).forEach((ws) => ws.disconnect()); + websocketsRef.current = {}; + buffersRef.current = {}; + }; + // eslint-disable-next-line react-hooks/exhaustive-deps + }, []); + + // ============================================================ + // Effect 2: Periodic plot redraw timer + // Depends on pvNames + time settings. + // ============================================================ + useEffect(() => { + let updateCount = 0; + + updateTimerRef.current = setInterval(() => { + updateCount++; + + let xMin = null; + let xMax = null; + + // use the last point data time stamp to replace current time + + if (timeSyncEnabled) { + // Anchor the window's right edge to the newest DATA timestamp, + // not the front-end wall clock (they can differ if clocks aren't synced). + let latest = null; + pvNames.forEach((pvName) => { + const buffer = buffersRef.current[pvName]; + const ts = buffer ? buffer.getLatestTimestamp() : null; + if (ts && (!latest || ts > latest)) { + latest = ts; + } + }); + + // Use latest data time as the right edge; fall back to now if no data yet + xMax = latest || new Date(); + xMin = new Date(xMax.getTime() - globalTimeWindow * 1000); + } + + + + + //if (timeSyncEnabled) { + // const now = new Date(); + // xMax = now; + // xMin = new Date(now.getTime() - globalTimeWindow * 1000); + //} + + const traces = pvNames.map((pvName) => { + const buffer = buffersRef.current[pvName]; + let data = buffer ? buffer.getData() : { x: [], y: [] }; + + if (timeSyncEnabled && xMin && xMax && data.x.length > 0) { + const filteredIndices = []; + data.x.forEach((time, idx) => { + if (time >= xMin && time <= xMax) { + filteredIndices.push(idx); + } + }); + + data = { + x: filteredIndices.map(i => data.x[i]), + y: filteredIndices.map(i => data.y[i]) + }; + } + + return { + x: data.x, + y: data.y, + type: 'scatter', + mode: 'lines+markers', + name: pvName, + line: { width: 2, color: getTraceColor(pvName) }, + marker: { size: 4 } + }; + }); + + let allValues = []; + traces.forEach((trace) => { + allValues = allValues.concat(trace.y); + }); + + if (allValues.length > 0) { + const min = Math.min(...allValues); + const max = Math.max(...allValues); + const range = max - min; + const padding = range * 0.2 || 0.0001; + setYAxisRange([min - padding * 0.8, max + padding * 3.]); + } + + if (timeSyncEnabled && xMin && xMax) { + setXAxisRange([xMin, xMax]); + } else { + setXAxisRange(null); + } + + if (updateCount % 10 === 0) { + const totalPoints = pvNames.reduce((sum, pvName) => { + const buffer = buffersRef.current[pvName]; + return sum + (buffer ? buffer.getPointCount() : 0); + }, 0); + console.log(`🔄 Plot update #${updateCount}: ${totalPoints} total points across ${pvNames.length} PV(s)`); + + if (timeSyncEnabled) { + console.log(`🕐 Time window: ${xMin?.toLocaleTimeString()} - ${xMax?.toLocaleTimeString()}`); + } + } + + setPlotData(traces); + setRevision(prev => prev + 1); + }, PLOT_CONFIG.UPDATE_INTERVAL); + + console.log(`✅ Plot update timer started (interval: ${PLOT_CONFIG.UPDATE_INTERVAL}ms)`); + + return () => { + if (updateTimerRef.current) { + clearInterval(updateTimerRef.current); + console.log(`✅ Plot update timer stopped`); + } + }; + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [pvNames, timeSyncEnabled, globalTimeWindow]); + + // ============================================================ + // Connection status icon helper (used by pv-tags when enabled) + // ============================================================ + const getStatusIcon = (status) => { + switch (status) { + case 'connected': + return ; + case 'error': + return ; + default: + return ; + } + }; + + // ============================================================ + // Plot layout + // ============================================================ + const plotLayout = { + ...PLOT_LAYOUT_TEMPLATE, + // title removed intentionally + datarevision: revision, + showlegend: true, + + yaxis: { + ...PLOT_LAYOUT_TEMPLATE.yaxis, + autorange: yAxisRange ? false : true, + range: yAxisRange, + exponentformat: 'e', + tickformat: '.2e' + }, + xaxis: { + ...PLOT_LAYOUT_TEMPLATE.xaxis, + type: 'date', + tickformat: '%H:%M:%S', + autorange: xAxisRange ? false : true, + range: xAxisRange + }, + margin: { l: 85, r: 30, t: 10, b: 50 } + }; + + return ( +
+
+ + {/* + Per-PV tags. Controlled by SHOW_PV_TAGS (top of component). + Hidden by default; logic preserved and re-enabled by SHOW_PV_TAGS = true. + */} +
+ {SHOW_PV_TAGS && pvNames.map((pvName) => ( +
+ {getStatusIcon(connectionStatus[pvName])} + {pvName} + {buffersRef.current[pvName] && ( + + ({buffersRef.current[pvName].getPointCount()}) + + )} + {pvNames.length > 1 && ( + + )} +
+ ))} +
+ +
+ + + +
+
+ +
+ +
+
+ ); +} diff --git a/src/services/PVConnectionPool.js b/src/services/PVConnectionPool.js index 659d7ca..8b11121 100644 --- a/src/services/PVConnectionPool.js +++ b/src/services/PVConnectionPool.js @@ -6,10 +6,15 @@ class PVConnectionPool { this.pool = new Map(); } + //called in MultiPVPlot.jsx, e.g. pool.subscribe("ADC1", { onData, onError, onConnect }) + //returns on unsubscribe() function subscribe(pvName, callbacks) { let entry = this.pool.get(pvName); if (!entry) { + + + console.log(`[POOL] creating new connection for ${pvName}`); const subscribers = new Set(); // Create ONE shared PVWebSocket for this PV @@ -30,6 +35,9 @@ class PVConnectionPool { ws.connect(); entry = { ws, subscribers }; this.pool.set(pvName, entry); + } else{ + console.log(`[POOL] reusing connection for ${pvName}`); + } // Register this subscriber (one per plot per PV) @@ -40,6 +48,7 @@ class PVConnectionPool { }; entry.subscribers.add(subscriber); + console.log(`[POOL] ${pvName} subscribers=${entry.subscribers.size}, activeConns=${this.pool.size}`); // Return an unsubscribe function return () => { diff --git a/src/services/WebSocketManager.js b/src/services/WebSocketManager.js index 4aee699..a1d6f7e 100644 --- a/src/services/WebSocketManager.js +++ b/src/services/WebSocketManager.js @@ -1,46 +1,55 @@ // src/services/WebSocketManager.js -// To Manage a Single PV connection -// -import { WS_CONFIG } from '../utils/constants'; +// Manages a single PV WebSocket connection (used by the global PV connection pool). + +import { WS_CONFIG } from "../utils/constants"; export class PVWebSocket { constructor(pvName, onData, onError, onConnect) { - this.pvName = pvName; + this.pvName = pvName; - //callback functions, separation of passing and handling - this.onData = onData; //call when data is received - this.onError = onError; //call when error happens - this.onConnect = onConnect; //call when connected + // Callback functions (separation of transport and handling) + this.onData = onData; // called when data is received + this.onError = onError; // called when an error happens + this.onConnect = onConnect; // called when the connection is opened - this.ws = null; + this.ws = null; this.reconnectAttempts = 0; this.reconnectTimer = null; - this.messageCount = 0; // Track number of messages received + this.messageCount = 0; // number of messages received in the current session + + // When true, an unexpected close will trigger auto-reconnect. + // Set to false on intentional disconnect() to prevent reconnecting. + this.shouldReconnect = true; } - //called when the WebSocket connection is opened + + // Open the WebSocket connection and attach event handlers. connect() { try { - //WS_CONFIG.BASE_URL saved in utils/constant.js IP address + Port Number - //encodeURIComponent: convert special characters into UTF-8 + percent-encoding + // Any manual/previous close should not block a new connect attempt. + this.shouldReconnect = true; + + // WS_CONFIG.BASE_URL (IP + port) is defined in utils/constants.js + // encodeURIComponent: safely encode PV name (colons, etc.) into the URL. const wsUrl = `${WS_CONFIG.BASE_URL}/ws?pv=${encodeURIComponent(this.pvName)}`; - + console.log(`🔗 Connecting to: ${wsUrl}`); this.ws = new WebSocket(wsUrl); - //Fire when the WebSocket handshake completes and the connection is open + // Fired when the WebSocket handshake completes and the connection is OPEN. this.ws.onopen = () => { console.log(`✅ Connected: ${this.pvName}`); this.reconnectAttempts = 0; this.messageCount = 0; - //call onConnect callbackfunction/Notify the caller if it is passed if (this.onConnect) this.onConnect(); }; - //WebSocket Message Event Handler + // Fired for every message pushed by the server. + // event is a MessageEvent; event.data is the raw payload (JSON string here). this.ws.onmessage = (event) => { try { const data = JSON.parse(event.data); - + + // Server-side error message: { error, pv, timestamp } if (data.error) { console.error(`❌ PV Error (${this.pvName}):`, data.error); if (this.onError) this.onError(data.error); @@ -48,47 +57,53 @@ export class PVWebSocket { } this.messageCount++; - - // Log every 10th message to show updates are arriving + + // Periodic log to confirm updates are arriving. if (this.messageCount % 10 === 0) { console.log(`📡 ${this.pvName}: Received ${this.messageCount} messages`); } - // Call data callback + // Forward the value to the caller (UI / buffer). if (this.onData) { this.onData(data.value, data.timestamp); } } catch (err) { - console.error('Parse error:', err); - if (this.onError) this.onError('Data parse error'); + console.error("Parse error:", err); + if (this.onError) this.onError("Data parse error"); } }; - //WebSocket Error Handler + + // Fired on transport-level errors (network/handshake issues). this.ws.onerror = (error) => { - // Only log error if not initial connection attempt + // Reduce noise: only report errors after a reconnect attempt. if (this.reconnectAttempts > 0) { console.error(`❌ WebSocket error (${this.pvName}):`, error); - } - if (this.onError && this.reconnectAttempts > 0) { - this.onError('Connection error'); + if (this.onError) this.onError("Connection error"); } }; - // when connection is closed, reconnect + // Fired when the connection closes (either intentional or unexpected). this.ws.onclose = () => { - console.log(`🔌 Connection closed: ${this.pvName} (received ${this.messageCount} messages)`); - this.attemptReconnect(); + console.log( + `🔌 Connection closed: ${this.pvName} (received ${this.messageCount} messages)` + ); + // Only auto-reconnect if the close was NOT intentional. + if (this.shouldReconnect) { + this.attemptReconnect(); + } }; - } catch (err) { - console.error('WebSocket creation failed:', err); + console.error("WebSocket creation failed:", err); if (this.onError) this.onError(err.message); } } + // Try to reconnect after an unexpected close, up to MAX_RECONNECT_ATTEMPTS. attemptReconnect() { if (this.reconnectAttempts >= WS_CONFIG.MAX_RECONNECT_ATTEMPTS) { - console.error(`❌ ${this.pvName}: Reconnection failed (${WS_CONFIG.MAX_RECONNECT_ATTEMPTS} attempts)`); + console.error( + `❌ ${this.pvName}: Reconnection failed (${WS_CONFIG.MAX_RECONNECT_ATTEMPTS} attempts)` + ); if (this.onError) { this.onError(`Reconnection failed (${WS_CONFIG.MAX_RECONNECT_ATTEMPTS} attempts)`); } @@ -96,15 +111,23 @@ export class PVWebSocket { } this.reconnectAttempts++; - console.log(`🔄 ${this.pvName}: Reconnecting in ${WS_CONFIG.RECONNECT_DELAY/1000}s (attempt ${this.reconnectAttempts}/${WS_CONFIG.MAX_RECONNECT_ATTEMPTS})...`); + console.log( + `🔄 ${this.pvName}: Reconnecting in ${WS_CONFIG.RECONNECT_DELAY / 1000}s ` + + `(attempt ${this.reconnectAttempts}/${WS_CONFIG.MAX_RECONNECT_ATTEMPTS})...` + ); this.reconnectTimer = setTimeout(() => { this.connect(); }, WS_CONFIG.RECONNECT_DELAY); } + // Intentionally close the connection and stop any reconnect logic. disconnect() { - this.reconnectAttemps = 0; + // Prevent onclose from triggering auto-reconnect. + this.shouldReconnect = false; + + // Fixed typo: was this.reconnectAttemps + this.reconnectAttempts = 0; if (this.reconnectTimer) { clearTimeout(this.reconnectTimer); diff --git a/src/services/WebSocketManager_oldwq.js b/src/services/WebSocketManager_oldwq.js new file mode 100644 index 0000000..e334bd1 --- /dev/null +++ b/src/services/WebSocketManager_oldwq.js @@ -0,0 +1,119 @@ +// src/services/WebSocketManager.js +// To Manage a Single PV connection +// +import { WS_CONFIG } from '../utils/constants'; + +export class PVWebSocket { + constructor(pvName, onData, onError, onConnect) { + this.pvName = pvName; + + //callback functions, separation of passing and handling + this.onData = onData; //call when data is received + this.onError = onError; //call when error happens + this.onConnect = onConnect; //call when connected + + this.ws = null; + this.reconnectAttempts = 0; + this.reconnectTimer = null; + this.messageCount = 0; // Track number of messages received + } + //called when the WebSocket connection is opened + connect() { + try { + //WS_CONFIG.BASE_URL saved in utils/constant.js IP address + Port Number + //encodeURIComponent: convert special characters into UTF-8 + percent-encoding + const wsUrl = `${WS_CONFIG.BASE_URL}/ws?pv=${encodeURIComponent(this.pvName)}`; + + console.log(`🔗 Connecting to: ${wsUrl}`); + this.ws = new WebSocket(wsUrl); + + //Fire when the WebSocket handshake completes and the connection is open + this.ws.onopen = () => { + console.log(`✅ Connected: ${this.pvName}`); + this.reconnectAttempts = 0; + this.messageCount = 0; + //call onConnect callbackfunction/Notify the caller if it is passed + if (this.onConnect) this.onConnect(); + }; + + //WebSocket Message Event Handler + this.ws.onmessage = (event) => { + try { + const data = JSON.parse(event.data); + + if (data.error) { + console.error(`❌ PV Error (${this.pvName}):`, data.error); + if (this.onError) this.onError(data.error); + return; + } + + this.messageCount++; + + // Log every 10th message to show updates are arriving + if (this.messageCount % 10 === 0) { + console.log(`📡 ${this.pvName}: Received ${this.messageCount} messages`); + } + + // Call data callback + if (this.onData) { + this.onData(data.value, data.timestamp); + } + } catch (err) { + console.error('Parse error:', err); + if (this.onError) this.onError('Data parse error'); + } + }; + //WebSocket Error Handler + this.ws.onerror = (error) => { + // Only log error if not initial connection attempt + if (this.reconnectAttempts > 0) { + console.error(`❌ WebSocket error (${this.pvName}):`, error); + } + if (this.onError && this.reconnectAttempts > 0) { + this.onError('Connection error'); + } + }; + + // when connection is closed, reconnect + this.ws.onclose = () => { + console.log(`🔌 Connection closed: ${this.pvName} (received ${this.messageCount} messages)`); + this.attemptReconnect(); + }; + + } catch (err) { + console.error('WebSocket creation failed:', err); + if (this.onError) this.onError(err.message); + } + } + + attemptReconnect() { + if (this.reconnectAttempts >= WS_CONFIG.MAX_RECONNECT_ATTEMPTS) { + console.error(`❌ ${this.pvName}: Reconnection failed (${WS_CONFIG.MAX_RECONNECT_ATTEMPTS} attempts)`); + if (this.onError) { + this.onError(`Reconnection failed (${WS_CONFIG.MAX_RECONNECT_ATTEMPTS} attempts)`); + } + return; + } + + this.reconnectAttempts++; + console.log(`🔄 ${this.pvName}: Reconnecting in ${WS_CONFIG.RECONNECT_DELAY/1000}s (attempt ${this.reconnectAttempts}/${WS_CONFIG.MAX_RECONNECT_ATTEMPTS})...`); + + this.reconnectTimer = setTimeout(() => { + this.connect(); + }, WS_CONFIG.RECONNECT_DELAY); + } + + disconnect() { + this.reconnectAttempts = 0; + + if (this.reconnectTimer) { + clearTimeout(this.reconnectTimer); + this.reconnectTimer = null; + } + + if (this.ws) { + this.ws.close(); + this.ws = null; + } + } +}