Create mission from file, use enums

Author: tedtasman

docs/source/conf.py

@@ -11,7 +11,7 @@
 project = "MAVez"
 copyright = "2025, The Pennsylvania State University Unmanned Aerial Systems Club"
 author = "The Pennsylvania State University Unmanned Aerial Systems Club"
-version = "3.6.1"
+version = "3.7.0"
 
 # -- General configuration ---------------------------------------------------
 # https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name='MAVez'
-version='3.6.1'
+version='3.7.0'
 description='An intuitive MAVLink library for ArduPilot via pymavlink'
 authors=[
     {name='The Pennsylvania State University Unmanned Aerial Systems Club', email='psuuasofficial@gmail.com'}

src/MAVez/controller.py

@@ -26,6 +26,7 @@
 from MAVez.enums.mav_landed_state import MAVLandedState
 from MAVez.enums.mav_mission_result import MAVMissionResult
 from MAVez.enums.mav_result import MAVResult
+from MAVez.enums.mav_message import MAVMessage
 
 
 class Controller:
@@ -362,7 +363,7 @@ def send_clear_mission(self) -> int:
         self.logger.info("[Controller] Sent clear mission")
         return 0
 
-    async def set_mode(self, mode) -> int:
+    async def set_mode(self, mode: str) -> int:
         """
         Set the ardupilot mode.
 
@@ -638,7 +639,7 @@ async def set_home(self, home_coordinate=Coordinate(0, 0, 0)) -> int:
             self.logger.error("[Controller] Bad response received for set home location")
             return self.BAD_RESPONSE_ERROR
 
-    async def set_servo(self, servo_number, pwm) -> int:
+    async def set_servo(self, servo_number: int, pwm: int) -> int:
         """
         Set the a servo to a specified PWM value.
 
@@ -782,7 +783,7 @@ async def receive_landing_status(self, timeout=TIMEOUT_DURATION) -> int:
             self.logger.error("[Controller] Bad response received for landing status")
             return self.BAD_RESPONSE_ERROR
 
-    async def set_message_interval(self, message_type: int, interval: int) -> int:
+    async def set_message_interval(self, message_type: MAVMessage, interval: int) -> int:
         """
         Set the message interval for the specified message type.
 
@@ -799,7 +800,7 @@ async def set_message_interval(self, message_type: int, interval: int) -> int:
             0,  # target_component
             mavutil.mavlink.MAV_CMD_SET_MESSAGE_INTERVAL,  # command
             0,  # confirmation
-            message_type,  # param1
+            message_type.value,  # param1
             interval,  # param2
             0,  # param3
             0,  # param4
@@ -825,7 +826,7 @@ async def set_message_interval(self, message_type: int, interval: int) -> int:
             self.logger.error("[Controller] Bad response received for set message interval")
             return self.BAD_RESPONSE_ERROR
 
-    async def disable_message_interval(self, message_type) -> int:
+    async def disable_message_interval(self, message_type: MAVMessage) -> int:
         """
         Disable the message interval for the specified message type.
 
@@ -841,7 +842,7 @@ async def disable_message_interval(self, message_type) -> int:
             0,  # target_component
             mavutil.mavlink.MAV_CMD_SET_MESSAGE_INTERVAL,  # command
             0,  # confirmation
-            message_type,  # param1
+            message_type.value,  # param1
             -1,  # param2 # -1 disables the message
             0,  # param3
             0,  # param4

src/MAVez/flight_controller.py

@@ -15,6 +15,7 @@
 from logging import Logger
 from MAVez.mission import Mission
 from MAVez.controller import Controller
+from MAVez.enums.mav_message import MAVMessage
 import time
 
 class FlightController(Controller):
@@ -137,7 +138,7 @@ async def takeoff(self, takeoff_mission_filename: str) -> int:
 
         return 0
 
-    def append_mission(self, filename) -> int:
+    def append_mission(self, filename: str) -> int:
         """
         Append a mission to the mission list.
 
@@ -159,7 +160,7 @@ def append_mission(self, filename) -> int:
         self.mission_queue.append(mission)
         return 0
 
-    async def wait_for_waypoint(self, target) -> int:
+    async def wait_for_waypoint(self, target: int) -> int:
         """
         Wait for ardupilot to reach the current waypoint.
 
@@ -252,8 +253,8 @@ async def wait_for_landing(self, timeout=60) -> int:
 
         # start receiving landing status
         response = await self.set_message_interval(
-            message_type=245, interval=100000
-        )  # 245 is landing status (EXTENDED_SYS_STATE), 1e6 is 1 second
+            message_type=MAVMessage.EXTENDED_SYS_STATE, interval=100000 # 1 second
+        )
         if response:
             self.logger.critical("[Flight] Failed waiting for landing.")
             return response
@@ -281,8 +282,8 @@ async def wait_for_landing(self, timeout=60) -> int:
 
         # stop receiving landing status
         response = await self.disable_message_interval(
-            message_type=245
-        )  # 245 is landing status (EXTENDED_SYS_STATE)
+            message_type=MAVMessage.EXTENDED_SYS_STATE
+        )
         if response:
             self.logger.error("[Flight] Error waiting for landing.")
             return response

src/MAVez/mission.py

@@ -9,6 +9,7 @@
 An ardupilot mission.
 """
 
+from typing import Optional
 from uas_messenger.message import Message
 
 from MAVez.mission_item import MissionItem
@@ -74,6 +75,16 @@ def decode_error(self, error_code: int) -> str:
 
         return error_codes.get(error_code, f"\nUNKNOWN ERROR ({error_code})\n")
 
+    @classmethod
+    def from_file(cls, controller: Controller, filepath: str, type: int=0) -> Optional['Mission']:
+        mission = cls(controller, type)
+        res = mission.load_mission_from_file(filepath)
+        if res != 0:
+            if controller.logger:
+                controller.logger.error(f"[Mission] Failed to load mission: {mission.decode_error(res)}")
+            return None
+        return mission
+
     def load_mission_from_file(
         self, filename: str, start: int=0, end: int=-1, first_seq: int=-1, overwrite: bool=True
     ):
@@ -323,7 +334,7 @@ def get_mission_length(filepath: str, logger: logging.Logger | None = None) -> i
         with open(filepath, 'r') as f:
             return len(f.readlines()) - 1  # Subtract 1 for the header line
     except Exception as e:
-        print(f"Error reading mission file {filepath}: {e}")
+        logger.error(f"[Mission] Error reading mission file {filepath}: {e}") if logger else None
         return -1