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
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