mirror of
https://github.com/azaion/gps-denied-onboard.git
synced 2026-04-22 22:46:36 +00:00
Oleksandr Bezdieniezhnykh
39 lines
1.0 KiB
Python
39 lines
1.0 KiB
Python
from typing import AsyncIterator
|
|
|
|
from .base import SSEEventStreamerBase
|
|
from models.results import FrameResult
|
|
from models.recovery import UserInputRequest
|
|
|
|
|
|
class SSEEventStreamer(SSEEventStreamerBase):
|
|
async def emit_frame_result(
|
|
self, flight_id: str, result: FrameResult
|
|
) -> None:
|
|
raise NotImplementedError
|
|
|
|
async def emit_status_update(
|
|
self, flight_id: str, status: dict
|
|
) -> None:
|
|
raise NotImplementedError
|
|
|
|
async def emit_user_input_request(
|
|
self, flight_id: str, request: UserInputRequest
|
|
) -> None:
|
|
raise NotImplementedError
|
|
|
|
async def emit_error(
|
|
self, flight_id: str, error: str
|
|
) -> None:
|
|
raise NotImplementedError
|
|
|
|
async def subscribe(self, flight_id: str) -> AsyncIterator[dict]:
|
|
raise NotImplementedError
|
|
yield
|
|
|
|
async def unsubscribe(self, flight_id: str, subscriber_id: str) -> None:
|
|
raise NotImplementedError
|
|
|
|
async def get_subscriber_count(self, flight_id: str) -> int:
|
|
raise NotImplementedError
|
|
|