Jellyfish Python Server SDK

 36class RoomApi(BaseApi):
 37    """Allows for managing rooms"""
 38
 39    def __init__(
 40        self,
 41        server_address: str = "localhost:5002",
 42        server_api_token: str = "development",
 43        secure: bool = False,
 44    ):
 45        """
 46        Create RoomApi instance, providing the jellyfish address and api token.
 47        Set secure to `True` for `https` and `False` for `http` connection (default).
 48        """
 49        super().__init__(
 50            server_address=server_address,
 51            server_api_token=server_api_token,
 52            secure=secure,
 53        )
 54
 55    def create_room(
 56        self,
 57        room_id: str = None,
 58        max_peers: int = None,
 59        video_codec: Literal["h264", "vp8"] = None,
 60        webhook_url: str = None,
 61    ) -> (str, Room):
 62        """
 63        Creates a new room
 64
 65        Returns a tuple (`jellyfish_address`, `Room`) - the address of the Jellyfish
 66        in which the room has been created and the created `Room`
 67
 68        The returned address may be different from the current `RoomApi` instance.
 69        In such case, a new `RoomApi` instance has to be created using
 70        the returned address in order to interact with the room.
 71        """
 72
 73        if video_codec is not None:
 74            video_codec = RoomConfigVideoCodec(video_codec)
 75        else:
 76            video_codec = None
 77
 78        room_config = RoomConfig(
 79            room_id=room_id,
 80            max_peers=max_peers,
 81            video_codec=video_codec,
 82            webhook_url=webhook_url,
 83        )
 84
 85        resp = self._request(room_create_room, json_body=room_config)
 86        return (resp.data.jellyfish_address, resp.data.room)
 87
 88    def delete_room(self, room_id: str) -> None:
 89        """Deletes a room"""
 90
 91        return self._request(room_delete_room, room_id=room_id)
 92
 93    def get_all_rooms(self) -> list:
 94        """Returns list of all rooms"""
 95
 96        return self._request(room_get_all_rooms).data
 97
 98    def get_room(self, room_id: str) -> Room:
 99        """Returns room with the given id"""
100
101        return self._request(room_get_room, room_id=room_id).data
102
103    def add_peer(self, room_id: str, options: PeerOptionsWebRTC) -> (str, Peer):
104        """
105        Creates peer in the room
106
107        Currently only `webrtc` peer is supported
108
109        Returns a tuple (`peer_token`, `Peer`) - the token needed by Peer
110        to authenticate to Jellyfish and the new `Peer`
111        """
112
113        peer_type = "webrtc"
114        json_body = AddPeerJsonBody(type=peer_type, options=options)
115
116        resp = self._request(room_add_peer, room_id=room_id, json_body=json_body)
117        return (resp.data.token, resp.data.peer)
118
119    def delete_peer(self, room_id: str, peer_id: str) -> None:
120        """Deletes peer"""
121
122        return self._request(room_delete_peer, id=peer_id, room_id=room_id)
123
124    def add_component(
125        self,
126        room_id: str,
127        options: Union[ComponentOptionsFile, ComponentOptionsHLS, ComponentOptionsRTSP],
128    ) -> Union[ComponentFile, ComponentHLS, ComponentRTSP]:
129        """Creates component in the room"""
130
131        if isinstance(options, ComponentOptionsFile):
132            component_type = "file"
133        elif isinstance(options, ComponentOptionsHLS):
134            component_type = "hls"
135        elif isinstance(options, ComponentOptionsRTSP):
136            component_type = "rtsp"
137        else:
138            raise ValueError(
139                "options must be ComponentFile, ComponentOptionsHLS"
140                "or ComponentOptionsRTSP"
141            )
142
143        json_body = AddComponentJsonBody(type=component_type, options=options)
144
145        return self._request(
146            room_add_component, room_id=room_id, json_body=json_body
147        ).data
148
149    def delete_component(self, room_id: str, component_id: str) -> None:
150        """Deletes component"""
151
152        return self._request(room_delete_component, id=component_id, room_id=room_id)
153
154    def hls_subscribe(self, room_id: str, origins: [str]):
155        """
156        In order to subscribe to HLS peers/components,
157        the HLS component should be initialized with the subscribe_mode set to manual.
158        This mode proves beneficial when you do not wish to record or stream
159        all the available streams within a room via HLS.
160        It allows for selective addition instead –
161        you can manually select specific streams.
162        For instance, you could opt to record only the stream of an event's host.
163        """
164
165        return self._request(
166            hls_subscribe_hls_to,
167            room_id=room_id,
168            json_body=SubscriptionConfig(origins=origins),
169        )

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)

14def receive_json(json: Dict) -> betterproto.Message:
15    """
16    Transform received json notification to adequate notification instance.
17
18    The available notifications are listed in `jellyfish.events` module.
19    """
20    msg = json["notification"]
21    msg = bytes(msg, "utf-8")
22    message = ServerMessage().parse(msg)
23    _which, message = betterproto.which_one_of(message, "content")
24    return message

 18@_attrs_define
 19class Room:
 20    """Description of the room state"""
 21
 22    components: List[Union["ComponentFile", "ComponentHLS", "ComponentRTSP"]]
 23    """List of all components"""
 24    config: "RoomConfig"
 25    """Room configuration"""
 26    id: str
 27    """Room ID"""
 28    peers: List["Peer"]
 29    """List of all peers"""
 30    additional_properties: Dict[str, Any] = _attrs_field(init=False, factory=dict)
 31    """@private"""
 32
 33    def to_dict(self) -> Dict[str, Any]:
 34        """@private"""
 35        from ..models.component_hls import ComponentHLS
 36        from ..models.component_rtsp import ComponentRTSP
 37
 38        components = []
 39        for components_item_data in self.components:
 40            components_item: Dict[str, Any]
 41
 42            if isinstance(components_item_data, ComponentHLS):
 43                components_item = components_item_data.to_dict()
 44
 45            elif isinstance(components_item_data, ComponentRTSP):
 46                components_item = components_item_data.to_dict()
 47
 48            else:
 49                components_item = components_item_data.to_dict()
 50
 51            components.append(components_item)
 52
 53        config = self.config.to_dict()
 54
 55        id = self.id
 56        peers = []
 57        for peers_item_data in self.peers:
 58            peers_item = peers_item_data.to_dict()
 59
 60            peers.append(peers_item)
 61
 62        field_dict: Dict[str, Any] = {}
 63        field_dict.update(self.additional_properties)
 64        field_dict.update(
 65            {
 66                "components": components,
 67                "config": config,
 68                "id": id,
 69                "peers": peers,
 70            }
 71        )
 72
 73        return field_dict
 74
 75    @classmethod
 76    def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
 77        """@private"""
 78        from ..models.component_file import ComponentFile
 79        from ..models.component_hls import ComponentHLS
 80        from ..models.component_rtsp import ComponentRTSP
 81        from ..models.peer import Peer
 82        from ..models.room_config import RoomConfig
 83
 84        d = src_dict.copy()
 85        components = []
 86        _components = d.pop("components")
 87        for components_item_data in _components:
 88
 89            def _parse_components_item(
 90                data: object,
 91            ) -> Union["ComponentFile", "ComponentHLS", "ComponentRTSP"]:
 92                try:
 93                    if not isinstance(data, dict):
 94                        raise TypeError()
 95                    componentsschemas_component_type_0 = ComponentHLS.from_dict(data)
 96
 97                    return componentsschemas_component_type_0
 98                except:  # noqa: E722
 99                    pass
100                try:
101                    if not isinstance(data, dict):
102                        raise TypeError()
103                    componentsschemas_component_type_1 = ComponentRTSP.from_dict(data)
104
105                    return componentsschemas_component_type_1
106                except:  # noqa: E722
107                    pass
108                if not isinstance(data, dict):
109                    raise TypeError()
110                componentsschemas_component_type_2 = ComponentFile.from_dict(data)
111
112                return componentsschemas_component_type_2
113
114            components_item = _parse_components_item(components_item_data)
115
116            components.append(components_item)
117
118        config = RoomConfig.from_dict(d.pop("config"))
119
120        id = d.pop("id")
121
122        peers = []
123        _peers = d.pop("peers")
124        for peers_item_data in _peers:
125            peers_item = Peer.from_dict(peers_item_data)
126
127            peers.append(peers_item)
128
129        room = cls(
130            components=components,
131            config=config,
132            id=id,
133            peers=peers,
134        )
135
136        room.additional_properties = d
137        return room
138
139    @property
140    def additional_keys(self) -> List[str]:
141        """@private"""
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

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    room_id: Union[Unset, None, str] = UNSET
20    """Custom id used for identifying room within Jellyfish. Must be unique across all rooms. If not provided, random UUID is generated."""
21    video_codec: Union[Unset, None, RoomConfigVideoCodec] = UNSET
22    """Enforces video codec for each peer in the room"""
23    webhook_url: Union[Unset, None, str] = UNSET
24    """URL where Jellyfish notifications will be sent"""
25    additional_properties: Dict[str, Any] = _attrs_field(init=False, factory=dict)
26    """@private"""
27
28    def to_dict(self) -> Dict[str, Any]:
29        """@private"""
30        max_peers = self.max_peers
31        room_id = self.room_id
32        video_codec: Union[Unset, None, str] = UNSET
33        if not isinstance(self.video_codec, Unset):
34            video_codec = self.video_codec.value if self.video_codec else None
35
36        webhook_url = self.webhook_url
37
38        field_dict: Dict[str, Any] = {}
39        field_dict.update(self.additional_properties)
40        field_dict.update({})
41        if max_peers is not UNSET:
42            field_dict["maxPeers"] = max_peers
43        if room_id is not UNSET:
44            field_dict["roomId"] = room_id
45        if video_codec is not UNSET:
46            field_dict["videoCodec"] = video_codec
47        if webhook_url is not UNSET:
48            field_dict["webhookUrl"] = webhook_url
49
50        return field_dict
51
52    @classmethod
53    def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
54        """@private"""
55        d = src_dict.copy()
56        max_peers = d.pop("maxPeers", UNSET)
57
58        room_id = d.pop("roomId", UNSET)
59
60        _video_codec = d.pop("videoCodec", UNSET)
61        video_codec: Union[Unset, None, RoomConfigVideoCodec]
62        if _video_codec is None:
63            video_codec = None
64        elif isinstance(_video_codec, Unset):
65            video_codec = UNSET
66        else:
67            video_codec = RoomConfigVideoCodec(_video_codec)
68
69        webhook_url = d.pop("webhookUrl", UNSET)
70
71        room_config = cls(
72            max_peers=max_peers,
73            room_id=room_id,
74            video_codec=video_codec,
75            webhook_url=webhook_url,
76        )
77
78        room_config.additional_properties = d
79        return room_config
80
81    @property
82    def additional_keys(self) -> List[str]:
83        """@private"""
84        return list(self.additional_properties.keys())
85
86    def __getitem__(self, key: str) -> Any:
87        return self.additional_properties[key]
88
89    def __setitem__(self, key: str, value: Any) -> None:
90        self.additional_properties[key] = value
91
92    def __delitem__(self, key: str) -> None:
93        del self.additional_properties[key]
94
95    def __contains__(self, key: str) -> bool:
96        return key in self.additional_properties

 5class RoomConfigVideoCodec(str, Enum):
 6    """Enforces video codec for each peer in the room"""
 7
 8    H264 = "h264"
 9    VP8 = "vp8"
10
11    def __str__(self) -> str:
12        return str(self.value)

12@_attrs_define
13class Peer:
14    """Describes peer status"""
15
16    id: str
17    """Assigned peer id"""
18    status: PeerStatus
19    """Informs about the peer status"""
20    type: str
21    """Peer type"""
22    additional_properties: Dict[str, Any] = _attrs_field(init=False, factory=dict)
23    """@private"""
24
25    def to_dict(self) -> Dict[str, Any]:
26        """@private"""
27        id = self.id
28        status = self.status.value
29
30        type = self.type
31
32        field_dict: Dict[str, Any] = {}
33        field_dict.update(self.additional_properties)
34        field_dict.update(
35            {
36                "id": id,
37                "status": status,
38                "type": type,
39            }
40        )
41
42        return field_dict
43
44    @classmethod
45    def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
46        """@private"""
47        d = src_dict.copy()
48        id = d.pop("id")
49
50        status = PeerStatus(d.pop("status"))
51
52        type = d.pop("type")
53
54        peer = cls(
55            id=id,
56            status=status,
57            type=type,
58        )
59
60        peer.additional_properties = d
61        return peer
62
63    @property
64    def additional_keys(self) -> List[str]:
65        """@private"""
66        return list(self.additional_properties.keys())
67
68    def __getitem__(self, key: str) -> Any:
69        return self.additional_properties[key]
70
71    def __setitem__(self, key: str, value: Any) -> None:
72        self.additional_properties[key] = value
73
74    def __delitem__(self, key: str) -> None:
75        del self.additional_properties[key]
76
77    def __contains__(self, key: str) -> bool:
78        return key in self.additional_properties

12@_attrs_define
13class PeerOptionsWebRTC:
14    """Options specific to the WebRTC peer"""
15
16    enable_simulcast: Union[Unset, bool] = True
17    """Enables the peer to use simulcast"""
18    additional_properties: Dict[str, Any] = _attrs_field(init=False, factory=dict)
19    """@private"""
20
21    def to_dict(self) -> Dict[str, Any]:
22        """@private"""
23        enable_simulcast = self.enable_simulcast
24
25        field_dict: Dict[str, Any] = {}
26        field_dict.update(self.additional_properties)
27        field_dict.update({})
28        if enable_simulcast is not UNSET:
29            field_dict["enableSimulcast"] = enable_simulcast
30
31        return field_dict
32
33    @classmethod
34    def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
35        """@private"""
36        d = src_dict.copy()
37        enable_simulcast = d.pop("enableSimulcast", UNSET)
38
39        peer_options_web_rtc = cls(
40            enable_simulcast=enable_simulcast,
41        )
42
43        peer_options_web_rtc.additional_properties = d
44        return peer_options_web_rtc
45
46    @property
47    def additional_keys(self) -> List[str]:
48        """@private"""
49        return list(self.additional_properties.keys())
50
51    def __getitem__(self, key: str) -> Any:
52        return self.additional_properties[key]
53
54    def __setitem__(self, key: str, value: Any) -> None:
55        self.additional_properties[key] = value
56
57    def __delitem__(self, key: str) -> None:
58        del self.additional_properties[key]
59
60    def __contains__(self, key: str) -> bool:
61        return key in self.additional_properties