mirror of
https://github.com/azaion/gps-denied-onboard.git
synced 2026-04-22 21:46:36 +00:00
Oleksandr Bezdieniezhnykh
35 lines
864 B
Python
35 lines
864 B
Python
from abc import ABC, abstractmethod
|
|
from typing import Optional, AsyncIterator
|
|
|
|
from models.images import ImageData, ImageBatch, ProcessingStatus
|
|
from models.core import ValidationResult
|
|
|
|
|
|
class ImageInputPipelineBase(ABC):
|
|
@abstractmethod
|
|
async def receive_batch(
|
|
self, flight_id: str, batch: ImageBatch
|
|
) -> ValidationResult:
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def get_next_image(self, flight_id: str) -> Optional[ImageData]:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def stream_images(self, flight_id: str) -> AsyncIterator[ImageData]:
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def get_status(self, flight_id: str) -> ProcessingStatus:
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def clear_queue(self, flight_id: str) -> bool:
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def get_queue_size(self, flight_id: str) -> int:
|
|
pass
|
|
|