diff --git a/README.md b/README.md index 6dd30cf..2086356 100644 --- a/README.md +++ b/README.md @@ -11,36 +11,50 @@ This is useful if you want to use `tqdm` to track the progress of a long-running ```bash pip install tqdm_publisher ``` +## Getting Started +### Basic Usage +To monitor the progress of an existing `tqdm` progress bar, simply swap the `tqdm`and `TQDMPublisher` constructors. Then, declare a callback function to handle progress updates, and subscribe it to the `TQDMPublisher` updates using the `subscribe` method _before iteration begins_. -## Usage +#### Original Code ```python import random -import asyncio +import time -from tqdm_publisher import TQDMPublisher +from tqdm import tqdm + +N_TASKS = 100 + +# Create a list of tasks +durations = [ random.uniform(0, 1.0) for _ in range(N_TASKS) ] + +# Create a progress bar +progress_bar = tqdm(durations) -async def sleep_func(sleep_duration = 1): - await asyncio.sleep(delay=sleep_duration) +# Iterate over the progress bar +for duration in progress_bar: + time.sleep(duration) # Execute the task +``` -async def run_multiple_sleeps(sleep_durations): +#### Modified Code - tasks = [] +```python +import random +import time - for sleep_duration in sleep_durations: - task = asyncio.create_task(sleep_func(sleep_duration=sleep_duration)) - tasks.append(task) +from tqdm_publisher import TQDMPublisher - progress_bar = TQDMPublisher(asyncio.as_completed(tasks), total=len(tasks)) - callback_id = progress_bar.subscribe(lambda info: print('Progress Update', info)) +N_TASKS = 100 +durations = [ random.uniform(0, 1.0) for _ in range(N_TASKS) ] +progress_bar = TQDMPublisher(durations) - for f in progress_bar: - await f +# Declare a callback function to handle progress updates +on_update = lambda info: print('Progress Update', info) - progress_bar.unsubscribe(callback_id) +# Subscribe the callback to the TQDMPublisher +progress_bar.subscribe(on_update) -number_of_tasks = 10**5 -sleep_durations = [random.uniform(0, 5.0) for _ in range(number_of_tasks)] -asyncio.run(run_multiple_sleeps(sleep_durations=sleep_durations)) +for duration in progress_bar: + time.sleep(duration) ``` ## Demo diff --git a/demo/client.py b/demo/client.py deleted file mode 100644 index 73bf883..0000000 --- a/demo/client.py +++ /dev/null @@ -1,12 +0,0 @@ -import subprocess -from pathlib import Path - -client_path = Path(__file__).parent / "client.html" - - -def main(): - subprocess.run(["open", client_path]) - - -if __name__ == "__main__": - main() diff --git a/demo/demo_cli.py b/demo/demo_cli.py deleted file mode 100644 index 766d806..0000000 --- a/demo/demo_cli.py +++ /dev/null @@ -1,41 +0,0 @@ -import subprocess -import sys -from pathlib import Path - -demo_base_path = Path(__file__).parent - -client_path = demo_base_path / "client.html" -server_path = demo_base_path / "server.py" - - -def main(): - if len(sys.argv) <= 1: - print("No command provided. Please specify a command (e.g. 'demo').") - return - - command = sys.argv[1] - - flags_list = sys.argv[2:] - - client_flag = "--client" in flags_list - server_flag = "--server" in flags_list - both_flags = "--server" in flags_list and "--client" in flags_list - - flags = dict( - client=not server_flag or both_flags, - server=not client_flag or both_flags, - ) - - if command == "demo": - if flags["client"]: - subprocess.run(["open", client_path]) - - if flags["server"]: - subprocess.run(["python", server_path]) - - else: - print(f"{command} is an invalid command.") - - -if __name__ == "__main__": - main() diff --git a/demo/server.py b/demo/server.py deleted file mode 100644 index 02d68dd..0000000 --- a/demo/server.py +++ /dev/null @@ -1,152 +0,0 @@ -#!/usr/bin/env python - -import asyncio -import json -import random -import threading -from typing import List -from uuid import uuid4 - -import websockets - -from tqdm_publisher import TQDMPublisher - - -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 = str(uuid4()) - 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 spawn_server(): - handler = WebSocketHandler().handler - async with websockets.serve(handler, "", 8000): - await asyncio.Future() # run forever - - -def main(): - asyncio.run(spawn_server()) - - -if __name__ == "__main__": - main() diff --git a/pyproject.toml b/pyproject.toml index 2518f55..942e7fa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,6 +31,7 @@ classifiers = [ "Operating System :: MacOS", "Operating System :: Unix", ] + dependencies = [ "tqdm>=4.49.0" ] @@ -50,8 +51,8 @@ demo = [ "Homepage" = "https://github.com/catalystneuro/tqdm_publisher" "Bug Tracker" = "https://github.com/catalystneuro/tqdm_publisher/issues" -[project.gui-scripts] -tqdm_publisher = "demo.demo_cli:main" +[project.scripts] +tqdm_publisher = "tqdm_publisher._demo._demo_command_line_interface:_command_line_interface" [tool.black] line-length = 120 diff --git a/src/tqdm_publisher/_demo/__init__.py b/src/tqdm_publisher/_demo/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/demo/client.html b/src/tqdm_publisher/_demo/_client.html similarity index 57% rename from demo/client.html rename to src/tqdm_publisher/_demo/_client.html index 67ed769..0e514eb 100644 --- a/demo/client.html +++ b/src/tqdm_publisher/_demo/_client.html @@ -12,6 +12,7 @@ Concurrent Client Demo + + +

tqdm_progress

Create multiple progress bars to test concurrent subscriptions
+ +
-
- -
- + +
- - - diff --git a/src/tqdm_publisher/_demo/_client.js b/src/tqdm_publisher/_demo/_client.js new file mode 100644 index 0000000..edf3670 --- /dev/null +++ b/src/tqdm_publisher/_demo/_client.js @@ -0,0 +1,72 @@ + +// Grab bar container from HTML +const barContainer = document.querySelector('#bars'); + +// Create a progress bar and append it to the bar container +const createProgressBar = () => { + const element = document.createElement('div'); + element.classList.add('progress'); + const progress = document.createElement('div'); + element.appendChild(progress); + barContainer.appendChild(element); + return { element, progress }; +} + +// Create a simple WebSocket client wrapper class +class ProgressClient { + + #connect = (props = {}) => { + + const { + onopen = () => {}, + onclose = () => {}, + onmessage = () => {} + } = props; + + this.socket = new WebSocket('ws://localhost:8000'); + this.socket.addEventListener('open', onopen); + + // Attempt to reconnect every second if the connection is closed + this.socket.addEventListener('close', () => { + onclose(); + setTimeout(() => this.#connect(props), 1000); + }); + + this.socket.addEventListener('message', onmessage); + } + + constructor(props) { + this.#connect(props); + } + + close() { + this.socket.close(); + } + +} + + +const bars = {} // Track progress bars + + +// Update the specified progress bar when a message is received from the server +const onProgressUpdate = (event) => { + const { progress_bar_id, format_dict } = JSON.parse(event.data); + bars[progress_bar_id].style.width = 100 * (format_dict.n / format_dict.total) + '%'; +} + +// Create a new WebSocket client +const client = new ProgressClient({ onmessage: onProgressUpdate }); + +// Declare that the HTML Button should create a new progress bar when clicked +const button = document.querySelector('button'); +button.addEventListener('click', () => { + const { element, progress } = createProgressBar(); // Create a progress bar + + barContainer.appendChild(element); // Render the progress bar + + const progress_bar_id = Math.random().toString(36).substring(7); // Create a unique ID for the progress bar + bars[progress_bar_id] = progress; // Track the progress bar + + client.socket.send(JSON.stringify({ command: 'start', progress_bar_id })); // Send a message to the server to start the progress bar +}) diff --git a/src/tqdm_publisher/_demo/_demo_command_line_interface.py b/src/tqdm_publisher/_demo/_demo_command_line_interface.py new file mode 100644 index 0000000..2f1764c --- /dev/null +++ b/src/tqdm_publisher/_demo/_demo_command_line_interface.py @@ -0,0 +1,43 @@ +import os +import subprocess +import sys +from pathlib import Path + +from ._server import run_demo + +DEMO_BASE_FOLDER_PATH = Path(__file__).parent + +CLIENT_FILE_PATH = DEMO_BASE_FOLDER_PATH / "_client.html" +SERVER_FILE_PATH = DEMO_BASE_FOLDER_PATH / "_server.py" + + +def _command_line_interface(): + """A simple command line interface for running the demo for TQDM Publisher.""" + if len(sys.argv) <= 1: + print("No input provided. Please specify a command (e.g. 'demo').") + return + + command = sys.argv[1] + if "-" in command: + print( + f"No command provided, but flag {command} was received. " + "Please specify a command before the flag (e.g. 'demo')." + ) + return + + flags_list = sys.argv[2:] + if len(flags_list) > 0: + print(f"No flags are accepted at this time, but flags {flags_list} were received.") + return + + if command == "demo": + # For convenience - automatically pop-up a browser window on the locally hosted HTML page + if sys.platform == "win32": + os.system(f'start "" "{CLIENT_FILE_PATH}"') + else: + subprocess.run(["open", CLIENT_FILE_PATH]) + + run_demo() + + else: + print(f"{command} is an invalid command.") diff --git a/src/tqdm_publisher/_demo/_server.py b/src/tqdm_publisher/_demo/_server.py new file mode 100644 index 0000000..ac821dc --- /dev/null +++ b/src/tqdm_publisher/_demo/_server.py @@ -0,0 +1,74 @@ +import asyncio +import json +import threading +import time + +import websockets + +import tqdm_publisher + + +def start_progress_bar(*, progress_bar_id: str, client_callback: callable) -> None: + """ + Emulate running the specified number of tasks by sleeping the specified amount of time on each iteration. + + Defaults are chosen for a deterministic and regular update period of one second for a total time of one minute. + """ + all_task_durations_in_seconds = [1.0 for _ in range(60)] # One minute at one second per update + progress_bar = tqdm_publisher.TQDMPublisher(iterable=all_task_durations_in_seconds) + + def run_function_on_progress_update(format_dict: dict) -> None: + """ + This is the injected callback that will be run on each update of the TQDM object. + + Its first and only positional argument must be the `format_dict` of the TQDM instance. Additional customization + on outside parameters must be achieved by defining those fields at an outer scope and defining this + server-specific callback inside the local scope. + + In this demo, we will execute the `client_callback` whose protocol is known only to the WebSocketHandler. + """ + client_callback(progress_bar_id=progress_bar_id, format_dict=format_dict) + + progress_bar.subscribe(callback=run_function_on_progress_update) + + for task_duration in progress_bar: + time.sleep(task_duration) + + +async def handler(websocket: websockets.WebSocketServerProtocol) -> None: + """Handle messages from the client and manage the client connections.""" + + def forward_progress_to_client(*, progress_bar_id: str, format_dict: dict) -> None: + """ + This is the function that will run on every update of the TQDM object. + + It will forward the progress to the client. + """ + asyncio.run( + websocket.send(message=json.dumps(obj=dict(progress_bar_id=progress_bar_id, format_dict=format_dict))) + ) + + # Wait for messages from the client + async for message in websocket: + message_from_client = json.loads(message) + + if message_from_client["command"] == "start": + thread = threading.Thread( + target=start_progress_bar, + kwargs=dict( + progress_bar_id=message_from_client["progress_bar_id"], + client_callback=forward_progress_to_client, + ), + ) + thread.start() + + +async def spawn_server() -> None: + """Spawn the server asynchronously.""" + async with websockets.serve(ws_handler=handler, host="", port=8000): + await asyncio.Future() + + +def run_demo() -> None: + """Trigger the execution of the asynchronous spawn.""" + asyncio.run(spawn_server())