Mercurial > libervia-backend
diff libervia/backend/plugins/plugin_xep_0343.py @ 4270:0d7bb4df2343
Reformatted code base using black.
author | Goffi <goffi@goffi.org> |
---|---|
date | Wed, 19 Jun 2024 18:44:57 +0200 |
parents | 79c8a70e1813 |
children | 39ac821ebbdb |
line wrap: on
line diff
--- a/libervia/backend/plugins/plugin_xep_0343.py Tue Jun 18 12:06:45 2024 +0200 +++ b/libervia/backend/plugins/plugin_xep_0343.py Wed Jun 19 18:44:57 2024 +0200 @@ -30,7 +30,10 @@ from libervia.backend.core.i18n import _ from libervia.backend.core.log import getLogger from libervia.backend.core.xmpp import SatXMPPEntity -from libervia.backend.plugins.plugin_xep_0166.models import BaseTransportHandler, ContentData +from libervia.backend.plugins.plugin_xep_0166.models import ( + BaseTransportHandler, + ContentData, +) from libervia.backend.tools.common import data_format from .plugin_xep_0167 import mapping @@ -50,9 +53,9 @@ C.PI_HANDLER: "yes", C.PI_DESCRIPTION: _("""Use WebRTC to create a generic data transport."""), } -NS_JINGLE_WEBRTC_DATACHANNELS: Final[ - str -] = "urn:xmpp:jingle:transports:webrtc-datachannel:1" +NS_JINGLE_WEBRTC_DATACHANNELS: Final[str] = ( + "urn:xmpp:jingle:transports:webrtc-datachannel:1" +) class XEP_0343(BaseTransportHandler): @@ -73,9 +76,7 @@ ) host.trigger.add("XEP-0176_jingle_handler_send_buffer", self._on_send_ice_buffer) host.trigger.add("XEP-0176_ice_candidate_send", self._on_ice_candidate_send) - host.trigger.add( - "XEP-0234_file_jingle_send", self._file_jingle_send - ) + host.trigger.add("XEP-0234_file_jingle_send", self._file_jingle_send) def get_handler(self, client: SatXMPPEntity): return XEP0343Handler() @@ -115,7 +116,7 @@ application_data: dict, app_data_key: str, media_data: dict, - media: str + media: str, ) -> None: """Generate "sctp-port" and "max-message-size" attributes""" transport_data = content_data["transport_data"] @@ -127,10 +128,7 @@ if max_message_size is not None: sdp_lines.append(f"a=max-message-size:{max_message_size}") - def _wrap_transport_element( - self, - transport_elt: domish.Element - ) -> None: + def _wrap_transport_element(self, transport_elt: domish.Element) -> None: """Wrap the XEP-0176 transport in a transport with this XEP namespace @param transport_elt: ICE UDP <transport>. Must be already a child of a <content> @@ -152,7 +150,7 @@ content_name: str, content_data: dict, transport_elt: domish.Element, - iq_elt: domish.Element + iq_elt: domish.Element, ) -> bool: if content_data["transport"].handler == self: self._wrap_transport_element(transport_elt) @@ -165,7 +163,7 @@ media_ice_data: dict[str, dict], content_name: str, content_data: dict, - iq_elt: domish.Element + iq_elt: domish.Element, ) -> bool: if content_data["transport"].handler == self: transport_elt = iq_elt.jingle.content.transport @@ -175,10 +173,7 @@ return True async def _file_jingle_send( - self, - client: SatXMPPEntity, - peer_jid: jid.JID, - content: dict + self, client: SatXMPPEntity, peer_jid: jid.JID, content: dict ) -> None: call_data = content["app_kwargs"]["extra"].pop("call_data", None) if call_data: @@ -186,9 +181,7 @@ try: application_data = call_data["application"] except KeyError: - raise exceptions.DataError( - '"call_data" must have an application media.' - ) + raise exceptions.DataError('"call_data" must have an application media.') try: content["transport_data"] = { "sctp-port": metadata["sctp-port"], @@ -198,7 +191,7 @@ "pwd": metadata["ice-pwd"], "candidates": application_data.pop("ice-candidates"), "fingerprint": application_data.pop("fingerprint", {}), - } + }, } except KeyError as e: raise exceptions.DataError(f"Mandatory key is missing: {e}") @@ -212,16 +205,14 @@ content_data = session["contents"][content_name] transport_data = content_data["transport_data"] ice_transport_elt = await self._ice_udp.jingle_session_init( - client, - session, - content_name + client, session, content_name ) transport_elt = domish.Element( (NS_JINGLE_WEBRTC_DATACHANNELS, "transport"), attribs={ "sctp-port": str(transport_data["sctp-port"]), - "max-message-size": str(transport_data["max-message-size"]) - } + "max-message-size": str(transport_data["max-message-size"]), + }, ) transport_elt.addChild(ice_transport_elt) return transport_elt