Mercurial > libervia-backend
comparison libervia/backend/plugins/plugin_xep_0167/mapping.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 | 00a9316547ed |
comparison
equal
deleted
inserted
replaced
4269:64a85ce8be70 | 4270:0d7bb4df2343 |
---|---|
41 return "a=sendonly" | 41 return "a=sendonly" |
42 else: | 42 else: |
43 return "a=recvonly" | 43 return "a=recvonly" |
44 | 44 |
45 | 45 |
46 def generate_sdp_from_session( | 46 def generate_sdp_from_session(session: dict, local: bool = False, port: int = 9) -> str: |
47 session: dict, local: bool = False, port: int = 9 | |
48 ) -> str: | |
49 """Generate an SDP string from session data. | 47 """Generate an SDP string from session data. |
50 | 48 |
51 @param session: A dictionary containing the session data. It should have the | 49 @param session: A dictionary containing the session data. It should have the |
52 following structure: | 50 following structure: |
53 | 51 |
109 host.trigger.point( | 107 host.trigger.point( |
110 "XEP-0167_generate_sdp_session", | 108 "XEP-0167_generate_sdp_session", |
111 session, | 109 session, |
112 local, | 110 local, |
113 sdp_lines, | 111 sdp_lines, |
114 triggers_no_cancel=True | 112 triggers_no_cancel=True, |
115 ) | 113 ) |
116 content_names = sorted(contents) | 114 content_names = sorted(contents) |
117 | 115 |
118 for content_name, content_data in [(n, contents[n]) for n in content_names]: # contents.items(): | 116 for content_name, content_data in [ |
117 (n, contents[n]) for n in content_names | |
118 ]: # contents.items(): | |
119 app_data_key = "local_data" if local else "peer_data" | 119 app_data_key = "local_data" if local else "peer_data" |
120 application_data = content_data["application_data"] | 120 application_data = content_data["application_data"] |
121 media_data = application_data[app_data_key] | 121 media_data = application_data[app_data_key] |
122 media = application_data["media"] | 122 media = application_data["media"] |
123 payload_types = media_data.get("payload_types", {}) | 123 payload_types = media_data.get("payload_types", {}) |
140 | 140 |
141 # stream direction | 141 # stream direction |
142 if senders is None: | 142 if senders is None: |
143 sdp_lines.append(senders_to_sdp(content_data["senders"], session)) | 143 sdp_lines.append(senders_to_sdp(content_data["senders"], session)) |
144 | 144 |
145 | |
146 # Generate a= lines for rtpmap and fmtp | 145 # Generate a= lines for rtpmap and fmtp |
147 for pt_id, pt in payload_types.items(): | 146 for pt_id, pt in payload_types.items(): |
148 name = pt["name"] | 147 name = pt["name"] |
149 clockrate = pt.get("clockrate", "") | 148 clockrate = pt.get("clockrate", "") |
150 | 149 |
220 | 219 |
221 crypto_line = f"a=crypto:{tag} {crypto_suite} {key_params}" | 220 crypto_line = f"a=crypto:{tag} {crypto_suite} {key_params}" |
222 if session_params: | 221 if session_params: |
223 crypto_line += f" {session_params}" | 222 crypto_line += f" {session_params}" |
224 sdp_lines.append(crypto_line) | 223 sdp_lines.append(crypto_line) |
225 | |
226 | 224 |
227 host.trigger.point( | 225 host.trigger.point( |
228 "XEP-0167_generate_sdp_content", | 226 "XEP-0167_generate_sdp_content", |
229 session, | 227 session, |
230 local, | 228 local, |
233 sdp_lines, | 231 sdp_lines, |
234 application_data, | 232 application_data, |
235 app_data_key, | 233 app_data_key, |
236 media_data, | 234 media_data, |
237 media, | 235 media, |
238 triggers_no_cancel=True | 236 triggers_no_cancel=True, |
239 ) | 237 ) |
240 | 238 |
241 # Combine SDP lines and return the result | 239 # Combine SDP lines and return the result |
242 return "\r\n".join(sdp_lines) + "\r\n" | 240 return "\r\n".join(sdp_lines) + "\r\n" |
243 | 241 |
296 if ice_ufrag is not None: | 294 if ice_ufrag is not None: |
297 transport_data["ufrag"] = ice_ufrag | 295 transport_data["ufrag"] = ice_ufrag |
298 media_data = call_data[media_type] = { | 296 media_data = call_data[media_type] = { |
299 "application_data": application_data, | 297 "application_data": application_data, |
300 "transport_data": transport_data, | 298 "transport_data": transport_data, |
301 "senders": senders | 299 "senders": senders, |
302 } | 300 } |
303 | 301 |
304 elif prefix == "a=": | 302 elif prefix == "a=": |
305 if ":" in parts[0]: | 303 if ":" in parts[0]: |
306 attribute, parts[0] = parts[0].split(":", 1) | 304 attribute, parts[0] = parts[0].split(":", 1) |
455 call_data, | 453 call_data, |
456 metadata, | 454 metadata, |
457 media_type, | 455 media_type, |
458 application_data, | 456 application_data, |
459 transport_data, | 457 transport_data, |
460 triggers_no_cancel=True | 458 triggers_no_cancel=True, |
461 ) | 459 ) |
462 | 460 |
463 except ValueError as e: | 461 except ValueError as e: |
464 raise ValueError(f"Could not parse line. Invalid format ({e}): {line}") from e | 462 raise ValueError(f"Could not parse line. Invalid format ({e}): {line}") from e |
465 except IndexError as e: | 463 except IndexError as e: |
472 del call_data[key] | 470 del call_data[key] |
473 | 471 |
474 # FIXME: is this really useful? | 472 # FIXME: is this really useful? |
475 # ICE candidates may only be specified for the first media, this | 473 # ICE candidates may only be specified for the first media, this |
476 # duplicate the candidate for the other in this case | 474 # duplicate the candidate for the other in this case |
477 all_media = {k:v for k,v in call_data.items() if k in ("audio", "video")} | 475 all_media = {k: v for k, v in call_data.items() if k in ("audio", "video")} |
478 if len(all_media) > 1 and not all( | 476 if len(all_media) > 1 and not all( |
479 "candidates" in c["transport_data"] for c in all_media.values() | 477 "candidates" in c["transport_data"] for c in all_media.values() |
480 ): | 478 ): |
481 first_content = next(iter(all_media.values())) | 479 first_content = next(iter(all_media.values())) |
482 try: | 480 try: |
546 "XEP-0167_build_description_payload_type", | 544 "XEP-0167_build_description_payload_type", |
547 desc_elt, | 545 desc_elt, |
548 media_data, | 546 media_data, |
549 pt_data, | 547 pt_data, |
550 payload_type_elt, | 548 payload_type_elt, |
551 triggers_no_cancel=True | 549 triggers_no_cancel=True, |
552 ) | 550 ) |
553 | 551 |
554 if "bandwidth" in media_data: | 552 if "bandwidth" in media_data: |
555 bandwidth_elt = desc_elt.addElement("bandwidth") | 553 bandwidth_elt = desc_elt.addElement("bandwidth") |
556 bandwidth_elt["type"] = media_data["bandwidth"] | 554 bandwidth_elt["type"] = media_data["bandwidth"] |
572 host.trigger.point( | 570 host.trigger.point( |
573 "XEP-0167_build_description", | 571 "XEP-0167_build_description", |
574 desc_elt, | 572 desc_elt, |
575 media_data, | 573 media_data, |
576 session, | 574 session, |
577 triggers_no_cancel=True | 575 triggers_no_cancel=True, |
578 ) | 576 ) |
579 | 577 |
580 return desc_elt | 578 return desc_elt |
581 | 579 |
582 | 580 |
630 "XEP-0167_parse_description_payload_type", | 628 "XEP-0167_parse_description_payload_type", |
631 desc_elt, | 629 desc_elt, |
632 media_data, | 630 media_data, |
633 payload_type_elt, | 631 payload_type_elt, |
634 payload_type_data, | 632 payload_type_data, |
635 triggers_no_cancel=True | 633 triggers_no_cancel=True, |
636 ) | 634 ) |
637 payload_types[pt_id] = payload_type_data | 635 payload_types[pt_id] = payload_type_data |
638 | 636 |
639 # bandwidth | 637 # bandwidth |
640 media_data["payload_types"] = payload_types | 638 media_data["payload_types"] = payload_types |
676 | 674 |
677 if encryption_data: | 675 if encryption_data: |
678 media_data["encryption"] = encryption_data | 676 media_data["encryption"] = encryption_data |
679 | 677 |
680 host.trigger.point( | 678 host.trigger.point( |
681 "XEP-0167_parse_description", | 679 "XEP-0167_parse_description", desc_elt, media_data, triggers_no_cancel=True |
682 desc_elt, | |
683 media_data, | |
684 triggers_no_cancel=True | |
685 ) | 680 ) |
686 | 681 |
687 return media_data | 682 return media_data |