Merge pull request #235 from hotosm/develop

Production release

.pre-commit-config.yaml

@@ -87,7 +87,7 @@ repos:
   # Lint / autoformat: Python code
   - repo: https://github.com/astral-sh/ruff-pre-commit
     # Ruff version.
-    rev: "v0.6.4"
+    rev: "v0.6.7"
     hooks:
       # Run the linter
       - id: ruff

src/backend/app/config.py

@@ -105,6 +105,8 @@ def assemble_db_connection(cls, v: Optional[str], info: ValidationInfo) -> Any:
     EMAILS_FROM_EMAIL: Optional[EmailStr] = None
     EMAILS_FROM_NAME: Optional[str] = "Drone Tasking Manager"
 
+    NODE_ODM_URL: Optional[str] = "http://odm-api:3000"
+
     @computed_field
     @property
     def emails_enabled(self) -> bool:

src/backend/app/db/db_models.py

@@ -74,6 +74,9 @@ class DbTask(Base):
     )
     project_task_index = cast(int, Column(Integer))
     outline = cast(WKBElement, Column(Geometry("POLYGON", srid=4326)))
+    take_off_point = cast(
+        WKBElement, Column(Geometry("POINT", srid=4326), nullable=True)
+    )
 
 
 class DbProject(Base):

src/backend/app/drones/drone_schemas.py

@@ -35,7 +35,6 @@ class DbDrone(BaseDrone):
     @staticmethod
     async def one(db: Connection, drone_id: int):
         """Get a single drone by it's ID"""
-        print("drone_id = ", drone_id)
         async with db.cursor(row_factory=class_row(DbDrone)) as cur:
             await cur.execute(
                 """

src/backend/app/migrations/versions/aec0d408df01_added_take_off_point_in_task.py

@@ -0,0 +1,43 @@
+"""added take_off_point in task
+
+Revision ID: aec0d408df01
+Revises: 2b92f8a9bbec
+Create Date: 2024-09-24 03:57:03.760365
+
+"""
+
+import geoalchemy2
+import sqlalchemy as sa
+from alembic import op
+from typing import Sequence, Union
+
+
+# revision identifiers, used by Alembic.
+revision: str = "aec0d408df01"
+down_revision: Union[str, None] = "2b92f8a9bbec"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column(
+        "tasks",
+        sa.Column(
+            "take_off_point",
+            geoalchemy2.types.Geometry(
+                geometry_type="POINT",
+                srid=4326,
+                from_text="ST_GeomFromEWKT",
+                name="geometry",
+            ),
+            nullable=True,
+        ),
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column("tasks", "take_off_point")
+    # ### end Alembic commands ###

src/backend/app/projects/image_processing.py

@@ -0,0 +1,155 @@
+import uuid
+import tempfile
+import shutil
+from pathlib import Path
+from pyodm import Node
+from app.s3 import get_file_from_bucket, list_objects_from_bucket, add_file_to_bucket
+from loguru import logger as log
+from concurrent.futures import ThreadPoolExecutor
+
+
+class DroneImageProcessor:
+    def __init__(
+        self,
+        node_odm_url: str,
+        project_id: uuid.UUID,
+        task_id: uuid.UUID,
+    ):
+        """
+        Initializes the connection to the ODM node.
+        """
+        # self.node = Node(node_odm_host, node_odm_port)
+        self.node = Node.from_url(node_odm_url)
+        self.project_id = project_id
+        self.task_id = task_id
+
+    def options_list_to_dict(self, options=[]):
+        """
+        Converts options formatted as a list ([{'name': optionName, 'value': optionValue}, ...])
+        to a dictionary {optionName: optionValue, ...}
+        """
+        opts = {}
+        if options is not None:
+            for o in options:
+                opts[o["name"]] = o["value"]
+        return opts
+
+    def download_object(self, bucket_name: str, obj, images_folder: str):
+        if obj.object_name.endswith((".jpg", ".jpeg", ".JPG", ".png", ".PNG")):
+            local_path = Path(images_folder) / Path(obj.object_name).name
+            local_path.parent.mkdir(parents=True, exist_ok=True)
+            get_file_from_bucket(bucket_name, obj.object_name, local_path)
+
+    def download_images_from_s3(self, bucket_name, local_dir):
+        """
+        Downloads images from MinIO under the specified path.
+
+        :param bucket_name: Name of the MinIO bucket.
+        :param project_id: The project UUID.
+        :param task_id: The task UUID.
+        :param local_dir: Local directory to save the images.
+        :return: List of local image file paths.
+        """
+        prefix = f"projects/{self.project_id}/{self.task_id}"
+
+        objects = list_objects_from_bucket(bucket_name, prefix)
+
+        # Process images concurrently
+        with ThreadPoolExecutor() as executor:
+            executor.map(
+                lambda obj: self.download_object(bucket_name, obj, local_dir),
+                objects,
+            )
+
+    def list_images(self, directory):
+        """
+        Lists all images in the specified directory.
+
+        :param directory: The directory containing the images.
+        :return: List of image file paths.
+        """
+        images = []
+        path = Path(directory)
+
+        for file in path.rglob("*"):
+            if file.suffix.lower() in {".jpg", ".jpeg", ".png"}:
+                images.append(str(file))
+        return images
+
+    def process_new_task(self, images, name=None, options=[], progress_callback=None):
+        """
+        Sends a set of images via the API to start processing.
+
+        :param images: List of image file paths.
+        :param name: Name of the task.
+        :param options: Processing options ([{'name': optionName, 'value': optionValue}, ...]).
+        :param progress_callback: Callback function to report upload progress.
+        :return: The created task object.
+        """
+        opts = self.options_list_to_dict(options)
+
+        # FIXME: take this from the function above
+        opts = {"dsm": True}
+
+        task = self.node.create_task(images, opts, name, progress_callback)
+        return task
+
+    def monitor_task(self, task):
+        """
+        Monitors the task progress until completion.
+
+        :param task: The task object.
+        """
+        log.info(f"Monitoring task {task.uuid}...")
+        task.wait_for_completion(interval=5)
+        log.info("Task completed.")
+        return task
+
+    def download_results(self, task, output_path):
+        """
+        Downloads all results of the task to the specified output path.
+
+        :param task: The task object.
+        :param output_path: The directory where results will be saved.
+        """
+        log.info(f"Downloading results to {output_path}...")
+        path = task.download_zip(output_path)
+        log.info("Download completed.")
+        return path
+
+    def process_images_from_s3(self, bucket_name, name=None, options=[]):
+        """
+        Processes images from MinIO storage.
+
+        :param bucket_name: Name of the MinIO bucket.
+        :param project_id: The project UUID.
+        :param task_id: The task UUID.
+        :param name: Name of the task.
+        :param options: Processing options ([{'name': optionName, 'value': optionValue}, ...]).
+        :return: The task object.
+        """
+        # Create a temporary directory to store downloaded images
+        temp_dir = tempfile.mkdtemp()
+        try:
+            self.download_images_from_s3(bucket_name, temp_dir)
+
+            images_list = self.list_images(temp_dir)
+
+            # Start a new processing task
+            task = self.process_new_task(images_list, name=name, options=options)
+            # Monitor task progress
+            self.monitor_task(task)
+
+            # Optionally, download results
+            output_file_path = f"/tmp/{self.project_id}"
+            path_to_download = self.download_results(task, output_path=output_file_path)
+
+            # Upload the results into s3
+            s3_path = f"projects/{self.project_id}/{self.task_id}/assets.zip"
+            add_file_to_bucket(bucket_name, path_to_download, s3_path)
+            return task
+
+        finally:
+            # Clean up temporary directory
+            shutil.rmtree(temp_dir)
+            pass

src/backend/app/projects/project_logic.py

@@ -2,15 +2,23 @@
 import uuid
 from loguru import logger as log
 from fastapi import HTTPException, UploadFile
-from app.tasks.splitter import split_by_square
+from app.tasks.task_splitter import split_by_square
 from fastapi.concurrency import run_in_threadpool
 from psycopg import Connection
 from app.utils import merge_multipolygon
 import shapely.wkb as wkblib
 from shapely.geometry import shape
 from io import BytesIO
-from app.s3 import add_obj_to_bucket
+from app.s3 import (
+    add_obj_to_bucket,
+    list_objects_from_bucket,
+    get_presigned_url,
+    get_object_metadata,
+)
 from app.config import settings
+from app.projects.image_processing import DroneImageProcessor
+from app.projects import project_schemas
+from minio import S3Error
 
 
 async def upload_file_to_s3(
@@ -155,3 +163,68 @@ async def preview_split_by_square(boundary: str, meters: int):
             meters=meters,
         )
     )
+
+
+def process_drone_images(project_id: uuid.UUID, task_id: uuid.UUID):
+    # Initialize the processor
+    processor = DroneImageProcessor(settings.NODE_ODM_URL, project_id, task_id)
+
+    # Define processing options
+    options = [
+        {"name": "dsm", "value": True},
+        {"name": "orthophoto-resolution", "value": 5},
+    ]
+
+    processor.process_images_from_s3(
+        settings.S3_BUCKET_NAME, name=f"DTM-Task-{task_id}", options=options
+    )
+
+
+async def get_project_info_from_s3(project_id: uuid.UUID, task_id: uuid.UUID):
+    """
+    Helper function to get the number of images and the URL to download the assets.
+    """
+    try:
+        # Prefix for the images
+        images_prefix = f"projects/{project_id}/{task_id}/images/"
+
+        # List and count the images
+        objects = list_objects_from_bucket(
+            settings.S3_BUCKET_NAME, prefix=images_prefix
+        )
+        image_extensions = (".jpg", ".jpeg", ".png", ".tif", ".tiff")
+        image_count = sum(
+            1 for obj in objects if obj.object_name.lower().endswith(image_extensions)
+        )
+
+        # Generate a presigned URL for the assets ZIP file
+        try:
+            # Check if the object exists
+            assets_path = f"projects/{project_id}/{task_id}/assets.zip"
+            get_object_metadata(settings.S3_BUCKET_NAME, assets_path)
+
+            # If it exists, generate the presigned URL
+            presigned_url = get_presigned_url(
+                settings.S3_BUCKET_NAME, assets_path, expires=2
+            )
+        except S3Error as e:
+            if e.code == "NoSuchKey":
+                # The object does not exist
+                log.info(
+                    f"Assets ZIP file not found for project {project_id}, task {task_id}."
+                )
+                presigned_url = None
+            else:
+                # An unexpected error occurred
+                log.error(f"An error occurred while accessing assets file: {e}")
+                raise HTTPException(status_code=500, detail=str(e))
+
+        return project_schemas.AssetsInfo(
+            project_id=str(project_id),
+            task_id=str(task_id),
+            image_count=image_count,
+            assets_url=presigned_url,
+        )
+    except Exception as e:
+        log.exception(f"An error occurred while retrieving assets info: {e}")
+        raise HTTPException(status_code=500, detail=str(e))

src/backend/app/projects/project_routes.py

@@ -1,4 +1,5 @@
 import os
+import uuid
 from typing import Annotated, Optional
 from uuid import UUID
 import geojson
@@ -13,6 +14,7 @@
     File,
     Form,
     Response,
+    BackgroundTasks,
 )
 from geojson_pydantic import FeatureCollection
 from loguru import logger as log
@@ -28,6 +30,7 @@
 from app.users.user_schemas import AuthUser
 from app.tasks import task_schemas
 
+
 router = APIRouter(
     prefix=f"{settings.API_PREFIX}/projects",
     responses={404: {"description": "Not found"}},
@@ -295,3 +298,33 @@ async def read_project(
 ):
     """Get a specific project and all associated tasks by ID."""
     return project
+
+
+@router.post("/process_imagery/{project_id}/{task_id}/", tags=["Image Processing"])
+async def process_imagery(
+    task_id: uuid.UUID,
+    project: Annotated[
+        project_schemas.DbProject, Depends(project_deps.get_project_by_id)
+    ],
+    user_data: Annotated[AuthUser, Depends(login_required)],
+    background_tasks: BackgroundTasks,
+):
+    background_tasks.add_task(project_logic.process_drone_images, project.id, task_id)
+    return {"message": "Processing started"}
+
+
+@router.get(
+    "/assets/{project_id}/{task_id}/",
+    tags=["Image Processing"],
+    response_model=project_schemas.AssetsInfo,
+)
+async def get_assets_info(
+    project: Annotated[
+        project_schemas.DbProject, Depends(project_deps.get_project_by_id)
+    ],
+    task_id: uuid.UUID,
+):
+    """
+    Endpoint to get the number of images and the URL to download the assets for a given project and task.
+    """
+    return await project_logic.get_project_info_from_s3(project.id, task_id)

src/backend/app/projects/project_schemas.py

@@ -457,3 +457,10 @@ class PresignedUrlRequest(BaseModel):
     task_id: uuid.UUID
     image_name: List[str]
     expiry: int  # Expiry time in hours
+
+
+class AssetsInfo(BaseModel):
+    project_id: str
+    task_id: str
+    image_count: int
+    assets_url: Optional[str]