Skip to content

RJ45_px6x Interface Documentation

Introduction

Overview: This file provides TCP communication and device management capabilities for connecting to the PX6x series drone swarm via RJ45 Ethernet ports. It implements network port device abstraction and TCP server functionality, supporting swarm networking and communication for drones.

In RflySim drone swarm simulation and real-flight testing scenarios, multiple PX6x flight controller drones typically require stable data communication with the ground station control end through wired RJ45 Ethernet ports to enable functions such as flight control command transmission and flight status reporting. As the network communication component of the RflySimSDK swarm module, this module uses the RJ_Device class to abstract communication devices for individual PX6x drones connected via RJ45, and the TCPServer class provides TCP server support for multi-device access. It is applicable to scenarios such as real-machine RJ45-based swarm testing and fixed-IP swarm communication in simulation environments, offering foundational communication interfaces for upper-layer swarm task scheduling and status monitoring.

Quick Start

Minimal working example; copy and modify minimal configuration to run.

from RflySimSDK.swarm import RJ45_px6x

# 1. Initialize RJ45 swarm device object using default broadcast address and port
rj_device = RJ45_px6x.RJ_Device()

# 2. Search for all PX6x devices on the local network, set search timeout to 2 seconds
found_devices = rj_device.discover_devices(timeout=2)
print("Discovered devices:", found_devices)

# 3. Example: Restart a device with a specified ID (if devices are discovered)
if found_devices:
    first_device_id = found_devices[0]
    print(f"Restarting device {first_device_id}")
    rj_device.reboot_device_by_id(device_id=first_device_id)

# 4. Start all discovered devices
rj_device.start_all_devices()
print("All devices started")

Environment and Dependencies

  • Python Environment: >= 3.8.10
  • Dependencies: re, socket, time
  • Prerequisites: Before calling this interface, ensure that the RJ45 network connection is properly configured, PX6x device hardware wiring is completed, and the RflySimSDK dependency environment is correctly installed.

Core Interface Description

The module RJ45_px6x.py includes configuration variables, helper functions, and core business classes.

Global Constants and Enumerations

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

Standalone Constants

None

Global/Standalone Functions

None

RJ_Device Class

Used to discover and manage RJ45-connected PX6x series devices via UDP broadcast, supporting device discovery, reboot, network configuration, and other functions. Suitable for network management of drone swarm devices.

__init__(broadcast_ip='255.255.255.255', broadcast_port=5000)

Function Description: Initializes the RJ_Device device management object and sets the IP address and port for UDP broadcast. Parameters (Args):

Parameter Name Type Required Default Value Description
broadcast_ip str No '255.255.255.255' IP address used for UDP broadcast
broadcast_port int No 5000 Port number used for UDP broadcast

Return Value (Returns):

  • RJ_Device instance object

Exceptions (Raises):

  • None

discover_devices(timeout=1)

Function Description: Broadcasts to search for available RJ45 devices on the local network and retrieves a list of discovered device information. Parameters (Args):

Parameter Name Type Required Default Value Description
timeout float No 1 Timeout duration for waiting device responses, in seconds

Return Value (Returns):

  • list: A list of dictionaries, each containing information about a discovered device (e.g., device ID, IP address)

Exceptions (Raises):

  • None

Example:

from RflySimSDK.swarm.RJ45_px6x import RJ_Device
device_manager = RJ_Device()
# Search for devices on the local network with a 2-second timeout
devices = device_manager.discover_devices(timeout=2)
print(f"Discovered {len(devices)} devices: {devices}")

---

#### `reboot_device_by_id(device_id)`

**Function Description**: Sends a reboot command to the specified device based on its device ID.  
**Parameters (Args)**:

| Parameter Name | Type | Required | Default Value | Description |
| :--- | :--- | :---: | :--- | :--- |
| `device_id` | `int/str` | Yes | - | ID of the target device |

**Returns**:

- `bool`: Returns `True` if the command is sent successfully; `False` otherwise.

**Raises**:

- None

---

#### `start_all_devices()`

**Function Description**: Sends a startup command to all discovered devices on the local network.  
**Parameters (Args)**:  
None  
**Returns**:

- `None`

**Raises**:

- None

---

#### `get_current_ip(device_id)`

**Function Description**: Retrieves the current IP address of the device corresponding to the given device ID.  
**Parameters (Args)**:

| Parameter Name | Type | Required | Default Value | Description |
| :--- | :--- | :---: | :--- | :--- |
| `device_id` | `int/str` | Yes | - | ID of the target device |

**Returns**:

- `str | None`: Returns the IP address string of the device upon success; returns `None` if the device is not found.

**Raises**:

- None

---

#### `set_dhcp(device_id, enable)`

**Function Description**: Configures whether the specified device uses DHCP to automatically obtain its IP address.  
**Parameters (Args)**:

| Parameter Name | Type | Required | Default Value | Description |
| :--- | :--- | :---: | :--- | :--- |
| `device_id` | `int/str` | Yes | - | ID of the target device |
| `enable` | `bool` | Yes | - | `True` enables DHCP; `False` disables DHCP |

**Returns**:

- `bool`: Returns `True` if the configuration is successful; `False` otherwise.

**Raises**:

- None

**Example**:

```python
from RflySimSDK.swarm.RJ45_px6x import RJ_Device
device_manager = RJ_Device()
devices = device_manager.discover_devices()
# Enable DHCP for the first device
if devices:
    device_id = devices[0]['device_id']
    success = device_manager.set_dhcp(device_id, enable=True)
    print(f"DHCP setting result: {success}")

---

#### `set_static_ip(device_id, ip_address, gateway, mask)`

**Function Description**: Sets a static IP address, gateway, and subnet mask for a specified device.  
**Parameters (Args)**:

| Parameter Name | Type | Required | Default Value | Description |
| :--- | :--- | :---: | :--- | :--- |
| `device_id` | `int/str` | Yes | - | ID of the target device |
| `ip_address` | `str` | Yes | - | Static IP address to be set |
| `gateway` | `str` | Yes | - | Gateway address to be set |
| `mask` | `str` | Yes | - | Subnet mask to be set |

**Return Value (Returns)**:

- `bool`: Returns `True` on success, `False` on failure.

**Exceptions (Raises)**:

- None

---

#### `modify_device_id(device_id)`

**Function Description**: Modifies the device ID of a specified device.  
**Parameters (Args)**:

| Parameter Name | Type | Required | Default Value | Description |
| :--- | :--- | :---: | :--- | :--- |
| `device_id` | `int/str` | Yes | - | New device ID to be assigned |

**Return Value (Returns)**:

- `bool`: Returns `True` on success, `False` on failure.

**Exceptions (Raises)**:

- None

---

#### `modify_device_port(device_id)`

**Function Description**: Modifies the communication port of a specified device.  
**Parameters (Args)**:

| Parameter Name | Type | Required | Default Value | Description |
| :--- | :--- | :---: | :--- | :--- |
| `device_id` | `int/str` | Yes | - | New communication port to be assigned |

**Return Value (Returns)**:

- `bool`: Returns `True` on success, `False` on failure.

**Exceptions (Raises)**:

- None

---

### `TCPServer` Class

A multi-mode TCP/UDP network communication server utility class, used to establish network communication between PX6x flight controllers and the simulation environment in RflySim drone swarm simulations. It supports switching among four connection modes: TCP server, TCP client, UDP client, and UDP server.

#### `__init__()`

**Function Description**: Initializes a `TCPServer` instance.  
**Parameters (Args)**:

- None

**Return Value (Returns)**:

- `TCPServer` instance object

**Exceptions (Raises)**:

- None

---

#### `get_local_ip()`

**Function Description**: Retrieves the local LAN IP address available for network communication binding.  
**Parameters (Args)**:

- None

**Return Value (Returns)**:

- `str`: Local LAN IP address

**Exceptions (Raises)**:

- None

---

#### `setup_server()`

**Function Description**: Starts the network service based on the currently configured connection mode, completing socket binding and initialization of listening (for server modes) or connection (for client modes).  
**Parameters (Args)**:

- None

**Return Value (Returns)**:

- None

**Exceptions (Raises)**:

- None

---

#### `print_current_mode()`

**Function Description**: Prints the currently active network connection mode.  
**Parameters (Args)**:

- None

**Return Value (Returns)**:

- None

**Exceptions (Raises)**:

- None

---

#### `change_connection_mode()`

**Function Description**: Modifies the network connection mode, allowing the user to interactively select the desired mode.  
**Parameters (Args)**:

- None

**Return Value (Returns)**:

- None

**Exceptions (Raises)**:

- None

---

#### `set_default_cleanup()`

**Function Description**: Releases network resources and closes any open sockets.  
**Parameters (Args)**:

- None

**Return Value (Returns)**:

- None

**Exceptions (Raises)**:

- None

---

#### `set_default_mode()`

**Function Description**: Sets the default network connection mode to `UDP_SERVER_MODE`.  
**Parameters (Args)**:

- None

**Return Value (Returns)**:

- None

**Exceptions (Raises)**:

- None

**Example**:

```python
from RflySimSDK.swarm import TCPServer

# Create a TCPServer instance with default configuration
tcp_server = TCPServer()
tcp_server.set_default_mode()
tcp_server.setup_server()

# Print the current connection mode
tcp_server.print_current_mode()

# Release resources upon exit
tcp_server.set_default_cleanup()

---

### Class Variable Description

| Class Variable Name | Type | Value | Description |
| :--- | :--- | :--- | :--- |
| `TCP_SERVER_MODE` | `int` | `0` | TCP server connection mode |
| `TCP_CLIENT_MODE` | `int` | `1` | TCP client connection mode |
| `UDP_CLIENT_MODE` | `int` | `2` | UDP client connection mode |
| `UDP_SERVER_MODE` | `int` | `3` | UDP server connection mode |

## Advanced Usage Example

> Demonstrates complex composite scenarios (e.g., multi-class collaboration, asynchronous control, batch operations)

```python
import RflySimSDK.swarm.RJ45_px6x as rj_sdk
import asyncio

async def batch_configure_devices(async_interval=1):
    # Batch scan to discover all online RJ45 devices
    device_list = rj_sdk.RJ_Device.discover_devices(timeout=5)
    if not device_list:
        print("No configurable devices found")
        return None
    # Start TCP server to establish device communication links
    tcp_server = rj_sdk.TCPServer()
    local_ip = tcp_server.get_local_ip()
    tcp_server.setup_server(local_ip, 8899)
    tcp_server.set_default_cleanup()
    # Asynchronously batch-modify device IDs and static IPs to avoid blocking due to synchronous waits
    config_tasks = []
    for idx, dev in enumerate(device_list):
        new_id = 100 + idx
        new_ip = f"{'.'.join(local_ip.split('.')[:3])}.{200 + idx}"
        # Asynchronously encapsulate device configuration tasks
        async def config_task(dev, new_id, new_ip):
            rj_sdk.RJ_Device.modify_device_id(dev, new_id)
            rj_sdk.RJ_Device.set_static_ip(dev, new_ip, "255.255.255.0", local_ip)
            await asyncio.sleep(async_interval)
        config_tasks.append(config_task(dev, new_id, new_ip))
    # Concurrently execute all configuration tasks
    await asyncio.gather(*config_tasks)
    # Batch-start all configured devices
    rj_sdk.RJ_Device.start_all_devices()
    # Switch to auto-reconnect mode to ensure link stability
    tcp_server.change_connection_mode("auto_reconnect")
    tcp_server.print_current_mode()
    print(f"Batch configuration and startup completed for {len(device_list)} devices")
    return device_list

if __name__ == "__main__":
    asyncio.run(batch_configure_devices())

Notes and Pitfall Avoidance Guide

  • Device Scan Scope Limitation: discover_devices can only discover PX6X devices within the same LAN subnet. Cross-subnet scanning will directly return an empty list; ensure all target devices and the control endpoint are on the same subnet before use.
  • Static IP Configuration Conflict Risk: When calling set_static_ip for batch device configuration, verify in advance that the target IPs are not already occupied by other devices on the LAN; IP conflicts will cause subsequent communication failures for the affected devices.
  • TCP Server Resource Cleanup: Before program exit, ensure set_default_cleanup is correctly invoked; otherwise, the port will remain occupied, leading to port binding failure errors upon subsequent server startups.
  • Device ID Modification Activation Condition: After calling modify_device_id to change a device’s ID, reboot_device_by_id must be invoked to restart the device for the new ID to take effect; attempting subsequent operations with the new ID before rebooting will result in a “device not found” error.

Changelog

  • 2024-12-10: fix: Updated comments for swarm control interfaces
  • 2024-11-13: fix: Adapted to the latest hardware interface program
  • 2024-08-30: feat: Added factory-default TCP connection mode, local identification interface, and TCP-to-UDP mode interface
  • 2024-08-28: (1) feat: Added delay time function, DHCP enable/disable, static IP setting, device port configuration, and device ID setting
  • 2024-07-31: feat: Added RJ45 network relay-related interfaces