[feat] re-add samgeo with segment-anything-fast

README.md

@@ -9,6 +9,9 @@ docker stop $(docker ps -a -q); docker rm $(docker ps -a -q)
 # build the base docker image with the docker aws repository tag
 docker build . -f dockerfiles/dockerfile-lambda-gdal-runner --tag 686901913580.dkr.ecr.eu-west-1.amazonaws.com/lambda-gdal-runner
 
+# OPTIONAL: to build the lambda-gdal-runner image on a x86 machine use the build arg `RIE="https://github.com/aws/aws-lambda-runtime-interface-emulator/releases/latest/download/aws-lambda-rie"`:
+docker build . -f dockerfiles/dockerfile-lambda-gdal-runner --build-arg RIE="https://github.com/aws/aws-lambda-runtime-interface-emulator/releases/latest/download/aws-lambda-rie" --tag 686901913580.dkr.ecr.eu-west-1.amazonaws.com/lambda-gdal-runner --progress=plain
+
 # build the final docker image
 docker build . -f dockerfiles/dockerfile-lambda-samgeo-api --tag 686901913580.dkr.ecr.eu-west-1.amazonaws.com/lambda-samgeo-api
 ```

dockerfiles/dockerfile-lambda-gdal-runner

@@ -5,12 +5,17 @@ ARG LAMBDA_TASK_ROOT="/var/task"
 ARG PYTHONPATH="${LAMBDA_TASK_ROOT}:${PYTHONPATH}:/usr/local/lib/python3/dist-packages"
 ARG RIE="https://github.com/aws/aws-lambda-runtime-interface-emulator/releases/latest/download/aws-lambda-rie-arm64"
 
+RUN echo "ENV RIE: $RIE ..."
+
 # Set working directory to function root directory
 WORKDIR ${LAMBDA_TASK_ROOT}
-COPY requirements_dev.txt ${LAMBDA_TASK_ROOT}/requirements_dev.txt
+COPY requirements.txt ${LAMBDA_TASK_ROOT}/requirements.txt
 
 RUN apt update && apt install -y curl python3-pip
-RUN python -m pip install -r ${LAMBDA_TASK_ROOT}/requirements_dev.txt --target ${LAMBDA_TASK_ROOT}
+RUN which python
+RUN python --version
+RUN python -m pip install torch torchvision --index-url https://download.pytorch.org/whl/cpu
+RUN python -m pip install -r ${LAMBDA_TASK_ROOT}/requirements.txt --target ${LAMBDA_TASK_ROOT}
 
 RUN curl -Lo /usr/local/bin/aws-lambda-rie ${RIE}
 RUN chmod +x /usr/local/bin/aws-lambda-rie

events/example_output.json

@@ -0,0 +1,19 @@
+{
+  "type": "FeatureCollection",
+  "features": [
+    {
+      "type": "Feature",
+      "geometry": {
+        "type": "Polygon",
+        "coordinates": [
+          [
+            [46.143, 9.361],
+            [46.151, 9.401],
+            [46.137, 9.353]
+          ]
+        ]
+      },
+      "properties": {"name": "Colico"}
+    }
+  ]
+}

requirements.txt

@@ -1,4 +1,7 @@
+awslambdaric
 aws-lambda-powertools
 bson
+geojson-pydantic
 python-dotenv
+segment-anything-fast
 segment-geospatial

requirements_dev.txt

@@ -1,6 +1,7 @@
 awslambdaric
 aws_lambda_powertools
 fastjsonschema
+geojson-pydantic
 jmespath
 pydantic
 requests

src/__init__.py

@@ -0,0 +1,4 @@
+from aws_lambda_powertools import Logger
+
+
+app_logger = Logger()

src/app.py

@@ -1,28 +1,42 @@
 import json
 import time
 from http import HTTPStatus
-from aws_lambda_powertools import Logger
+from typing import Dict, List
+
 from aws_lambda_powertools.event_handler import content_types
 from aws_lambda_powertools.utilities.typing import LambdaContext
+from geojson_pydantic import FeatureCollection, Feature, Polygon
 from pydantic import BaseModel, ValidationError
 
-from src.utilities.constants import CUSTOM_RESPONSE_MESSAGES
-from src.utilities.type_hints import input_floatlist, input_floatlist2
+from src import app_logger
+from src.prediction_api.predictor import base_predict
+from src.utilities.constants import CUSTOM_RESPONSE_MESSAGES, MODEL_NAME, ZOOM
 from src.utilities.utilities import base64_decode
 
+PolygonFeatureCollectionModel = FeatureCollection[Feature[Polygon, Dict]]
+
+
+class LatLngTupleLeaflet(BaseModel):
+    lat: float
+    lng: float
+
 
-logger = Logger()
+class RequestBody(BaseModel):
+    ne: LatLngTupleLeaflet
+    sw: LatLngTupleLeaflet
+    points: List[LatLngTupleLeaflet]
+    model: str = MODEL_NAME
+    zoom: float = ZOOM
 
 
-class BBoxWithPointInput(BaseModel):
-    bbox: input_floatlist
-    points: input_floatlist2
-    duration_run: float = 0
-    message: str = ""
-    request_id: str = ""
+class ResponseBody(BaseModel):
+    geojson: Dict = None
+    request_id: str
+    duration_run: float
+    message: str
 
 
-def get_response(status: int, start_time: float, request_id: str, output: BBoxWithPointInput = None) -> str:
+def get_response(status: int, start_time: float, request_id: str, response_body: ResponseBody = None) -> str:
     """
     Return a response for frontend clients.
 
@@ -30,68 +44,80 @@ def get_response(status: int, start_time: float, request_id: str, output: BBoxWi
         status: status response
         start_time: request start time (float)
         request_id: str
-        output: dict we embed into our response
+        response_body: dict we embed into our response
 
     Returns:
-        dict: response
+        str: json response
 
     """
-    duration_run = time.time() - start_time
-    body = f"{CUSTOM_RESPONSE_MESSAGES[status]} - duration_run: {duration_run}, request_id: {request_id}."
-    if output is not None and status == 200:
-        output.duration_run = duration_run
-        output.message = CUSTOM_RESPONSE_MESSAGES[status]
-        output.request_id = request_id
-        body = output.model_dump_json()
-    elif status == 200:
-        # should never be here...
-        raise KeyError("status 200, but missing BBoxWithPointInput argument.")
+    response_body.duration_run = time.time() - start_time
+    response_body.message = CUSTOM_RESPONSE_MESSAGES[status]
+    response_body.request_id = request_id
+
     response = {
         "statusCode": status,
-        "header": {"Content-Type": content_types.APPLICATION_JSON if status == 200 else content_types.TEXT_PLAIN},
-        "body": body,
+        "header": {"Content-Type": content_types.APPLICATION_JSON},
+        "body": response_body.model_dump_json(),
         "isBase64Encoded": False
     }
-    logger.info(f"response type:{type(response)} => {response}.")
+    app_logger.info(f"response type:{type(response)} => {response}.")
     return json.dumps(response)
 
 
+def get_parsed_bbox_points(request_input: RequestBody) -> Dict:
+    return {
+        "bbox": [
+            request_input.ne.lat, request_input.sw.lat,
+            request_input.ne.lng, request_input.sw.lng
+        ],
+        "points": [[p.lat, p.lng] for p in request_input.points]
+    }
+
+
 def lambda_handler(event: dict, context: LambdaContext):
-    logger.info(f"start with aws_request_id:{context.aws_request_id}.")
+    app_logger.info(f"start with aws_request_id:{context.aws_request_id}.")
     start_time = time.time()
 
     if "version" in event:
-        logger.info(f"event version: {event['version']}.")
+        app_logger.info(f"event version: {event['version']}.")
 
     try:
-        logger.info(f"event:{json.dumps(event)}...")
-        logger.info(f"context:{context}...")
+        app_logger.info(f"event:{json.dumps(event)}...")
+        app_logger.info(f"context:{context}...")
 
         try:
             body = event["body"]
         except Exception as e_constants1:
-            logger.error(f"e_constants1:{e_constants1}.")
+            app_logger.error(f"e_constants1:{e_constants1}.")
             body = event
 
-        logger.info(f"body: {type(body)}, {body}...")
+        app_logger.info(f"body: {type(body)}, {body}...")
 
         if isinstance(body, str):
             body_decoded_str = base64_decode(body)
-            logger.info(f"body_decoded_str: {type(body_decoded_str)}, {body_decoded_str}...")
+            app_logger.info(f"body_decoded_str: {type(body_decoded_str)}, {body_decoded_str}...")
             body = json.loads(body_decoded_str)
 
-        logger.info(f"body:{body}...")
+        app_logger.info(f"body:{body}...")
 
         try:
-            bbox_points = BBoxWithPointInput(bbox=body["bbox"], points=body["points"])
-            logger.info(f"validation ok, bbox_points:{bbox_points}...")
-            response = get_response(HTTPStatus.OK.value, start_time, context.aws_request_id, bbox_points)
+            model_name = body["model"] if "model" in body else MODEL_NAME
+            zoom = body["zoom"] if "zoom" in body else ZOOM
+            body_request_validated = RequestBody(ne=body["ne"], sw=body["sw"], points=body["points"], model=model_name, zoom=zoom)
+            body_request = get_parsed_bbox_points(body_request_validated)
+            app_logger.info(f"validation ok - body_request:{body_request}, starting prediction...")
+            output_geojson_dict = base_predict(bbox=body_request["bbox"], model_name=body_request_validated["model"], zoom=body_request_validated["zoom"])
+
+            # raise ValidationError in case this is not a valid geojson by GeoJSON specification rfc7946
+            PolygonFeatureCollectionModel(**output_geojson_dict)
+            body_response = ResponseBody(geojson=output_geojson_dict)
+            response = get_response(HTTPStatus.OK.value, start_time, context.aws_request_id, body_response)
         except ValidationError as ve:
-            logger.error(f"validation error:{ve}.")
+            app_logger.error(f"validation error:{ve}.")
             response = get_response(HTTPStatus.UNPROCESSABLE_ENTITY.value, start_time, context.aws_request_id)
     except Exception as e:
-        logger.error(f"exception:{e}.")
+        app_logger.error(f"exception:{e}.")
         response = get_response(HTTPStatus.INTERNAL_SERVER_ERROR.value, start_time, context.aws_request_id)
 
-    logger.info(f"response_dumped:{response}...")
+    app_logger.info(f"response_dumped:{response}...")
     return response

src/main.py

@@ -1,147 +1,57 @@
-import json
-import uuid
+from typing import Dict
 
-from fastapi import FastAPI, HTTPException, Request, status
-from fastapi.exceptions import RequestValidationError
-from fastapi.responses import FileResponse, JSONResponse
-from fastapi.staticfiles import StaticFiles
+from geojson_pydantic import Feature, Polygon, FeatureCollection
 from pydantic import BaseModel
 
-from src import app_logger
-from src.utilities.type_hints import input_floatlist, input_floatlist2
-
-
-app = FastAPI()
-
-
-@app.middleware("http")
-async def request_middleware(request, call_next):
-    request_id = str(uuid.uuid4())
-    with app_logger.contextualize(request_id=request_id):
-        app_logger.info("Request started")
-
-        try:
-            response = await call_next(request)
-
-        except Exception as ex:
-            app_logger.error(f"Request failed: {ex}")
-            response = JSONResponse(content={"success": False}, status_code=500)
-
-        finally:
-            response.headers["X-Request-ID"] = request_id
-            app_logger.info(f"Request ended")
-            return response
-
-
-class Input(BaseModel):
-    x1: float
-    y1: float
-    x2: float
-    y2: float
-    x: float
-    y: float
-
-
-class BBoxWithPoint(BaseModel):
-    bbox: input_floatlist
-    point: input_floatlist2
-
-
-def get_parsed_bbox_points(request_input: Input) -> BBoxWithPoint:
-    return {
-        "bbox": [
-            request_input.x1, request_input.x2,
-            request_input.y1, request_input.y2
-        ],
-        "point": [[request_input.x, request_input.y]]
-    }
-
-
-@app.post("/post_test")
-async def post_test(request_input: Input) -> JSONResponse:
-    app_logger.info(f"start:{request_input}.")
-    request_body = get_parsed_bbox_points(request_input)
-    app_logger.info(f"request_body:{request_body}.")
-    return JSONResponse(
-        status_code=200,
-        content=request_body
-    )
-
-
-@app.get("/hello")
-async def hello() -> JSONResponse:
-    app_logger.info(f"hello")
-    return JSONResponse(status_code=200, content={"msg": "hello"})
-
-
-@app.post("/infer_samgeo")
-def samgeo(request_input: Input):
-    import subprocess
-
-    from src.prediction_api.predictor import base_predict
-
-    app_logger.info("starting inference request...")
-
-    try:
-        import time
-
-        time_start_run = time.time()
-        request_body = get_parsed_bbox_points(request_input)
-        app_logger.info(f"request_body:{request_body}.")
-        try:
-            output = base_predict(
-                bbox=request_body["bbox"],
-                point_coords=request_body["point"]
-            )
-            duration_run = time.time() - time_start_run
-            app_logger.info(f"duration_run:{duration_run}.")
-            body = {
-                "duration_run": duration_run,
-                "output": output
-            }
-            return JSONResponse(status_code=200, content={"body": json.dumps(body)})
-        except Exception as inference_exception:
-            home_content = subprocess.run("ls -l /home/user", shell=True, universal_newlines=True, stdout=subprocess.PIPE)
-            app_logger.error(f"/home/user ls -l: {home_content.stdout}.")
-            app_logger.error(f"inference error:{inference_exception}.")
-            return HTTPException(status_code=500, detail="Internal server error on inference")
-    except Exception as generic_exception:
-        app_logger.error(f"generic error:{generic_exception}.")
-        return HTTPException(status_code=500, detail="Generic internal server error")
-
-
-@app.exception_handler(RequestValidationError)
-async def request_validation_exception_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
-    app_logger.error(f"exception errors: {exc.errors()}.")
-    app_logger.error(f"exception body: {exc}.")
-    headers = request.headers.items()
-    app_logger.error(f"request header: {dict(headers)}.")
-    params = request.query_params.items()
-    app_logger.error(f'request query params: {dict(params)}.')
-    return JSONResponse(
-        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
-        content={"msg": "Error - Unprocessable Entity"}
-    )
-
-
-@app.exception_handler(HTTPException)
-async def http_exception_handler(request: Request, exc: HTTPException) -> JSONResponse:
-    app_logger.error(f"exception: {str(exc)}.")
-    headers = request.headers.items()
-    app_logger.error(f'request header: {dict(headers)}.' )
-    params = request.query_params.items()
-    app_logger.error(f'request query params: {dict(params)}.')
-    return JSONResponse(
-        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-        content={"msg": "Error - Internal Server Error"}
-    )
-
-
-# important: the index() function and the app.mount MUST be at the end
-app.mount("/", StaticFiles(directory="static", html=True), name="static")
-
-
-@app.get("/")
-def index() -> FileResponse:
-    return FileResponse(path="/app/static/index.html", media_type="text/html")
-
+g1 = {
+    "type": "FeatureCollection",
+    "features": [{
+        "type": "Feature",
+        "geometry": {
+            "type": "Polygon",
+            "coordinates": [
+                [
+                    [13.1, 52.46385],
+                    [13.42786, 52.6],
+                    [13.2, 52.5],
+                    [13.38272, 52.4],
+                    [13.43, 52.46385],
+                    [13.1, 52.46385]
+                ]
+            ],
+        },
+        "properties": {
+            "name": "uno",
+        },
+    }, {
+        "type": "Feature",
+        "geometry": {
+            "type": "Polygon",
+            "coordinates": [
+                [
+                    [13.77, 52.8],
+                    [13.88, 52.77],
+                    [13.99, 52.66],
+                    [13.11, 52.55],
+                    [13.33, 52.44],
+                    [13.77, 52.8]
+                ]
+            ],
+        },
+        "properties": {
+            "name": "due",
+        },
+    }]
+}
+
+PolygonFeatureCollectionModel = FeatureCollection[Feature[Polygon, Dict]]
+
+if __name__ == '__main__':
+    feat = PolygonFeatureCollectionModel(**g1)
+    print(feat)
+    print("feat")
+    """
+    
+    point:  {"lat":12.425847783029134,"lng":53.887939453125} 
+    map:ne:{"lat":17.895114303749143,"lng":58.27148437500001} sw:{"lat":0.6591651462894632,"lng":34.01367187500001}.
+    """

src/prediction_api/predictor.py

@@ -2,15 +2,14 @@
 import json
 
 from src import app_logger
-from src.utilities.constants import ROOT
+from src.utilities.constants import ROOT, MODEL_NAME, ZOOM
 from src.utilities.type_hints import input_floatlist, input_floatlist2
 
 
-def base_predict(
-        bbox: input_floatlist, point_coords: input_floatlist2, point_crs: str = "EPSG:4326", zoom: float = 16, model_name: str = "vit_h", root_folder: str = ROOT
-) -> str:
+def base_predict(bbox: input_floatlist, zoom: float = ZOOM, model_name: str = MODEL_NAME, root_folder: str = ROOT) -> dict:
     import tempfile
-    from samgeo import SamGeo, tms_to_geotiff
+    from samgeo import tms_to_geotiff
+    from samgeo.fast_sam import SamGeo
 
     with tempfile.NamedTemporaryFile(prefix="satellite_", suffix=".tif", dir=root_folder) as image_input_tmp:
         app_logger.info(f"start tms_to_geotiff using bbox:{bbox}, type:{type(bbox)}, download image to {image_input_tmp.name} ...")
@@ -32,7 +31,7 @@ def base_predict(
 
         with tempfile.NamedTemporaryFile(prefix="output_", suffix=".tif", dir=root_folder) as image_output_tmp:
             app_logger.info(f"done set_image, start prediction using {image_output_tmp.name} as output...")
-            predictor.predict(point_coords, point_labels=len(point_coords), point_crs=point_crs, output=image_output_tmp.name)
+            predictor.everything_prompt(output=image_output_tmp.name)
 
             # geotiff to geojson
             with tempfile.NamedTemporaryFile(prefix="feats_", suffix=".geojson", dir=root_folder) as vector_tmp:

src/utilities/constants.py

@@ -21,4 +21,6 @@ CUSTOM_RESPONSE_MESSAGES = {
     200: "ok",
     422: "Missing required parameter",
     500: "Internal server error"
-}
+}
+MODEL_NAME = "FastSAM-s.pt"
+ZOOM = 13