Make method for sending mavlink commands public in pyflix

This commit is contained in:
Oleg Kalachev
2026-08-29 17:32:18 +03:00
parent 3861c4438a
commit 97a01db7b4
2 changed files with 21 additions and 21 deletions
+19 -19
View File
@@ -255,23 +255,6 @@ class Flix:
def _flu_to_mavlink(v: Sequence[float]) -> List[float]: def _flu_to_mavlink(v: Sequence[float]) -> List[float]:
return Flix._mavlink_to_flu(v) # flu to mavlink is the same as mavlink to flu return Flix._mavlink_to_flu(v) # flu to mavlink is the same as mavlink to flu
def _command_send(self, command: int, params: Sequence[float]):
if len(params) != 7:
raise ValueError('Command must have 7 parameters')
for attempt in range(3):
try:
logger.debug(f'Send command {command} with params {params} (attempt #{attempt + 1})')
self.mavlink.command_long_send(self.system_id, 0, command, 0, *params) # type: ignore
ack = self.wait('mavlink.COMMAND_ACK', value=lambda msg: msg.command == command, timeout=0.1)
if ack.result != mavlink.MAV_RESULT_ACCEPTED:
name = getattr(mavlink.enums['MAV_CMD'].get(command, {}), 'name', f'UNKNOWN({command})')
result = getattr(mavlink.enums['MAV_RESULT'].get(ack.result, {}), 'name', f'UNKNOWN({ack.result})')
raise RuntimeError(f'Command {name} failed with result {result}')
return
except TimeoutError:
continue
raise RuntimeError(f'Failed to send command {command} after 3 attempts')
def _connected(self): def _connected(self):
# Reset disconnection timer # Reset disconnection timer
self._disconnected_timer.cancel() self._disconnected_timer.cancel()
@@ -288,6 +271,23 @@ class Flix:
self.connected = False self.connected = False
self._trigger('disconnected') self._trigger('disconnected')
def send_command(self, command: int, params: Sequence[float]):
if len(params) != 7:
raise ValueError('Command must have 7 parameters')
for attempt in range(3):
try:
logger.debug(f'Send command {command} with params {params} (attempt #{attempt + 1})')
self.mavlink.command_long_send(self.system_id, 0, command, 0, *params) # type: ignore
ack: mavlink.MAVLink_command_ack_message = self.wait('mavlink.COMMAND_ACK', value=lambda msg: msg.command == command, timeout=0.1)
if ack.result != mavlink.MAV_RESULT_ACCEPTED:
name = getattr(mavlink.enums['MAV_CMD'].get(command, {}), 'name', f'UNKNOWN({command})')
result = getattr(mavlink.enums['MAV_RESULT'].get(ack.result, {}), 'name', f'UNKNOWN({ack.result})')
raise RuntimeError(f'Command {name} failed with result {result}')
return
except TimeoutError:
continue
raise RuntimeError(f'Failed to send command {command} after 3 attempts')
def get_param(self, name: str) -> float: def get_param(self, name: str) -> float:
if len(name.encode('ascii')) > 16: if len(name.encode('ascii')) > 16:
raise ValueError('Parameter name must be 16 characters or less') raise ValueError('Parameter name must be 16 characters or less')
@@ -319,10 +319,10 @@ class Flix:
def set_mode(self, mode: Union[str, int]): def set_mode(self, mode: Union[str, int]):
if isinstance(mode, str): if isinstance(mode, str):
mode = self._modes.index(mode.upper()) mode = self._modes.index(mode.upper())
self._command_send(mavlink.MAV_CMD_DO_SET_MODE, (0, mode, 0, 0, 0, 0, 0)) self.send_command(mavlink.MAV_CMD_DO_SET_MODE, (0, mode, 0, 0, 0, 0, 0))
def set_armed(self, armed: bool): def set_armed(self, armed: bool):
self._command_send(mavlink.MAV_CMD_COMPONENT_ARM_DISARM, (1 if armed else 0, 0, 0, 0, 0, 0, 0)) self.send_command(mavlink.MAV_CMD_COMPONENT_ARM_DISARM, (1 if armed else 0, 0, 0, 0, 0, 0, 0))
def set_position(self, position: Sequence[float], yaw: Optional[float] = None, wait: bool = False, tolerance: float = 0.1): def set_position(self, position: Sequence[float], yaw: Optional[float] = None, wait: bool = False, tolerance: float = 0.1):
raise NotImplementedError('Position control is not implemented yet') raise NotImplementedError('Position control is not implemented yet')
+2 -2
View File
@@ -48,6 +48,6 @@ def test():
print("=== Check command errors") print("=== Check command errors")
with raises(RuntimeError, match='MAV_RESULT_DENIED'): with raises(RuntimeError, match='MAV_RESULT_DENIED'):
flix._command_send(mavlink.MAV_CMD_DO_SET_MODE, [0, 99, 0, 0, 0, 0, 0]) # invalid mode flix.send_command(mavlink.MAV_CMD_DO_SET_MODE, [0, 99, 0, 0, 0, 0, 0]) # invalid mode
with raises(RuntimeError, match='MAV_RESULT_UNSUPPORTED'): with raises(RuntimeError, match='MAV_RESULT_UNSUPPORTED'):
flix._command_send(mavlink.MAV_CMD_DO_PARACHUTE, [0, 0, 0, 0, 0, 0, 0]) # unsupported command flix.send_command(mavlink.MAV_CMD_DO_PARACHUTE, [0, 0, 0, 0, 0, 0, 0]) # unsupported command