Merge pull request #43 from codeguru42/35-worker-queue

35 worker queue

Author: codeguru42

docker-compose-dev.yml

@@ -0,0 +1,29 @@
+services:
+  api:
+    build: .
+    command: "poetry run python3 /api/manage.py runserver 0.0.0.0:8000"
+    env_file:
+      - .env
+    environment:
+      - CELERY_BROKER_URL=amqp://guest:guest@broker:5672
+    ports:
+      - 8000:8000
+    volumes:
+      - .:/api
+  worker:
+    build: .
+    command: "poetry run celery -A go_capture.tasks worker"
+    depends_on:
+      - broker
+    environment:
+      - CELERY_BROKER_URL=amqp://guest:guest@broker:5672
+      - CELERY_RESULT_BACKEND=rpc
+    env_file:
+      - .env
+    volumes:
+      - .:/api
+  broker:
+    image: rabbitmq:3-management
+    ports:
+      - 5672:5672
+      - 15672:15672

docker-compose-prod.yml

@@ -0,0 +1,24 @@
+services:
+  api:
+    image: ${account_id}.dkr.ecr.us-west-1.amazonaws.com/go-capture:${image_hash}
+    env_file:
+      - .env
+    environment:
+      - CELERY_BROKER_URL=amqp://guest:guest@broker:5672
+    ports:
+      - 8000:8000
+  worker:
+    image: ${account_id}.dkr.ecr.us-west-1.amazonaws.com/go-capture:${image_hash}
+    command: "poetry run celery -A go_capture.tasks worker"
+    depends_on:
+      - broker
+    environment:
+      - CELERY_BROKER_URL=amqp://guest:guest@broker:5672
+      - CELERY_RESULT_BACKEND=rpc
+    env_file:
+      - .env
+  broker:
+    image: rabbitmq:3-management
+    ports:
+      - 5672:5672
+      - 15672:15672

docker-compose.yml

@@ -0,0 +1,3 @@
+from .tasks import app as celery_app
+
+__all__ = ('celery_app',)

go_capture/__init__.py

@@ -123,3 +123,9 @@
 # https://docs.djangoproject.com/en/4.1/ref/settings/#default-auto-field
 
 DEFAULT_AUTO_FIELD = 'django.db.models.BigAutoField'
+
+IMAGES_DIR = BASE_DIR / 'images'
+IMAGES_DIR.mkdir(exist_ok=True)
+
+SGF_DIR = BASE_DIR / 'sgf'
+SGF_DIR.mkdir(exist_ok=True)

go_capture/settings.py

@@ -0,0 +1,13 @@
+import cv2
+import numpy as np
+
+from go_capture.sgf import perspective, find_stones
+from go_capture.sgf.make_sgf import make_sgf
+
+
+def process_image(image_file, output_file):
+    image_data = np.asarray(bytearray(image_file.read()), dtype="uint8")
+    image = cv2.imdecode(image_data, cv2.IMREAD_COLOR)
+    board = perspective.get_grid(image)
+    black, white = find_stones.find_stones(board)
+    make_sgf(output_file, black, white)

go_capture/sgf/process_image.py

@@ -0,0 +1,35 @@
+import os
+from pathlib import Path
+
+from celery import Celery
+from django.conf import settings
+
+from go_capture.sgf.process_image import process_image
+
+# Set the default Django settings module for the 'celery' program.
+os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'go_capture.settings')
+
+app = Celery('go_capture')
+
+# Using a string here means the worker doesn't have to serialize
+# the configuration object to child processes.
+# - namespace='CELERY' means all celery-related configuration keys
+#   should have a `CELERY_` prefix.
+app.config_from_object('django.conf:settings', namespace='CELERY')
+
+# Load task modules from all registered Django apps.
+app.autodiscover_tasks()
+
+
+@app.task(bind=True)
+def debug_task(self):
+    print(f'Request: {self.request!r}')
+
+
+@app.task
+def process_image_task(image_filename):
+    sgf_filename = f'{Path(image_filename).stem}.sgf'
+    sgf_path = settings.SGF_DIR / sgf_filename
+    with open(image_filename, 'rb') as image_file:
+        with open(sgf_path, 'w') as sgf_file:
+            process_image(image_file, sgf_file)

go_capture/tasks.py

@@ -21,5 +21,6 @@
 urlpatterns = [
     path('admin/', admin.site.urls),
     path('capture/', views.capture),
+    path('capture_async/', views.capture_async),
     path('health_check/', views.health_check),
 ]

go_capture/urls.py

@@ -1,24 +1,19 @@
 import io
 from pathlib import Path
 
-import cv2
-import numpy as np
-from django.http import FileResponse, JsonResponse
+from django.conf import settings
+from django.http import JsonResponse, HttpResponse, FileResponse
 from django.views.decorators.http import require_POST, require_GET
 
-from go_capture.sgf import perspective, find_stones
-from go_capture.sgf.make_sgf import make_sgf
+from go_capture.sgf.process_image import process_image
+from go_capture.tasks import process_image_task
 
 
 @require_POST
 def capture(request):
     image_file = request.FILES['image']
-    image_data = np.asarray(bytearray(image_file.read()), dtype="uint8")
-    image = cv2.imdecode(image_data, cv2.IMREAD_COLOR)
-    board = perspective.get_grid(image)
-    black, white = find_stones.find_stones(board)
     output_file = io.StringIO()
-    make_sgf(output_file, black, white)
+    process_image(image_file, output_file)
     output_file.seek(0)
     filename = Path(image_file.name).stem
     return FileResponse(
@@ -30,6 +25,17 @@ def capture(request):
     )
 
 
+@require_POST
+def capture_async(request):
+    image_file = request.FILES['image']
+    filename = Path(image_file.name)
+    output_path = settings.IMAGES_DIR / filename
+    with output_path.open('wb') as output_file:
+        output_file.write(image_file.read())
+    process_image_task.delay(str(output_path.absolute()))
+    return HttpResponse(status=201)
+
+
 @require_GET
 def health_check(request):
     body = {