aioice里面candidate固定UDP端口测试

环境：

aioice0.9.0

问题描述：

aioice里面candidate固定UDP端口测试

解决方案：

/miniconda3/envs/nerfstream/lib/python3.10/site-packages/aioice

import hashlib
import ipaddress
import random
from typing import Optional
import logging# 配置日志级别，默认为 INFO，可改为 DEBUG 查看详细调试信息
logging.basicConfig(level=logging.INFO)class Candidate:"""表示一个 ICE 候选者，包括固定端口功能和多种优化。"""FIXED_PORT = 59990  # 定义固定端口def __init__(self,foundation: str,component: int,transport: str,priority: int,host: str,port: Optional[int] = None,  # 端口可以是 None，将被替换为固定端口type: str = "host",related_address: Optional[str] = None,related_port: Optional[int] = None,tcptype: Optional[str] = None,generation: Optional[int] = None,) -> None:"""初始化候选者对象，并强制使用固定端口。"""self.foundation = foundationself.component = componentself.transport = transportself.priority = priorityself.host = hostself.port = self.get_fixed_port() if port is None else port  # 设置为固定端口或提供的端口self.type = typeself.related_address = related_addressself.related_port = related_portself.tcptype = tcptypeself.generation = generation# 输出候选者信息到日志logging.info(f"创建候选者：host={self.host}, port={self.port}, type={self.type}")@classmethoddef get_fixed_port(cls) -> int:"""返回固定端口值，并记录日志。"""fixed_port = cls.FIXED_PORTlogging.debug(f"Returning fixed port: {fixed_port}")return fixed_port@classmethoddef from_sdp(cls, sdp: str):"""从 SDP 字符串中解析一个 ICE 候选者。示例 SDP：'6815297761 1 udp 659136 192.168.1.1 12345 typ host'"""bits = sdp.split()if len(bits) < 8:raise ValueError("SDP 描述字段不足")kwargs = {"foundation": bits[0],"component": int(bits[1]),"transport": bits[2],"priority": int(bits[3]),"host": bits[4],"port": cls.get_fixed_port(),  # 使用固定端口"type": bits[7],}# 提取 SDP 中的附加信息for i in range(8, len(bits), 2):if bits[i] == "raddr":kwargs["related_address"] = bits[i + 1]elif bits[i] == "rport":kwargs["related_port"] = int(bits[i + 1])elif bits[i] == "tcptype":kwargs["tcptype"] = bits[i + 1]elif bits[i] == "generation":kwargs["generation"] = int(bits[i + 1])return cls(**kwargs)def to_sdp(self) -> str:"""返回一个适用于 SDP 的字符串表示形式。"""sdp = f"{self.foundation} {self.component} {self.transport} {self.priority} {self.host} {self.port} typ {self.type}"if self.related_address is not None:sdp += f" raddr {self.related_address}"if self.related_port is not None:sdp += f" rport {self.related_port}"if self.tcptype is not None:sdp += f" tcptype {self.tcptype}"if self.generation is not None:sdp += f" generation {self.generation}"return sdpdef can_pair_with(self, other) -> bool:"""判断本地候选者是否可以与远程候选者配对。配对条件：- 组件相同- 使用相同的传输协议- IP 地址版本相同"""a = ipaddress.ip_address(self.host)b = ipaddress.ip_address(other.host)return (self.component == other.componentand self.transport.lower() == other.transport.lower()and a.version == b.version)def __repr__(self) -> str:"""返回候选者的字符串表示形式。"""return f"Candidate({self.to_sdp()})"# 工具函数部分
def candidate_foundation(candidate_type: str, candidate_transport: str, base_address: str) -> str:"""计算候选者的 Foundation（基础标识符）。"""key = f"{candidate_type}|{candidate_transport}|{base_address}"return hashlib.md5(key.encode("ascii")).hexdigest()def candidate_priority(candidate_component: int, candidate_type: str, local_pref: int = 65535) -> int:"""计算候选者优先级。优先级从高到低的顺序为：host > srflx > relay。"""type_preferences = {"host": 126,   # Host 类型优先级最高"srflx": 100,  # Server reflexive 类型"relay": 0,    # Relay 类型优先级最低}type_pref = type_preferences.get(candidate_type, 0)return (1 << 24) * type_pref + (1 << 8) * local_pref + (256 - candidate_component)# 测试代码部分
if __name__ == "__main__":# 测试创建候选者candidate = Candidate(foundation="1",component=1,transport="udp",priority=candidate_priority(1, "host"),host="192.168.168.77",  # 使用你的本地 IP 地址type="host",)print(candidate)  # 打印候选者信息print(candidate.to_sdp())  # 打印 SDP 字符串# 测试从 SDP 创建候选者sdp_example = "1 1 udp 100 192.168.1.1 12345 typ host"candidate_from_sdp = Candidate.from_sdp(sdp_example)print(candidate_from_sdp)  # 打印解析后的候选者print(candidate_from_sdp.to_sdp())  # 打印 SDP 字符串# 模拟 WebRTC 应用程序创建多个候选者candidates = [Candidate(foundation=str(i),component=1,transport="udp",priority=candidate_priority(1, "host"),host=f"192.168.1.{i % 255}",  # 使用不同的 IP 地址模拟不同设备type="host",)for i in range(1, 6)]for c in candidates:print(c)

运行没有成功