Fishjam Python Server SDK

117class FishjamClient(Client):
118    """Allows for managing rooms"""
119
120    def __init__(
121        self,
122        fishjam_id: str,
123        management_token: str,
124    ):
125        """
126        Create a FishjamClient instance, providing the fishjam id and management token.
127        """
128        super().__init__(fishjam_id=fishjam_id, management_token=management_token)
129
130    def create_peer(
131        self,
132        room_id: str,
133        options: PeerOptions | None = None,
134    ) -> tuple[Peer, str]:
135        """
136        Creates peer in the room
137
138        Returns a tuple (`Peer`, `PeerToken`) - the token is needed by Peer
139        to authenticate to Fishjam.
140
141        The possible options to pass for peer are `PeerOptions`.
142        """
143        options = options or PeerOptions()
144
145        peer_metadata = self.__parse_peer_metadata(options.metadata)
146        peer_options = PeerOptionsWebRTC(
147            enable_simulcast=options.enable_simulcast,
148            metadata=peer_metadata,
149            subscribe_mode=PeerOptionsWebRTCSubscribeMode(options.subscribe_mode),
150        )
151        body = AddPeerBody(type_=PeerType.WEBRTC, options=peer_options)
152
153        resp = cast(
154            PeerDetailsResponse,
155            self._request(room_add_peer, room_id=room_id, body=body),
156        )
157
158        return (resp.data.peer, resp.data.token)
159
160    def create_agent(self, room_id: str, options: AgentOptions | None = None):
161        options = options or AgentOptions()
162        body = AddPeerBody(
163            type_=PeerType.AGENT,
164            options=PeerOptionsAgent(
165                output=PeerOptionsAgentOutput(
166                    audio_format=PeerOptionsAgentOutputAudioFormat(
167                        options.output.audio_format
168                    ),
169                    audio_sample_rate=PeerOptionsAgentOutputAudioSampleRate(
170                        options.output.audio_sample_rate
171                    ),
172                ),
173                subscribe_mode=PeerOptionsAgentSubscribeMode(options.subscribe_mode),
174            ),
175        )
176
177        resp = cast(
178            PeerDetailsResponse,
179            self._request(room_add_peer, room_id=room_id, body=body),
180        )
181
182        return Agent(resp.data.peer.id, room_id, resp.data.token, self._fishjam_url)
183
184    def create_room(self, options: RoomOptions | None = None) -> Room:
185        """
186        Creates a new room
187        Returns the created `Room`
188        """
189        options = options or RoomOptions()
190
191        if options.video_codec is None:
192            codec = UNSET
193        else:
194            codec = RoomConfigVideoCodec(options.video_codec)
195
196        config = RoomConfig(
197            max_peers=options.max_peers,
198            video_codec=codec,
199            webhook_url=options.webhook_url,
200            room_type=RoomConfigRoomType(options.room_type),
201            public=options.public,
202        )
203
204        room = cast(
205            RoomCreateDetailsResponse, self._request(room_create_room, body=config)
206        ).data.room
207
208        return Room(config=room.config, id=room.id, peers=room.peers)
209
210    def get_all_rooms(self) -> list[Room]:
211        """Returns list of all rooms"""
212
213        rooms = cast(RoomsListingResponse, self._request(room_get_all_rooms)).data
214
215        return [
216            Room(config=room.config, id=room.id, peers=room.peers) for room in rooms
217        ]
218
219    def get_room(self, room_id: str) -> Room:
220        """Returns room with the given id"""
221
222        room = cast(
223            RoomDetailsResponse, self._request(room_get_room, room_id=room_id)
224        ).data
225
226        return Room(config=room.config, id=room.id, peers=room.peers)
227
228    def delete_peer(self, room_id: str, peer_id: str) -> None:
229        """Deletes peer"""
230
231        return self._request(room_delete_peer, id=peer_id, room_id=room_id)
232
233    def delete_room(self, room_id: str) -> None:
234        """Deletes a room"""
235
236        return self._request(room_delete_room, room_id=room_id)
237
238    def refresh_peer_token(self, room_id: str, peer_id: str) -> str:
239        """Refreshes peer token"""
240
241        response = cast(
242            PeerRefreshTokenResponse,
243            self._request(room_refresh_token, id=peer_id, room_id=room_id),
244        )
245
246        return response.data.token
247
248    def create_livestream_viewer_token(self, room_id: str) -> str:
249        """Generates viewer token for livestream rooms"""
250
251        response = cast(
252            ViewerToken, self._request(viewer_generate_viewer_token, room_id=room_id)
253        )
254
255        return response.token
256
257    def create_livestream_streamer_token(self, room_id: str) -> str:
258        """Generates streamer token for livestream rooms"""
259
260        response = cast(
261            StreamerToken,
262            self._request(streamer_generate_streamer_token, room_id=room_id),
263        )
264
265        return response.token
266
267    def subscribe_peer(self, room_id: str, peer_id: str, target_peer_id: str):
268        """Subscribe a peer to all tracks of another peer."""
269
270        self._request(
271            room_subscribe_peer,
272            room_id=room_id,
273            id=peer_id,
274            peer_id=target_peer_id,
275        )
276
277    def subscribe_tracks(self, room_id: str, peer_id: str, track_ids: list[str]):
278        """Subscribe a peer to specific tracks of another peer."""
279
280        self._request(
281            room_subscribe_tracks,
282            room_id=room_id,
283            id=peer_id,
284            body=SubscribeTracksBody(track_ids=track_ids),
285        )
286
287    def __parse_peer_metadata(self, metadata: dict | None) -> PeerOptionsWebRTCMetadata:
288        peer_metadata = PeerOptionsWebRTCMetadata()
289
290        if not metadata:
291            return peer_metadata
292
293        for key, value in metadata.items():
294            peer_metadata.additional_properties[key] = value
295
296        return peer_metadata

 35class FishjamNotifier:
 36    """
 37    Allows for receiving WebSocket messages from Fishjam.
 38    """
 39
 40    def __init__(
 41        self,
 42        fishjam_id: str,
 43        management_token: str,
 44    ):
 45        """
 46        Create FishjamNotifier instance, providing the fishjam id and management token.
 47        """
 48
 49        websocket_url = get_fishjam_url(fishjam_id).replace("http", "ws")
 50        self._fishjam_url = f"{websocket_url}/socket/server/websocket"
 51        self._management_token: str = management_token
 52        self._websocket: client.ClientConnection | None = None
 53        self._ready: bool = False
 54
 55        self._ready_event: asyncio.Event | None = None
 56
 57        self._notification_handler: NotificationHandler | None = None
 58
 59    def on_server_notification(self, handler: NotificationHandler):
 60        """
 61        Decorator used for defining handler for Fishjam Notifications
 62        """
 63        self._notification_handler = handler
 64        return handler
 65
 66    async def connect(self):
 67        """
 68        A coroutine which connects FishjamNotifier to Fishjam and listens for
 69        all incoming messages from the Fishjam.
 70
 71        It runs until the connection isn't closed.
 72
 73        The incoming messages are handled by the functions defined using the
 74        `on_server_notification` decorator.
 75
 76        The handler have to be defined before calling `connect`,
 77        otherwise the messages won't be received.
 78        """
 79        async with client.connect(self._fishjam_url) as websocket:
 80            try:
 81                self._websocket = websocket
 82                await self._authenticate()
 83
 84                if self._notification_handler:
 85                    await self._subscribe_event(
 86                        event=ServerMessageEventType.EVENT_TYPE_SERVER_NOTIFICATION
 87                    )
 88
 89                self._ready = True
 90                if self._ready_event:
 91                    self._ready_event.set()
 92
 93                await self._receive_loop()
 94            finally:
 95                self._websocket = None
 96
 97    async def wait_ready(self) -> None:
 98        """
 99        Waits until the notifier is connected and authenticated to Fishjam.
100
101        If already connected, returns immediately.
102        """
103        if self._ready:
104            return
105
106        if self._ready_event is None:
107            self._ready_event = asyncio.Event()
108
109        await self._ready_event.wait()
110
111    async def _authenticate(self):
112        if not self._websocket:
113            raise RuntimeError("Websocket is not connected")
114
115        msg = ServerMessage(
116            auth_request=ServerMessageAuthRequest(token=self._management_token)
117        )
118        await self._websocket.send(bytes(msg))
119
120        try:
121            message = await self._websocket.recv(decode=False)
122        except ConnectionClosed as exception:
123            if "invalid token" in str(exception):
124                raise RuntimeError("Invalid management token") from exception
125            raise
126
127        message = ServerMessage().parse(message)
128
129        _type, message = betterproto.which_one_of(message, "content")
130        assert isinstance(message, ServerMessageAuthenticated)
131
132    async def _receive_loop(self):
133        if not self._websocket:
134            raise RuntimeError("Websocket is not connected")
135        if not self._notification_handler:
136            raise RuntimeError("Notification handler is not defined")
137
138        while True:
139            message = cast(bytes, await self._websocket.recv())
140            message = ServerMessage().parse(message)
141            _which, message = betterproto.which_one_of(message, "content")
142
143            if isinstance(message, ALLOWED_NOTIFICATIONS):
144                res = self._notification_handler(message)
145                if inspect.isawaitable(res):
146                    await res
147
148    async def _subscribe_event(self, event: ServerMessageEventType):
149        if not self._websocket:
150            raise RuntimeError("Websocket is not connected")
151
152        request = ServerMessage(subscribe_request=ServerMessageSubscribeRequest(event))
153
154        await self._websocket.send(bytes(request))
155        message = cast(bytes, await self._websocket.recv())
156        message = ServerMessage().parse(message)
157        _which, message = betterproto.which_one_of(message, "content")
158        assert isinstance(message, ServerMessageSubscribeResponse)

18def receive_binary(binary: bytes) -> Union[AllowedNotification, None]:
19    """
20    Transform received protobuf notification to adequate notification instance.
21    The available notifications are listed in `fishjam.events` module.
22    """
23    message = ServerMessage().parse(binary)
24    _which, message = betterproto.which_one_of(message, "content")
25
26    if isinstance(message, ALLOWED_NOTIFICATIONS):
27        return message
28
29    return None

11@_attrs_define
12class PeerMetadata:
13    """Custom metadata set by the peer
14
15    Example:
16        {'name': 'FishjamUser'}
17
18    """
19
20    additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)
21
22    def to_dict(self) -> dict[str, Any]:
23        field_dict: dict[str, Any] = {}
24        field_dict.update(self.additional_properties)
25
26        return field_dict
27
28    @classmethod
29    def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T:
30        d = dict(src_dict)
31        peer_metadata = cls()
32
33        peer_metadata.additional_properties = d
34        return peer_metadata
35
36    @property
37    def additional_keys(self) -> list[str]:
38        return list(self.additional_properties.keys())
39
40    def __getitem__(self, key: str) -> Any:
41        return self.additional_properties[key]
42
43    def __setitem__(self, key: str, value: Any) -> None:
44        self.additional_properties[key] = value
45
46    def __delitem__(self, key: str) -> None:
47        del self.additional_properties[key]
48
49    def __contains__(self, key: str) -> bool:
50        return key in self.additional_properties

88@dataclass
89class PeerOptions:
90    """Options specific to a WebRTC Peer"""
91
92    enable_simulcast: bool = True
93    """Enables the peer to use simulcast"""
94    metadata: dict[str, Any] | None = None
95    """Peer metadata"""
96    subscribe_mode: Literal["auto", "manual"] = "auto"
97    """Configuration of peer's subscribing policy"""

65@dataclass
66class RoomOptions:
67    """Description of a room options"""
68
69    max_peers: int | None = None
70    """Maximum amount of peers allowed into the room"""
71    video_codec: Literal["h264", "vp8"] | None = None
72    """Enforces video codec for each peer in the room"""
73    webhook_url: str | None = None
74    """URL where Fishjam notifications will be sent"""
75    room_type: Literal[
76        "conference",
77        "audio_only",
78        "livestream",
79        "full_feature",
80        "broadcaster",
81        "audio_only_livestream",
82    ] = "conference"
83    """The use-case of the room. If not provided, this defaults to conference."""
84    public: bool = False
85    """True if livestream viewers can omit specifying a token."""

108@dataclass
109class AgentOptions:
110    """Options specific to a WebRTC Peer"""
111
112    output: AgentOutputOptions = field(default_factory=AgentOutputOptions)
113
114    subscribe_mode: Literal["auto", "manual"] = "auto"

100@dataclass
101class AgentOutputOptions:
102    """Options of the desired format of audio tracks going from Fishjam to the agent."""
103
104    audio_format: Literal["pcm16"] = "pcm16"
105    audio_sample_rate: Literal[16000, 24000] = 16000

53@dataclass
54class Room:
55    """Description of the room state"""
56
57    config: RoomConfig
58    """Room configuration"""
59    id: str
60    """Room ID"""
61    peers: list[Peer]
62    """List of all peers"""

 26@_attrs_define
 27class Peer:
 28    """Describes peer status
 29
 30    Attributes:
 31        id (str): Assigned peer id Example: 4a1c1164-5fb7-425d-89d7-24cdb8fff1cf.
 32        metadata (Union['PeerMetadata', None]): Custom metadata set by the peer Example: {'name': 'FishjamUser'}.
 33        status (PeerStatus): Informs about the peer status Example: disconnected.
 34        subscribe_mode (SubscribeMode): Configuration of peer's subscribing policy
 35        subscriptions (list[str]): Describes peer's subscriptions in manual mode
 36        tracks (list['Track']): List of all peer's tracks
 37        type_ (PeerType): Peer type Example: webrtc.
 38    """
 39
 40    id: str
 41    metadata: Union["PeerMetadata", None]
 42    status: PeerStatus
 43    subscribe_mode: SubscribeMode
 44    subscriptions: list[str]
 45    tracks: list["Track"]
 46    type_: PeerType
 47    additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)
 48
 49    def to_dict(self) -> dict[str, Any]:
 50        from ..models.peer_metadata import PeerMetadata
 51
 52        id = self.id
 53
 54        metadata: Union[None, dict[str, Any]]
 55        if isinstance(self.metadata, PeerMetadata):
 56            metadata = self.metadata.to_dict()
 57        else:
 58            metadata = self.metadata
 59
 60        status = self.status.value
 61
 62        subscribe_mode = self.subscribe_mode.value
 63
 64        subscriptions = self.subscriptions
 65
 66        tracks = []
 67        for tracks_item_data in self.tracks:
 68            tracks_item = tracks_item_data.to_dict()
 69            tracks.append(tracks_item)
 70
 71        type_ = self.type_.value
 72
 73        field_dict: dict[str, Any] = {}
 74        field_dict.update(self.additional_properties)
 75        field_dict.update(
 76            {
 77                "id": id,
 78                "metadata": metadata,
 79                "status": status,
 80                "subscribeMode": subscribe_mode,
 81                "subscriptions": subscriptions,
 82                "tracks": tracks,
 83                "type": type_,
 84            }
 85        )
 86
 87        return field_dict
 88
 89    @classmethod
 90    def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T:
 91        from ..models.peer_metadata import PeerMetadata
 92        from ..models.track import Track
 93
 94        d = dict(src_dict)
 95        id = d.pop("id")
 96
 97        def _parse_metadata(data: object) -> Union["PeerMetadata", None]:
 98            if data is None:
 99                return data
100            try:
101                if not isinstance(data, dict):
102                    raise TypeError()
103                componentsschemas_peer_metadata_type_0 = PeerMetadata.from_dict(data)
104
105                return componentsschemas_peer_metadata_type_0
106            except:  # noqa: E722
107                pass
108            return cast(Union["PeerMetadata", None], data)
109
110        metadata = _parse_metadata(d.pop("metadata"))
111
112        status = PeerStatus(d.pop("status"))
113
114        subscribe_mode = SubscribeMode(d.pop("subscribeMode"))
115
116        subscriptions = cast(list[str], d.pop("subscriptions"))
117
118        tracks = []
119        _tracks = d.pop("tracks")
120        for tracks_item_data in _tracks:
121            tracks_item = Track.from_dict(tracks_item_data)
122
123            tracks.append(tracks_item)
124
125        type_ = PeerType(d.pop("type"))
126
127        peer = cls(
128            id=id,
129            metadata=metadata,
130            status=status,
131            subscribe_mode=subscribe_mode,
132            subscriptions=subscriptions,
133            tracks=tracks,
134            type_=type_,
135        )
136
137        peer.additional_properties = d
138        return peer
139
140    @property
141    def additional_keys(self) -> list[str]:
142        return list(self.additional_properties.keys())
143
144    def __getitem__(self, key: str) -> Any:
145        return self.additional_properties[key]
146
147    def __setitem__(self, key: str, value: Any) -> None:
148        self.additional_properties[key] = value
149
150    def __delitem__(self, key: str) -> None:
151        del self.additional_properties[key]
152
153    def __contains__(self, key: str) -> bool:
154        return key in self.additional_properties