"""
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,
)