""" EMScope WebSocket API Client ============================= Client library for the EMScope EMI measurement instrument. Communicates with the EMScope WebSocket server running on port 8010. Protocol overview ----------------- - All messages are JSON over WebSocket. - Configuration is set by sending ``{field: value}`` messages. - The server streams measurement data as ``{"values": [...], "overload": bool}`` after each sweep completes. - Read-only data (temperatures, standards, etc.) is requested with specific command messages and received via the normal message stream. Quick start ----------- :: import asyncio from api import EmscopeConnection async def main(): async with EmscopeConnection("ws://10.10.14.45:8010") as emscope: await emscope.set_rbw("9") await emscope.set_value("detector_type", "pk") await emscope.set_value("amp_units", "dbuv") for _ in range(10): msg = await emscope.parse_messages() if msg and "values" in msg: print("Frequencies:", [v[0] for v in emscope.values[:5]]) asyncio.run(main()) Requirements ------------ - Python 3.8+ - ``websockets`` library (``pip install websockets``) """ import asyncio import json from typing import Any, Dict, List, Optional, Union import websockets class EmscopeConnection: """ Async WebSocket client for the EMScope measurement server. Attributes are updated in place each time :meth:`parse_messages` is called. The most important ones are: - ``values`` – last received spectrum: list of ``[freq_index, amplitude]`` pairs - ``overload`` – True if the ADC was saturated during the last sweep - ``SN`` – device serial number (populated on connect) - ``temperatures`` – ``[pcb_temp_C, fpga_temp_C]`` after :meth:`request_temps` """ # Fields the client is allowed to set on the server _WRITABLE_FIELDS = { "measure_channel", "detector_type", "trace_type", "average", "rbw", "mode", "reference_level", "input_attenuator", "amp_units", "sweep_time", "external_loss", "session_UUID", "visible", "display_range", "osc_us_div", "osc_volts_div", "osc_sweep", "trigger", "osc_trigger_offset", "osc_trigger_level", "osc_vertical_offset", "osc_horizontal_offset", } # All fields the server may send (used by parse_messages) _ALL_FIELDS = _WRITABLE_FIELDS | { "values", "SN", "SFP_SN", "MAC", "measurement_uncertainty", "num_points", "input_attenuator_auto_value", "standards", "sessions", "external_losses", "licenses", "report", "frase", "overload", "temperatures", "repetition", "threephase", } def __init__(self, url: str, session_uuid: str = "emscope-api"): """ Create a new connection instance (does not connect yet). Args: url: WebSocket URL of the EMScope server, e.g. ``"ws://10.10.14.45:8010"``. session_uuid: Unique string that locks the device to this session. Passed automatically to :meth:`connect`. Defaults to ``"emscope-api"``. Change it if multiple independent clients may connect to the same instrument. """ self.url = url self.session_uuid = session_uuid self.websocket = None self._reset_values() # ── Lifecycle ───────────────────────────────────────────────────────── async def connect(self) -> None: """ Open the WebSocket connection and activate the client session. The server requires a ``session_UUID`` message before it will send any data or accept configuration. This method sends one automatically using the ``session_uuid`` value passed to :meth:`__init__`. After ``session_UUID`` is accepted, the server replies with device info (``SN``, ``MAC``, ``SFP_SN``). This method waits for that reply so that ``self.SN`` and ``self.MAC`` are populated before returning. Raises: websockets.exceptions.WebSocketException: on connection failure. websockets.exceptions.ConnectionClosedError: if the server closes the connection with code 4003, meaning another session already holds the device lock with a different UUID. """ self.websocket = await websockets.connect(self.url) # The server activates the client only after receiving session_UUID, # and sends SN+MAC as a direct response to it. await self.set_session_uuid(self.session_uuid) await self.parse_messages() # Receives {"SN": ..., "MAC": ..., "SFP_SN": ...} async def disconnect(self) -> None: """Close the WebSocket connection gracefully.""" if self.websocket is not None: await self.websocket.close() self.websocket = None async def __aenter__(self) -> "EmscopeConnection": await self.connect() return self async def __aexit__(self, exc_type, exc_val, exc_tb) -> bool: await self.disconnect() return False # ── Message parsing ─────────────────────────────────────────────────── async def parse_messages(self) -> Optional[Dict[str, Any]]: """ Receive and parse one WebSocket message from the server. Updates the corresponding instance attributes automatically. Responds to server ``ping`` messages with ``pong`` transparently. Returns: The parsed message as a dict, or ``None`` if the connection was closed. Example:: while True: msg = await emscope.parse_messages() if msg is None: break if "values" in msg: process(emscope.values) """ try: raw = await self.websocket.recv() except websockets.ConnectionClosed: return None data = json.loads(raw) for key, value in data.items(): if key == "ping": await self.websocket.send(json.dumps({"pong": True})) continue # The server sends "input_attenuator" to report the resolved # auto-attenuator value; store it separately so we don't overwrite # the user-set value (which may be "auto"). if key == "input_attenuator": self.input_attenuator_auto_value = value continue if key in self._ALL_FIELDS: setattr(self, key, value) else: print(f"WARNING: Received unknown key from server: {key!r}") return data # ── Generic configuration setter ────────────────────────────────────── async def set_value(self, field: str, value: Any) -> None: """ Set a writable configuration field on the server. This is the generic setter. Prefer the specific helpers (e.g. :meth:`set_rbw`, :meth:`set_detector_type`) when available, as they include parameter validation and documentation. Args: field: One of the writable field names (see ``_WRITABLE_FIELDS``). value: The new value. Must match the type expected by the server. Raises: ValueError: if ``field`` is not a valid writable field. Valid writable fields and their accepted values: ======================= ================================================ Field Valid values ======================= ================================================ ``measure_channel`` ``"lg"``, ``"ng"``, ``"cm"``, ``"dm"``, ``"l1"``, ``"l2"``, ``"l3"``, ``"n"`` (RX4) ``detector_type`` ``"pk"``, ``"qp"``, ``"av"`` ``trace_type`` ``"clearwrite"``, ``"maxhold"``, ``"minhold"``, ``"freeze"``, ``"average"`` ``average`` Integer, typically 10–20 ``mode`` ``"circuit"``, ``"modal"`` ``reference_level`` Integer (units depend on ``amp_units``) ``input_attenuator`` ``0``–``78`` dB, or ``"auto"`` ``amp_units`` ``"dbuv"``, ``"dbmv"``, ``"dbm"``, ``"volts"``, ``"watts"`` ``sweep_time`` Float, 1–15 (seconds) ``external_loss`` Name of an existing external loss table, or ``""`` ``session_UUID`` Arbitrary string; locks the device to this session ``visible`` Boolean ``display_range`` ``[from_hz, to_hz]`` — subset of the current band ======================= ================================================ """ if field not in self._WRITABLE_FIELDS: raise ValueError( f"Cannot set field {field!r}. " f"Writable fields: {sorted(self._WRITABLE_FIELDS)}" ) setattr(self, field, value) await self.websocket.send(json.dumps({field: value})) # ── RBW / band ──────────────────────────────────────────────────────── # Default frequency range for each RBW value, in Hz. # Used by set_rbw() when no custom display_range is provided. _RBW_DEFAULT_DISPLAY_RANGE = { "200": (9_000, 150_000), "9": (150_000, 30_000_000), "120": (30_000_000, 110_000_000), "1": (10_000, 150_000), "10": (150_000, 30_000_000), "200_9": (9_000, 30_000_000), "1_10": (10_000, 30_000_000), } async def set_rbw( self, rbw: str, threephase: bool = False, display_range: Optional[tuple] = None, ) -> None: """ Set the resolution bandwidth (RBW) and frequency band, and wait for the server to confirm the change before returning. Changing RBW may trigger a firmware reload on the instrument (up to ~5 seconds for bands like ``"120"`` that require a different bitstream). The server echoes ``{"rbw": ""}`` once the firmware is ready. This method blocks until that echo is received, so it is safe to send further configuration immediately after it returns. The display range is set automatically to the natural band of the selected RBW (see ``_RBW_DEFAULT_DISPLAY_RANGE``). Pass a custom ``display_range`` tuple to override it. Args: rbw: Resolution bandwidth selector. Options: ======== ========= =================== ============== Value RBW Frequency range Standard ======== ========= =================== ============== ``"200"`` 200 Hz 9 kHz – 150 kHz CISPR 16-1-1 ``"9"`` 9 kHz 150 kHz – 30 MHz CISPR 16-1-1 ``"120"`` 120 kHz 30 MHz – 110 MHz CISPR 16-1-1 ``"1"`` 1 kHz 10 kHz – 150 kHz MIL-STD-461 ``"10"`` 10 kHz 150 kHz – 30 MHz MIL-STD-461 ``"200_9"`` dual 9 kHz – 30 MHz CISPR dual-band ``"1_10"`` dual 10 kHz – 30 MHz MIL dual-band ======== ========= =================== ============== threephase: Set to ``True`` for three-phase (RX4) models. display_range: Optional ``(from_hz, to_hz)`` tuple to override the default frequency window for the selected band. Only effective for freq_mode RBWs (``"200"``, ``"9"``, ``"1"``, ``"10"``, ``"200_9"``, ``"1_10"``); the server ignores ``display_range`` for ``"120"``. """ _FREQ_MODES = {"200", "9", "1", "10", "200_9", "1_10"} _SPECIAL_MODES = {"120", "osc", "imp", "clickmeter"} # Determine the effective display range for the target band. effective_range = display_range or self._RBW_DEFAULT_DISPLAY_RANGE.get(rbw) # When switching from a freq_mode to a special mode, the server uses the # current display_start/display_stop to slice the new band's data. Send # a full-spectrum range while still in the freq_mode so those values are # reset before the firmware switches. This is what the web frontend does. if self.rbw in _FREQ_MODES and rbw in _SPECIAL_MODES: await self.websocket.send(json.dumps({"display_range": [0, 10_000_000_000]})) self.rbw = rbw self.threephase = threephase await self.websocket.send(json.dumps({"rbw": rbw, "threephase": threephase})) # Wait for the server to echo the RBW value, which signals that the # firmware is loaded and ready. Other messages may arrive first (e.g. # configuration echoes) — keep reading until we get the RBW confirmation. while True: msg = await self.parse_messages() if msg is None or "rbw" in msg: break # After the RBW is confirmed, send the display range for freq_modes. # Store it on self.display_range so callers can use it for client-side filtering. if effective_range is not None: self.display_range = list(effective_range) if rbw in _FREQ_MODES: await self.websocket.send(json.dumps({"display_range": list(effective_range)})) # ── Convenience setters ─────────────────────────────────────────────── async def set_detector_type(self, detector: str) -> None: """ Set the detector type. Args: detector: ``"pk"`` (Peak), ``"qp"`` (Quasi-Peak), or ``"av"`` (Average). """ await self.set_value("detector_type", detector) async def set_measure_channel(self, channel: str) -> None: """ Set the input channel. Args: channel: For RX2 models: ``"lg"``, ``"ng"``, ``"cm"``, ``"dm"``. For RX4 models: ``"l1"``, ``"l2"``, ``"l3"``, ``"n"``. """ await self.set_value("measure_channel", channel) async def set_trace_type(self, trace: str) -> None: """ Set the trace display mode. Args: trace: ``"clearwrite"``, ``"maxhold"``, ``"minhold"``, ``"freeze"``, or ``"average"``. """ await self.set_value("trace_type", trace) async def set_amp_units(self, units: str) -> None: """ Set the amplitude units for displayed values. Args: units: ``"dbuv"``, ``"dbmv"``, ``"dbm"``, ``"volts"``, or ``"watts"``. """ await self.set_value("amp_units", units) async def set_reference_level(self, level: Union[int, str]) -> None: """ Set the reference level (top of the display scale). The value must be consistent with the current ``amp_units``. For example, 100 dBuV or 70 dBmV are typical values. Args: level: Reference level as integer (or numeric string). """ await self.set_value("reference_level", level) async def set_input_attenuator(self, value: Union[int, str]) -> None: """ Set the input attenuator. Args: value: Attenuation in dB (0–78), or ``"auto"`` for automatic selection. """ await self.set_value("input_attenuator", value) async def set_sweep_time(self, seconds: Union[float, str]) -> None: """ Set the sweep time per band. Args: seconds: Sweep duration, 1–15 seconds (float or numeric string). """ await self.set_value("sweep_time", seconds) async def set_average(self, n: int) -> None: """ Set the number of averages for ``trace_type="average"``. Args: n: Number of averages (typically 10–20). """ await self.set_value("average", n) async def set_mode(self, mode: str) -> None: """ Set the measurement mode. Args: mode: ``"circuit"`` (default) or ``"modal"``. """ await self.set_value("mode", mode) async def set_external_loss(self, name: str) -> None: """ Apply an external loss (cable / LISN attenuation) table. Args: name: Name of an existing external loss table, or ``""`` to disable. """ await self.set_value("external_loss", name) async def set_display_range(self, from_hz: int, to_hz: int) -> None: """ Restrict the displayed frequency range within the current band. Both values must fall within the frequency range of the active RBW. .. note:: ``display_range`` is **not supported** for RBW ``"120"`` (30 MHz – 110 MHz band). The server ignores this command in that mode and always returns the full band. Supported RBWs: ``"200"``, ``"9"``, ``"1"``, ``"10"``, ``"200_9"``, ``"1_10"``. Args: from_hz: Lower frequency boundary in Hz. to_hz: Upper frequency boundary in Hz. Example:: await emscope.set_display_range(150_000, 30_000_000) # 9 kHz band """ if self.rbw == "120": print(f"WARNING: display_range is not supported for RBW '120' and will be ignored by the server.") return await self.set_value("display_range", [from_hz, to_hz]) async def set_visible(self, visible: bool) -> None: """ Show or hide the trace on the instrument front panel. Args: visible: ``True`` to show, ``False`` to hide. """ await self.set_value("visible", visible) async def set_session_uuid(self, uuid: str) -> None: """ Lock the device to this session. If another client already holds a lock with a different UUID, the server will close the WebSocket with close code **4003**. Args: uuid: Arbitrary unique string identifying this session. """ await self.set_value("session_UUID", uuid) # ── Data request commands ───────────────────────────────────────────── async def request_standards(self) -> None: """ Request the list of limit standards from the server. After calling this, call :meth:`parse_messages` until ``self.standards`` is populated. Each entry is a dict ``{name: {rbw, data}}``. """ await self.websocket.send(json.dumps({"get_standards": True})) async def request_sessions(self) -> None: """ Request the list of saved preset sessions from the server. After calling this, call :meth:`parse_messages` until ``self.sessions`` is populated. Each entry is a dict ``{name: json_string}``. """ await self.websocket.send(json.dumps({"get_sessions": True})) async def request_external_losses(self) -> None: """ Request the list of external loss attenuator tables from the server. After calling this, call :meth:`parse_messages` until ``self.external_losses`` is populated. Each entry is a dict ``{name: [[freq, line_loss, neutral_loss], ...]}``. """ await self.websocket.send(json.dumps({"get_external_losses": True})) async def request_licenses(self) -> None: """ Request the list of activated licenses from the server. After calling this, call :meth:`parse_messages` until ``self.licenses`` is populated. Example: ``["emi", "osc"]``. """ await self.websocket.send(json.dumps({"get_licenses": True})) async def request_temps(self) -> None: """ Request the current PCB and FPGA temperatures from the server. After calling this, call :meth:`parse_messages` until ``self.temperatures`` is populated as ``[pcb_temp_C, fpga_temp_C]``. """ await self.websocket.send(json.dumps({"get_temps": True})) async def request_report( self, standard: str, subranges: int = 10, margin: int = 10 ) -> None: """ Request an EMC compliance report for a given limit standard. After calling this, call :meth:`parse_messages` until ``self.report`` is populated. Each entry in ``self.report`` is a list:: [marker, freq_MHz, pk_dBuV, qp_dBuV, qp_limit_dBuV, qp_margin_dB, av_dBuV, av_limit_dBuV, av_margin_dB, channel, "PASS"/"FAIL"] ``self.frase`` will be ``True`` if any emission is within ``margin`` dB of a limit. Args: standard: Name of the limit standard to evaluate against. subranges: Number of sub-ranges for peak selection (default 10). margin: Pass/fail margin in dB (default 10). """ await self.websocket.send(json.dumps({ "standard": standard, "subranges": subranges, "margin": margin, })) # ── Standards CRUD ──────────────────────────────────────────────────── async def create_standard(self, name: str, rbw: str, values: List[List[str]]) -> None: """ Create a new limit standard. Args: name: Name for the new standard. rbw: RBW this standard applies to (e.g. ``"9"`` or ``"120"``). values: List of frequency rows. Each row must be a list of 6 strings:: [from_MHz, to_MHz, qp_from_dBuV, qp_to_dBuV, av_from_dBuV, av_to_dBuV] """ await self.websocket.send(json.dumps({ "name": name, "modify": False, "standard_rbw": rbw, "values": values, })) async def edit_standard( self, original_name: str, new_name: str, rbw: str, values: List[List[str]], ) -> None: """ Edit an existing limit standard. Args: original_name: Current name of the standard to edit. new_name: New name (can be the same as original to keep the name). rbw: New RBW value. values: New frequency rows (same format as :meth:`create_standard`). """ await self.websocket.send(json.dumps({ "original_name": original_name, "name": new_name, "modify": True, "standard_rbw": rbw, "values": values, })) async def delete_standard(self, name: str) -> None: """ Delete a limit standard by name. Args: name: Name of the standard to delete. """ await self.websocket.send(json.dumps({"delete_standard": name})) async def reset_standards(self) -> None: """Reset all limit standards to factory defaults.""" await self.websocket.send(json.dumps({"reset_standards": True})) # ── Sessions CRUD ───────────────────────────────────────────────────── async def create_session(self, name: str, values: str) -> None: """ Save a new preset session. Args: name: Name for the new session. values: Session configuration as a JSON string. """ await self.websocket.send(json.dumps({ "session_name": name, "modify": False, "session_values": values, })) async def edit_session( self, original_name: str, new_name: str, values: str ) -> None: """ Edit an existing preset session. Args: original_name: Current name of the session. new_name: New name (can be the same to keep it). values: Updated session configuration as a JSON string. """ await self.websocket.send(json.dumps({ "original_session": original_name, "session_name": new_name, "modify": True, "values": values, })) async def delete_session(self, name: str) -> None: """ Delete a preset session by name. Args: name: Name of the session to delete. """ await self.websocket.send(json.dumps({"delete_session": name})) async def reset_sessions(self) -> None: """Delete all saved sessions.""" await self.websocket.send(json.dumps({"reset_sessions": True})) # ── External loss CRUD ──────────────────────────────────────────────── async def create_external_loss(self, name: str, values: List[List[str]]) -> None: """ Create a new external loss (cable / LISN) attenuation table. Args: name: Name for the new attenuator table. values: List of frequency rows. Each row is a list of 3 strings:: [frequency_MHz, line_loss_dB, neutral_loss_dB] Example:: await emscope.create_external_loss("LISN 50uH", [ ["0.009", "1.2", "1.1"], ["30.000", "2.5", "2.4"], ]) """ await self.websocket.send(json.dumps({ "external_loss_name": name, "modify": False, "values": values, })) async def edit_external_loss( self, original_name: str, new_name: str, values: List[List[str]] ) -> None: """ Edit an existing external loss table. Args: original_name: Current name of the table. new_name: New name (can be the same to keep it). values: Updated rows (same format as :meth:`create_external_loss`). """ await self.websocket.send(json.dumps({ "original_external_loss_name": original_name, "external_loss_name": new_name, "modify": True, "values": values, })) async def delete_external_loss(self, name: str) -> None: """ Delete an external loss table by name. Args: name: Name of the table to delete. """ await self.websocket.send(json.dumps({"delete_external_loss": name})) # ── Private helpers ─────────────────────────────────────────────────── def _reset_values(self) -> None: """Initialize all instance attributes to their defaults.""" # Device info (populated on connect, read-only) self.SN: str = "" self.SFP_SN: str = "" self.MAC: str = "" self.measurement_uncertainty: str = "" self.num_points: int = 0 # Measurement configuration (writable) self.measure_channel: str = "lg" self.detector_type: str = "pk" self.trace_type: str = "clearwrite" self.average: int = 10 self.rbw: str = "9" self.mode: str = "circuit" self.reference_level: int = 100 self.input_attenuator: Union[int, str] = "auto" self.input_attenuator_auto_value: int = 0 self.amp_units: str = "dbuv" self.sweep_time: float = 1.0 self.external_loss: str = "" self.session_UUID: str = "" self.visible: bool = True self.display_range: Optional[List[int]] = None self.threephase: bool = False # Oscilloscope settings self.osc_us_div: Optional[float] = None self.osc_volts_div: Optional[float] = None self.osc_sweep: Optional[str] = None self.trigger: Optional[str] = None self.osc_trigger_offset: Optional[float] = None self.osc_trigger_level: Optional[float] = None self.osc_vertical_offset: Optional[float] = None self.osc_horizontal_offset: Optional[float] = None # Measurement data (updated each sweep) self.values: List[List[float]] = [] self.overload: bool = False # Requested data (populated after the corresponding request_* call) self.standards: List[dict] = [] self.sessions: List[dict] = [] self.external_losses: List[dict] = [] self.licenses: List[str] = [] self.report: List[list] = [] self.frase: bool = False self.temperatures: List[float] = [] self.repetition: Optional[Any] = None # ── Backwards-compatibility aliases ─────────────────────────────────── async def setValue(self, field: str, value: Any) -> None: """Alias for :meth:`set_value`. Kept for backwards compatibility.""" await self.set_value(field, value) async def setRBW(self, rbw: str, threephase: bool = False) -> None: """Alias for :meth:`set_rbw`. Kept for backwards compatibility.""" await self.set_rbw(rbw, threephase) async def parseMessages(self) -> Optional[Dict[str, Any]]: """Alias for :meth:`parse_messages`. Kept for backwards compatibility.""" return await self.parse_messages() async def updateStandardsList(self) -> None: """Alias for :meth:`request_standards`. Kept for backwards compatibility.""" await self.request_standards()