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 = foundation
        self.component = component
        self.transport = transport
        self.priority = priority
        self.host = host
        self.port = self.get_fixed_port() if port is None else port  # 设置为固定端口或提供的端口
        self.type = type
        self.related_address = related_address
        self.related_port = related_port
        self.tcptype = tcptype
        self.generation = generation

        # 输出候选者信息到日志
        logging.info(f"创建候选者：host={self.host}, port={self.port}, type={self.type}")

    @classmethod
    def get_fixed_port(cls) -> int:
        """
        返回固定端口值，并记录日志。
        """
        fixed_port = cls.FIXED_PORT
        logging.debug(f"Returning fixed port: {fixed_port}")
        return fixed_port

    @classmethod
    def 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 sdp

    def can_pair_with(self, other) -> bool:
        """
        判断本地候选者是否可以与远程候选者配对。

        配对条件：
        - 组件相同
        - 使用相同的传输协议
        - IP 地址版本相同
        """
        a = ipaddress.ip_address(self.host)
        b = ipaddress.ip_address(other.host)
        return (
            self.component == other.component
            and 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)

运行没有成功