Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_xep_0167/mapping.py @ 4276:00a9316547ed

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
plugin XEP-0167: move ICE candidates parsing/SDP line building in separate methods: These methods will be used by incoming "Conferences" component. rel 445
author Goffi <goffi@goffi.org>
date Fri, 05 Jul 2024 17:18:37 +0200
parents 0d7bb4df2343
children 33ecebb2c02f
comparison
equal deleted inserted replaced
4275:29fda1340078 4276:00a9316547ed
181 181
182 sdp_lines.append(f"a=ice-ufrag:{ice_data['ufrag']}") 182 sdp_lines.append(f"a=ice-ufrag:{ice_data['ufrag']}")
183 sdp_lines.append(f"a=ice-pwd:{ice_data['pwd']}") 183 sdp_lines.append(f"a=ice-pwd:{ice_data['pwd']}")
184 184
185 for candidate in ice_data["candidates"]: 185 for candidate in ice_data["candidates"]:
186 foundation = candidate["foundation"] 186 candidate_line = generate_candidate_line(candidate)
187 component_id = candidate["component_id"] 187 sdp_lines.append(f"a={candidate_line}")
188 transport = candidate["transport"]
189 priority = candidate["priority"]
190 address = candidate["address"]
191 candidate_port = candidate["port"]
192 candidate_type = candidate["type"]
193
194 candidate_line = (
195 f"a=candidate:{foundation} {component_id} {transport} {priority} "
196 f"{address} {candidate_port} typ {candidate_type}"
197 )
198
199 if "rel_addr" in candidate and "rel_port" in candidate:
200 candidate_line += (
201 f" raddr {candidate['rel_addr']} rport {candidate['rel_port']}"
202 )
203
204 if "generation" in candidate:
205 candidate_line += f" generation {candidate['generation']}"
206
207 if "network" in candidate:
208 candidate_line += f" network {candidate['network']}"
209
210 sdp_lines.append(candidate_line)
211 188
212 # Generate a= lines for encryption 189 # Generate a= lines for encryption
213 if "encryption" in media_data: 190 if "encryption" in media_data:
214 for enc_data in media_data["encryption"]: 191 for enc_data in media_data["encryption"]:
215 crypto_suite = enc_data["crypto-suite"] 192 crypto_suite = enc_data["crypto-suite"]
236 triggers_no_cancel=True, 213 triggers_no_cancel=True,
237 ) 214 )
238 215
239 # Combine SDP lines and return the result 216 # Combine SDP lines and return the result
240 return "\r\n".join(sdp_lines) + "\r\n" 217 return "\r\n".join(sdp_lines) + "\r\n"
218
219
220 def generate_candidate_line(candidate: dict) -> str:
221 """Generate a ``candidate:`` attribute line from candidate data.
222
223 @param candidate: ICE candidate data.
224 @return ICE candidate attribute line.
225 """
226 foundation = candidate["foundation"]
227 component_id = candidate["component_id"]
228 transport = candidate["transport"]
229 priority = candidate["priority"]
230 address = candidate["address"]
231 candidate_port = candidate["port"]
232 candidate_type = candidate["type"]
233
234 candidate_line = (
235 f"candidate:{foundation} {component_id} {transport} {priority} "
236 f"{address} {candidate_port} typ {candidate_type}"
237 )
238
239 if "rel_addr" in candidate and "rel_port" in candidate:
240 candidate_line += f" raddr {candidate['rel_addr']} rport {candidate['rel_port']}"
241
242 if "generation" in candidate:
243 candidate_line += f" generation {candidate['generation']}"
244
245 if "network" in candidate:
246 candidate_line += f" network {candidate['network']}"
247
248 return candidate_line
249
250
251 def parse_candidate(parts: list[str]) -> dict:
252 """Parse parts of a ICE candidate
253
254 @param parts: Parts of the candidate line.
255 @return: candidate data
256 """
257 candidate = {
258 "foundation": parts[0],
259 "component_id": int(parts[1]),
260 "transport": parts[2],
261 "priority": int(parts[3]),
262 "address": parts[4],
263 "port": int(parts[5]),
264 "type": parts[7],
265 }
266
267 for part in parts[8:]:
268 if part == "raddr":
269 candidate["rel_addr"] = parts[parts.index(part) + 1]
270 elif part == "rport":
271 candidate["rel_port"] = int(parts[parts.index(part) + 1])
272 elif part == "generation":
273 candidate["generation"] = parts[parts.index(part) + 1]
274 elif part == "network":
275 candidate["network"] = parts[parts.index(part) + 1]
276
277 return candidate
241 278
242 279
243 def parse_sdp(sdp: str, role: str) -> dict: 280 def parse_sdp(sdp: str, role: str) -> dict:
244 """Parse SDP string. 281 """Parse SDP string.
245 282
375 except ValueError: 412 except ValueError:
376 payload_type.setdefault("exra-parameters", []).extend(params) 413 payload_type.setdefault("exra-parameters", []).extend(params)
377 414
378 elif attribute == "candidate": 415 elif attribute == "candidate":
379 assert transport_data is not None 416 assert transport_data is not None
380 candidate = { 417 candidate = parse_candidate(parts)
381 "foundation": parts[0],
382 "component_id": int(parts[1]),
383 "transport": parts[2],
384 "priority": int(parts[3]),
385 "address": parts[4],
386 "port": int(parts[5]),
387 "type": parts[7],
388 }
389
390 for part in parts[8:]:
391 if part == "raddr":
392 candidate["rel_addr"] = parts[parts.index(part) + 1]
393 elif part == "rport":
394 candidate["rel_port"] = int(parts[parts.index(part) + 1])
395 elif part == "generation":
396 candidate["generation"] = parts[parts.index(part) + 1]
397 elif part == "network":
398 candidate["network"] = parts[parts.index(part) + 1]
399 418
400 transport_data.setdefault("candidates", []).append(candidate) 419 transport_data.setdefault("candidates", []).append(candidate)
401 420
402 elif attribute == "fingerprint": 421 elif attribute == "fingerprint":
403 algorithm, fingerprint = parts[0], parts[1] 422 algorithm, fingerprint = parts[0], parts[1]