SDK Loop¶

SDK Task¶

A task that determinants a set of biometric features estimated from input images or samples. All features are divided into 3 groups:

face detection features: face detection, landmarks, eyes, gaze direction, ags.

samples(warp) features: emotions, mouth state, warp quality.

face attributes: age, gender, ethnicity, descriptor.

Face detection features are estimated on images, other features are estimated on warped images.

Task pipeline¶

SDK loop is a structure for task processing. There exist 3 stages of task processing:

detector stage

warp estimator stage

extractor stage

All features in estimated in the corresponding stage. Loop determines a pipeline for every task which is an order list of stages. Communication between stage handlers is done through queues. Loop sends a task to a queue which is corresponding first task pipeline stage. The task is put to a queue which corresponding next stage in the pipeline after stage processing. A task is put to a common results queue if it is failed or done. Loop gets a task from the results queue and transfers to a task customer.

Module contains task loop

class luna_handlers.sdk.sdk_loop.task_loop.SDKTaskLoop(settings)[source]¶

Loop for processing sdk task (extract attributes, detect faces, estiimate attributes)

_faceExtractorQueue¶

queue for extract tasks face attributes

Type: NamedQueue

_faceDetectorQueue¶

queue for detect face tasks

Type: NamedQueue

_faceEstimatorQueue¶

queue for estimate warp attributes tasks

Type: NamedQueue

_humanDetectorQueue¶

queue for detect face tasks

Type: NamedQueue

_humanExtractorQueue¶

queue for extract tasks human attributes

Type: NamedQueue

_resultsQueue¶

queue for tasks results

Type: NamedQueue

_running¶

loop status

Type: bool

_results¶

map a task id to a future which waiting the task result

Type: Dict[str, Future]

logger¶

loop logger

Type: Logger

loopScheduler¶

Scheduler which contains all running coroutines (loop, queue workers)

Type: AsyncRunner

checkStages(task)[source]¶

Check detector/extractor/estimator using in task’s pipeline and its availability :param task: task

Raises: ForbiddenProcess if one of pipeline stages needs service which turned off –
Return type: None

async close()[source]¶: Close loop

async collectResultLoop()[source]¶: Infinity loop for collect results from the results queue

async initialize()[source]¶

Initialize loop.

Start coroutine with infinite loop collect results loop, start task workers

async processTask(task, logger=None)[source]¶

Process a sdk task.

Parameters

task – task
logger – logger

Returns

task with results

Raises

RuntimeError – if “.pipeline” return unknown stage.

Return type

SDKTask

async validate()[source]¶

Validate SDK loop.

Start SDK tasks to validate SDK services (detector, extractor, estimator).

Raises: RuntimeError indicating the stage that failed to validate if one of the tasks returns an error –

class luna_handlers.sdk.sdk_loop.task_loop.TaskData(image, target)¶

image¶: Alias for field number 0

target¶: Alias for field number 1

Module contains sdk task

class luna_handlers.sdk.sdk_loop.sdk_task.AggregatedDetectionEstimations(face=None, body=None)[source]¶: Container for aggregated detection result

class luna_handlers.sdk.sdk_loop.sdk_task.AggregatedDetectionFaceAttributes(warps, liveness, filtered)[source]¶: Container for aggregated face detection result

class luna_handlers.sdk.sdk_loop.sdk_task.AggregatedDetectionHumanAttributes[source]¶: Container for aggregated human detection result

class luna_handlers.sdk.sdk_loop.sdk_task.AggregatedEstimations(detection=<factory>, extraction=<factory>, estimation=<factory>)[source]¶: Container for aggregated estimation result

class luna_handlers.sdk.sdk_loop.sdk_task.AggregatedExtractionEstimations(face=None, body=None)[source]¶: Container for aggregated extraction result

class luna_handlers.sdk.sdk_loop.sdk_task.AggregatedFaceWarpAttributes(emotions=None, mask=None)[source]¶: Container for the aggregated result of face warp attributes

class luna_handlers.sdk.sdk_loop.sdk_task.AggregatedHumanWarpAttributes[source]¶: Container for the aggregated result of human warp attributes

class luna_handlers.sdk.sdk_loop.sdk_task.AggregatedWarpEstimations(face=None, body=None)[source]¶: Container for aggregated estimation result

class luna_handlers.sdk.sdk_loop.sdk_task.BaseEstimation(warp, rect=None, attributes=None, extractedAttributes=None, quality=None, filter=None)[source]¶: Base container for estimation result

class luna_handlers.sdk.sdk_loop.sdk_task.BoundingBox(width, height, x, y)[source]¶

Face bounding box.

width¶

width

Type: float

height¶

height

Type: float

x¶

x coordinate of top left angle

Type: float

y¶

y coordinate of top left angle

Type: float

class luna_handlers.sdk.sdk_loop.sdk_task.FaceAttributes(warps, basicAttributes=None, descriptor=None, attributeId=None)[source]¶

Face attributes: descriptor, gender, ethnicity, age.

basicAttributes¶

dict with basic attributes from lunavl

Type: Optional[dict]

descriptor¶

dict with descriptor from lunavl

Type: Optional[dict]

warps¶

list of warps

Type: List[str]

filtered¶

attributes is filtered or not in a task processing (by garbage score)

Type: bool

attributeId¶

attribute id

Type: Optional[str]

class luna_handlers.sdk.sdk_loop.sdk_task.FaceEstimation(warp, rect=None, attributes=None, extractedAttributes=None, quality=None, filter=None, landmarks5=None, landmarks68=None)[source]¶: Container for face estimation result

class luna_handlers.sdk.sdk_loop.sdk_task.FaceWarp(body, sampleId=None, filename=None, imageId=None, detectTime=None, imageOrigin=None)[source]¶: Face warp

class luna_handlers.sdk.sdk_loop.sdk_task.FilteredEstimation(filename, estimation)[source]¶

Container for filtered image estimation result

property filterReasons¶: Estimation filter reason.

class luna_handlers.sdk.sdk_loop.sdk_task.HumanAttributes(warps, descriptor=None, attributeId=None)[source]¶

Face attributes: descriptor, gender, ethnicity, age.

descriptor¶

dict with descriptor from lunavl

Type: Optional[dict]

warps¶

warp list

Type: List[HumanWarp]

filtered¶

attributes is filtered or not in a task processing (by garbage score)

Type: bool

attributeId¶

attribute id

Type: Optional[str]

class luna_handlers.sdk.sdk_loop.sdk_task.HumanEstimation(warp, rect=None, attributes=None, extractedAttributes=None, quality=None, filter=None, landmarks17=None, score=None)[source]¶: Container for human body estimation result

class luna_handlers.sdk.sdk_loop.sdk_task.HumanWarp(body, sampleId=None, filename=None, imageId=None, detectTime=None, imageOrigin=None)[source]¶: Human warp

class luna_handlers.sdk.sdk_loop.sdk_task.ImageEstimation(image)[source]¶

Container for image estimation result.

image¶: source image

filename¶: source image filename

estimations¶: image estimations

property error: Optional[luna_handlers.sdk.sdk_loop.crutches_on_wheels.errors.errors.ErrorInfo]¶

Estimation error.

Returns: first estimation error if any errors occur, otherwise None
Return type: Optional[ErrorInfo]

property errors: List[luna_handlers.sdk.sdk_loop.crutches_on_wheels.errors.errors.ErrorInfo]¶

Estimation errors.

Returns: list of all estimation errors
Return type: List[ErrorInfo]

class luna_handlers.sdk.sdk_loop.sdk_task.SDKDetectableImage(body, filename=None, faceBoundingBoxes=None, humanBoundingBoxes=None, url=None, detectTime=None, imageOrigin=None)[source]¶

Image for detect faces.

id¶

image id

Type: str

body¶

body

Type: Union[bytes, bytearray, PilImage]

faceBoundingBoxes¶

list face detections

Type: Optional[List[BoundingBox]]

humanBoundingBoxes¶

list human body detections

Type: Optional[List[BoundingBox]]

filename¶

image filename (meta data)

Type: Optional[str]

error¶

error which is occurred during task processing

Type: Optional[SDKErrorInfo]

url¶

image source

Type: Optional[str]

detectTime¶

detection time in ISO format

Type: Optional[str]

imageOrigin¶

detection image origin

Type: Optional[str]

static asBytes(imageData, imageFormat=None, exif=None)[source]¶

Get image data as bytes. Convert Pil image to bytes if needed.

Parameters

imageData – pillow object
imageFormat – image format (determined from the filename extension)

Returns

image data as bytes

Return type

bytes

asPillow()[source]¶

Get image data as pillow object.

Returns: PIL image
Return type: Optional[Image]

exifTranspose()[source]¶

Based on genuine function from pillow https://pillow.readthedocs.io/en/latest/_modules/PIL/ImageOps.html#exif_transpose

The only difference is that manipulation is being done inplace, allowing us to leverage internal caching of exif data.

NOTE: Transposed image might have incorrect values in particular tags. For example, orientation, length, width.

Return type: None

class luna_handlers.sdk.sdk_loop.sdk_task.SDKEstimation(id=None, body=None, face=None)[source]¶

Container for sdk estimation result.

id¶

unique estimation ID

Type: str

property errors: List[luna_handlers.sdk.sdk_loop.crutches_on_wheels.errors.errors.ErrorInfo]¶

Estimation errors :returns: list of all estimation errors

Return type: List[ErrorInfo]

class luna_handlers.sdk.sdk_loop.sdk_task.SDKTask(toEstimation, data, filters=None, aggregateAttributes=False, multifacePolicy=MultifacePolicy.allowed)[source]¶

Task for execution in the sdk loop.

createTime¶

task create time, timestamp

Type: float

taskId¶

unique task id

Type: int

source¶

source of data for task

Type: TaskDataSource

toEstimation¶

set of estimations for task

Type: SDKEstimationTargets

aggregateAttributes¶

aggregate face attributes to one from input data or not

Type: bool

_pipeline¶

list of stages for task

Type: List[Stages]

filters¶

filters for processing estimation values

Type: Optional[SDKTaskFilters]

error¶

error which is occurred during task processing

Type: Optional[SDKErrorInfo]

multifacePolicy¶

multiple face detection policy

Type: MultifacePolicy

monitoringData¶

task processing monitoring data

Type: TaskMonitoringData

startExecutionTime¶: start execution time

lastCheckpointTime¶: last checkpoint time

images¶

task image processing results

Type: List[ImageEstimation]

aggregatedEstimations¶

task aggregated estimations

Type: AggregatedEstimations

filteredEstimations¶

task filtered estimations

Type: List[FilteredEstimation]

Raises: RuntimeError – if input data is not correct

checkMultiFacesRule()[source]¶

Check task multiple face detection.

Returns: True if multiple faces or humans not detected else False

property faceWarps: List[luna_handlers.sdk.sdk_loop.sdk_task.FaceWarp]¶

Task face warps.

Return type: List[FaceWarp]

getNextStage(currentStage)[source]¶

Get next stage of the task.

Parameters: currentStage – current stage
Returns: next stage or None (task need put to results queue)
Return type: Optional[Stages]

getObjectCountForStage(stage)[source]¶

Get count objects (images or warps) for task processing for stage.

Parameters: stage – stage
Returns: count images if stage is detector otherwise warps count
Return type: int

getPipeline()[source]¶

Get task pipeline based on estimations.

Returns: order list of stages list
Return type: List[Stages]

property humanWarps: List[luna_handlers.sdk.sdk_loop.sdk_task.HumanWarp]¶

Task human body warps.

Return type: List[HumanWarp]

property pipeline: List[luna_handlers.sdk.sdk_loop.enums.Stages]¶

Getter for pipe line.

Returns: order list of stages
Return type: List[Stages]

updateEstimationsWithExtractionResult(subtask)[source]¶

Update task estimations with attribute extraction result.

Parameters: subtask – extractor subtask

class luna_handlers.sdk.sdk_loop.sdk_task.SDKTaskFilters(yawThreshold=None, rollThreshold=None, pitchThreshold=None, livenessStates=None, maskStates=None, garbageScoreThreshold=None)[source]¶

Task filter’s.

yawThreshold¶

filter by yaw head angle for detection ([0…180])

Type: Optional[int]

pitchThreshold¶

filter by pitch head angle for detection ([0…180])

Type: Optional[int]

rollThreshold¶

filter by roll head angle for detection ([0…180])

Type: Optional[int]

maskStates¶

filter by medical mask states for detection

Type: Optional[List[MaskState]]

livenessStates¶

filter by liveness states for detection

Type: Optional[List[LivenessPrediction]]

garbageScoreThreshold¶

filter by garbage score for descriptor ([0…1])

Type: Optional[float]

checkFilterByAngles(headPose)[source]¶

Check filter by angles.

Parameters

headPose – estimate head pose

Returns

dict in api format:

{
    "is_filtered": <bool>,
    "filter_reasons": [<reasons>]
}

Return type

dict

checkFilterByGS(gs)[source]¶

Check garbage score filter.

Parameters

gs – estimated garbage score

Returns

dict in api format:

{
    "is_filtered": <bool>,
    "filter_reasons": [<reasons>]
}

Return type

dict

checkFilterByLiveness(liveness)[source]¶

Check filter by liveness.

Parameters

liveness – liveness estimation

Returns

dict in api format:

{
    "is_filtered": <bool>,
    "filter_reasons": [<reasons>]
}

Return type

dict

checkFilterByMaskState(mask)[source]¶

Check filter by medical mask state.

Parameters

mask – medical mask estimation

Returns

dict in api format:

{
    "is_filtered": <bool>,
    "filter_reasons": [<reasons>]
}

Return type

dict

needFilterByAngles()[source]¶

Need or not filter gy head angles.

Returns: True if any of angle threshold is not none.
Return type: bool

class luna_handlers.sdk.sdk_loop.sdk_task.SDKWarp(body, sampleId=None, filename=None, imageId=None, detectTime=None, imageOrigin=None)[source]¶

SDK Warp.

body¶

binary body

Type: Union[bytes, bytearray, PilImage]

outerSampleId¶

id of sample if warps is obtained from the sample storage

Type: Optional[str]

filename¶

warp source filename

Type: Optional[str]

detectTime¶

detection time in ISO format

Type: Optional[str]

id¶

warp id

Type: str

isFiltered¶

warp is filtered or not in a task processing (by garbage score)

Type: bool

imageId¶

image id which is a source for warp

Type: Optional[str]

error¶

error which is occurred during task processing

Type: Optional[Union[ErrorInfo, SDKErrorInfo]]

isSaved¶

whether sample is saved in luna-image-store

Type: bool

imageOrigin¶

detection image origin

Type: Optional[str]

asBytes()[source]¶

Get warped image as bytes.

Raises: RuntimeError – if body has incorrect type
Returns: image as bytes
Return type: bytes

asPillow()[source]¶

Get image data as pillow object.

Raises: RuntimeError – if body has incorrect type
Returns: PIL image
Return type: Image

property sampleId: str¶

Get sample id of warp.

Returns: outerSampleId if it is not None otherwise self.id
Return type: str

class luna_handlers.sdk.sdk_loop.sdk_task.TaskDataSource(value)[source]¶

Enum for a task data source

faceWarps = 2¶: face warped images

humanWarps = 3¶: human body warped images

class luna_handlers.sdk.sdk_loop.sdk_task.WarpAttributes[source]¶

Warp attributes.

emotions¶

dict with emotions from lunavl

Type: dict

mouthState¶

dict with mouth state from lunavl

Type: dict

warpQuality¶

dict with quality state from lunavl

Type: dict

mask¶

dict with mask state from lunavl

Type: dict

glasses¶

dict with glasses state from lunavl

Type: dict

asDict()[source]¶

Convert to dict without not extracted attributes.

Returns

dict in api format:

{
    "emotions": <self.emotions>,
    "mouth_attributes": <self.mouthState>,
    "quality": <self.warpQuality>
}

Return type

Dict[str, dict]

luna_handlers.sdk.sdk_loop.sdk_task.tasksTimeMonitoring(fieldName, tasks)[source]¶

Context manager for updating execution time of tasks.

Parameters

fieldName – monitoring field name
tasks – tasks

Loop task monitoring module.

class luna_handlers.sdk.sdk_loop.monitoring.MonitoringField(fieldName, value=None)[source]¶

Monitoring field.

fieldName¶

Type: str

value¶: value

class luna_handlers.sdk.sdk_loop.monitoring.TaskMonitoringData[source]¶

Container with list of all monitoring fields for sdk task.

executionTime¶

task execution Time

Type: float

resultsQueueTransportTime¶

task transport time in results queue

Type: float

faceDetectorQueueTransportTime¶

task transport time in detector queue

Type: float

faceExtractorQueueTransportTime¶

task transport time in extractor queue

Type: float

faceWarpEstimatorQueueTransportTime¶

task transport time in warp estimator queue

Type: float

faceDetectorStageTime¶

face detector stage time

Type: float

faceExtractorStageTime¶

extractor stage time

Type: float

faceWarpEstimatorStageTime¶

warp estimator stage time

Type: float

faceDetectTime¶

face detection time

Type: float

faceDescriptorExtractTime¶

face descriptor extact time

Type: float

basicAttributesExtractTime¶

basic attributes extract time

Type: float

gazeEstimationTime¶

gaze direction estimation time

Type: float

headPoseEstimationTime¶

head pose direction estimation time

Type: float

eyesEstimationTime¶

eyes attribute estimation time

Type: float

agsEstimationTime¶

ags estimation time

Type: float

warpQualityEstimationTime¶

warp quality direction estimation time

Type: float

emotionsEstimationTime¶

emotions estimation time

Type: float

mouthStateEstimationTime¶

mouth state estimation time

Type: float

maskEstimationTime¶

mask estimation time

Type: float

glassesEstimationTime¶

glasses estimation time

Type: float

livenessEstimationTime¶

liveness estimation time

Type: float

toDataForMonitoring()[source]¶

Convert to DataForMonitoring format.

Returns: data without empty fields
Return type: DataForMonitoring

Helpers enums.

class luna_handlers.sdk.sdk_loop.enums.MultifacePolicy(value)[source]¶

Multiple face detection policy enum.

allowed = 1¶: multiple face detection allowed

getBest = 2¶: get best detection from the image

notAllowed = 0¶: multiple face detection not allowed

class luna_handlers.sdk.sdk_loop.enums.SDKFaceEstimations(value)[source]¶

SDK face estimations enum. Name - estimation api name, value - estimation lunavl name.

ags = 'ags'¶: ags

basicAttributes = 'basicAttributes'¶: basic attributes (age, gender, ethnicity)

emotions = 'emotions'¶: emotions

eyes = 'eyes'¶: eyes state, points

faceDescriptor = 'faceDescriptor'¶: face descriptor

gaze = 'gaze'¶: faze

glasses = 'glasses'¶: glasses

headPose = 'headPose'¶: head pose

liveness = 'liveness'¶: liveness

mask = 'mask'¶: mask

mouthAttributes = 'mouthState'¶: mouth state

quality = 'warpQuality'¶: warp quality

class luna_handlers.sdk.sdk_loop.enums.SDKHumanEstimations(value)[source]¶

SDK human estimations enum. Name - estimation api name, value - estimation lunavl name.

humanDescriptor = 'humanDescriptor'¶: human body descriptor

class luna_handlers.sdk.sdk_loop.enums.Stages(value)[source]¶

Sdk task stages enum.

faceDetector = 'faceDetector'¶: face detection and estimate detection attributes

faceEstimator = 'faceWarpEstimator'¶: estimate face warp attributes (emotions, mouth, quality)

faceExtractor = 'faceExtractor'¶: extract face descriptor and basic attributes

humanDetector = 'humanDetector'¶: human detection and estimate detection attributes

humanExtractor = 'humanExtractor'¶: extract human descriptor and basic attributes

Module contains a structure for a task estimation targets

class luna_handlers.sdk.sdk_loop.estimation_targets.BaseEstimationTargets[source]¶

Base class for estimation targets

isEmpty()[source]¶

Check targets is empty or not. :returns: True if exists a targets to estimation otherwise False

Return type: bool

class luna_handlers.sdk.sdk_loop.estimation_targets.SDKEstimationTargets(estimateFace=1, estimateHuman=1, faceEstimationTargets=None, humanEstimationTargets=None)[source]¶

SDK estimation target structure.

estimateFace¶

estimate or not face detection/face attributes

Type: int

estimateHuman¶

estimate or not human detection/human attributes

Type: int

faceEstimationTargets¶

face estimation targets

Type: SDKFaceEstimationTargets

humanEstimationTargets¶

human estimation targets

Type: SDKHumanEstimationTargets

Notes

landmarks68 estimated always for face detection (for more precision warping).

class luna_handlers.sdk.sdk_loop.estimation_targets.SDKFaceEstimationTargets(estimateGaze=0, estimateHeadPose=0, estimateEyesAttributes=0, estimateAGS=0, estimateQuality=0, estimateMouthAttributes=0, estimateEmotions=0, estimateFaceDescriptor=0, estimateBasicAttributes=0, estimateMask=0, estimateGlasses=0, estimateLiveness=SDKLivenessEstimationPolicy({'estimate': 0, 'scoreThreshold': None, 'qualityThreshold': None}))[source]¶

Face estimations targets

estimateGaze¶

estimate or not gaze direction

Type: int

estimateHeadPose¶

estimate or not head pose

Type: int

estimateEyesAttributes¶

estimate or not eyes attributes

Type: int

estimateAGS¶

estimate or not ags

Type: int

estimateQuality¶

estimate or not head pose

Type: int

estimateMouthAttributes¶

estimate or not mouth state

Type: int

estimateEmotions¶

estimate or not emotions

Type: int

estimateBasicAttributes¶

estimate or not basic attributes

Type: int

estimateFaceDescriptor¶

estimate or not face descriptor 0 - not estimate face descriptor 1 - estimate default version descriptor other values - descriptor version

Type: int

estimateMask¶

estimate or not mask

Type: int

estimateGlasses¶

estimate or not glasses

Type: int

estimateLiveness¶

liveness estimation policy

Type: SDKLivenessEstimationPolicy

getTargetForWarpEstimator()[source]¶

Get targets for detector.

Returns: set of estimations that warp estimator must do
Return type: Set[SDKFaceEstimations]

getTargetsForFaceDetector()[source]¶

Get targets for detector.

Returns: set of estimations that detector must do
Return type: Set[SDKFaceEstimations]

getTargetsForFaceExtractor()[source]¶

Get targets for detector.

Returns: set of estimations that detector must do
Return type: Set[SDKFaceEstimations]

class luna_handlers.sdk.sdk_loop.estimation_targets.SDKHumanEstimationTargets(estimateHumanDescriptor=0)[source]¶

Human body estimations targets

estimateHumanDescriptor¶

estimate or not face descriptor 0 - not estimate face descriptor 1 - estimate default version descriptor other values - descriptor version

Type: int

getTargetsForHumanExtractor()[source]¶

Get targets for detector.

Returns: set of estimations that detector must do
Return type: Set[SDKHumanEstimations]

class luna_handlers.sdk.sdk_loop.estimation_targets.SDKLivenessEstimationPolicy(estimate=0, scoreThreshold=None, qualityThreshold=None)[source]¶

Liveness estimations policy

estimate¶

whether to estimate liveness

Type: int

scoreThreshold¶

liveness estimation score threshold ([0…1])

Type: Optional[float]

qualityThreshold¶

liveness estimation quality threshold ([0…1])

Type: Optional[float]

Module contains stage handlers

class luna_handlers.sdk.sdk_loop.sdk_stages.AsyncStageHandler(stage, optimalStageBatchSize, inputQueue, executor, workerSettings, workerInitializer, workerCount=1, poolSize=1)[source]¶

Asynchronous stage handler

isRun¶: state of the handler

listenerCount¶

count of queue listeners

Type: int

asyncExecutorPool¶

pool executor

Type: concurrent.futures.ProcessPoolExecutor

scheduler¶

container for queue listeners

Type: AsyncRunner

logger¶

logger

Type: Logger

poolSize¶

process pool size

Type: int

processTaskError(task, error, resultQueue)[source]¶

Process task error

Parameters

task – task
error – error
resultQueue – result queue

recreatePool()[source]¶

Create or recreate process pool.

Return type: None

run(resultQueue, routingQueueMap)[source]¶

Run Handler.

Parameters

resultQueue – result queue
routingQueueMap – routing queue map

async stop()[source]¶: Stop handler

updateQueueTransportMonitoringData(tasks)[source]¶: Update queue transport time for task monitoring data :param tasks: tasks for update

class luna_handlers.sdk.sdk_loop.sdk_stages.BaseStageHandler(stage, optimalStageBatchSize, inputQueue, executor)[source]¶

Base class for stage handler .. attribute:: stage

stage type

type

Stages

optimalStageBatchSize¶

optimal stage batch size

Type: int

inputQueue¶

input queue for tasks

Type: NamedQueue

executor¶

executor

Type: Callable[[List[SDKTask]], List[SDKTask]]

execute(tasks)[source]¶

Execute tasks :param tasks: sdk tasks

Returns: processing tasks
Return type: List[SDKTask]

Common human and face detections functions

class luna_handlers.sdk.sdk_loop.base_detector.BaseDetectorState[source]¶

Detector worker state, process local. Final class.

State contains:

logger for worker

instance of detector

abstract property detector: Union[VLFaceDetector, lunavl.sdk.detectors.humandetector.HumanDetector]¶

Get detector :returns: detector from state

Return type: Union[ForwardRef, HumanDetector]

property faceEngine: VLFaceEngine¶

Get instance of FaceEngine :returns: _faceEngine

Return type: VLFaceEngine

class luna_handlers.sdk.sdk_loop.base_detector.BaseDetectorSubTask(image, getBestDetection, taskId, taskMonitoring, estimations)[source]¶

Sub task for detector.

taskId¶

original task id

Type: int

image¶

an image for detection and further estimation of the detect attributes.

Type: SDKDetectableImage

detections¶

human detections from the image

Type: List[VLFaceDetection]

sdkImage¶

loaded sdk image if the image is a valid image otherwise None

Type: Optional[VLImage, ImageForRedetection]

isValid¶

whether the image is loaded into a sdkImage

Type: bool

estimations¶

estimation targets, set this param for estimations.

Type: SDKEstimationTargets

getBestDetection¶

whether to get best detection

Type: bool

originTaskMonitoring¶

monitoring data of the original task

Type: TaskMonitoringData

error¶

error occurred during task execution

Type: LunaVLError

imageOrientation¶

image orientation mode

Type: Optional[OrientationType]

abstract static createImageForDetection(img)[source]¶

Create sdk image from SDKDetectableImage :param img: image

Returns: VLImage if image has not bounding box, ImageForDetection if has bounding box, None if loading is failed
Return type: Union[ImageForRedetection, VLImage, None]

abstract classmethod createImageSubTasksForDetector(tasks)[source]¶

Create sub tasks for each image in tasks :param tasks: tasks

Returns: list of sub tasks
Return type: List[~DetectorSubTask]

abstract static getEstimationTargets(estimations)[source]¶

Get estimation targets for sub task.

Parameters: estimations – task estimations
Return type: BaseEstimationTargets

updateMonitoringData(estimation, estimationTime)[source]¶

Update monitoring data of the origin task. Add estimation time of a estimation to times from other subtasks.

Parameters

estimation – estimation type
estimationTime – estimation time

class luna_handlers.sdk.sdk_loop.base_detector.BaseLoopDetector[source]¶

Base class for detectors

batchDetect(subTasks)[source]¶

Batch detect faces for valid sub tasks.

Parameters: subTasks – sub tasks
Return type: None

abstract static collectResultsFromSubTasksToTasks(tasks, subTasks)[source]¶

Collect result from sub tasks to corresponding tasks.

Parameters

tasks – tasks
subTasks – sub tasks

detectAndRetry(vlImages, toDetectSubTasks)[source]¶

Batch detect and if any errors occurred - collect errors and try to one more time for sub tasks without errors :param vlImages: images for batch detection :param toDetectSubTasks: detect subtasks

Return type: None

property detector: Union[VLFaceDetector, lunavl.sdk.detectors.humandetector.HumanDetector]¶

Get detector :returns: detector from state

Return type: Union[ForwardRef, HumanDetector]

estimateDetectionAttributes(subTask)[source]¶: Estimate detection attributes for a sub task. :param subTask: sub task

getImageOrientation(image)[source]¶

Get image orientation mode :param image: original image

Returns: image orientation
Return type: OrientationType

abstract static getResultDetectionAsDict(subTask, detection)[source]¶

Get detection result as dict. Rotate detection and all its coordinates if original image has been rotated. :param subTask: detector subtask :param detection: detection

Returns: dict with detection results

property logger: luna_handlers.sdk.sdk_loop.crutches_on_wheels.utils.log.Logger¶

Get Logger. :returns: logger from state

Return type: Logger

redetectAndRetry(vlRedectImages, toRedetectSubTasks)[source]¶

Batch redetect and if any errors occurred - collect errors and try to one more time for sub tasks without errors :param vlImages: images for batch detection :param toRedetectSubTasks: redetect subtasks

Return type: None

property state: luna_handlers.sdk.sdk_loop.base_detector.DetectorState¶

Get state of detector.

Returns: state instance
Return type: ~DetectorState

class luna_handlers.sdk.sdk_loop.base_detector.Detections(image, detections)[source]¶

Detection comparison class

confThreshold¶: confidence detection comparison threshold

areaThreshold¶: area detection comparison threshold

image¶: image

detections¶: face detections from the image sorted by combined criteria of detection confidence, detection area and manhattan distance from detection center to the center of the image

bestDetection¶: best face detection from the image

getBestDetection()[source]¶

Get best detection from the image.

Returns: best face detection from the image
Return type: ~DetectorType

luna_handlers.sdk.sdk_loop.base_detector.DetectorState¶

generic type for detector state

alias of TypeVar(‘DetectorState’, bound=BaseDetectorState)

luna_handlers.sdk.sdk_loop.base_detector.DetectorSubTask¶

generic type for detector subtasks

alias of TypeVar(‘DetectorSubTask’, bound=BaseDetectorSubTask)

luna_handlers.sdk.sdk_loop.base_detector.DetectorType¶

generic type for allowed values type of detections

alias of TypeVar(‘DetectorType’, VLFaceDetection, lunavl.sdk.detectors.humandetector.HumanDetection)

Base module for extractors

class luna_handlers.sdk.sdk_loop.base_extractor.BaseExtractor[source]¶

Base extractor class

batchExtractDescriptors(subTasks)[source]¶

Batch extract descriptors for sub tasks with one warp.

Batches are chunked to reduce FSDK memory consumption.

Parameters: subTasks – sub tasks
Return type: None

batchExtractDescriptorsWithAggregation(subTasks)[source]¶

Batch extract descriptors for sub tasks with several warps

Parameters: subTasks – sub tasks

property logger: luna_handlers.sdk.sdk_loop.crutches_on_wheels.utils.log.Logger¶

Get Logger. :returns: logger from state

Return type: Logger

property state: luna_handlers.sdk.sdk_loop.base_extractor.ExtractorState¶

Get extractor state :returns: state

Return type: ~ExtractorState

class luna_handlers.sdk.sdk_loop.base_extractor.BaseExtractorState[source]¶

Extractor state, process local.

State contains:

logger for worker

instance of basic attributes estimator

instance of FaceEngine

maps a descriptor version to instance of extractor

property faceEngine: VLFaceEngine¶

Get instance of FaceEngine :returns: _faceEngine

Return type: VLFaceEngine

getExtractor(descriptorVersion=None)[source]¶

Get extractor for corresponding descriptor version.

If there is no extractor of corresponding version, it will be created. If descriptorVersion is equal to None or 1, an extractor with default version will be returned. An extractor with the corresponding version will be returned otherwise.

Parameters: descriptorVersion – descriptor version.
Returns: extractor
Return type: Union[FaceDescriptorEstimator, HumanDescriptorEstimator]

classmethod initialize(workerName, settings)[source]¶

Initialize state. Singleton. Initialize FaceEngine, extractor and basic attributes.

Parameters

workerName – worker name
settings – settings for worker

Returns

True if it is first call of initialize (for process) otherwise False

Return type

bool

class luna_handlers.sdk.sdk_loop.base_extractor.BaseExtractorSubTask(attributes, warps, taskId, estimationId=None, descriptorVersion=0)[source]¶

Extractor sub task. .. attribute:: warps

list of warp for extract aggregate attributes.

type

List[WarpedImage]

attributes¶

result attributes

Type: Attributes

taskId¶

original task id

Type: int

estimationId¶

unique estimation id

Type: str

estimationDescriptorVersion¶

0 means do not estimate descriptor, -1 - default, other - estimate descriptor with specified version

Type: int

error¶

error occurred during task execution

Type: LunaVLError

needEstimateDescriptor()[source]¶

Need or not estimate descriptor :returns: true if estimationDescriptorVersion != 0

Return type: bool

luna_handlers.sdk.sdk_loop.base_extractor.ExtractorState¶

generic type for extractor state

alias of TypeVar(‘ExtractorState’, bound=BaseExtractorState)

Module contains a functional for batch face detection and estimation detection attributes

class luna_handlers.sdk.sdk_loop.face_detector.DetectorState[source]¶

Detector worker state, process local. Final class.

State contains:

logger for worker

instance of detector and estimators

property detector: VLFaceDetector¶

Get detector

Returns: vl detector
Return type: VLFaceDetector

classmethod initialize(workerName, settings)[source]¶

Initialize state. Singleton. Initialize FaceEngine, detector.

Parameters

workerName – worker name
settings – settings for worker

Returns

True if it is first call of initialize (for process) otherwise False

Return type

bool

class luna_handlers.sdk.sdk_loop.face_detector.FaceDetector[source]¶

Loop face detector

static aggregateLivenessAndCheckState(task)[source]¶

Check whether extracted attribute is filtered, and update task with filtered result if it is.

task¶: task

Returns: True if attribute is filtered, otherwise False
Return type: None

static collectResultsFromSubTasksToTasks(tasks, subTasks)[source]¶

Collect result from sub tasks to corresponding tasks.

Parameters

tasks – tasks
subTasks – sub tasks

static estimateDetectionAttributes(subTask)[source]¶: Estimate detection attributes for a sub task. :param subTask: sub task

static getResultDetectionAsDict(subTask, detection)[source]¶

Get detection result as dict. Rotate detection if needed :param subTask: face detector subtask :param detection: face detection

Returns: dict with detection results
Return type: dict

static handleSubtaskDetections(subTask, task)[source]¶

Handle subtask detections and collect results to task :param subTask: detection subtask :param task: detection task

Return type: None

static makeLivenessPrediction(task, detection)[source]¶

Update SDK liveness estimation using task settings :param task: SDK task :param detection: SDK detection

Return type: None

class luna_handlers.sdk.sdk_loop.face_detector.FaceDetectorSubTask(image, estimations, getBestDetection, taskId, taskMonitoring)[source]¶

Sub task for face detector. .. attribute:: detections

list face detections

type

List[VLFaceDetection]

static createImageForDetection(img)[source]¶

Create sdk image from SDKDetectableImage :param img: image

Returns: VLImage if image has not bounding box, ImageForDetection if has bounding box, None if loading is failed
Return type: Union[ImageForRedetection, VLImage, None]

classmethod createImageSubTasksForDetector(tasks)[source]¶

Create sub tasks for each image in tasks :param tasks: tasks

Returns: list of sub tasks
Return type: List[~DetectorSubTask]

static getEstimationTargets(estimations)[source]¶

Get estimation targets :param estimations: task estimations

Returns: task face estimation targets
Return type: SDKFaceEstimationTargets

luna_handlers.sdk.sdk_loop.face_detector.aggregateLivenessFromEstimates(livenessEstimates, qualityThreshold, scoreThreshold=None)[source]¶

Get aggregated liveness from estimates. :param livenessEstimates: liveness input estimates :param qualityThreshold: liveness quality threshold :param scoreThreshold: liveness score threshold

Returns: LivenessPredicted
Return type: class
Return type: LivenessPredicted

luna_handlers.sdk.sdk_loop.face_detector.detect(tasks)[source]¶

Detect faces and estimate detection attributes.

Parameters: tasks – tasks
Returns: task with estimated attributes and cropped warps
Return type: List[SDKTask]

luna_handlers.sdk.sdk_loop.face_detector.initWorker(settings)[source]¶

Initialize face detector worker. Init logger, initialize FSDK, create detector, initialize estimators.

Parameters: settings – detector settings

luna_handlers.sdk.sdk_loop.face_detector.updateTaskWithDetectionResults(task, warp, detection)[source]¶

Update task with collected face detection results.

Parameters

task – task
warp – face warp
detection – face detection result

Module contains a functional for batch estimation descriptors and basic attributes

class luna_handlers.sdk.sdk_loop.face_extractor.ExtractorState[source]¶

Extractor state, process local. Final class.

State contains:

logger for worker

instance of basic attributes estimator

instance of FaceEngine

maps a descriptor version to instance of extractor

property basicAttributesEstimator: lunavl.sdk.estimators.face_estimators.basic_attributes.BasicAttributesEstimator¶

Get basic attributes estimators.

Returns: _basicAttributesEstimator
Return type: BasicAttributesEstimator

classmethod initialize(workerName, settings)[source]¶

Initialize state. Singleton. Initialize FaceEngine, extractor and basic attributes.

Parameters

workerName – worker name
settings – settings for worker

Returns

True if it is first call of initialize (for process) otherwise False

Return type

bool

class luna_handlers.sdk.sdk_loop.face_extractor.FaceExtractor[source]¶

Face extractor

batchExtractBasicAttributes(subTasks)[source]¶

Batch extract basic attributes for sub tasks with one warp

Batches are chunked to reduce FSDK memory consumption.

Parameters: subTasks – sub tasks

batchExtractBasicAttributesWithAggregation(subTasks)[source]¶

Batch extract basic attributes for sub tasks with several warps

Parameters: subTasks – sub tasks

extractAggregateBasicAttribute(subTask)[source]¶

Batch extract descriptors for sub task with several warps

Parameters: subTask – sub tasks

class luna_handlers.sdk.sdk_loop.face_extractor.FaceExtractorSubTask(attributes, warps, taskId, estimationId=None, estimateBasicAttributes=False, descriptorVersion=0)[source]¶

Face extractor sub task. .. attribute:: estimateBasicAttributes

estimate or not basic attributes

type

bool

luna_handlers.sdk.sdk_loop.face_extractor.checkAttributesFiltered(task, attribute)[source]¶

Check whether extracted attribute is filtered, and update task with filtered result if it is.

luna_handlers.sdk.sdk_loop.face_extractor.task¶: task

luna_handlers.sdk.sdk_loop.face_extractor.attribute¶: extracted attribute

Returns: True if attribute is filtered, otherwise False
Return type: bool

luna_handlers.sdk.sdk_loop.face_extractor.extract(tasks)[source]¶

Extract face attributes attributes.

Parameters: tasks – tasks
Returns: processed tasks

luna_handlers.sdk.sdk_loop.face_extractor.getSubTasksForTasks(tasks)[source]¶

Separate tasks on sub tasks. :param tasks: tasks

Returns: tuple sub tasks without aggregation and with
Return type: Tuple[List[BaseExtractorSubTask], List[BaseExtractorSubTask]]

luna_handlers.sdk.sdk_loop.face_extractor.initWorker(settings)[source]¶

Initialize extractor worker. Init logger, initialize FSDK, create basic attributes estimator and descriptor extractors.

Parameters: settings – detector settings

luna_handlers.sdk.sdk_loop.face_extractor.updateTaskWithFilteredResults(task, filtrationRes, attribute)[source]¶

Update task with results filtered.

Parameters

task – task
filtrationRes – filtration result
attribute – filtered attribute

Module contains a functional for batch estimation basic attributes

class luna_handlers.sdk.sdk_loop.face_warp_estimator.AggregateAttributesCounter(score, index, count=1, operator=ScoreOperator.concat)[source]¶

Container class for counting aggregate attributes.

update(score, index)[source]¶

Update attribute score and set top index. :param score: next attribute score :param index: attribute index

Return type: None

class luna_handlers.sdk.sdk_loop.face_warp_estimator.ScoreOperator(value)[source]¶: Score operator enum.

class luna_handlers.sdk.sdk_loop.face_warp_estimator.WarpEstimatorState[source]¶

Warp estimator worker state, process local. Final class.

State contains:

logger for worker

instance of estimator collections

property estimatorCollections: FaceEstimatorsCollection¶

Get estimator collections :returns: get estimators collections

Return type: FaceEstimatorsCollection

classmethod initialize(workerName, settings)[source]¶

Initialize class.

Return type: bool

luna_handlers.sdk.sdk_loop.face_warp_estimator.aggregateWarpAttributes(task)[source]¶

Aggregate face attributes.

Parameters: task – sdk task
Return type: None

luna_handlers.sdk.sdk_loop.face_warp_estimator.estimate(tasks)[source]¶

Estimate warp attributes (emotions, quality, mouth).

Parameters: tasks – tasks
Returns: processed tasks
Return type: List[SDKTask]

luna_handlers.sdk.sdk_loop.face_warp_estimator.estimateWarpAttributes(task)[source]¶

Estimate warp attributes (emotions, quality, mouth) for task.

Parameters: task – task
Return type: None

luna_handlers.sdk.sdk_loop.face_warp_estimator.getMostCommonAttribute(attributes, predominantKey, scoreOperator=ScoreOperator.concat)[source]¶

Get most common attribute.

Parameters

attributes – incoming attributes for aggregate
predominantKey – predominant attribute key
scoreOperator – score operator for calculating aggregation

Returns

most popular attribute, if there is the most popular, otherwise attribute with the highest score (from most popular attribute)

Return type

Dict[str, Any]

luna_handlers.sdk.sdk_loop.face_warp_estimator.initWorker(settings)[source]¶

Initialize warp estimator worker. Init logger, initialize FSDK, create estimators.

Parameters: settings – detector settings

Module contains a functional for batch human detection

class luna_handlers.sdk.sdk_loop.human_detector.HumanDetectorState[source]¶

Detector worker state, process local. Final class.

State contains:

logger for worker

instance of detector

property detector: lunavl.sdk.detectors.humandetector.HumanDetector¶

Get detector

Returns: vl detector
Return type: HumanDetector

classmethod initialize(workerName, settings)[source]¶

Initialize state. Singleton. Initialize FaceEngine, detector, orientation mode estimator.

Parameters

workerName – worker name
settings – settings for worker

Returns

True if it is first call of initialize (for process) otherwise False

Return type

bool

property warper: lunavl.sdk.estimators.body_estimators.humanwarper.HumanWarper¶

Get human warper :returns: warper

Return type: HumanWarper

class luna_handlers.sdk.sdk_loop.human_detector.HumanDetectorSubTask(image, estimations, getBestDetection, taskId, taskMonitoring)[source]¶

Sub task for human detector.

Parameters: detections (List[HumanDetection]) – human detection list

static createImageForDetection(img)[source]¶

Create sdk image from SDKDetectableImage :param img: image

Returns: VLImage if image has not bounding box, ImageForDetection if has bounding box, None if loading is failed
Return type: Union[ImageForRedetection, VLImage, None]

classmethod createImageSubTasksForDetector(tasks)[source]¶

Create sub tasks for each image in tasks :param tasks: tasks

Returns: list of sub tasks
Return type: List[~DetectorSubTask]

static getEstimationTargets(estimations)[source]¶

Get estimation from task. :param estimations: task estimations

Returns: human estimations
Return type: SDKHumanEstimationTargets

class luna_handlers.sdk.sdk_loop.human_detector.LoopHumanDetector[source]¶

Loop human detector

static collectResultsFromSubTasksToTasks(tasks, subTasks)[source]¶

Collect result from sub tasks to corresponding tasks.

Parameters

tasks – tasks
subTasks – sub tasks

static getResultDetectionAsDict(subTask, detection)[source]¶

Get detection result as dict. Rotate detection if needed :param subTask: human detector subtask :param detection: human detection

Returns: dict with detection results
Return type: dict

property warper: lunavl.sdk.estimators.body_estimators.humanwarper.HumanWarper¶

Get warper :returns: warper from state

Return type: HumanWarper

luna_handlers.sdk.sdk_loop.human_detector.checkDetectionsRelated(bodyDetection, faceDetection)[source]¶

Check the body detection and face detection are related to the same event.

Parameters

bodyDetection – body detection
faceDetection – face detection

Returns

True if face landmarks of body detection are in face detection rect, otherwise False

Return type

bool

luna_handlers.sdk.sdk_loop.human_detector.checkLandmarkInRect(landmark, rect)[source]¶

Check landmark in detection rect :param landmark: landmark :param rect: detection rect

Returns: True if landmark in detection rect, otherwise False

luna_handlers.sdk.sdk_loop.human_detector.checkLandmarkReliable(landmark)[source]¶

Check landmark reliable.

There may be unreliable landmarks among the landmarks17 that do not need to be taken into account when check detections related. Unreliable landmark takes the value (0, 0) and its score is below the threshold.

Parameters: landmark – landmark
Returns: True if landmark reliable, otherwise False
Return type: bool

luna_handlers.sdk.sdk_loop.human_detector.detect(tasks)[source]¶

Detect human bodies

Parameters: tasks – tasks
Returns: task with estimated attributes and cropped warps
Return type: List[SDKTask]

luna_handlers.sdk.sdk_loop.human_detector.initWorker(settings)[source]¶

Initialize detector worker. Init logger, initialize FSDK, create detector, initialize estimators.

Parameters: settings – detector settings

luna_handlers.sdk.sdk_loop.human_detector.updateTaskWithDetectionResults(task, warp, detection, isFiltered)[source]¶

Update task with collected human body detection results.

Parameters

task – task
warp – human body warp
detection – human body detection result
isFiltered – whether the detection is filtered

Module contains a functional for batch estimation descriptors and basic attributes

class luna_handlers.sdk.sdk_loop.human_extractor.HumanExtractor[source]¶: Human extractor

class luna_handlers.sdk.sdk_loop.human_extractor.HumanExtractorState[source]¶

Extractor state, process local. Final class.

State contains:

logger for worker

instance of basic attributes estimator

instance of FaceEngine

maps a descriptor version to instance of extractor

class luna_handlers.sdk.sdk_loop.human_extractor.HumanExtractorSubTask(attributes, warps, taskId, estimationId=None, descriptorVersion=0)[source]¶: Human Extractor sub task.

luna_handlers.sdk.sdk_loop.human_extractor.extract(tasks)[source]¶

Extract human body attributes attributes.

Parameters: tasks – tasks
Returns: processed tasks

luna_handlers.sdk.sdk_loop.human_extractor.getSubTasksForTasks(tasks)[source]¶

Separate tasks on sub tasks. :param tasks: tasks

Returns: tuple sub tasks without aggregation and with
Return type: Tuple[List[BaseExtractorSubTask], List[BaseExtractorSubTask]]

luna_handlers.sdk.sdk_loop.human_extractor.initWorker(settings)[source]¶

Initialize extractor worker. Init logger, initialize FSDK, create basic attributes estimator and descriptor extractors.

Parameters: settings – detector settings