Source code for luna_lambda_tools.public.clients.lpm

"""
Module contains luna-python-matcher client adapted for usage in lambda
"""
from typing import Awaitable

from luna3.common.luna_response import LunaResponse
from luna3.python_matcher.match_objects import (
    AttributeFilters,
    BinaryReference,
    Candidates,
    EventFilters,
    FaceFilters,
    Reference,
)
from luna3.python_matcher.python_matcher import PythonMatcherApi


[docs] class LPM: """Luna-python-matcher client"""
[docs] def __init__(self, lpmClient: PythonMatcherApi, accountId: str): self._accountId = accountId self._lpmClient = lpmClient
[docs] def getAddress(self) -> str: """Get luna-python-matcher address""" return self._lpmClient.baseUri
[docs] def matchRaw( self, candidates: list[BinaryReference], references: list[BinaryReference], raiseError=True, **kwargs, ) -> Awaitable[LunaResponse] | LunaResponse: """ Matcher raw API allows to match each candidate with each reference descriptors. Args: candidates: list of BinaryReference objects for candidates references: list of BinaryReference objects for references raiseError: whether to raise LunaApiException in case of failure Returns: class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*. In body of :class: `~.LunaResponse`: results of matching for each reference """ return self._lpmClient.matchRaw(candidates=candidates, references=references, raiseError=raiseError, **kwargs)
[docs] def matchFaces( self, candidates: list[Candidates], references: list[Reference | BinaryReference], raiseError=True, **kwargs, ) -> Awaitable[LunaResponse] | LunaResponse: """ Matcher API allows to submit tasks to a service that searches for faces similar to a given reference(s) by matching them. An attribute descriptor should be extracted for reference(s) and candidate(s). A source for references are events, attributes and faces. A source for candidates are faces and lists. Args: candidates: Candidates object with info about candidates references: list of Reference objects. Each object has entity id and it type (possible type variants: events, attributes, faces, raw descriptors) raiseError: whether to raise LunaApiException in case of failure Returns: class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*. In body of :class: `~.LunaResponse`: results of matching (failed and succeeded matches) """ return self._lpmClient.matchFaces( candidates=candidates, references=references, raiseError=raiseError, accountId=self._accountId, **kwargs )
[docs] def matchBodies( self, candidates: list[Candidates], references: list[Reference | BinaryReference], raiseError=True, **kwargs, ) -> Awaitable[LunaResponse] | LunaResponse: """ Matcher API allows to submit tasks to a service that searches for faces similar to a given reference(s) by matching them. An attribute descriptor should be extracted for reference(s) and candidate(s). A source for references are events, attributes and faces. A source for candidates are faces and lists. Args: candidates: Candidates object with info about candidates (possible type variants: events) references: list of Reference objects. Each object has entity id and it type (possible type variants: events, raw descriptors) raiseError: whether to raise LunaApiException in case of failure Returns: class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*. In body of :class: `~.LunaResponse`: results of matching (failed and succeeded matches) """ return self._lpmClient.matchBodies( candidates=candidates, references=references, raiseError=raiseError, accountId=self._accountId, **kwargs )
[docs] def crossmatchFaces( self, candidateFilters: FaceFilters | EventFilters | AttributeFilters, referenceFilters: FaceFilters | EventFilters | AttributeFilters, limit: int | None = None, threshold: int | None = None, sorting: int | None = None, raiseError=True, **kwargs, ) -> Awaitable[LunaResponse] | LunaResponse: """ Cross Matcher API allows to submit tasks to a service that searches for faces/events by given filters and matches them with each other. Resource implements face descriptors matching. Args: candidateFilters: candidate filters referenceFilters: reference filters limit: matching limit threshold: matching threshold sorting: whether to sort match results raiseError: whether to raise LunaApiException in case of failure Returns: class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*. In body of :class: `~.LunaResponse`: results of matching (failed and succeeded matches) """ return self._lpmClient.crossmatchFaces( candidateFilters=candidateFilters, referenceFilters=referenceFilters, limit=limit, threshold=threshold, sorting=sorting, raiseError=raiseError, accountId=self._accountId, **kwargs, )
[docs] def crossmatchBodies( self, candidateFilters: EventFilters, referenceFilters: EventFilters, limit: int | None = None, threshold: int | None = None, sorting: int | None = None, raiseError=True, **kwargs, ) -> Awaitable[LunaResponse] | LunaResponse: """ Cross Matcher API allows to submit tasks to a service that searches for events by given filters and matches them with each other. Resource implements body descriptors matching. Args: candidateFilters: candidate events filters referenceFilters: reference events filters limit: matching limit threshold: matching threshold sorting: whether to sort match results raiseError: whether to raise LunaApiException in case of failure Returns: class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*. In body of :class: `~.LunaResponse`: results of matching (failed and succeeded matches) """ return self._lpmClient.crossmatchBodies( candidateFilters=candidateFilters, referenceFilters=referenceFilters, limit=limit, threshold=threshold, sorting=sorting, raiseError=raiseError, **kwargs, )