IpManager 接口文档

简介

简述：该文件定义了IpManager类，用于统一管理RflySim仿真平台中通信交互所需的IP地址与端口配置。

在RflySim无人机仿真框架中，地面站、飞控、仿真环境等不同组件往往运行在不同的网络节点或进程上，需要统一的IP地址信息管理来保障各模块间的网络通信正常建立。该模块负责维护各类通信链路对应的地址配置信息，为SDK中其他通信控制模块提供统一的地址查询与管理入口，适用于多机仿真、分布式仿真等需要区分不同节点通信地址的场景，简化了IP配置的维护与调用流程。

快速开始

最简可用示例，复制后修改最少配置即可运行。

from IpManager import IpManager

# 检查IP配置文件是否存在且路径有效
exists, config_path = IpManager.is_valid()
print(f"IP配置文件状态: 存在={exists}, 路径={config_path}")

# 检测当前程序是否运行在Docker/K8s容器中
in_container = IpManager.is_container()
print(f"当前是否运行在容器中: {in_container}")

# 获取配置文件中的本机IP，获取失败使用默认127.0.0.1
local_ip = IpManager.get_local_ip()
print(f"获取到的本机IP地址为: {local_ip}")

# 也可自定义获取失败时的默认IP
custom_default_ip = "192.168.1.100"
local_ip_custom = IpManager.get_local_ip(custom_default_ip)
print(f"自定义默认IP后的本机IP地址为: {local_ip_custom}")

环境与依赖

• Python 环境：>= 3.8.10
• 依赖库：datetime、json、logging、os、threading、time、typing
• 前置准备：如需从局域网配置解析地址，应将环境变量 RFLYSIM_IP_MANAGER_PATH 指向包含 rflysim_lan_appinfo.json 的目录。配置中的 updateTime 格式必须为 %Y-%m-%d %H:%M:%S，且默认不能早于当前时间 10 秒以上。

核心接口说明

该模块 IpManager.py 包含了配置变量、辅助函数及核心业务类。

全局常量与枚举定义

本节列出模块中所有可直接引用的全局常量和枚举定义。

独立常量

变量名	默认值	说明
_cache_ttl	5.0	JSON 配置在内存中的缓存有效期，单位为秒。
_stale_seconds	10.0	updateTime 允许的最大时间差，超过后配置被视为过期并忽略。

---

全局/独立函数

无

---

IpManager 类

IP地址管理工具类，用于从配置文件中获取RflySim平台各组件（RflySim3D、CopterSim、QGroundControl）的IP地址，同时支持环境检测和配置校验，适配本地和分布式仿真场景。

---

is_valid()

功能说明：检查环境变量 RFLYSIM_IP_MANAGER_PATH 指向的目录是否存在，并确认该目录下存在 rflysim_lan_appinfo.json。此方法不校验 JSON 内容、IP 格式、时间戳或网络连通性。 参数列表 (Args)： 无参数 返回值 (Returns)：

• tuple[bool, str | None]: 第一个元素表示文件是否存在；有效时第二个元素为配置文件路径，否则为 None

异常 (Raises)：

• 无

---

is_container()

功能说明：检测当前程序是否运行在容器环境（Docker/K8s等）中。 参数列表 (Args)： 无参数 返回值 (Returns)：

• bool: 若运行在容器中返回True，否则返回False

异常 (Raises)：

• 无

---

get_local_ip(original_ip='127.0.0.1')

功能说明：从有效且未过期的配置文件读取顶层 localIp 字段。配置缺失、解析失败或 updateTime 过期时返回 original_ip；该方法不会枚举本机网卡。 参数列表 (Args)：

参数名	类型	是否必填	默认值	说明
original_ip	str	否	127.0.0.1	配置不可用时返回的默认IP地址

返回值 (Returns)：

• str: 本机IP地址字符串

异常 (Raises)：

• 无

示例：

from RflySimSDK.ctrl import IpManager
local_ip = IpManager.get_local_ip()
print("本机IP为：", local_ip)

---

get_rflysim3d_ip(original_ip='127.0.0.1')

功能说明：当 original_ip 为 127.0.0.1 时，在有效配置的 apps 列表中查找 appName == "RflySim3D" 的条目并返回其 ip；调用方已传入其他地址时直接原样返回。 参数列表 (Args)：

参数名	类型	是否必填	默认值	说明
original_ip	str	否	127.0.0.1	查找失败或不需要查找时返回的默认IP地址

返回值 (Returns)：

• str: RflySim3D IP地址字符串

异常 (Raises)：

• 无

示例：

from RflySimSDK.ctrl import IpManager
rflysim3d_ip = IpManager.get_rflysim3d_ip()
print("RflySim3D IP为：", rflysim3d_ip)

---

get_coptersim_ip(copter_id, original_ip='127.0.0.1')

功能说明：获取指定飞机ID对应的CopterSim IP地址，当CopterSim与SDK不在同一台计算机时，会自动请求CopterSim将数据发送到本机。 参数列表 (Args)：

参数名	类型	是否必填	默认值	说明
copter_id	int	是	-	目标飞机的ID号，与配置中的 appInstance 按字符串形式比较
original_ip	str	否	127.0.0.1	查找失败或不需要查找时返回的默认IP地址

返回值 (Returns)：

• str: CopterSim IP地址字符串

异常 (Raises)：

• 无

示例：

from RflySimSDK.ctrl import IpManager
# 获取ID为1的无人机对应的CopterSim IP
coptersim_ip = IpManager.get_coptersim_ip(1)
print("CopterSim IP为：", coptersim_ip)

---

get_qgc_ip(original_ip='127.0.0.1')

功能说明：获取QGroundControl地面站的IP地址。 参数列表 (Args)：

参数名	类型	是否必填	默认值	说明
original_ip	str	否	127.0.0.1	查找失败或不需要查找时返回的默认IP地址

返回值 (Returns)：

• str: QGC IP地址字符串

异常 (Raises)：

• 无

---

进阶用法示例

展示复杂组合场景（如多类协作、异步控制、批量操作）

针对多无人机协同仿真集群的组网场景，我们可以结合IpManager实现批量IP有效性校验，同时配合多机控制模块完成异构仿真组件的IP地址自动适配。例如针对同时运行PX4仿真、RflySim3D渲染以及QGC地面站的多节点分布式部署场景，可以批量获取不同服务的IP并提前完成连通性校验，避免仿真启动后出现连接失败问题，示例代码如下：

from IpManager import IpManager
import concurrent.futures
import ipaddress
import os
import subprocess


def is_ip_reachable(ip, timeout_seconds=1):
    """通过一次 ICMP ping 检查目标主机是否可达。"""
    try:
        ipaddress.ip_address(ip)
    except (TypeError, ValueError):
        return False

    if os.name == "nt":
        command = ["ping", "-n", "1", "-w", str(timeout_seconds * 1000), ip]
    else:
        command = ["ping", "-c", "1", "-W", str(timeout_seconds), ip]

    try:
        result = subprocess.run(
            command,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            timeout=timeout_seconds + 1,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired):
        return False
    return result.returncode == 0

# 异步批量校验多个仿真组件IP的连通性
def check_ip_valid(service_name, get_ip_method):
    ip = get_ip_method()
    return service_name, ip, is_ip_reachable(ip)

# 需要校验的服务列表，覆盖不同类型仿真组件
service_list = [
    ("RflySim3D", IpManager.get_rflysim3d_ip),
    ("CopterSim #1", lambda: IpManager.get_coptersim_ip(1)),
    ("QGC地面站", IpManager.get_qgc_ip)
]

# 异步批量校验，提升大规模集群场景下的启动效率
with concurrent.futures.ThreadPoolExecutor() as executor:
    futures = [executor.submit(check_ip_valid, name, method) for name, method in service_list]
    for future in concurrent.futures.as_completed(futures):
        service_name, ip, is_valid = future.result()
        if is_valid:
            print(f"{service_name} IP {ip} 主机可达")
        else:
            print(f"警告：{service_name} IP {ip} 主机不可达，请检查网络配置")

# 获取本机IP用于跨设备分布式仿真组网
local_ip = IpManager.get_local_ip()
print(f"本机组网IP为：{local_ip}")

该方案实现了异步批量IP校验，能够适配多机分布式集群仿真、异构组件协作的复杂场景，提前排查网络配置问题。

注意事项与避坑指南

• 配置时效：JSON 的 updateTime 超过当前时间 10 秒会被忽略；即使 5 秒缓存尚未到期，缓存内容也会重新做时效检查。
• 校验范围：is_valid() 只检查环境变量、目录和文件是否存在，不检查 JSON 内容、IP 格式、端口状态或网络连通性。
• 容器检测范围：is_container() 检查 /.dockerenv、/proc/self/cgroup 关键字以及 DOCKER_CONTAINER、KUBERNETES_SERVICE_HOST 环境变量，无法覆盖所有自定义容器环境。
• 远端请求去重：解析到远端 CopterSim 后，同一 copter_id 在当前进程中只会调用一次 ReqCopterSim.sendReSimIP()；该状态由类级集合保存并通过锁保护。

更新日志

• 2026-05-14: ✨ fix: 修复读取文件 updateTime 过期仍返回问题 [P2]
• 2026-03-03: feat：SDK增加IP处理机制，兼容本地版上云