107 lines
2.7 KiB
Python
107 lines
2.7 KiB
Python
import asyncio, time
|
|
import threading
|
|
|
|
from fastapi import APIRouter, Query, BackgroundTasks
|
|
from djitellopy import Tello
|
|
|
|
|
|
class FlightStatsCollector:
|
|
def __init__(self):
|
|
self.stats_thread = None
|
|
self.stop_event = threading.Event()
|
|
|
|
def start_collecting(self):
|
|
self.stats_thread = threading.Thread(target=self.collect_stats)
|
|
self.stats_thread.start()
|
|
|
|
def stop_collecting(self):
|
|
self.stop_event.set()
|
|
if self.stats_thread:
|
|
self.stats_thread.join()
|
|
|
|
def collect_stats(self):
|
|
while not self.stop_event.is_set():
|
|
tello.send_command_with_return("command")
|
|
bat = tello.get_battery()
|
|
temp = tello.get_temperature()
|
|
# wsnr = tello.query_wifi_signal_noise_ratio()
|
|
print(f"bat: {bat} - temp: {temp}") #- wsnr: {wsnr}")
|
|
time.sleep(5)
|
|
|
|
router = APIRouter()
|
|
tello = Tello()
|
|
stats_collector = FlightStatsCollector()
|
|
|
|
|
|
def land_on_low_battery():
|
|
bat_level = tello.get_battery()
|
|
if bat_level < 20:
|
|
tello.land()
|
|
|
|
|
|
|
|
@router.get("/takeoff")
|
|
def takeoff():
|
|
try:
|
|
tello.connect()
|
|
except Exception as e:
|
|
return {"msg": "failed to connect"}
|
|
|
|
if tello.is_flying:
|
|
return {"msg": "already flying"}
|
|
|
|
tello.takeoff()
|
|
stats_collector.start_collecting()
|
|
|
|
if tello.is_flying:
|
|
return {"msg": "ok"}
|
|
else:
|
|
return {"msg": "takeoff failed"}
|
|
|
|
@router.get("/land")
|
|
def land():
|
|
if not tello.is_flying:
|
|
return {"msg": "already on land"}
|
|
|
|
tello.land()
|
|
stats_collector.stop_collecting()
|
|
|
|
if not tello.is_flying:
|
|
return {"msg": "ok"}
|
|
|
|
@router.get("/turn/{direction}/{degree}")
|
|
def turn(direction: str, degree: int):
|
|
if direction not in ["left", "right"]:
|
|
return {"direction must be only left or right"}
|
|
if degree < 1 or degree > 360:
|
|
return {"degree must be between 1 and 360"}
|
|
|
|
if direction == "right":
|
|
tello.rotate_clockwise(degree)
|
|
elif direction == "left":
|
|
tello.rotate_counter_clockwise(degree)
|
|
|
|
return {"msg": "ok"}
|
|
|
|
@router.get("/move/{direction}/{distance}")
|
|
def turn(direction: str, distance: int):
|
|
if direction not in ["back", "forward", "left", "right", "up" , "down"]:
|
|
return {"direction must be only back, forward, left, right up or down"}
|
|
if distance < 20 or distance > 200:
|
|
return {"distance must be between 20 and 500"}
|
|
|
|
try:
|
|
tello.move(direction, distance)
|
|
except Exception as e:
|
|
return {
|
|
"msg": "command failed",
|
|
"reason": e
|
|
}
|
|
|
|
return {"msg": "ok"}
|
|
|
|
@router.get("/end")
|
|
def end_flight_session():
|
|
tello.end()
|
|
return {"msg": "ok"}
|