# Generated by the protocol buffer compiler.  DO NOT EDIT!
# sources: tracking.proto
# plugin: python-betterproto
from dataclasses import dataclass
from typing import Dict, List, Optional

import betterproto
from betterproto.grpc.grpclib_server import ServiceBase
import grpclib


@dataclass(eq=False, repr=False)
class Empty(betterproto.Message):
    pass


@dataclass(eq=False, repr=False)
class Frame(betterproto.Message):
    image: bytes = betterproto.bytes_field(1)
    timestamp: int = betterproto.uint64_field(2)
    frame_number: int = betterproto.uint64_field(3)


@dataclass(eq=False, repr=False)
class Detections(betterproto.Message):
    frame: "Frame" = betterproto.message_field(1)
    detections: List["DetectionsDetection"] = betterproto.message_field(2)


@dataclass(eq=False, repr=False)
class DetectionsDetection(betterproto.Message):
    bbox: List[int] = betterproto.int64_field(1)
    conf: float = betterproto.float_field(2)
    clazz: str = betterproto.string_field(3)


class ObjectTrackingStub(betterproto.ServiceStub):
    async def add_detections(
        self,
        *,
        frame: "Frame" = None,
        detections: Optional[List["DetectionsDetection"]] = None,
    ) -> "Empty":
        detections = detections or []

        request = Detections()
        if frame is not None:
            request.frame = frame
        if detections is not None:
            request.detections = detections

        return await self._unary_unary(
            "/tracking.ObjectTracking/AddDetections", request, Empty
        )

    async def track_objects(
        self, *, image: bytes = b"", timestamp: int = 0, frame_number: int = 0
    ) -> "Empty":

        request = Frame()
        request.image = image
        request.timestamp = timestamp
        request.frame_number = frame_number

        return await self._unary_unary(
            "/tracking.ObjectTracking/TrackObjects", request, Empty
        )


class ObjectTrackingBase(ServiceBase):
    async def add_detections(
        self, frame: "Frame", detections: Optional[List["DetectionsDetection"]]
    ) -> "Empty":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def track_objects(
        self, image: bytes, timestamp: int, frame_number: int
    ) -> "Empty":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def __rpc_add_detections(self, stream: grpclib.server.Stream) -> None:
        request = await stream.recv_message()

        request_kwargs = {
            "frame": request.frame,
            "detections": request.detections,
        }

        response = await self.add_detections(**request_kwargs)
        await stream.send_message(response)

    async def __rpc_track_objects(self, stream: grpclib.server.Stream) -> None:
        request = await stream.recv_message()

        request_kwargs = {
            "image": request.image,
            "timestamp": request.timestamp,
            "frame_number": request.frame_number,
        }

        response = await self.track_objects(**request_kwargs)
        await stream.send_message(response)

    def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
        return {
            "/tracking.ObjectTracking/AddDetections": grpclib.const.Handler(
                self.__rpc_add_detections,
                grpclib.const.Cardinality.UNARY_UNARY,
                Detections,
                Empty,
            ),
            "/tracking.ObjectTracking/TrackObjects": grpclib.const.Handler(
                self.__rpc_track_objects,
                grpclib.const.Cardinality.UNARY_UNARY,
                Frame,
                Empty,
            ),
        }

# Generated by the protocol buffer compiler.  DO NOT EDIT!
# sources: tracking.proto
# plugin: python-betterproto
from dataclasses import dataclass
from typing import Dict, List, Optional

import betterproto
from betterproto.grpc.grpclib_server import ServiceBase
import grpclib


@dataclass(eq=False, repr=False)
class Empty(betterproto.Message):
    pass


@dataclass(eq=False, repr=False)
class Frame(betterproto.Message):
    image: bytes = betterproto.bytes_field(1)
    timestamp: int = betterproto.uint64_field(2)
    frame_number: int = betterproto.uint64_field(3)


@dataclass(eq=False, repr=False)
class Detections(betterproto.Message):
    frame: "Frame" = betterproto.message_field(1)
    detections: List["DetectionsDetection"] = betterproto.message_field(2)


@dataclass(eq=False, repr=False)
class DetectionsDetection(betterproto.Message):
    bbox: List[int] = betterproto.int64_field(1)
    conf: float = betterproto.float_field(2)
    clazz: str = betterproto.string_field(3)


class ObjectTrackingStub(betterproto.ServiceStub):
    async def add_detections(
        self,
        *,
        frame: "Frame" = None,
        detections: Optional[List["DetectionsDetection"]] = None,
    ) -> "Empty":
        detections = detections or []

        request = Detections()
        if frame is not None:
            request.frame = frame
        if detections is not None:
            request.detections = detections

        return await self._unary_unary(
            "/tracking.ObjectTracking/AddDetections", request, Empty
        )

    async def track_objects(
        self, *, image: bytes = b"", timestamp: int = 0, frame_number: int = 0
    ) -> "Empty":

        request = Frame()
        request.image = image
        request.timestamp = timestamp
        request.frame_number = frame_number

        return await self._unary_unary(
            "/tracking.ObjectTracking/TrackObjects", request, Empty
        )


class ObjectTrackingBase(ServiceBase):
    async def add_detections(
        self, frame: "Frame", detections: Optional[List["DetectionsDetection"]]
    ) -> "Empty":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def track_objects(
        self, image: bytes, timestamp: int, frame_number: int
    ) -> "Empty":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def __rpc_add_detections(self, stream: grpclib.server.Stream) -> None:
        request = await stream.recv_message()

        request_kwargs = {
            "frame": request.frame,
            "detections": request.detections,
        }

        response = await self.add_detections(**request_kwargs)
        await stream.send_message(response)

    async def __rpc_track_objects(self, stream: grpclib.server.Stream) -> None:
        request = await stream.recv_message()

        request_kwargs = {
            "image": request.image,
            "timestamp": request.timestamp,
            "frame_number": request.frame_number,
        }

        response = await self.track_objects(**request_kwargs)
        await stream.send_message(response)

    def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
        return {
            "/tracking.ObjectTracking/AddDetections": grpclib.const.Handler(
                self.__rpc_add_detections,
                grpclib.const.Cardinality.UNARY_UNARY,
                Detections,
                Empty,
            ),
            "/tracking.ObjectTracking/TrackObjects": grpclib.const.Handler(
                self.__rpc_track_objects,
                grpclib.const.Cardinality.UNARY_UNARY,
                Frame,
                Empty,
            ),
        }

# Generated by the protocol buffer compiler.  DO NOT EDIT! # sources: tracking.proto # plugin: python-betterproto