# HG changeset patch
# User Goffi <goffi@goffi.org>
# Date 1720192717 -7200
# Node ID 00a9316547ed5b3f9b4f4df94e217c1a6a5ab03f
# Parent  29fda1340078f2e66d2bb80825585163d70cb972
plugin XEP-0167: move ICE candidates parsing/SDP line building in separate methods:


These methods will be used by incoming "Conferences" component.

rel 445

diff -r 29fda1340078 -r 00a9316547ed libervia/backend/plugins/plugin_xep_0167/mapping.py
--- a/libervia/backend/plugins/plugin_xep_0167/mapping.py	Fri Jul 05 17:18:32 2024 +0200
+++ b/libervia/backend/plugins/plugin_xep_0167/mapping.py	Fri Jul 05 17:18:37 2024 +0200
@@ -183,31 +183,8 @@
         sdp_lines.append(f"a=ice-pwd:{ice_data['pwd']}")
 
         for candidate in ice_data["candidates"]:
-            foundation = candidate["foundation"]
-            component_id = candidate["component_id"]
-            transport = candidate["transport"]
-            priority = candidate["priority"]
-            address = candidate["address"]
-            candidate_port = candidate["port"]
-            candidate_type = candidate["type"]
-
-            candidate_line = (
-                f"a=candidate:{foundation} {component_id} {transport} {priority} "
-                f"{address} {candidate_port} typ {candidate_type}"
-            )
-
-            if "rel_addr" in candidate and "rel_port" in candidate:
-                candidate_line += (
-                    f" raddr {candidate['rel_addr']} rport {candidate['rel_port']}"
-                )
-
-            if "generation" in candidate:
-                candidate_line += f" generation {candidate['generation']}"
-
-            if "network" in candidate:
-                candidate_line += f" network {candidate['network']}"
-
-            sdp_lines.append(candidate_line)
+            candidate_line = generate_candidate_line(candidate)
+            sdp_lines.append(f"a={candidate_line}")
 
         # Generate a= lines for encryption
         if "encryption" in media_data:
@@ -240,6 +217,66 @@
     return "\r\n".join(sdp_lines) + "\r\n"
 
 
+def generate_candidate_line(candidate: dict) -> str:
+    """Generate a ``candidate:`` attribute line from candidate data.
+
+    @param candidate: ICE candidate data.
+    @return ICE candidate attribute line.
+    """
+    foundation = candidate["foundation"]
+    component_id = candidate["component_id"]
+    transport = candidate["transport"]
+    priority = candidate["priority"]
+    address = candidate["address"]
+    candidate_port = candidate["port"]
+    candidate_type = candidate["type"]
+
+    candidate_line = (
+        f"candidate:{foundation} {component_id} {transport} {priority} "
+        f"{address} {candidate_port} typ {candidate_type}"
+    )
+
+    if "rel_addr" in candidate and "rel_port" in candidate:
+        candidate_line += f" raddr {candidate['rel_addr']} rport {candidate['rel_port']}"
+
+    if "generation" in candidate:
+        candidate_line += f" generation {candidate['generation']}"
+
+    if "network" in candidate:
+        candidate_line += f" network {candidate['network']}"
+
+    return candidate_line
+
+
+def parse_candidate(parts: list[str]) -> dict:
+    """Parse parts of a ICE candidate
+
+    @param parts: Parts of the candidate line.
+    @return: candidate data
+    """
+    candidate = {
+        "foundation": parts[0],
+        "component_id": int(parts[1]),
+        "transport": parts[2],
+        "priority": int(parts[3]),
+        "address": parts[4],
+        "port": int(parts[5]),
+        "type": parts[7],
+    }
+
+    for part in parts[8:]:
+        if part == "raddr":
+            candidate["rel_addr"] = parts[parts.index(part) + 1]
+        elif part == "rport":
+            candidate["rel_port"] = int(parts[parts.index(part) + 1])
+        elif part == "generation":
+            candidate["generation"] = parts[parts.index(part) + 1]
+        elif part == "network":
+            candidate["network"] = parts[parts.index(part) + 1]
+
+    return candidate
+
+
 def parse_sdp(sdp: str, role: str) -> dict:
     """Parse SDP string.
 
@@ -377,25 +414,7 @@
 
                 elif attribute == "candidate":
                     assert transport_data is not None
-                    candidate = {
-                        "foundation": parts[0],
-                        "component_id": int(parts[1]),
-                        "transport": parts[2],
-                        "priority": int(parts[3]),
-                        "address": parts[4],
-                        "port": int(parts[5]),
-                        "type": parts[7],
-                    }
-
-                    for part in parts[8:]:
-                        if part == "raddr":
-                            candidate["rel_addr"] = parts[parts.index(part) + 1]
-                        elif part == "rport":
-                            candidate["rel_port"] = int(parts[parts.index(part) + 1])
-                        elif part == "generation":
-                            candidate["generation"] = parts[parts.index(part) + 1]
-                        elif part == "network":
-                            candidate["network"] = parts[parts.index(part) + 1]
+                    candidate = parse_candidate(parts)
 
                     transport_data.setdefault("candidates", []).append(candidate)