Jellyfish Python Server SDK

 43class RoomApi(BaseApi):
 44    """Allows for managing rooms"""
 45
 46    def __init__(
 47        self,
 48        server_address: str = "localhost:5002",
 49        server_api_token: str = "development",
 50        secure: bool = False,
 51    ):
 52        """
 53        Create RoomApi instance, providing the jellyfish address and api token.
 54        Set secure to `True` for `https` and `False` for `http` connection (default).
 55        """
 56        super().__init__(
 57            server_address=server_address,
 58            server_api_token=server_api_token,
 59            secure=secure,
 60        )
 61
 62    def create_room(
 63        self,
 64        room_id: str = None,
 65        max_peers: int = None,
 66        video_codec: Literal["h264", "vp8"] = None,
 67        webhook_url: str = None,
 68        peerless_purge_timeout: int = None,
 69        peer_disconnected_timeout: int = None,
 70    ) -> Tuple[str, Room]:
 71        """
 72        Creates a new room
 73
 74        Returns a tuple (`jellyfish_address`, `Room`) - the address of the Jellyfish
 75        in which the room has been created and the created `Room`
 76
 77        The returned address may be different from the current `RoomApi` instance.
 78        In such case, a new `RoomApi` instance has to be created using
 79        the returned address in order to interact with the room.
 80        """
 81
 82        if video_codec is not None:
 83            video_codec = RoomConfigVideoCodec(video_codec)
 84        else:
 85            video_codec = None
 86
 87        room_config = RoomConfig(
 88            room_id=room_id,
 89            max_peers=max_peers,
 90            video_codec=video_codec,
 91            webhook_url=webhook_url,
 92            peerless_purge_timeout=peerless_purge_timeout,
 93            peer_disconnected_timeout=peer_disconnected_timeout,
 94        )
 95
 96        resp = self._request(room_create_room, json_body=room_config)
 97        return (resp.data.jellyfish_address, resp.data.room)
 98
 99    def delete_room(self, room_id: str) -> None:
100        """Deletes a room"""
101
102        return self._request(room_delete_room, room_id=room_id)
103
104    def get_all_rooms(self) -> list:
105        """Returns list of all rooms"""
106
107        return self._request(room_get_all_rooms).data
108
109    def get_room(self, room_id: str) -> Room:
110        """Returns room with the given id"""
111
112        return self._request(room_get_room, room_id=room_id).data
113
114    def add_peer(
115        self, room_id: str, options: PeerOptionsWebRTC
116    ) -> PeerDetailsResponseData:
117        """
118        Creates peer in the room
119
120        Currently only `webrtc` peer is supported
121
122        Returns a tuple (`peer_token`, `Peer`) - the token needed by Peer
123        to authenticate to Jellyfish and the new `Peer`.
124
125        The possible options to pass for peer are `PeerOptionsWebRTC`.
126        """
127
128        peer_type = "webrtc"
129        json_body = AddPeerJsonBody(type=peer_type, options=options)
130
131        resp = self._request(room_add_peer, room_id=room_id, json_body=json_body)
132        return PeerDetailsResponseData(
133            peer=resp.data.peer,
134            token=resp.data.token,
135            peer_websocket_url=resp.data.peer_websocket_url,
136        )
137
138    def delete_peer(self, room_id: str, peer_id: str) -> None:
139        """Deletes peer"""
140
141        return self._request(room_delete_peer, id=peer_id, room_id=room_id)
142
143    def add_component(
144        self,
145        room_id: str,
146        options: Union[
147            ComponentOptionsFile,
148            ComponentOptionsHLS,
149            ComponentOptionsRecording,
150            ComponentOptionsRTSP,
151            ComponentOptionsSIP,
152        ],
153    ) -> Union[
154        ComponentFile, ComponentHLS, ComponentRecording, ComponentRTSP, ComponentSIP
155    ]:
156        """
157        Creates component in the room.
158        Currently there are 4 different components:
159        * File Component for which the options are `ComponentOptionsFile`
160        * HLS Component which options are `ComponentOptionsHLS`
161        * Recording Component which options are `ComponentOptionsRecording`
162        * RTSP Component which options are `ComponentOptionsRTSP`
163        * SIP Component which options are `ComponentOptionsSIP`
164        """
165
166        if isinstance(options, ComponentOptionsFile):
167            component_type = "file"
168        elif isinstance(options, ComponentOptionsHLS):
169            component_type = "hls"
170        elif isinstance(options, ComponentOptionsRecording):
171            component_type = "recording"
172        elif isinstance(options, ComponentOptionsRTSP):
173            component_type = "rtsp"
174        elif isinstance(options, ComponentOptionsSIP):
175            component_type = "sip"
176        else:
177            raise ValueError(
178                "options must be ComponentOptionsFile, ComponentOptionsHLS,"
179                "ComponentOptionsRTSP, ComponentOptionsRecording or ComponentOptionsSIP"
180            )
181
182        json_body = AddComponentJsonBody(type=component_type, options=options)
183
184        return self._request(
185            room_add_component, room_id=room_id, json_body=json_body
186        ).data
187
188    def delete_component(self, room_id: str, component_id: str) -> None:
189        """Deletes component"""
190
191        return self._request(room_delete_component, id=component_id, room_id=room_id)
192
193    def subscribe(self, room_id: str, component_id: str, origins: List[str]):
194        """
195        In order to subscribe the component to peers/components,
196        the component should be initialized with the subscribe_mode set to manual.
197        This mode proves beneficial when you do not wish to record or stream
198        all the available streams within a room.
199        It allows for selective addition instead –
200        you can manually select specific streams.
201        For instance, you could opt to record only the stream of an event's host.
202        """
203
204        return self._request(
205            subscribe_to,
206            room_id=room_id,
207            component_id=component_id,
208            json_body=SubscriptionConfig(origins=origins),
209        )
210
211    def sip_dial(self, room_id: str, component_id: str, phone_number: str):
212        """
213        Starts a phone call from a specified component to a provided phone number.
214
215        This is asynchronous operation.
216        In case of providing incorrect phone number you will receive
217        notification `ComponentCrashed`.
218        """
219
220        return self._request(
221            sip_dial,
222            room_id=room_id,
223            component_id=component_id,
224            json_body=DialConfig(phone_number=phone_number),
225        )
226
227    def sip_end_call(self, room_id: str, component_id: str):
228        """
229        End a phone call on a specified SIP component.
230
231        This is asynchronous operation.
232        """
233
234        return self._request(
235            sip_end_call,
236            room_id=room_id,
237            component_id=component_id,
238        )

10class RecordingApi(BaseApi):
11    """Allows for managing recordings"""
12
13    def __init__(
14        self,
15        server_address: str = "localhost:5002",
16        server_api_token: str = "development",
17        secure: bool = False,
18    ):
19        """
20        Create RecordingApi instance, providing the jellyfish address and api token.
21        Set secure to `True` for `https` and `False` for `http` connection (default).
22        """
23
24        super().__init__(
25            server_address=server_address,
26            server_api_token=server_api_token,
27            secure=secure,
28        )
29
30    def get_list(self) -> list:
31        """Returns a list of available recordings"""
32
33        return self._request(get_recordings).data
34
35    def delete(self, recording_id: str):
36        """Deletes recording with given id"""
37
38        return self._request(delete_recording, recording_id=recording_id)

 24class Notifier:
 25    """
 26    Allows for receiving WebSocket messages from Jellyfish.
 27    """
 28
 29    def __init__(
 30        self,
 31        server_address: str = "localhost:5002",
 32        server_api_token: str = "development",
 33        secure: bool = False,
 34    ):
 35        """
 36        Create Notifier instance, providing the jellyfish address and api token.
 37        Set secure to `True` for `wss` and `False` for `ws` connection (default).
 38        """
 39
 40        protocol = "wss" if secure else "ws"
 41        self._server_address = f"{protocol}://{server_address}/socket/server/websocket"
 42        self._server_api_token = server_api_token
 43        self._websocket = None
 44        self._ready = False
 45
 46        self._ready_event: asyncio.Event = None
 47
 48        self._notification_handler: Callable = None
 49        self._metrics_handler: Callable = None
 50
 51    def on_server_notification(self, handler: Callable[[Any], None]):
 52        """
 53        Decorator used for defining handler for ServerNotifications
 54        i.e. all messages other than `ServerMessageMetricsReport`.
 55        """
 56        self._notification_handler = handler
 57        return handler
 58
 59    def on_metrics(self, handler: Callable[[ServerMessageMetricsReport], None]):
 60        """
 61        Decorator used for defining handler for `ServerMessageMetricsReport`.
 62        """
 63        self._metrics_handler = handler
 64        return handler
 65
 66    async def connect(self):
 67        """
 68        A coroutine which connects Notifier to Jellyfish and listens for all incoming
 69        messages from the Jellyfish.
 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` and `on_metrics` decorators.
 75
 76        The handlers have to be defined before calling `connect`,
 77        otherwise the messages won't be received.
 78        """
 79        async with client.connect(self._server_address) 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                if self._metrics_handler:
 90                    await self._subscribe_event(
 91                        event=ServerMessageEventType.EVENT_TYPE_METRICS
 92                    )
 93
 94                self._ready = True
 95                if self._ready_event:
 96                    self._ready_event.set()
 97
 98                await self._receive_loop()
 99            finally:
100                self._websocket = None
101
102    async def wait_ready(self) -> True:
103        """
104        Waits until the notifier is connected and authenticated to Jellyfish.
105
106        If already connected, returns `True` immediately.
107        """
108        if self._ready:
109            return True
110
111        if self._ready_event is None:
112            self._ready_event = asyncio.Event()
113
114        await self._ready_event.wait()
115
116    async def _authenticate(self):
117        msg = ServerMessage(
118            auth_request=ServerMessageAuthRequest(token=self._server_api_token)
119        )
120        await self._websocket.send(bytes(msg))
121
122        try:
123            message = await self._websocket.recv()
124        except ConnectionClosed as exception:
125            if "invalid token" in str(exception):
126                raise RuntimeError("Invalid server_api_token") from exception
127            raise
128
129        message = ServerMessage().parse(message)
130
131        _type, message = betterproto.which_one_of(message, "content")
132        assert isinstance(message, ServerMessageAuthenticated)
133
134    async def _receive_loop(self):
135        while True:
136            message = await self._websocket.recv()
137            message = ServerMessage().parse(message)
138            _which, message = betterproto.which_one_of(message, "content")
139
140            if isinstance(message, ServerMessageMetricsReport):
141                self._metrics_handler(message)
142            else:
143                self._notification_handler(message)
144
145    async def _subscribe_event(self, event: ServerMessageEventType):
146        request = ServerMessage(subscribe_request=ServerMessageSubscribeRequest(event))
147
148        await self._websocket.send(bytes(request))
149        message = await self._websocket.recv()
150        message = ServerMessage().parse(message)
151        _which, message = betterproto.which_one_of(message, "content")
152        assert isinstance(message, ServerMessageSubscribeResponse)

12def receive_binary(binary: bytes) -> betterproto.Message:
13    """
14    Transform received protobuf notification to adequate notification instance.
15
16    The available notifications are listed in `jellyfish.events` module.
17    """
18    message = ServerMessage().parse(binary)
19    _which, message = betterproto.which_one_of(message, "content")
20    return message

 28@_attrs_define
 29class Room:
 30    """Description of the room state"""
 31
 32    components: List[
 33        Union[
 34            "ComponentFile",
 35            "ComponentHLS",
 36            "ComponentRTSP",
 37            "ComponentRecording",
 38            "ComponentSIP",
 39        ]
 40    ]
 41    """List of all components"""
 42    config: "RoomConfig"
 43    """Room configuration"""
 44    id: str
 45    """Room ID"""
 46    peers: List["Peer"]
 47    """List of all peers"""
 48    additional_properties: Dict[str, Any] = _attrs_field(init=False, factory=dict)
 49    """@private"""
 50
 51    def to_dict(self) -> Dict[str, Any]:
 52        """@private"""
 53        from ..models.component_file import ComponentFile
 54        from ..models.component_hls import ComponentHLS
 55        from ..models.component_rtsp import ComponentRTSP
 56        from ..models.component_sip import ComponentSIP
 57
 58        components = []
 59        for components_item_data in self.components:
 60            components_item: Dict[str, Any]
 61
 62            if isinstance(components_item_data, ComponentHLS):
 63                components_item = components_item_data.to_dict()
 64
 65            elif isinstance(components_item_data, ComponentRTSP):
 66                components_item = components_item_data.to_dict()
 67
 68            elif isinstance(components_item_data, ComponentFile):
 69                components_item = components_item_data.to_dict()
 70
 71            elif isinstance(components_item_data, ComponentSIP):
 72                components_item = components_item_data.to_dict()
 73
 74            else:
 75                components_item = components_item_data.to_dict()
 76
 77            components.append(components_item)
 78
 79        config = self.config.to_dict()
 80
 81        id = self.id
 82        peers = []
 83        for peers_item_data in self.peers:
 84            peers_item = peers_item_data.to_dict()
 85
 86            peers.append(peers_item)
 87
 88        field_dict: Dict[str, Any] = {}
 89        field_dict.update(self.additional_properties)
 90        field_dict.update(
 91            {
 92                "components": components,
 93                "config": config,
 94                "id": id,
 95                "peers": peers,
 96            }
 97        )
 98
 99        return field_dict
100
101    @classmethod
102    def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
103        """@private"""
104        from ..models.component_file import ComponentFile
105        from ..models.component_hls import ComponentHLS
106        from ..models.component_recording import ComponentRecording
107        from ..models.component_rtsp import ComponentRTSP
108        from ..models.component_sip import ComponentSIP
109        from ..models.peer import Peer
110        from ..models.room_config import RoomConfig
111
112        d = src_dict.copy()
113        components = []
114        _components = d.pop("components")
115        for components_item_data in _components:
116
117            def _parse_components_item(
118                data: object,
119            ) -> Union[
120                "ComponentFile",
121                "ComponentHLS",
122                "ComponentRTSP",
123                "ComponentRecording",
124                "ComponentSIP",
125            ]:
126                try:
127                    if not isinstance(data, dict):
128                        raise TypeError()
129                    componentsschemas_component_type_0 = ComponentHLS.from_dict(data)
130
131                    return componentsschemas_component_type_0
132                except:  # noqa: E722
133                    pass
134                try:
135                    if not isinstance(data, dict):
136                        raise TypeError()
137                    componentsschemas_component_type_1 = ComponentRTSP.from_dict(data)
138
139                    return componentsschemas_component_type_1
140                except:  # noqa: E722
141                    pass
142                try:
143                    if not isinstance(data, dict):
144                        raise TypeError()
145                    componentsschemas_component_type_2 = ComponentFile.from_dict(data)
146
147                    return componentsschemas_component_type_2
148                except:  # noqa: E722
149                    pass
150                try:
151                    if not isinstance(data, dict):
152                        raise TypeError()
153                    componentsschemas_component_type_3 = ComponentSIP.from_dict(data)
154
155                    return componentsschemas_component_type_3
156                except:  # noqa: E722
157                    pass
158                if not isinstance(data, dict):
159                    raise TypeError()
160                componentsschemas_component_type_4 = ComponentRecording.from_dict(data)
161
162                return componentsschemas_component_type_4
163
164            components_item = _parse_components_item(components_item_data)
165
166            components.append(components_item)
167
168        config = RoomConfig.from_dict(d.pop("config"))
169
170        id = d.pop("id")
171
172        peers = []
173        _peers = d.pop("peers")
174        for peers_item_data in _peers:
175            peers_item = Peer.from_dict(peers_item_data)
176
177            peers.append(peers_item)
178
179        room = cls(
180            components=components,
181            config=config,
182            id=id,
183            peers=peers,
184        )
185
186        room.additional_properties = d
187        return room
188
189    @property
190    def additional_keys(self) -> List[str]:
191        """@private"""
192        return list(self.additional_properties.keys())
193
194    def __getitem__(self, key: str) -> Any:
195        return self.additional_properties[key]
196
197    def __setitem__(self, key: str, value: Any) -> None:
198        self.additional_properties[key] = value
199
200    def __delitem__(self, key: str) -> None:
201        del self.additional_properties[key]
202
203    def __contains__(self, key: str) -> bool:
204        return key in self.additional_properties

 13@_attrs_define
 14class RoomConfig:
 15    """Room configuration"""
 16
 17    max_peers: Union[Unset, None, int] = UNSET
 18    """Maximum amount of peers allowed into the room"""
 19    peer_disconnected_timeout: Union[Unset, None, int] = UNSET
 20    """Duration (in seconds) after which the peer will be removed if it is disconnected. If not provided, this feature is disabled."""
 21    peerless_purge_timeout: Union[Unset, None, int] = UNSET
 22    """Duration (in seconds) after which the room will be removed if no peers are connected. If not provided, this feature is disabled."""
 23    room_id: Union[Unset, None, str] = UNSET
 24    """Custom id used for identifying room within Jellyfish. Must be unique across all rooms. If not provided, random UUID is generated."""
 25    video_codec: Union[Unset, None, RoomConfigVideoCodec] = UNSET
 26    """Enforces video codec for each peer in the room"""
 27    webhook_url: Union[Unset, None, str] = UNSET
 28    """URL where Jellyfish notifications will be sent"""
 29    additional_properties: Dict[str, Any] = _attrs_field(init=False, factory=dict)
 30    """@private"""
 31
 32    def to_dict(self) -> Dict[str, Any]:
 33        """@private"""
 34        max_peers = self.max_peers
 35        peer_disconnected_timeout = self.peer_disconnected_timeout
 36        peerless_purge_timeout = self.peerless_purge_timeout
 37        room_id = self.room_id
 38        video_codec: Union[Unset, None, str] = UNSET
 39        if not isinstance(self.video_codec, Unset):
 40            video_codec = self.video_codec.value if self.video_codec else None
 41
 42        webhook_url = self.webhook_url
 43
 44        field_dict: Dict[str, Any] = {}
 45        field_dict.update(self.additional_properties)
 46        field_dict.update({})
 47        if max_peers is not UNSET:
 48            field_dict["maxPeers"] = max_peers
 49        if peer_disconnected_timeout is not UNSET:
 50            field_dict["peerDisconnectedTimeout"] = peer_disconnected_timeout
 51        if peerless_purge_timeout is not UNSET:
 52            field_dict["peerlessPurgeTimeout"] = peerless_purge_timeout
 53        if room_id is not UNSET:
 54            field_dict["roomId"] = room_id
 55        if video_codec is not UNSET:
 56            field_dict["videoCodec"] = video_codec
 57        if webhook_url is not UNSET:
 58            field_dict["webhookUrl"] = webhook_url
 59
 60        return field_dict
 61
 62    @classmethod
 63    def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
 64        """@private"""
 65        d = src_dict.copy()
 66        max_peers = d.pop("maxPeers", UNSET)
 67
 68        peer_disconnected_timeout = d.pop("peerDisconnectedTimeout", UNSET)
 69
 70        peerless_purge_timeout = d.pop("peerlessPurgeTimeout", UNSET)
 71
 72        room_id = d.pop("roomId", UNSET)
 73
 74        _video_codec = d.pop("videoCodec", UNSET)
 75        video_codec: Union[Unset, None, RoomConfigVideoCodec]
 76        if _video_codec is None:
 77            video_codec = None
 78        elif isinstance(_video_codec, Unset):
 79            video_codec = UNSET
 80        else:
 81            video_codec = RoomConfigVideoCodec(_video_codec)
 82
 83        webhook_url = d.pop("webhookUrl", UNSET)
 84
 85        room_config = cls(
 86            max_peers=max_peers,
 87            peer_disconnected_timeout=peer_disconnected_timeout,
 88            peerless_purge_timeout=peerless_purge_timeout,
 89            room_id=room_id,
 90            video_codec=video_codec,
 91            webhook_url=webhook_url,
 92        )
 93
 94        room_config.additional_properties = d
 95        return room_config
 96
 97    @property
 98    def additional_keys(self) -> List[str]:
 99        """@private"""
100        return list(self.additional_properties.keys())
101
102    def __getitem__(self, key: str) -> Any:
103        return self.additional_properties[key]
104
105    def __setitem__(self, key: str, value: Any) -> None:
106        self.additional_properties[key] = value
107
108    def __delitem__(self, key: str) -> None:
109        del self.additional_properties[key]
110
111    def __contains__(self, key: str) -> bool:
112        return key in self.additional_properties