better ecg server impl

build/lib/medtrace_synth/__main__.py

@@ -1,34 +1,46 @@
+import time
 import websockets
 import asyncio
+from pojagi_dsp.channel.ecg.generator.wavetable import ECGWaveTableSynthesizer
+from pojagi_dsp.channel.ecg.generator.wavetable.sinus import SinusWaveTable, TachycardiaWaveTable
 
-# Server data
 PORT = 7890
-print("Server listening on Port " + str(PORT))
 
-# A set of connected ws clients
-connected = set()
-
-# The main behavior function for this server
-async def echo(websocket, path):
-    print("A client just connected")
-    print(f"path: {path}")
-    # Store a copy of the connected client
-    connected.add(websocket)
-    # Handle incoming messages
-    try:
+async def consumer_handler(websocket):
     async for message in websocket:
-            print("Received message from client: " + message)
-            # Send a response to all connected clients except sender
-            for conn in connected:
-                if conn != websocket:
-                    await conn.send("Someone said: " + message)
-    # Handle disconnecting clients 
+        print(f"message received: {message}")
+
+async def producer_handler(websocket):
+    ecg = ECGWaveTableSynthesizer(
+        tables={
+            (0,90): SinusWaveTable(),
+            (70, 300): TachycardiaWaveTable(),
+        },
+        heart_rate=70,
+        srate=50,
+    )
+    while True:
+        try:
+            message = next(ecg)
+            await websocket.send(str(message))
+            time.sleep(0.01)
         except websockets.exceptions.ConnectionClosed as e:
             print("A client just disconnected")
-    finally:
-        connected.remove(websocket)
+            break
+
+async def handler(websocket, path):
+    while True:
+        print(f"New connection. Path: {path}")
+        consumer_task = asyncio.create_task(consumer_handler(websocket))
+        producer_task = asyncio.create_task(producer_handler(websocket))
+        done, pending = await asyncio.wait(
+            [consumer_task, producer_task],
+            return_when=asyncio.FIRST_COMPLETED,
+        )
+        for task in pending:
+            task.cancel()
 
 # Start the server
-start_server = websockets.serve(echo, "0.0.0.0", PORT)
+start_server = websockets.serve(handler, "0.0.0.0", PORT)
 asyncio.get_event_loop().run_until_complete(start_server)
 asyncio.get_event_loop().run_forever()

src/medtrace_synth.egg-info/SOURCES.txt

@@ -1,3 +1,4 @@
+.gitignore
 pyproject.toml
 src/medtrace_synth/__main__.py
 src/medtrace_synth.egg-info/PKG-INFO

src/medtrace_synth/__main__.py

@@ -1,36 +1,46 @@
+import time
 import websockets
 import asyncio
+from pojagi_dsp.channel.ecg.generator.wavetable import ECGWaveTableSynthesizer
+from pojagi_dsp.channel.ecg.generator.wavetable.sinus import SinusWaveTable, TachycardiaWaveTable
 
-# Server data
 PORT = 7890
-print("Server listening on Port " + str(PORT))
 
-# A set of connected ws clients
-connected = set()
-
-# The main behavior function for this server
-async def echo(websocket, path):
-    print("A client just connected")
-    print(f"path: {path}")
-    # Store a copy of the connected client
-    connected.add(websocket)
-    # Handle incoming messages
-    try:
+async def consumer_handler(websocket):
     async for message in websocket:
-            print("Received message from client: " + message)
-            # Send a response to all connected clients except sender
-            for conn in connected:
-                if conn != websocket:
-                    await conn.send("Someone said: " + message)
-                else:
-                    await conn.send(f"You said: {message}")
-    # Handle disconnecting clients 
+        print(f"message received: {message}")
+
+async def producer_handler(websocket):
+    ecg = ECGWaveTableSynthesizer(
+        tables={
+            (0,90): SinusWaveTable(),
+            (70, 300): TachycardiaWaveTable(),
+        },
+        heart_rate=70,
+        srate=50,
+    )
+    while True:
+        try:
+            message = next(ecg)
+            await websocket.send(str(message))
+            time.sleep(0.01)
         except websockets.exceptions.ConnectionClosed as e:
             print("A client just disconnected")
-    finally:
-        connected.remove(websocket)
+            break
+
+async def handler(websocket, path):
+    while True:
+        print(f"New connection. Path: {path}")
+        consumer_task = asyncio.create_task(consumer_handler(websocket))
+        producer_task = asyncio.create_task(producer_handler(websocket))
+        done, pending = await asyncio.wait(
+            [consumer_task, producer_task],
+            return_when=asyncio.FIRST_COMPLETED,
+        )
+        for task in pending:
+            task.cancel()
 
 # Start the server
-start_server = websockets.serve(echo, "0.0.0.0", PORT)
+start_server = websockets.serve(handler, "0.0.0.0", PORT)
 asyncio.get_event_loop().run_until_complete(start_server)
 asyncio.get_event_loop().run_forever()