Parallel demo to test ProgressHandler queue strategy (#51)

* Parallel demo to test ProgressHandler queue strategy

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Separate ws and non-ws implementation

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Show global job bar

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Update style and handle more on the client

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* working windows version for global bar

* Swap from 8000 to a non-dedicated port

* Limit websocket reconnection to 3 times

* Receive all job bar events from the server

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: codycbakerphd <[email protected]>

pyproject.toml

@@ -44,7 +44,9 @@ test = [
 ]
 
 demo = [
-    "websockets==12.0"
+    "websockets==12.0",
+    "flask==2.3.2",
+    "flask-cors==4.0.0"
 ]
 
 [project.urls]

src/tqdm_publisher/__init__.py

@@ -1,3 +1,5 @@
+from ._handler import TQDMProgressHandler
 from ._publisher import TQDMPublisher
+from ._subscriber import TQDMProgressSubscriber
 
-__all__ = ["TQDMPublisher"]
+__all__ = ["TQDMPublisher", "TQDMProgressSubscriber", "TQDMProgressHandler"]

src/tqdm_publisher/_demos/_client.css

@@ -1,6 +1,7 @@
 
 html, body {
     font-family: sans-serif;
+    box-sizing: border-box;
 }
 
 h1 {
@@ -28,10 +29,30 @@ header {
     width: 100%;
     height: 20px;
     background-color: #ddd;
+    box-sizing: border-box;
 }
 
 .progress div {
     height: 100%;
     background-color: #4caf50;
     width: 0%;
 }
+
+.progress[data-small="true"] {
+    height: 10px;
+    border-bottom: 1px solid gainsboro;
+}
+
+.progress[data-small="true"]:last-child {
+    border-bottom: none;
+}
+
+.progress[data-small="false"] {
+    border-bottom: 1px solid black;
+    padding: 4px;
+}
+
+
+.bar-container {
+    border: 1px solid black;
+}

src/tqdm_publisher/_demos/_demo_command_line_interface.py

@@ -5,14 +5,19 @@
 from pathlib import Path
 
 from tqdm_publisher._demos._multiple_bars._server import run_multiple_bar_demo
+from tqdm_publisher._demos._parallel_bars._server import run_parallel_bar_demo
+from tqdm_publisher._demos._parallel_bars._server_ws import (
+    run_parallel_bar_demo as run_parallel_bar_demo_ws,
+)
 from tqdm_publisher._demos._single_bar._server import run_single_bar_demo
 
 CLIENT_PORT = 1234
 
 DEMOS = {
     "demo_single": dict(subpath="_single_bar", server=run_single_bar_demo),
     "demo_multiple": dict(subpath="_multiple_bars", server=run_multiple_bar_demo),
-    # "parallel": "_parallel",
+    "demo_parallel": dict(subpath="_parallel_bars", server=run_parallel_bar_demo),
+    "demo_parallel_ws": dict(subpath="_parallel_bars", server=run_parallel_bar_demo_ws),
 }
 
 DEMO_BASE_FOLDER_PATH = Path(__file__).parent
@@ -43,13 +48,6 @@ def _command_line_interface():
 
         demo_info = DEMOS[command]
 
-        # if command == "parallel":
-        #     client_relative_path = Path(subpath) / "_client.py"
-        #     subprocess.Popen(['python', str(DEMO_BASE_FOLDER_PATH / subpath / "_server.py")])
-        #     subprocess.Popen(['python', str(DEMO_BASE_FOLDER_PATH / subpath / "_client.py")])
-
-        # else:
-
         client_relative_path = Path(demo_info["subpath"]) / "_client.html"
         subprocess.Popen(["python", "-m", "http.server", str(CLIENT_PORT), "-d", DEMO_BASE_FOLDER_PATH])
 

src/tqdm_publisher/_demos/_multiple_bars/_server.py

@@ -65,7 +65,7 @@ def start(self):
 
 async def spawn_server() -> None:
     """Spawn the server asynchronously."""
-    async with websockets.serve(ws_handler=handler, host="", port=8000):
+    async with websockets.serve(ws_handler=handler, host="", port=3768):
         await asyncio.Future()
 
 

src/tqdm_publisher/_demos/_parallel_bars/_client.html

@@ -0,0 +1,36 @@
+<!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>Parallel Bars Demo</title>
+
+    <!-- Basic Page Styling -->
+    <link rel="stylesheet" href="../_client.css">
+    <script src="./_client.js" type="module" defer></script>
+
+</head>
+
+<!-- Basic Page Structure -->
+<body>
+    <header>
+        <div>
+            <h1>TQDM Publisher Demo: Parallel Bars</h1>
+            <i><small>Click the button to create parallel progress bars</small></i>
+        </div>
+
+        <!-- Declare a button to create progress bars -->
+        <button>Create Progress Bars</button>
+    </header>
+
+    <!-- Declare a container to hold the progress bars -->
+    <div id="bars"></div>
+</body>
+</html>

src/tqdm_publisher/_demos/_parallel_bars/_client.js

@@ -0,0 +1,60 @@
+import { EventSourceManager } from '../utils/EventSourceManager.js';
+import { WebSocketManager } from '../utils/WebSocketManager.js';
+import { createProgressBar } from '../utils/elements.js';
+
+const bars = {} // Track progress bars
+const requests = {} // Track request containers
+
+function getRequestContainer(request_id) {
+    const existing = requests[request_id]
+    if (existing) return existing;
+
+    // Create a new container for the progress bar
+    const container = document.createElement('div');
+    container.id = request_id;
+    container.classList.add('request-container');
+    const h2 = document.createElement('h2');
+    h2.innerText = `Request ID: ${request_id}`;
+    container.append(h2);
+    document.body.append(container);
+    const barContainer = document.createElement('div');
+    barContainer.classList.add('bar-container');
+    container.append(barContainer);
+    requests[request_id] = barContainer;
+
+    return barContainer;
+}
+
+// Create and/or render a progress bar
+const getBar = (request_id, id) => {
+
+    if (bars[id]) return bars[id];
+
+    const container = getRequestContainer(request_id);
+    const bar = createProgressBar(container);
+
+    bar.parentElement.setAttribute('data-small', request_id !== id); // Add a small style to the progress bar if it is not the main request bar
+
+    return bars[id] = bar;
+
+}
+
+// Update the specified progress bar when a message is received from the server
+const onProgressUpdate = (event) => {
+    const { request_id, id, format_dict } = JSON.parse(event.data);
+    const bar = getBar(request_id, id);
+    bar.style.width = 100 * (format_dict.n / format_dict.total) + '%';
+}
+
+// Create a new message client
+const wsClient = new WebSocketManager({ onmessage: onProgressUpdate }, 3);
+const client = new EventSourceManager({ onmessage: onProgressUpdate });
+
+// Declare that the HTML Button should create a new progress bar when clicked
+const button = document.querySelector('button');
+button.addEventListener('click', async () => {
+    const request_id = Math.random().toString(36).substring(7); // Create a unique ID for the progress bar
+    getBar(request_id, request_id); // Create a bar specifically for this request
+    await client.send({ command: 'start', request_id }).catch(() => {}); // Send a message to the server to start the progress bar
+    wsClient.send({ command: 'start', request_id }); // Send a message to the server to start the progress bar
+})

src/tqdm_publisher/_demos/_parallel_bars/_client.py

@@ -0,0 +1,62 @@
+"""Demo of parallel tqdm client."""
+
+# HTTP server addition
+import http.server
+import json
+import signal
+import socket
+import socketserver
+import sys
+
+
+def find_free_port():
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(("", 0))  # Bind to a free port provided by the host
+        return s.getsockname()[1]  # Return the port number assigned
+
+
+def GLOBAL_CALLBACK(request_id, id, n, total, **kwargs):
+    print("Global Update", request_id, id, f"{n}/{total}")
+
+
+def create_http_server(port: int, callback):
+    class MyHttpRequestHandler(http.server.SimpleHTTPRequestHandler):
+
+        def do_POST(self):
+            content_length = int(self.headers["Content-Length"])
+            post_data = json.loads(self.rfile.read(content_length).decode("utf-8"))
+            callback(post_data["request_id"], post_data["id"], **post_data["data"])
+            self.send_response(200)
+            self.end_headers()
+
+    with socketserver.TCPServer(("", port), MyHttpRequestHandler) as httpd:
+
+        def signal_handler(signal, frame):
+            print("\n\nInterrupt signal received. Closing server...")
+            httpd.server_close()
+            httpd.socket.close()
+            print("Server closed.")
+            sys.exit(0)
+
+        try:
+            signal.signal(signal.SIGINT, signal_handler)
+        except:
+            pass  # Allow to work in thread
+
+        print(f"Serving HTTP on port {port}")
+        httpd.serve_forever()
+
+
+if __name__ == "__main__":
+
+    flags_list = sys.argv[1:]
+
+    port_flag = "--port" in flags_list
+
+    if port_flag:
+        port_index = flags_list.index("--port")
+        PORT = int(flags_list[port_index + 1])
+    else:
+        PORT = find_free_port()
+
+    create_http_server(port=PORT, callback=GLOBAL_CALLBACK)