Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_xep_0339.py @ 4071:4b842c1fb686

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
refactoring: renamed `sat` package to `libervia.backend`
author Goffi <goffi@goffi.org>
date Fri, 02 Jun 2023 11:49:51 +0200
parents sat/plugins/plugin_xep_0339.py@3f62c2445df1
children 0d7bb4df2343
comparison
equal deleted inserted replaced
4070:d10748475025 4071:4b842c1fb686
1 #!/usr/bin/env python3
2
3 # Libervia plugin
4 # Copyright (C) 2009-2023 Jérôme Poisson (goffi@goffi.org)
5
6 # This program is free software: you can redistribute it and/or modify
7 # it under the terms of the GNU Affero General Public License as published by
8 # the Free Software Foundation, either version 3 of the License, or
9 # (at your option) any later version.
10
11 # This program is distributed in the hope that it will be useful,
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 # GNU Affero General Public License for more details.
15
16 # You should have received a copy of the GNU Affero General Public License
17 # along with this program. If not, see <http://www.gnu.org/licenses/>.
18
19 from typing import List
20
21 from twisted.words.protocols.jabber.xmlstream import XMPPHandler
22 from twisted.words.xish import domish
23 from wokkel import disco, iwokkel
24 from zope.interface import implementer
25
26 from libervia.backend.core import exceptions
27 from libervia.backend.core.constants import Const as C
28 from libervia.backend.core.core_types import SatXMPPEntity
29 from libervia.backend.core.i18n import _
30 from libervia.backend.core.log import getLogger
31 from libervia.backend.tools import xml_tools
32
33 log = getLogger(__name__)
34
35 NS_JINGLE_RTP_SSMA = "urn:xmpp:jingle:apps:rtp:ssma:0"
36
37 PLUGIN_INFO = {
38 C.PI_NAME: "Source-Specific Media Attributes in Jingle",
39 C.PI_IMPORT_NAME: "XEP-0339",
40 C.PI_TYPE: "XEP",
41 C.PI_MODES: C.PLUG_MODE_BOTH,
42 C.PI_PROTOCOLS: ["XEP-0339"],
43 C.PI_DEPENDENCIES: ["XEP-0092", "XEP-0167"],
44 C.PI_RECOMMENDATIONS: [],
45 C.PI_MAIN: "XEP_0339",
46 C.PI_HANDLER: "yes",
47 C.PI_DESCRIPTION: _("""Source-Specific Media Attributes in Jingle"""),
48 }
49
50
51 class XEP_0339:
52 def __init__(self, host):
53 log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization")
54 self.host = host
55 host.trigger.add("XEP-0167_parse_sdp_a", self._parse_sdp_a_trigger)
56 host.trigger.add(
57 "XEP-0167_generate_sdp_content", self._generate_sdp_content_trigger
58 )
59 host.trigger.add("XEP-0167_parse_description", self._parse_description_trigger)
60 host.trigger.add("XEP-0167_build_description", self._build_description_trigger)
61
62 def get_handler(self, client):
63 return XEP_0339_handler()
64
65 def _parse_sdp_a_trigger(
66 self,
67 attribute: str,
68 parts: List[str],
69 call_data: dict,
70 metadata: dict,
71 media_type: str,
72 application_data: dict,
73 transport_data: dict,
74 ) -> None:
75 """Parse "ssrc" attributes"""
76 if attribute == "ssrc":
77 assert application_data is not None
78 ssrc_id = int(parts[0])
79
80 if len(parts) > 1:
81 name, *values = " ".join(parts[1:]).split(":", 1)
82 if values:
83 value = values[0] or None
84 else:
85 value = None
86 application_data.setdefault("ssrc", {}).setdefault(ssrc_id, {})[
87 name
88 ] = value
89 else:
90 log.warning(f"no attribute in ssrc: {' '.join(parts)}")
91 application_data.setdefault("ssrc", {}).setdefault(ssrc_id, {})
92 elif attribute == "ssrc-group":
93 assert application_data is not None
94 semantics, *ssrc_ids = parts
95 ssrc_ids = [int(ssrc_id) for ssrc_id in ssrc_ids]
96 application_data.setdefault("ssrc-group", {})[semantics] = ssrc_ids
97 elif attribute == "msid":
98 assert application_data is not None
99 application_data["msid"] = " ".join(parts)
100
101
102 def _generate_sdp_content_trigger(
103 self,
104 session: dict,
105 local: bool,
106 idx: int,
107 content_data: dict,
108 sdp_lines: List[str],
109 application_data: dict,
110 app_data_key: str,
111 media_data: dict,
112 media: str
113 ) -> None:
114 """Generate "msid" and "ssrc" attributes"""
115 if "msid" in media_data:
116 sdp_lines.append(f"a=msid:{media_data['msid']}")
117
118 ssrc_data = media_data.get("ssrc", {})
119 ssrc_group_data = media_data.get("ssrc-group", {})
120
121 for ssrc_id, attributes in ssrc_data.items():
122 if not attributes:
123 # there are no attributes for this SSRC ID, we add a simple line with only
124 # the SSRC ID
125 sdp_lines.append(f"a=ssrc:{ssrc_id}")
126 else:
127 for attr_name, attr_value in attributes.items():
128 if attr_value is not None:
129 sdp_lines.append(f"a=ssrc:{ssrc_id} {attr_name}:{attr_value}")
130 else:
131 sdp_lines.append(f"a=ssrc:{ssrc_id} {attr_name}")
132 for semantics, ssrc_ids in ssrc_group_data.items():
133 ssrc_lines = " ".join(str(ssrc_id) for ssrc_id in ssrc_ids)
134 sdp_lines.append(f"a=ssrc-group:{semantics} {ssrc_lines}")
135
136 def _parse_description_trigger(
137 self, desc_elt: domish.Element, media_data: dict
138 ) -> bool:
139 """Parse the <source> and <ssrc-group> elements"""
140 for source_elt in desc_elt.elements(NS_JINGLE_RTP_SSMA, "source"):
141 try:
142 ssrc_id = int(source_elt["ssrc"])
143 media_data.setdefault("ssrc", {})[ssrc_id] = {}
144 for param_elt in source_elt.elements(NS_JINGLE_RTP_SSMA, "parameter"):
145 name = param_elt["name"]
146 value = param_elt.getAttribute("value")
147 media_data["ssrc"][ssrc_id][name] = value
148 if name == "msid" and "msid" not in media_data:
149 media_data["msid"] = value
150 except (KeyError, ValueError) as e:
151 log.warning(f"Error while parsing <source>: {e}\n{source_elt.toXml()}")
152
153 for ssrc_group_elt in desc_elt.elements(NS_JINGLE_RTP_SSMA, "ssrc-group"):
154 try:
155 semantics = ssrc_group_elt["semantics"]
156 semantic_ids = media_data.setdefault("ssrc-group", {})[semantics] = []
157 for source_elt in ssrc_group_elt.elements(NS_JINGLE_RTP_SSMA, "source"):
158 semantic_ids.append(
159 int(source_elt["ssrc"])
160 )
161 except (KeyError, ValueError) as e:
162 log.warning(
163 f"Error while parsing <ssrc-group>: {e}\n{ssrc_group_elt.toXml()}"
164 )
165
166 return True
167
168 def _build_description_trigger(
169 self, desc_elt: domish.Element, media_data: dict, session: dict
170 ) -> bool:
171 """Build the <source> and <ssrc-group> elements if possible"""
172 for ssrc_id, parameters in media_data.get("ssrc", {}).items():
173 if "msid" not in parameters and "msid" in media_data:
174 parameters["msid"] = media_data["msid"]
175 source_elt = desc_elt.addElement((NS_JINGLE_RTP_SSMA, "source"))
176 source_elt["ssrc"] = str(ssrc_id)
177 for name, value in parameters.items():
178 param_elt = source_elt.addElement((NS_JINGLE_RTP_SSMA, "parameter"))
179 param_elt["name"] = name
180 if value is not None:
181 param_elt["value"] = value
182
183 for semantics, ssrc_ids in media_data.get("ssrc-group", {}).items():
184 ssrc_group_elt = desc_elt.addElement((NS_JINGLE_RTP_SSMA, "ssrc-group"))
185 ssrc_group_elt["semantics"] = semantics
186 for ssrc_id in ssrc_ids:
187 source_elt = ssrc_group_elt.addElement((NS_JINGLE_RTP_SSMA, "source"))
188 source_elt["ssrc"] = str(ssrc_id)
189
190 return True
191
192
193 @implementer(iwokkel.IDisco)
194 class XEP_0339_handler(XMPPHandler):
195 def getDiscoInfo(self, requestor, target, nodeIdentifier=""):
196 return [disco.DiscoFeature(NS_JINGLE_RTP_SSMA)]
197
198 def getDiscoItems(self, requestor, target, nodeIdentifier=""):
199 return []