github-actions[bot]
5c97379765
date: 2026-08-01T16:05:18 master commit: 3a05c03079d796f533f342489b3f681cfd21f98d
82 lines
2.8 KiB
Python
82 lines
2.8 KiB
Python
import abc
|
|
from typing import Dict, List, Optional
|
|
|
|
from teleoprtc.stream import RTCSessionDescription, WebRTCBaseStream, WebRTCOfferStream, WebRTCAnswerStream, ConnectionProvider
|
|
from teleoprtc.tracks import TiciVideoStreamTrack, TiciTrackWrapper
|
|
|
|
|
|
class WebRTCStreamBuilder(abc.ABC):
|
|
@abc.abstractmethod
|
|
def stream(self) -> WebRTCBaseStream:
|
|
raise NotImplementedError
|
|
|
|
|
|
class WebRTCOfferBuilder(WebRTCStreamBuilder):
|
|
def __init__(self, connection_provider: ConnectionProvider, bind_address: Optional[str] = None):
|
|
self.connection_provider = connection_provider
|
|
self.bind_address = bind_address
|
|
self.requested_camera_types: List[str] = []
|
|
self.requested_audio = False
|
|
self.audio_tracks: List[object] = []
|
|
self.messaging_enabled = False
|
|
|
|
def offer_to_receive_video_stream(self, camera_type: str):
|
|
assert camera_type in ["driver", "wideRoad", "road"]
|
|
self.requested_camera_types.append(camera_type)
|
|
|
|
def offer_to_receive_audio_stream(self):
|
|
self.requested_audio = True
|
|
|
|
def add_audio_stream(self, track: object):
|
|
assert len(self.audio_tracks) == 0
|
|
self.audio_tracks = [track]
|
|
|
|
def add_messaging(self):
|
|
self.messaging_enabled = True
|
|
|
|
def stream(self) -> WebRTCBaseStream:
|
|
return WebRTCOfferStream(
|
|
self.connection_provider,
|
|
consumed_camera_types=self.requested_camera_types,
|
|
consume_audio=self.requested_audio,
|
|
video_producer_tracks=[],
|
|
audio_producer_tracks=self.audio_tracks,
|
|
should_add_data_channel=self.messaging_enabled,
|
|
bind_address=self.bind_address,
|
|
)
|
|
|
|
|
|
class WebRTCAnswerBuilder(WebRTCStreamBuilder):
|
|
def __init__(self, offer_sdp: str, bind_address: Optional[str] = None):
|
|
self.offer_sdp = offer_sdp
|
|
self.bind_address = bind_address
|
|
self.video_tracks: Dict[str, TiciVideoStreamTrack] = {}
|
|
self.requested_audio = False
|
|
self.audio_tracks: List[object] = []
|
|
|
|
def offer_to_receive_audio_stream(self):
|
|
self.requested_audio = True
|
|
|
|
def add_video_stream(self, camera_type: str, track: object):
|
|
assert camera_type not in self.video_tracks
|
|
assert camera_type in ["driver", "wideRoad", "road"]
|
|
if not isinstance(track, TiciVideoStreamTrack):
|
|
track = TiciTrackWrapper(camera_type, track)
|
|
self.video_tracks[camera_type] = track
|
|
|
|
def add_audio_stream(self, track: object):
|
|
assert len(self.audio_tracks) == 0
|
|
self.audio_tracks = [track]
|
|
|
|
def stream(self) -> WebRTCBaseStream:
|
|
description = RTCSessionDescription(sdp=self.offer_sdp, type="offer")
|
|
return WebRTCAnswerStream(
|
|
description,
|
|
consumed_camera_types=[],
|
|
consume_audio=self.requested_audio,
|
|
video_producer_tracks=list(self.video_tracks.values()),
|
|
audio_producer_tracks=self.audio_tracks,
|
|
should_add_data_channel=False,
|
|
bind_address=self.bind_address,
|
|
)
|