working

Author: antbaez9

.DS_Store

@@ -0,0 +1,24 @@
+FROM --platform=linux/x86_64 ubuntu:24.04
+
+ENV DEBIAN_FRONTEND=noninteractive
+ENV TZ=UTC
+
+RUN apt-get update && \
+    apt-get install -y \
+    wget \
+    xz-utils \
+    bzip2 \
+    git \
+    python3-pip \
+    python3 \
+    && apt-get install -y software-properties-common \
+    && apt-get clean \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt .
+
+RUN pip install -r requirements.txt --break-system-packages
+
+COPY filtered/ ./filtered/
+
+CMD ["python3.11", "-m", "celery", "-A", "filtered.worker", "worker", "--loglevel=info", "--concurrency=1"]

Dockerfile

@@ -0,0 +1,4 @@
+from .distributed import *
+from .filter import *
+from .split import *
+from .worker import *

filtered/__init__.py

@@ -0,0 +1,74 @@
+
+from distributaur.distributaur import create_from_config
+from .filter import read_json_in_batches
+from .worker import run_job
+from tqdm import tqdm
+import time
+
+
+if __name__ == "__main__":
+
+    input_filename = 'datasets/cap3d_captions.json'
+    batch_size = 1000
+
+    distributaur = create_from_config()
+
+    max_price = 0.1
+    max_nodes = 50
+    docker_image = "antbaez/filter-worker:latest"
+    module_name = "filtered.worker"
+
+    redis_client = distributaur.get_redis_connection()
+
+    rented_nodes = distributaur.rent_nodes(max_price, max_nodes, docker_image, module_name)
+    print("Total nodes rented: ", len(rented_nodes))
+
+    distributaur.register_function(run_job)
+
+    while True:
+        user_input = input("press r when workers are ready: ")
+        if user_input == "r":
+            break
+
+
+    total_batches = 0
+    
+    print("Sending tasks")
+    tasks = []
+
+    json_batches = [batch for batch in read_json_in_batches(input_filename, batch_size)]
+    print(f"number of batches: {len(json_batches)}")
+
+    num_batches = len(json_batches)
+    for i in range(num_batches):
+
+        batch = json_batches[i]
+
+        total_batches += 1
+        task = distributaur.execute_function("run_job", {
+            "batch_index" : total_batches,
+            "batch" : batch
+        })
+
+        tasks.append(task)
+
+    first_task_done = False
+    print("Tasks sent. Starting monitoring")
+    with tqdm(total=len(tasks), unit="task") as pbar:
+        while not all(task.ready() for task in tasks):
+            current_tasks = sum([task.ready() for task in tasks])
+            pbar.update(current_tasks - pbar.n)
+            if current_tasks > 0:
+                if not first_task_done:
+                    first_task_done = True
+                    first_task_start_time = time.time()
+
+                end_time = time.time()
+                elapsed_time = end_time - first_task_start_time
+                time_per_tasks = elapsed_time / current_tasks
+                time_left = time_per_tasks * (len(tasks) - current_tasks)
+
+                pbar.set_postfix(
+                    elapsed=f"{elapsed_time:.2f}s", time_left=f"{time_left:.2f}"
+                )
+            time.sleep(2)

filtered/__pycache__/__init__.cpython-310.pyc

@@ -18,7 +18,7 @@ def detect_objects(self, caption):
         if not objects:
             objects = self._extract_noun_phrases(caption)
 
-        print("These are the objects:", objects)
+        # print("These are the objects:", objects)
         return objects
 
     def _extract_noun_phrases(self, text):
@@ -147,12 +147,12 @@ def test_caption_filtering():
 
         if total_filtered_count >= write_batch_size or current_batch == total_batches:
             write_filtered_json(output_filename, filtered_data, first_batch=first_batch, last_batch=(current_batch == total_batches))
-            print(f"Wrote batch {current_batch}/{total_batches} with {total_filtered_count} filtered captions")
+            # print(f"Wrote batch {current_batch}/{total_batches} with {total_filtered_count} filtered captions")
             filtered_data = {}
             total_filtered_count = 0
             first_batch = False
 
-    print("Filtering and writing completed.")
+    # print("Filtering and writing completed.")
 
     # Optionally, you can keep the test function call if you want to run tests
     # test_caption_filtering()

filtered/__pycache__/distributed.cpython-310.pyc

@@ -0,0 +1,23 @@
+import sys
+from .filter import filter_captions, write_filtered_json
+
+
+def run_job(batch_index, batch):
+
+    output_filename = f"batch_{batch_index}"
+
+    filtered_batch = filter_captions(batch)
+    write_filtered_json(output_filename, filtered_batch)
+
+    distributaur.upload_file(output_filename)
+
+    return "Task complete"
+
+
+if __name__ == "__main__" or any("celery" in arg for arg in sys.argv):
+    from distributaur.distributaur import create_from_config
+
+    distributaur = create_from_config()
+    distributaur.register_function(run_job)
+
+    celery = distributaur.app

filtered/__pycache__/filter.cpython-310.pyc

@@ -0,0 +1,10 @@
+requests
+fsspec
+celery
+redis
+huggingface_hub
+python-dotenv
+omegaconf
+tqdm
+gliner
+distributaur