aleks/flix
1
0
Fork 0
mirror of https://github.com/okalachev/flix.git synced 2026-08-16 00:38:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into stm

Browse Source
This commit is contained in:
Oleg Kalachev
2026-08-11 23:17:04 +03:00
parent c3b818c2ae
commit b8c687f3ed
117 changed files with 2264 additions and 810 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/pyflix/README.md
+42 -32
View File
@@ -1,8 +1,8 @@
# Flix Python library
The Flix Python library allows you to remotely connect to a Flix quadcopter. It provides access to telemetry data, supports executing CLI commands, and controlling the drone's flight.
The Flix Python library allows you to remotely connect to a Flix quadcopter. It provides access to telemetry data, supports executing console commands, and controlling the drone's flight.
To use the library, connect to the drone's Wi-Fi. To use it with the simulator, ensure the script runs on the same local network as the simulator.
To use the library, connect to the drone's Wi-Fi. To use it with the simulator, ensure the script runs on the same network as the simulator.
## Installation
@@ -24,30 +24,36 @@ pip install pyflix
The API is accessed through the `Flix` class:
```python
from flix import Flix
from pyflix import Flix
flix = Flix() # create a Flix object and wait for connection
```
If using ESP-NOW connection, specify the proxy device name in `FLIX_DEVICE` environment variable or pass it to the constructor: `Flix(device='/dev/cu.usbserial-0001')`.
### Telemetry
Basic telemetry is available through object properties. The properties names generally match the corresponding variables in the firmware itself:
Basic telemetry is available through object properties. The property names generally match the corresponding variables in the firmware code:
```python
print(flix.connected) # True if connected to the drone
print(flix.mode) # current flight mode (str)
print(flix.armed) # True if the drone is armed
print(flix.landed) # True if the drone is landed
print(flix.voltage) # battery voltage (NaN - unknown, ~0 - USB powered)
print(flix.attitude) # attitude quaternion [w, x, y, z]
print(flix.attitude_euler) # attitude as Euler angles [roll, pitch, yaw]
print(flix.rates) # angular rates [roll_rate, pitch_rate, yaw_rate]
print(flix.channels) # raw RC channels (list)
print(flix.motors) # motors outputs (list)
print(flix.motors) # motor outputs (list)
print(flix.acc) # accelerometer output (list)
print(flix.gyro) # gyroscope output (list)
```
> [!NOTE]
> The library uses the Front-Left-Up coordinate system — the same as in the firmware. All angles are in radians.
The library uses the Front-Left-Up coordinate system — the same as the firmware:
<img src="../../docs/img/drone-axes-rotate.svg" width="300">
All angles are in radians.
### Events
@@ -89,36 +95,37 @@ Full list of events:
|-----|-----------|----------------|
|`connected`|Connected to the drone||
|`disconnected`|Connection is lost||
|`armed`|Armed state update|Armed state (*bool*)|
|`mode`|Flight mode update|Flight mode (*str*)|
|`landed`|Landed state update|Landed state (*bool*)|
|`print`|The drone sends text to the console|Text|
|`attitude`|Attitude update|Attitude quaternion (*list*)|
|`attitude_euler`|Attitude update|Euler angles (*list*)|
|`rates`|Angular rates update|Angular rates (*list*)|
|`channels`|Raw RC channels update|Raw RC channels (*list*)|
|`motors`|Motors outputs update|Motors outputs (*list*)|
|`acc`|Accelerometer update|Accelerometer output (*list*)|
|`gyro`|Gyroscope update|Gyroscope output (*list*)|
|`armed`|Armed state update|Armed state *(bool)*|
|`mode`|Flight mode update|Flight mode *(str)*|
|`landed`|Landed state update|Landed state *(bool)*|
|`voltage`|Battery voltage update|Voltage *(float)*|
|`print`|The drone prints text to the console|Text|
|`attitude`|Attitude update|Attitude quaternion *(list)*|
|`attitude_euler`|Attitude update|Euler angles *(list)*|
|`rates`|Angular rates update|Angular rates *(list)*|
|`channels`|Raw RC channels update|Raw RC channels *(list)*|
|`motors`|Motor outputs update|Motor outputs *(list)*|
|`acc`|Accelerometer update|Accelerometer output *(list)*|
|`gyro`|Gyroscope update|Gyroscope output *(list)*|
|`mavlink`|Received MAVLink message|Message object|
|`mavlink.<message_name>`|Received specific MAVLink message|Message object|
|`mavlink.<message_id>`|Received specific MAVLink message|Message object|
|`value`|Named value update (see below)|Name, value|
|`value.<name>`|Specific named value update (see bellow)|Value|
|`value.<name>`|Specific named value update (see below)|Value|
> [!NOTE]
> Update events trigger on every new data from the drone, and do not mean the value is changed.
> Update events trigger on every new piece of data from the drone, and do not mean the value has changed.
### Common methods
### Basic methods
Get and set firmware parameters using `get_param` and `set_param` methods:
```python
pitch_p = flix.get_param('PITCH_P') # get parameter value
flix.set_param('PITCH_P', 5) # set parameter value
pitch_p = flix.get_param('CTL_P_P') # get parameter value
flix.set_param('CTL_P_P', 5) # set parameter value
```
Execute CLI commands using `cli` method. This method returns command response:
Execute console commands using `cli` method. This method returns the command response:
```python
imu = flix.cli('imu') # get detailed IMU data
@@ -136,7 +143,7 @@ flix.set_armed(True) # arm the drone
flix.set_armed(False) # disarm the drone
```
You can imitate pilot's controls using `set_controls` method:
You can pass pilot's controls using `set_controls` method:
```python
flix.set_controls(roll=0, pitch=0, yaw=0, throttle=0.6)
@@ -166,10 +173,10 @@ Setting angular rates target:
flix.set_rates([0.1, 0.2, 0.3], 0.6) # set target roll rate, pitch rate, yaw rate and thrust
```
You also can control raw motors outputs directly:
You also can control raw motor outputs directly:
```python
flix.set_motors([0.5, 0.5, 0.5, 0.5]) # set motors outputs in range [0, 1]
flix.set_motors([0.5, 0.5, 0.5, 0.5]) # set motor outputs in range [0, 1]
```
In *AUTO* mode, the drone will arm automatically if the thrust is greater than zero, and disarm if thrust is zero. Therefore, to disarm the drone, set thrust to zero:
@@ -183,7 +190,7 @@ The following methods are in development and are not functional yet:
* `set_position` — set target position.
* `set_velocity` — set target velocity.
To exit from *AUTO* mode move control sticks and the drone will switch to *STAB* mode.
To exit *AUTO* mode move control sticks and the drone will switch to *STAB* mode.
## Usage alongside QGroundControl
@@ -215,6 +222,13 @@ The following scripts demonstrate how to use the library:
* [`log.py`](../log.py) — download flight logs from the drone.
* [`example.py`](../example.py) — a simple example, prints telemetry data and waits for events.
> [!TIP]
> Set `FLIX_DEVICE` environment variable to use these tools with ESP-NOW connection, for example:
>
> ```bash
> FLIX_DEVICE=/dev/cu.usbserial-0001 tools/cli.py
> ```
## Advanced usage
### MAVLink
@@ -274,7 +288,3 @@ logger = logging.getLogger('flix')
logger.setLevel(logging.DEBUG) # be more verbose
logger.setLevel(logging.WARNING) # be less verbose
```
## Stability
The library is in development stage. The API is not stable.
tools/pyflix/flix.py
+28 -16
View File
@@ -5,6 +5,7 @@
import os
import time
import math
from queue import Queue, Empty
from typing import Optional, Callable, List, Dict, Any, Union, Sequence
import logging
@@ -17,7 +18,7 @@ from pymavlink.dialects.v20 import common as mavlink
logger = logging.getLogger('flix')
if not logger.hasHandlers():
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter('%(name)s - %(levelname)s - %(message)s'))
handler.setFormatter(logging.Formatter('%(name)s: %(message)s'))
logger.addHandler(handler)
logger.setLevel(logging.INFO)
@@ -26,6 +27,7 @@ class Flix:
mode: str = ''
armed: bool = False
landed: bool = False
voltage: float = math.nan
attitude: List[float]
attitude_euler: List[float] # roll, pitch, yaw
rates: List[float]
@@ -40,24 +42,29 @@ class Flix:
_connection_timeout = 3
_print_buffer: str = ''
_modes = ['MANUAL', 'ACRO', 'STAB', 'AUTO']
_modes = ['RAW', 'ACRO', 'STAB', 'AUTO']
def __init__(self, system_id: int=1, wait_connection: bool=True):
def __init__(self, system_id: int=1, wait_connection: bool=True, device=os.getenv('FLIX_DEVICE')):
if not (0 <= system_id < 256):
raise ValueError('system_id must be in range [0, 255]')
self._setup_mavlink()
self.system_id = system_id
self._init_state()
try:
# Direct connection
logger.debug('Listening on port 14550')
self.connection: mavutil.mavfile = mavutil.mavlink_connection('udpin:0.0.0.0:14550', source_system=255) # type: ignore
except OSError as e:
if e.errno != errno.EADDRINUSE:
raise
# Port busy - using proxy
logger.debug('Listening on port 14555 (proxy)')
self.connection: mavutil.mavfile = mavutil.mavlink_connection('udpin:0.0.0.0:14555', source_system=254) # type: ignore
if device is not None:
# User defined connection
logger.debug(f'Connecting to {device}')
self.connection: mavutil.mavfile = mavutil.mavlink_connection(device, source_system=255) # type: ignore
else:
try:
# Direct connection
logger.debug('Listening on port 14550')
self.connection: mavutil.mavfile = mavutil.mavlink_connection('udpin:0.0.0.0:14550', source_system=255) # type: ignore
except OSError as e:
if e.errno != errno.EADDRINUSE:
raise
# Port busy - using proxy
logger.debug('Listening on port 14555 (proxy)')
self.connection: mavutil.mavfile = mavutil.mavlink_connection('udpin:0.0.0.0:14555', source_system=254) # type: ignore
self.connection.target_system = system_id
self.mavlink: mavlink.MAVLink = self.connection.mav
self._event_listeners: Dict[str, List[Callable[..., Any]]] = {}
@@ -68,7 +75,7 @@ class Flix:
self._heartbeat_thread.start()
if wait_connection:
self.wait('mavlink.HEARTBEAT')
time.sleep(0.2) # give some time to receive initial state
time.sleep(0.6) # give some time to receive initial state
def _init_state(self):
self.attitude = [1, 0, 0, 0]
@@ -138,7 +145,7 @@ class Flix:
while True:
try:
msg: Optional[mavlink.MAVLink_message] = self.connection.recv_match(blocking=True)
if msg is None:
if msg is None or msg.get_srcSystem() != self.system_id:
continue
self._connected()
msg_dict = msg.to_dict()
@@ -185,11 +192,16 @@ class Flix:
self._trigger('motors', self.motors)
if isinstance(msg, mavlink.MAVLink_scaled_imu_message):
self.acc = self._mavlink_to_flu([msg.xacc / 1000, msg.yacc / 1000, msg.zacc / 1000])
ONE_G = 9.80665
self.acc = self._mavlink_to_flu([msg.xacc * ONE_G / 1000, msg.yacc * ONE_G / 1000, msg.zacc * ONE_G / 1000])
self.gyro = self._mavlink_to_flu([msg.xgyro / 1000, msg.ygyro / 1000, msg.zgyro / 1000])
self._trigger('acc', self.acc)
self._trigger('gyro', self.gyro)
if isinstance(msg, mavlink.MAVLink_battery_status_message):
self.voltage = msg.voltages[0] / 1000
self._trigger('voltage', self.voltage)
if isinstance(msg, mavlink.MAVLink_serial_control_message):
# new chunk of data
text = bytes(msg.data)[:msg.count].decode('utf-8', errors='ignore')