Add caption-image app

.github/workflows/main.yml

@@ -120,6 +120,7 @@ jobs:
           sql-server,
           label-studio,
           ocr-extraction,
+          caption-image,
           dataset-ingestion-movies
         ]
     runs-on: ubuntu-latest

apps/caption-image/Dockerfile

@@ -0,0 +1,15 @@
+# Pull base image.
+ARG VERSION=latest
+FROM aperturedata/workflows-base:${VERSION}
+
+ENV APP_NAME=workflows-caption-image
+
+COPY requirements.txt /
+RUN pip install -U pip
+RUN pip install --no-cache-dir -r /requirements.txt
+
+COPY warmup_validate.py /warmup_validate.py
+ARG PRELOAD_MODEL=false
+RUN if [ "$PRELOAD_MODEL" = "true" ]; then python /warmup_validate.py; fi && rm /warmup_validate.py
+
+COPY app /app/

apps/caption-image/README.md

@@ -0,0 +1,66 @@
+# Caption Image App
+
+This workflow retrieves all images from ApertureDB that have not been
+analyzed before, and runs them through a
+[BLIP (Bootstrapping Language-Image Pre-training)](https://github.com/salesforce/BLIP)
+model to generate a caption for each image.
+
+The workflow runs continuously by default, periodically checking for and processing new uncaptioned images. To run it only once, set `RUN_ONCE=true`.
+
+## Database details
+
+```mermaid
+sequenceDiagram
+    participant W as Caption Image
+    participant A as ApertureDB instance
+
+    W->>A: FindImage
+    A-->>W: count
+    loop Until done
+        W->>A: FindImage
+        A-->>W: images
+        W->>A: UpdateImage
+    end
+```
+
+Each image is updated with a caption property (`wf_caption_image`) containing the generated caption text. The BLIP model processes each image to generate descriptive text that describes the visual content of the image.
+
+## Running in Docker
+
+```
+docker run \
+           -e RUN_NAME=my_testing_run \
+           -e DB_HOST=workflowstesting.gcp.cloud.aperturedata.dev \
+           -e DB_PASS="password" \
+           -e NUM_WORKERS=4 \
+           -e BATCH_SIZE=32 \
+           -e WF_LOG_LEVEL=INFO \
+           aperturedata/workflows-caption-image
+```
+
+Parameters:
+* **`NUM_WORKERS`**: Specifies the number of worker threads that will be running simultaneously,
+retrieving and processing images in parallel. Default is `1`. Note that the BLIP model inference is serialized with a lock to prevent PyTorch intra-op threading conflicts on CPU, so increasing `NUM_WORKERS` only parallelizes image fetching and preprocessing, not inference itself.
+* **`BATCH_SIZE`**: Specifies the batch size for processing images. Default is `1`.
+* **`WF_LOG_LEVEL`**: Set log level for workflow code. Available options: DEBUG, INFO, WARNING, ERROR. Default is `WARNING`. `LOG_LEVEL` is also supported as a legacy alias.
+
+See [Common Parameters](../../README.md#common-parameters) for common parameters.
+
+## Cleaning up
+
+To clean all captions generated by this workflow, simply run the following query:
+
+```
+q = [
+        {
+            "UpdateImage": {
+                "constraints": {
+                    "wf_caption_image": ["!=", None]
+                },
+                "remove_props": ["wf_caption_image", "wf_caption_image_done", "wf_caption_image_failed", "wf_caption_image_error"]
+            }
+        }
+    ]
+```
+
+or manually remove the `wf_caption_image`, `wf_caption_image_done`, `wf_caption_image_failed`, and `wf_caption_image_error` properties from images that have been processed.

apps/caption-image/app/app.sh

@@ -0,0 +1,17 @@
+#!/bin/bash
+set -e
+
+SLEEPING_TIME=$(/app/wf_argparse.py --type non_negative_int --envar SLEEPING_TIME --default 30)
+RUN_ONCE=$(/app/wf_argparse.py --type bool --envar RUN_ONCE --default false)
+
+python3 status_tools.py --completed 0 --phases processing --phases sleeping --phase processing
+while true; do
+    python3 status_tools.py --completed 0 --phase processing
+    python3 log_processor.py "python3 caption_images.py"
+
+    if [ "$RUN_ONCE" = "true" ]; then
+        break
+    fi
+    python3 status_tools.py --completed 0 --phase sleeping
+    sleep $SLEEPING_TIME
+done

apps/caption-image/app/caption_images.py

@@ -0,0 +1,42 @@
+import logging
+
+import typer
+
+from images import FindImageQueryGenerator
+from aperturedb import ParallelQuery
+from connection_pool import ConnectionPool
+
+CAPTION_IMAGE_PROPERTY = 'wf_caption_image'
+
+def caption_images(
+    num_workers: int = typer.Option(1, envvar="NUM_WORKERS", help="Number of concurrent workers"),
+    batch_size: int = typer.Option(1, envvar="BATCH_SIZE", help="Batch size for fetching images"),
+    log_level: str = typer.Option("WARNING", envvar=["WF_LOG_LEVEL", "LOG_LEVEL"], help="Logging level")
+):
+    num_workers = int(num_workers)
+    if num_workers <= 0:
+        raise ValueError("num_workers must be > 0")
+
+    batch_size = int(batch_size)
+    if batch_size <= 0:
+        raise ValueError("batch_size must be > 0")
+
+    logging.basicConfig(level=log_level.upper(), force=True)
+    logger = logging.getLogger(__name__)
+    pool = ConnectionPool()
+    data = FindImageQueryGenerator(
+        pool,
+        batch_size=batch_size,
+        caption_image_property=CAPTION_IMAGE_PROPERTY)
+
+    logger.info("Running Caption Image...")
+    with pool.get_connection() as db:
+        querier = ParallelQuery.ParallelQuery(db)
+        querier.query(data, batchsize=1, numthreads=num_workers, stats=True)
+
+
+def main():
+    typer.run(caption_images)
+
+if __name__ == "__main__":
+    main()