IpManager Interface Documentation

Introduction

Overview: This file defines the IpManager class, used to uniformly manage IP addresses and port configurations required for communication and interaction within the RflySim simulation platform.

In the RflySim UAV simulation framework, components such as the ground station, flight controller, and simulation environment often run on different network nodes or processes. A unified IP address management system is required to ensure proper network communication between modules. This module maintains address configuration information for various communication links and provides a centralized entry point for address querying and management to other communication control modules in the SDK. It is particularly suitable for multi-robot simulation and distributed simulation scenarios where distinct node communication addresses must be distinguished, simplifying IP configuration maintenance and usage.

Quick Start

A minimal working example; copy and modify only the necessary configurations to run.

from IpManager import IpManager

# Check whether the IP configuration file exists and the path is valid
exists, config_path = IpManager.is_valid()
print(f"IP configuration file status: exists={exists}, path={config_path}")

# Detect whether the current program is running inside a Docker/K8s container
in_container = IpManager.is_container()
print(f"Running inside a container: {in_container}")

# Retrieve the local IP from the configuration file; fallback to default 127.0.0.1 on failure
local_ip = IpManager.get_local_ip()
print(f"Retrieved local IP address: {local_ip}")

# Alternatively, specify a custom default IP to use on failure
custom_default_ip = "192.168.1.100"
local_ip_custom = IpManager.get_local_ip(custom_default_ip)
print(f"Local IP address with custom default: {local_ip_custom}")

Environment and Dependencies

• Python Environment: >= 3.8.10
• Dependencies: datetime, json, logging, os, threading, time, typing
• Prerequisites: To resolve addresses from a LAN configuration, set the environment variable RFLYSIM_IP_MANAGER_PATH to the directory containing rflysim_lan_appinfo.json. The updateTime in the configuration must be in the format %Y-%m-%d %H:%M:%S, and by default it must not be more than 10 seconds earlier than the current time.

Core Interface Description

The module IpManager.py includes configuration variables, helper functions, and the core business class.

Global Constants and Enumerations

This section lists all globally accessible constants and enumerations defined in the module.

Standalone Constants

Variable Name	Default Value	Description
_cache_ttl	5.0	In-memory cache validity period for the JSON configuration, in seconds.
_stale_seconds	10.0	Maximum allowed time difference for updateTime; beyond this, the configuration is considered expired and ignored.

---

Global/Standalone Functions

None

---

IpManager Class

A utility class for IP address management, used to retrieve IP addresses for various RflySim platform components (RflySim3D, CopterSim, QGroundControl) from the configuration file. It also supports environment detection and configuration validation, enabling adaptation to both local and distributed simulation scenarios.

---

is_valid()

Function Description: Checks whether the directory pointed to by the environment variable RFLYSIM_IP_MANAGER_PATH exists, and confirms that rflysim_lan_appinfo.json is present in that directory. This method does not validate JSON content, IP format, timestamps, or network connectivity. Parameters (Args): None

Returns:

• tuple[bool, str | None]: The first element indicates whether the file exists; when valid, the second element is the configuration file path, otherwise None.

Raises:

• None

---

is_container()

Function Description: Detects whether the current program is running inside a container environment (e.g., Docker, K8s). Parameters (Args): None

Returns:

• bool: Returns True if running inside a container; otherwise returns False.

Raises:

• None

---

get_local_ip(original_ip='127.0.0.1')

Function Description: Reads the top-level localIp field from a valid and non-expired configuration file. Returns original_ip when the configuration is missing, parsing fails, or updateTime has expired; this method does not enumerate local network interfaces. Parameters (Args):

Parameter Name	Type	Required	Default Value	Description
original_ip	str	No	127.0.0.1	Default IP address to return when the configuration is unavailable

Returns:

• str: Local IP address as a string.

Raises:

• None

Example:

from RflySimSDK.ctrl import IpManager
local_ip = IpManager.get_local_ip()
print("Local IP:", local_ip)

---

get_rflysim3d_ip(original_ip='127.0.0.1')

Function Description: When original_ip is 127.0.0.1, searches the apps list of a valid configuration for the entry with appName == "RflySim3D" and returns its ip; if the caller has already passed another address, it is returned as-is. Parameters (Args):

Parameter Name	Type	Required	Default Value	Description
original_ip	str	No	127.0.0.1	Default IP address to return if lookup fails or lookup is skipped

Returns:

• str: RflySim3D IP address as a string.

Raises:

• None

Example:

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')

Function Description: Retrieves the CopterSim IP address corresponding to a specified aircraft ID. When CopterSim and the SDK are not running on the same machine, it automatically requests CopterSim to send data to the local host. Parameters (Args):

Parameter Name	Type	Required	Default Value	Description
copter_id	int	Yes	-	ID of the target aircraft, compared against appInstance in the configuration as a string
original_ip	str	No	127.0.0.1	Default IP address to return if lookup fails or lookup is skipped

Returns:

• str: CopterSim IP address string

Raises:

• None

Example:

from RflySimSDK.ctrl import IpManager
# Get the CopterSim IP corresponding to the drone with ID 1
coptersim_ip = IpManager.get_coptersim_ip(1)
print("CopterSim IP:", coptersim_ip)

---

get_qgc_ip(original_ip='127.0.0.1')

Function Description: Retrieves the IP address of the QGroundControl (QGC) ground station. Parameters (Args):

Parameter Name	Type	Required	Default Value	Description
original_ip	str	No	127.0.0.1	Default IP address to return if lookup fails or lookup is skipped

Returns:

• str: QGC IP address string

Raises:

• None

---

Advanced Usage Examples

Demonstrates complex collaborative scenarios (e.g., multi-component coordination, asynchronous control, batch operations)

For networked simulation clusters involving multi-vehicle coordination, IpManager can be combined to perform batch IP validity validation, while simultaneously supporting automatic IP adaptation for heterogeneous simulation components via the multi-vehicle control module. For example, in a distributed deployment scenario where PX4 simulation, RflySim3D rendering, and QGC ground station run concurrently across multiple nodes, one can batch-fetch IPs for different services and pre-validate connectivity to prevent connection failures after simulation startup. Example code is as follows:

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


def is_ip_reachable(ip, timeout_seconds=1):
    """Check if the target host is reachable via a single 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

# Asynchronously validate connectivity of multiple simulation component IPs in batch
def check_ip_valid(service_name, get_ip_method):
    ip = get_ip_method()
    return service_name, ip, is_ip_reachable(ip)

# List of services to validate, covering various simulation component types
service_list = [
    ("RflySim3D", IpManager.get_rflysim3d_ip),
    ("CopterSim #1", lambda: IpManager.get_coptersim_ip(1)),
    ("QGC Ground Station", IpManager.get_qgc_ip)
]

# Asynchronous batch validation to improve startup efficiency in large-scale cluster scenarios
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} host is reachable")
        else:
            print(f"Warning: {service_name} IP {ip} host is unreachable; please check network configuration")

# Retrieve local IP for cross-device distributed simulation networking
local_ip = IpManager.get_local_ip()
print(f"Local networking IP: {local_ip}")

This approach enables asynchronous batch IP validation, making it suitable for complex scenarios such as multi-vehicle distributed cluster simulation and heterogeneous component collaboration, while proactively identifying network configuration issues.

Notes and Best Practices

• Configuration Timeliness: If the JSON's updateTime exceeds the current time by 10 seconds, it will be ignored; even if the 5-second cache has not yet expired, the cached content will undergo a timeliness check again.
• Validation Scope: is_valid() only checks whether environment variables, directories, and files exist; it does not check JSON content, IP format, port status, or network connectivity.
• Container Detection Scope: is_container() checks for /.dockerenv, /proc/self/cgroup keywords, and the DOCKER_CONTAINER, KUBERNETES_SERVICE_HOST environment variables, and cannot cover all custom container environments.
• Remote Request Deduplication: After resolving to a remote CopterSim, the same copter_id will only call ReqCopterSim.sendReSimIP() once in the current process; this state is stored in a class-level set and protected by a lock.

Changelog

• 2026-05-14: ✨ fix: Fixed the issue where an expired updateTime in the read file still returned results [P2]
• 2026-03-03: feat: Added IP handling mechanism to the SDK, enabling compatibility with local version migration to the cloud.