alihan/tello-commander
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'master' of https://gitea.umutalihandikel.com/alihan/tello-commander

Browse Source
This commit is contained in:
Alihan
2023-05-19 03:01:33 +03:00
parent 2b5bbfa285 0b3523000c
commit 1cb1d6c7da
3 changed files with 34 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
brain/brain_base.py
View File

@@ -60,7 +60,7 @@ class BaseBrain:
self.response_to_chatui = msg
else:
msg = f"I will send this command: {command}"
logger.success(msg)
logger.info(msg)
self.response_to_chatui = msg
self.command_handler.handle(command)
else:

33
brain/commands.py
View File

@@ -2,6 +2,7 @@ import sys
import requests
from requests.exceptions import ConnectionError
from loguru import logger
from settings.config import settings
@@ -13,52 +14,50 @@ class CommandHandler:
self.COMMANDER_COMMANDS_URL = f"{settings.COMMANDER_ROOT_URL}/command"
self._check_commander_health()
def _parse_response(self, raw_resp):
json_resp = raw_resp.json()
if json_resp["msg"] == "ok": logger.success(json_resp)
else: logger.warning(json_resp["msg"])
def _check_commander_health(self):
try:
response = requests.get(f"{settings.COMMANDER_ROOT_URL}/test/health")
status = response.json()
except ConnectionError:
print(f"commander service is unavailable: {settings.COMMANDER_ROOT_URL}"); sys.exit(1)
logger.error(f"commander service is unavailable: {settings.COMMANDER_ROOT_URL}"); sys.exit(1)
#raise Exception(f"commander service is unavailable: {settings.COMMANDER_ROOT_URL}")
if status["msg"] != "ok":
print("connected to commander service")
logger.success("connected to commander service")
def _move(self, direction, distance):
response = requests.get(f"{self.COMMANDER_COMMANDS_URL}/move/{direction}/{distance}")
print(response.json())
self._parse_response(raw_resp=response)
def _turn(self, direction, degree):
response = requests.get(f"{self.COMMANDER_COMMANDS_URL}/turn/{direction}/{degree}")
print(response.json())
self._parse_response(raw_resp=response)
def _takeoff(self):
response = requests.get(f"{self.COMMANDER_COMMANDS_URL}/takeoff")
print(response.json())
self._parse_response(raw_resp=response)
def _land(self):
response = requests.get(f"{self.COMMANDER_COMMANDS_URL}/land")
print(response.json())
self._parse_response(raw_resp=response)
def _end_session(self):
response = requests.get(f"{self.COMMANDER_COMMANDS_URL}/end")
print(response.json())
self._parse_response(raw_resp=response)
def _emergency(self):
response = requests.get(f"{self.COMMANDER_COMMANDS_URL}/emergency")
print(response.json())
self._parse_response(raw_resp=response)
def handle(self, cmd: dict):
#print(f"commanding for: {cmd}")
if cmd["command"] == "move":
self._move(
direction=cmd["direction"],
distance=cmd["distance_angle"]
)
self._move(direction=cmd["direction"], distance=cmd["distance_angle"])
elif cmd["command"] == "turn":
self._turn(
direction=cmd["direction"],
degree=cmd["distance_angle"]
)
self._turn(direction=cmd["direction"], degree=cmd["distance_angle"])
elif cmd["command"] == "takeoff":
self._takeoff()
elif cmd["command"] == "land":

33
commander/routes/route_command.py
View File

@@ -26,7 +26,7 @@ class FlightStatsCollector:
temp = tello.get_temperature()
# wsnr = tello.query_wifi_signal_noise_ratio()
print(f"bat: {bat} - temp: {temp}") #- wsnr: {wsnr}")
time.sleep(5)
time.sleep(3)
router = APIRouter()
tello = Tello()
@@ -37,7 +37,6 @@ def land_on_low_battery():
bat_level = tello.get_battery()
if bat_level < 20:
tello.land()
@router.get("/takeoff")
@@ -48,7 +47,7 @@ def takeoff():
return {"msg": "command failed", "reason": "failed to connect"}
if tello.is_flying:
return {"msg": "already flying"}
return {"msg": "command failed", "reason": "already flying"}
try:
tello.takeoff()
@@ -64,7 +63,7 @@ def takeoff():
@router.get("/land")
def land():
if not tello.is_flying:
return {"msg": "already on land"}
return {"msg": "command failed", "reason": "already on land"}
try:
tello.land()
@@ -78,9 +77,9 @@ def land():
@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"}
return {"msg": "command failed", "reason": "direction must be only left or right"}
if degree < 1 or degree > 360:
return {"degree must be between 1 and 360"}
return {"msg": "command failed", "reason": "degree must be between 1 and 360"}
try:
if direction == "right":
@@ -88,31 +87,33 @@ def turn(direction: str, degree: int):
elif direction == "left":
tello.rotate_counter_clockwise(degree)
except Exception as e:
return { "msg": "command failed", "reason": e }
return {"msg": "command failed", "reason": e}
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"}
return {"msg": "command failed", "reason": "direction must be only back, forward, left, right up or down"}
if distance < 20 or distance > 250:
# tellonun kendi sınırı 500 ama nolur nolmaz diye kısıtladım
return {"msg": "command failed", "reason": "distance must be between 20 and 250"}
try:
tello.move(direction, distance)
except Exception as e:
return { "msg": "command failed", "reason": e }
return {"msg": "command failed", "reason": e}
return {"msg": "ok"}
@router.get("/emergency")
def emercengy_stop():
def emergency_stop():
try:
tello.emergency()
return {"msg": "ok"}
except Exception as e:
return { "msg": "command failed", "reason": e }
return {"msg": "command failed", "reason": e}
@router.get("/end")
def end_flight_session():
@@ -120,4 +121,4 @@ def end_flight_session():
tello.end()
return {"msg": "ok"}
except Exception as e:
return { "msg": "command failed", "reason": e }
return {"msg": "command failed", "reason": e}