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