catalystneuro · CodyCBakerPhD · Jan 19, 2024 · Jan 19, 2024 · Jan 19, 2024 · Jan 19, 2024
diff --git a/README.md b/README.md
@@ -41,3 +41,5 @@ n = 10**5
 sleep_durations = [random.uniform(0, 5.0) for _ in range(n)]
 asyncio.run(run_multiple_sleeps(sleep_durations=sleep_durations))
 ```
+
+A complete demo of `tqdm_publisher` can be found in the `demo` directory, which shows how to forward progress updates from the same `TQDMPublisher` instance to multiple clients.
diff --git a/demo/client.html b/demo/client.html
@@ -0,0 +1,110 @@
+<!DOCTYPE html>
+
+<html lang="en">
+
+<head>
+
+    <meta charset="UTF-8">
+
+    <meta http-equiv="X-UA-Compatible" content="IE=edge">
+
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+
+    <title>Concurrent Client Demo</title>
+
+    <style>
+
+        html, body {
+            font-family: sans-serif;
+        }
+
+        h1 {
+            margin: 0;
+            padding: 0;
+            font-size: 1.5rem;
+        }
+
+        header {
+            display: flex;
+            justify-content: space-between;
+            align-items: center;
+            padding: 20px;
+        }
+
+        #bars {
+            display: flex;
+            justify-content: center;
+            align-items: center;
+            gap: 20px;
+            flex-wrap: wrap;
+        }
+
+        .progress {
+            width: 100%;
+            height: 20px;
+            background-color: #ddd;
+        }
+
+
+
+        .progress div {
+            height: 100%;
+            background-color: #4caf50;
+            width: 0%;
+        }
+
+    </style>
+
+</head>
+
+<body>
+    <header>
+        <div>
+            <h1>tqdm_progress</h1>
+            <i><small>Create multiple progress bars to test concurrent subscriptions</small></i>
+        </div>
+        <button>Create Progress Bar</button>
+    </header>
+
+    <div id="bars">
+
+    </div>
+
+</body>
+
+<script>
+
+    const bars = document.querySelector('#bars');
+
+    class ProgressClient {
+        constructor() {
+            this.socket = new WebSocket('ws://localhost:8000');
+
+            this.element = document.createElement('div');
+            this.element.classList.add('progress');
+            const progress = document.createElement('div');
+            this.element.appendChild(progress);
+            bars.appendChild(this.element);
+
+            this.socket.addEventListener('message', function (event) {
+                const data = JSON.parse(event.data);
+                progress.style.width = 100 * (data.n / data.total) + '%';
+            });
+
+        }
+
+        close() {
+            this.socket.close();
+            this.element.remove()
+        }
+
+    }
+
+const button = document.querySelector('button');
+button.addEventListener('click', () => {
+    const client = new ProgressClient();
+})
+
+</script>
+
+</html>
diff --git a/demo/main.py b/demo/main.py
@@ -0,0 +1,157 @@
+#!/usr/bin/env python
+
+import random
+import asyncio
+from typing import List
+
+from tqdm_publisher import TQDMPublisher
+
+import websockets
+import threading
+from uuid import uuid4
+
+import json
+
+async def sleep_func(sleep_duration: float = 1) -> float:
+    await asyncio.sleep(delay=sleep_duration)
+
+
+def create_tasks():
+    n = 10**5
+    sleep_durations = [random.uniform(0, 5.0) for _ in range(n)]
+    tasks = []
+
+    for sleep_duration in sleep_durations:
+        task = asyncio.create_task(sleep_func(sleep_duration=sleep_duration))
+        tasks.append(task)
+
+    return tasks
+
+
+class ProgressHandler():
+
+    def __init__(self):
+        self.started = False
+        self.callbacks = []
+        self.callback_ids = []
+
+    def subscribe(self, callback):            
+        self.callbacks.append(callback)
+
+        if (hasattr(self, 'progress_bar')):
+            self._subscribe(callback)
+
+
+    def unsubscribe(self, callback_id):
+        self.progress_bar.unsubscribe(callback_id)
+
+    def clear(self):     
+        self.callbacks = []   
+        self._clear()
+
+    def _clear(self):
+
+        for callback_id in self.callback_ids:
+             self.unsubscribe(callback_id)
+
+        self.callback_ids = []
+
+    async def run(self):
+        for f in self.progress_bar:
+            await f
+
+    def stop(self):
+        self.started = False
+        self.clear()
+        self.thread.join()
+
+
+    def _subscribe(self, callback):
+        callback_id = self.progress_bar.subscribe(callback)
+        self.callback_ids.append(callback_id)
+
+
+    async def run(self):
+
+        if (hasattr(self, 'progress_bar')):
+            print("Progress bar already running")
+            return
+
+        self.tasks = create_tasks()
+        self.progress_bar = TQDMPublisher(asyncio.as_completed(self.tasks), total=len(self.tasks))
+
+        for callback in self.callbacks:
+            self._subscribe(callback)
+
+        for f in self.progress_bar:
+            await f
+
+        self._clear()
+        del self.progress_bar
+
+
+
+    def thread_loop(self):
+        while self.started:
+            asyncio.run(self.run())
+
+    def start(self):
+
+        if (self.started):
+            return
+
+        self.started = True
+
+        self.thread = threading.Thread(target=self.thread_loop) # Start infinite loop of progress bar thread
+        self.thread.start()
+
+
+progress_handler = ProgressHandler()
+
+class WebSocketHandler:
+    def __init__(self):
+
+        self.clients = {}
+
+        # Initialize with any state you need
+        pass
+
+    def handle_task_result(self, task):
+        try:
+            task.result()  # This will re-raise any exception that occurred in the task
+        except websockets.exceptions.ConnectionClosedOK:
+            print("WebSocket closed while sending message")
+        except Exception as e:
+            print(f"Error in task: {e}")
+
+    async def handler(self, websocket):
+        id = uuid4().hex
+        self.clients[id] = websocket # Register client connection
+
+        progress_handler.start() # Start if not started
+
+        def on_progress(info):
+            task = asyncio.create_task(websocket.send(json.dumps(info)))
+            task.add_done_callback(self.handle_task_result)  # Handle task result or exception
+
+        progress_handler.subscribe(on_progress)
+
+        try:
+            async for message in websocket:
+                print("Message from client received:", message)
+
+        finally:
+            # This is called when the connection is closed
+            del self.clients[id]
+            if (len(self.clients) == 0):
+                progress_handler.stop()
+
+
+
+async def main():
+    handler = WebSocketHandler().handler
+    async with websockets.serve(handler, "", 8000):
+        await asyncio.Future()  # run forever
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/main.py b/main.py
diff --git a/src/tqdm_publisher/publisher.py b/src/tqdm_publisher/publisher.py
@@ -21,7 +21,7 @@ def update(self, n=1, always_callback=False):
 
     # Subscribe to updates
     def subscribe(self, callback):
-        callback_id = uuid4()
+        callback_id = uuid4().hex
         self.callbacks[callback_id] = callback
         return callback_id