Standalone lambda development

Here is standalone lambda development description.

More information about lambda types and differences of standalone and others available at lambda types description.

Standalone lambda requirements

Standalone lambda requirements has no extra requirements and determines only by user: if lambda suggests Luna Events usage - it requires Luna Events service.

Standalone lambda development

The standalone lambda must be designed for detached from luna services requests processing, but it can access remote resources.

The request to standalone lambda can be used by user next ways:

lambda_main.py

from luna_lambda_tools import StandaloneLambdaRequest, logger

async def main(request: StandaloneLambdaRequest) -> dict:
    logger.info(request.json)   # print request json to log
    logger.info(request.getBody())  # print request body as raw bytes
    logger.info(request.headers)    # print request headers
    ...

It is required to use StandaloneLambdaRequest imported from luna_lambda_tools request as argument to main function.

Standalone lambda examples

Here is an example of standalone lambda which works with numbers, that getting from external resources:

lambda_main.py

import aiohttp
from luna_lambda_tools import StandaloneLambdaRequest, logger


class UserCtx:
    def __init__(self):
        self.session = None

    async def onStart(self):
        logger.info("start lambda")
        self.session = aiohttp.ClientSession()

    async def onShutdown(self):
        logger.info("shutdown lambda")
        await self.session.close()


async def main(request: StandaloneLambdaRequest) -> dict:
    """
    Supposed request structure:

    ```json
    {"url": "http://very-useful-url"}
    ```

    """
    logger.info("Request processing starts")
    url = request.json.get("url")

    async with request.userCtx.session.get(url) as resp:
        statusCode = resp.status

    logger.info("Request processing finished")
    result = f"{url} available" if statusCode == 200 else f"{url} is not available"
    return {"result": result}

request example

from luna3.luna_lambda.luna_lambda import LambdaApi

SERVER_ORIGIN = "http://lambda_address:lambda_port"  # Replace by your values before start
SERVER_API_VERSION = 1
lambdaApi = LambdaApi(origin=SERVER_ORIGIN, api=SERVER_API_VERSION)
lambdaId, accountId = "your_lambda_id", "your_account_id"  # Replace by your values before start


def makeRequest():
    data = {
        "url": "https://ya.ru/",
    }
    reply = lambdaApi.proxyLambdaPost(lambdaId=lambdaId, path="main", accountId=accountId, body=data)
    return reply


if __name__ == "__main__":
    response = makeRequest()
    print(response.json)

The luna-lambda-tools provides some stuff whose may be used to create standalone lambda:

standalone lambda stuff description

luna_lambda_tools.public.standalone

Module contains some stuff for standalone lambda

The lambda example which gets face detection and estimate mask directly using the LUNA SDK (available at https://github.com/VisionLabs/lunasdk)

LUNA python SDK must be added it to requirements.txt (requirements description available here)

requirements.txt

https://github.com/VisionLabs/lunasdk/archive/refs/tags/v.2.1.4.tar.gz

Note

To develop lambda with lunasdk locally it needs LUNA FSDK python bindings to be installed previously.

Warning

Such user lambda requires LUNA SDK data available from user lambda (the fsdk/data folder near the lambda_main.py main file) The LUNA SDK is available on VL release portal which must be extracted to fsdk folder (see archive file structure below). The only thing which is needed is data folder with used fsdk plans and config files (faceengine.conf and runtime.conf). It is recommended to not include not using plans from folder in archive to decrease result lambda image size.

Archive file structure with files required for the example

  ├──lambda_main.py
  ├──requirements.txt
  └──fsdk
     └──data
         ├──faceengine.conf
         ├──runtime.conf
         ├──FaceDet_v3_a5_cpu-avx2.plan
         ├──FaceDet_v3_redetect_v3_cpu-avx2.plan
         ├──LNet_precise_v2_cpu-avx2.plan
         ├──mask_clf_v3_cpu-avx2.plan
         └──slnet_v5_cpu-avx2.plan

lambda_main.py

from luna_lambda_tools import StandaloneLambdaRequest, UserException
from lunavl.sdk.faceengine.engine import VLFaceEngine
from lunavl.sdk.faceengine.setting_provider import DetectorType
from lunavl.sdk.image_utils.image import VLImage


class FaceDetectionException(UserException):
    statusCode = 400
    errorText = "failed to get face detection from image"


async def main(request: StandaloneLambdaRequest) -> dict:
    image = VLImage(body=request.getBody())
    faceEngine = VLFaceEngine()
    detector = faceEngine.createFaceDetector(DetectorType.FACE_DET_V3)
    detections = detector.detect([image])
    if not detections:
        raise FaceDetectionException
    faceDetections = detections[0]
    warper = faceEngine.createFaceWarper()
    warps = [warper.warp(faceDetection) for faceDetection in faceDetections]

    maskEstimator = faceEngine.createMaskEstimator()
    mask = await maskEstimator.estimate(warps[0].warpedImage, asyncEstimate=True)

    return {"results": mask.asDict()}

request example

from luna3.luna_lambda.luna_lambda import LambdaApi

SERVER_ORIGIN = "http://lambda_address:lambda_port"  # Replace by your values before start
SERVER_API_VERSION = 1
lambdaApi = LambdaApi(origin=SERVER_ORIGIN, api=SERVER_API_VERSION)
lambdaId, accountId = "your_lambda_id", "your_account_id"  # Replace by your values before start


def getImage(pathToImage):
    """
    Make sure pathToImage is valid path to specified image
    """
    with open(pathToImage, "rb") as file:
        return file.read()


def makeRequest():
    reply = lambdaApi.proxyLambdaPost(lambdaId=lambdaId, path="main", accountId=accountId, body=getImage("empty.jpeg"))
    return reply


if __name__ == "__main__":
    response = makeRequest()
    print(response.json)

Lambda example which extracts faces from video and matches them with faces from the list:

lambda_main.py

from luna3.public.matcher import Candidates, FaceFilters, SDKDescriptorReference
from luna3.public.sdk import (
    HumanTrackingAnalyticsParams as HTParams,
    Orientation,
    VideoSDKAnalytics,
    VideoSDKInput,
    VideoSDKProperties,
)
from luna_lambda_tools import StandaloneLambdaRequest

EVENTS_BATCH_SIZE = 50


def splitEvents(values: list) -> list[list]:
    """Split events into batches"""
    if not values:
        return []
    return [values[x : x + EVENTS_BATCH_SIZE] for x in range(0, len(values), EVENTS_BATCH_SIZE)]


async def main(request: StandaloneLambdaRequest) -> dict:
    results = []
    videosdkReply = (
        await request.clients.videoAgent.videosdk(
            VideoSDKInput(
                video=VideoSDKProperties(
                    url=request.json["video"]["url"], orientation=Orientation(request.json["video"].get("angle", 0))
                ),
                analytics=[
                    VideoSDKAnalytics(
                        name="human_tracking",
                        targets=["aggregated_face_descriptor"],
                        parameters=HTParams(tracking=HTParams.Tracking(detectorType="face")),
                    )
                ],
            ),
            asyncRequest=True,
            raiseError=True,
        )
    ).json

    for eventsBatch in splitEvents(videosdkReply["analytics"][0]["result"]["tracking"]["events"]):
        references = []
        eventsMap = {}
        for event in eventsBatch:
            descriptor = (
                event["aggregated_estimations"]["face"].get("descriptor")
                if event["aggregated_estimations"]["face"]
                else None
            )
            if descriptor is None:
                continue
            references.append(
                SDKDescriptorReference(
                    referenceId=event["event_id"],
                    descriptor=descriptor["sdk_descriptor"],
                )
            )
            eventsMap[event["event_id"]] = event

        matchReply = await request.clients.matcher.matchFaces(
            references=references,
            candidates=[
                Candidates(
                    FaceFilters(listId=request.json["list_id"]), threshold=request.json.get("similarity_threshold")
                )
            ],
            asyncRequest=True,
            raiseError=True,
        )

        for match in matchReply.json:
            results.append(
                {
                    "event": eventsMap[match["reference"]["id"]],
                    "matches": match,
                }
            )

    return {"result": results}

request example

from luna3.common.requests import RequestPayload
from luna3.luna_lambda.luna_lambda import LambdaApi

SERVER_ORIGIN = "http://lambda_address:lambda_port"  # Replace by your values before start
SERVER_API_VERSION = 1
lambdaApi = LambdaApi(origin=SERVER_ORIGIN, api=SERVER_API_VERSION)
lambdaId, accountId = "your_lambda_id", "your_account_id"  # Replace by your values before start
videoUrl = "your_video_url"
listId = "your_list_id"


def makeRequest():
    data = {
        "video": {"url": videoUrl, "angle": 0},
        "similarity_threshold": 0.9,
        "list_id": listId,
    }
    payload = RequestPayload.buildMsgpack(body=data)
    reply = lambdaApi.proxyLambdaPost(lambdaId=lambdaId, path="main", accountId=accountId, body=payload)
    return reply


if __name__ == "__main__":
    response = makeRequest()
    print(response.json)