import abc
from ..private.request import ClientsKwargsMixin
from ..private.tasks import _BaseLambdaTask
from .clients import TasksClients
from .clients.events import Events
from .clients.faces import Faces
from .clients.handlers import Handlers
from .clients.lis import LIS
from .clients.lpm import LPM
from .clients.lrs import LRS
from .clients.lva import LVA
from .clients.sender import Sender
[docs]
class BaseLambdaTask(_BaseLambdaTask, ClientsKwargsMixin):
"""Lambda task"""
[docs]
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@property
def clients(self):
facesKw = self._getFacesClientKwargs()
eventsKw = self._getEventsClientKwargs()
faceSamplesKw = self._getFacesSamplesClientKwargs()
bodySamplesKw = self._getBodySamplesClientKwargs()
imageOriginKw = self._getImageOriginClientKwargs()
lvaKw = self._getLVAClientKwargs()
senderKw = self._getSenderClientKwargs()
handlersKw = self._getHandlersClientKwargs()
return TasksClients(
faces=Faces(**facesKw) if facesKw else None,
events=Events(**eventsKw) if eventsKw else None,
matcher=LPM(**self._getLPMClientKwargs()),
faceSamplesStore=LIS(**faceSamplesKw) if faceSamplesKw else None,
bodySamplesStore=LIS(**bodySamplesKw) if bodySamplesKw else None,
imageOriginStore=LIS(**imageOriginKw) if imageOriginKw else None,
sdk=LRS(**self._getLRSClientKwargs()),
videoAgent=LVA(**lvaKw) if lvaKw else None,
sender=Sender(**senderKw) if senderKw else None,
handlers=Handlers(**handlersKw) if handlersKw else None,
)
@abc.abstractmethod
async def splitTasksContent(self, content: dict) -> list[dict]:
"""Split task content to sub task contents"""
...
@abc.abstractmethod
async def executeSubtask(self, subTaskContent: dict) -> dict | list:
"""Execute current sub task processing"""
...