API Client

Module: equser.api Dependencies: [analysis] extra (requests, websocket-client)

REST and WebSocket clients for EQ gateways (running EQ Coherence™ software). Install with:

pip install equser[analysis]

GatewayClient

REST client for one EQ gateway (EQ Coherence™ server). The gateway proxies the API under /api on the normal HTTP port, so the default base URL is http://localhost (no port needed).

from equser.api import GatewayClient

client = GatewayClient('http://gateway')   # or GatewayClient() for localhost

Previously named SynapseClient. That alias was removed in 0.0.5 and now raises on import; use GatewayClient.

list_devices() -> list[dict]

List all registered devices.

devices = client.list_devices()
for d in devices:
    print(f"{d['id']}: {d.get('name', 'unnamed')}")

get_pmon_data(device_id, **params) -> pa.Table

Fetch PMon data as an Arrow Table.

table = client.get_pmon_data('wave-001')
print(f"Rows: {table.num_rows}, Columns: {table.column_names}")

Optional query parameters: start_time, end_time, metrics, limit.

get_cpow_data(device_id, **params) -> pa.Table

Fetch CPOW waveform data as an Arrow Table.

table = client.get_cpow_data('wave-001')

Optional query parameters: start_time, end_time, limit.

get_events(device_id=None, limit=100) -> list[dict]

Fetch recent power quality events.

events = client.get_events(limit=10)
for e in events:
    print(f"{e['timestamp']}: {e['type']}")

query_sql(query, device_id=None, limit=None) -> pa.Table

Execute a SELECT query via the SQL endpoint. Only SELECT statements are allowed by the server. The result comes back as Arrow IPC and is returned as a pyarrow.Table (the server default row limit is 30).

table = client.query_sql(
    "SELECT time_us, FREQ, AVRMS FROM pmon_data ORDER BY time_us DESC",
    limit=10,
)
df = table.to_pandas()

WebSocket streaming

connect_cpow_stream(gateway_url) -> Generator

Connect to the CPOW waveform WebSocket and yield data in real time.

Each binary message is an Arrow IPC RecordBatch (~512 rows, 16 ms at 32 ksps). Text messages are JSON gap markers indicating dropped samples.

from equser.api import connect_cpow_stream

for item in connect_cpow_stream('http://gateway'):
    if isinstance(item, dict):
        print(f"Gap: {item['skipped_samples']} samples")
    else:
        # item is a pyarrow.RecordBatch
        va = item.column('VA').to_numpy()
        print(f"Batch: {item.num_rows} rows, VA peak: {va.max()}")

connect_spectral_stream(channels=('VA', 'IA'), ...) -> Generator

Connect to the spectral WebSocket and yield spectral windows. The spectral stream is a broadcast consumer of the live CPOW feed (it is not per-device, so there is no device_id). Each binary message is one complete Arrow IPC stream of FFT magnitudes for the subscribed channels, returned as a pyarrow.Table; per-window metadata (window_start_ts, fundamental_hz, THD, PLL state, …) is in table.schema.metadata. Text messages are JSON gap markers {"type": "gap", "skipped_samples": N}.

Args:

from equser.api import connect_spectral_stream

for item in connect_spectral_stream(channels=['VA', 'IA'], cycles=12):
    if isinstance(item, dict):
        print(f"Gap: {item['skipped_samples']} samples")
    else:
        # item is a pyarrow.Table of FFT magnitudes; metadata on the schema
        print(f"Window: {item.num_rows} bins, channels {item.column_names}")

API endpoints

The EQ Coherence™ server on the EQ gateway exposes these endpoints:

Data endpoints return Arrow IPC binary format for efficient transfer.