feat: image stitching in celery

.gitignore

@@ -3,4 +3,5 @@
 /.idea
 files/
 
-dump.rdb
+dump.rdb
+.DS_Store

pyproject.toml

@@ -12,6 +12,8 @@ django-rest-framework = "^0.1.0"
 django-polymorphic = "^3.1.0"
 drf-spectacular = "^0.27.2"
 pillow = "^10.3.0"
+opencv-python = "^4.0.0"
+stitching = "^0.6.0"
 requests = "^2.32.3"
 celery = "^5.4.0"
 channels = "^4.1.0"

src/gcom/settings.py

@@ -130,6 +130,8 @@
 
 STATIC_URL = "static/"
 
+MEDIA_ROOT = "media/"
+
 # Default primary key field type
 # https://docs.djangoproject.com/en/5.0/ref/settings/#default-auto-field
 

src/mapping/readme.md

@@ -0,0 +1,3 @@
+This module requires the presence of a docker image called odm.
+To get odm, open docker app and do the following command line.
+`docker pull opendronemap/odm`

src/mapping/tasks.py

@@ -0,0 +1,65 @@
+# get project structure
+from django.conf import settings
+
+# parallelzation
+from celery import shared_task
+
+import subprocess
+import shutil
+import os
+
+
+# pylint: disable=no-member
+def stitch_images_odm(output_path, input_pathes):
+    # that images should be in [data_sets_path]/[project_name]/images
+    data_sets_path = os.path.join(settings.MEDIA_ROOT, "odm_working_dir")
+
+    if os.path.exists(os.path.join(data_sets_path, "default")):
+        shutil.rmtree(os.path.join(data_sets_path, "default"))
+
+    os.makedirs(data_sets_path, exist_ok=True)
+    os.makedirs(os.path.join(data_sets_path, "default"), exist_ok=False)
+    os.makedirs(os.path.join(data_sets_path, "default", "images"), exist_ok=False)
+
+    for i, img_path in enumerate(input_pathes):
+        shutil.copy(
+            img_path,
+            os.path.join(data_sets_path, "default", "images", f"{i}.png"),
+        )
+
+    subprocess.Popen(
+        f'docker run -ti --rm -v "$(pwd)/{data_sets_path}":/datasets opendronemap/odm --project-path /datasets default --orthophoto-resolution 1 --orthophoto-png --skip-3dmodel --skip-report',
+        shell=True,
+        stdout=subprocess.DEVNULL,
+    ).wait()
+
+    generated_png_path = os.path.join(
+        data_sets_path, "default", "odm_orthophoto", "odm_orthophoto.png"
+    )
+
+    shutil.copy(
+        generated_png_path,
+        output_path,
+    )
+    return True
+
+
+@shared_task
+def stitch_images(output_path, input_pathes):
+    """
+    Stitch images at input path and save it at the out put path
+
+    Args:
+        output_path: This is the address that the stitched image would be written to.
+        input_pathes: these are where the images would be read from.
+
+    Returns:
+        A boolean about whether image stitching worked without any Issue.
+    """
+
+    success = stitch_images_odm(
+        output_path,
+        input_pathes,
+    )
+
+    return success

src/mapping/tests.py

@@ -1,7 +1,11 @@
 from django.test import TestCase
 from .serializers import AreaOfInterestSerializer
+from .tasks import stitch_images
 from rest_framework.test import APITestCase
 import json
+import os
+import subprocess
+from django.conf import settings
 
 
 class AreaOfInterestValidationTest(TestCase):
@@ -182,3 +186,72 @@ def test_post_n_get_area_of_interest(self):
         self.assertFalse("altitude" in returned_object["area_of_interest"][1])
         self.assertFalse("altitude" in returned_object["area_of_interest"][2])
         self.assertFalse("altitude" in returned_object["area_of_interest"][3])
+
+
+class StitchTest(TestCase):
+    # ./manage.py test mapping.tests.StitchTest --parallel
+    def stitch_test_helper(self, output_file, input_directory):
+
+        # clean up old test output
+        if os.path.exists(output_file):
+            os.remove(output_file)
+            self.assertFalse(os.path.exists(output_file), "file deletion unsuccessful")
+
+        input_filenames = [
+            os.path.join(input_directory, f)
+            for f in os.listdir(input_directory)
+            if os.path.isfile(os.path.join(input_directory, f))
+            and ("png" in f or "jpg" in f)
+        ]
+
+        # synchronous test of asynchronous function.
+        # source https://celery.school/unit-testing-celery-tasks
+        task_succeed = stitch_images.s(output_file, input_filenames).apply()
+
+        # to call it synchronously
+        # future = stitch_images.delay(output_file, input_filenames)
+        # the piece of information that need to be saved.
+        # task_id = future.id
+        # to get the result back out
+        # task_succeed = AsyncResult(id=task_id).get(timeout=10)
+
+        self.assertTrue(
+            task_succeed, "stiching algorithum failed on image set previously tested"
+        )
+        self.assertTrue(
+            os.path.exists(output_file),
+            "task is claimed to be successful yet no file is created",
+        )
+
+    def test_dependency_present(self):
+        present = subprocess.check_output(
+            "docker manifest inspect opendronemap/odm > /dev/null 2>&1 && echo 1 || echo 0",
+            shell=True,
+        )
+        present = int(present)
+        self.assertTrue(
+            present == 1,
+            "odm image is not present, run `docker pull opendronemap/odm` to get it",
+        )
+
+    def test_parking_lot_zoomed_in(self):
+
+        input_directory = os.path.join(
+            settings.MEDIA_ROOT, "test", "zoomed_in", "input"
+        )
+        output_file = os.path.join(
+            settings.MEDIA_ROOT, "test", "zoomed_in", "output.png"
+        )
+
+        self.stitch_test_helper(output_file, input_directory)
+
+    def test_parking_lot_zoomed_out(self):
+
+        input_directory = os.path.join(
+            settings.MEDIA_ROOT, "test", "zoomed_out", "input"
+        )
+        output_file = os.path.join(
+            settings.MEDIA_ROOT, "test", "zoomed_out", "output.png"
+        )
+
+        self.stitch_test_helper(output_file, input_directory)

src/media/odm_working_dir/.gitignore

@@ -0,0 +1,2 @@
+*
+!.gitignore

src/media/test/zoomed_in/.gitignore

@@ -0,0 +1 @@
+output.png

src/media/test/zoomed_out/.gitignore

@@ -0,0 +1 @@
+output.png