flight-gear-copilot/__init__.py
merspieler 180444e671 Make mycroft spell OK correctly
Signed-off-by: merspieler <merspieler@airmail.cc>
2018-09-11 18:56:30 +02:00

828 lines
23 KiB
Python

import sys
import socket
import re
import json
from time import sleep
from telnetlib import Telnet
from adapt.intent import IntentBuilder
from mycroft.audio import wait_while_speaking
from mycroft import MycroftSkill, intent_handler
from mycroft.skills.core import MycroftSkill
from mycroft.util.log import getLogger
from mycroft.util import normalize
LOGGER = getLogger(__name__)
class FlightGearCopilotSkill(MycroftSkill):
def __init__(self):
super(FlightGearCopilotSkill, self).__init__()
# Already existing options are here read only -> will not be overwritten
self.settings['host'] = "localhost"
self.settings['port'] = 8081
self.write_default_profiles()
# TODO make it possible, to play a .mp3 file instead of tts
#################################################################
# #
# Actions #
# #
#################################################################
#########################
# #
# Flaps #
# #
#########################
@intent_handler(IntentBuilder('FlapsIntent').require('flaps'))
def handle_flaps_intent(self, message):
flaps_request = normalize(message.data['utterance'])
if flaps_request == "flaps":
self.speak_dialog("no.setting")
sys.exit(0)
# extracting the flaps setting from the utterance
match = re.match(r'.*flaps.* (up|full|down|\d{1,2}).*', flaps_request, re.I)
if match == None:
self.speak_dialog("no.valid.flaps")
sys.exit(0)
flaps_request = match.group(1)
tn = self.connect()
# get acid
acid = self.get_prop(tn, "/sim/aircraft")
# read acid to know which profile to use
profile = None
for i_profiles in self.settings['profiles']:
for i_acid in i_profiles['acid']:
if i_acid == acid:
profile = i_profiles
break
if profile != None:
break
# check if the profile was found
if profile == None:
self.speak_dialog("no.profile")
self.exit(tn)
# get kias
kias = int(float(self.get_prop(tn, "/velocities/airspeed-kt")))
# find the flaps settings for the flaps id
o_flaps = None
for i_flaps in profile['flaps']:
if str(i_flaps['id']) == str(flaps_request):
o_flaps = i_flaps
break
# check if flaps setting is known
if o_flaps == None:
self.speak_dialog("flaps.setting.unknown")
self.exit(tn)
flaps = int(self.get_prop(tn, str(profile['flaps-path'])))
# check if extend or retract flaps
if str(flaps_request) == "down" or str(flaps_request) == "full":
if flaps == o_flaps['value']:
self.speak_dialog("keep.flaps")
self.exit(tn)
else:
flaps_mov = "extend"
flaps_name = flaps_request
flaps_request = o_flaps['value']
elif str(flaps_request) == "up":
if flaps == o_flaps['value']:
self.speak_dialog("keep.flaps")
self.exit(tn)
else:
flaps_mov = "retract"
flaps_name = flaps_request
flaps_request = o_flaps['value']
else:
flaps_name = str(flaps_request)
if int(flaps_request) > flaps:
flaps_mov = "extend"
elif int(flaps_request) < flaps:
flaps_mov = "retract"
else:
self.speak_dialog("keep.flaps")
self.exit(tn)
# get ground speed
gs = int(float(self.get_prop(tn, "/velocities/groundspeed-kt")))
# skip speed check is speed is <= 30
if gs > 30:
# check if speed is high/low enough for retraction/extention
if flaps_mov == "extend":
if o_flaps['max-spd'] < kias:
self.speak_dialog("spd.high")
self.exit(tn)
else:
self.speak("Speed checked.")
else:
if o_flaps['min-spd'] > kias:
self.speak_dialog("spd.low")
self.exit(tn)
else:
self.speak("Speed checked.")
flaps_request = int(flaps_request)
if flaps_mov == "extend":
pos_reached = 0
while pos_reached == 0:
self.nasal_exec(tn, "controls.flapsDown(1);")
flaps = int(self.get_prop(tn, str(profile['flaps-path'])))
if flaps == flaps_request:
pos_reached = 1
else:
pos_reached = 0
while pos_reached == 0:
self.nasal_exec(tn, "controls.flapsDown(-1);")
flaps = int(self.get_prop(tn, str(profile['flaps-path'])))
if flaps == flaps_request:
pos_reached = 1
self.speak("Flaps " + flaps_name)
tn.close
#########################
# #
# Gear #
# #
#########################
@intent_handler(IntentBuilder('GearUpIntent').require('gearup'))
def handle_gear_up_intent(self, message):
tn = self.connect()
# get acid
acid = self.get_prop(tn, "/sim/aircraft")
# read acid to know which profile to use
profile = None
for i_profiles in self.settings['profiles']:
for i_acid in i_profiles['acid']:
if i_acid == acid:
profile = i_profiles
break
if profile != None:
break
if profile == None:
self.speak_dialog("no.profile")
self.exit(tn)
if profile['gear-retractable'] == "true":
self.speak("Gear up")
self.nasal_exec(tn, "controls.gearDown(-1)")
else:
self.speak_dialog("gear.not.retractable")
@intent_handler(IntentBuilder('GearDownIntent').require('geardown'))
def handle_gear_down_intent(self, message):
tn = self.connect()
# get acid
acid = self.get_prop(tn, "/sim/aircraft")
# read acid to know which profile to use
profile = None
for i_profiles in self.settings['profiles']:
for i_acid in i_profiles['acid']:
if i_acid == acid:
profile = i_profiles
break
if profile != None:
break
if profile == None:
self.speak_dialog("no.profile")
self.exit(tn)
if profile['gear-retractable'] == "true":
self.speak("Gear down")
self.nasal_exec(tn, "controls.gearDown(1)")
else:
self.speak_dialog("gear.not.retractable")
#################################################################
# #
# Checklists #
# #
#################################################################
@intent_handler(IntentBuilder('CheckListIntent').require('check.list'))
def handle_check_list_intent(self, message):
tn = self.connect()
cl_request = message.data['utterance']
checklist_count = self.get_item_count(tn, "/sim/checklists")
if checklist_count == 0:
self.speak("No checklists has been found")
self.exit(tn)
for i in range(0, checklist_count):
checklist = i
checklist_title = self.get_prop(tn, "/sim/checklists/checklist[" + str(checklist) + "]/title")
checklist_title = checklist_title.replace('/', '|')
checklist_title = "^" + checklist_title
match = re.search(checklist_title, cl_request, re.I)
checklist_title = checklist_title[1:]
checklist_title = checklist_title.replace('|', ', ')
if match != None:
self.speak(checklist_title + " checklist")
break
if match == None:
self.speak("Required checklist wasn't found")
self.exit(tn)
# TODO make the programm use all pages, not only the first
page = ""
if self.prop_exist(tn, "page", "/sim/checklists/checklist[" + str(checklist) + "]") == 1:
page = "page/"
item_count = self.get_item_count(tn, "/sim/checklists/checklist[" + str(checklist) + "]/" + page)
if item_count == 0:
self.speak("The required checklist has no entrys")
self.exit(tn)
for i in range(0, item_count):
item = i
item_name = self.get_prop(tn, "/sim/checklists/checklist[" + str(checklist) + "]/" + page + "item[" + str(item) + "]/name")
item_value = self.get_prop(tn, "/sim/checklists/checklist[" + str(checklist) + "]/" + page + "item[" + str(item) + "]/value")
item_name = self.expand_adverbations(item_name)
item_name = item_name.replace('/', ' ') # maybe make the replace '/' -> ' and ' to be spoken out better
self.speak(item_name)
response = self.get_response("dummy")
if response == None:
self.speak("Checklist not completed")
sys.exit(0)
# check if F/O has to confirm as well
item_value_check = re.sub("\(BOTH\)", "", item_value, flags=re.I) # Add more indication of this case, that are used in other A/C
fo_conf = 0
if item_value_check != item_value:
fo_conf = 1
item_value = item_value_check
item_value = self.expand_adverbations(item_value)
item_value = re.sub('/', '|', item_value)
item_value = re.sub('_', '', item_value)
item_value = re.sub('\(.*\)', ' ', item_value)
response = self.expand_adverbations(response)
match = re.search(item_value, response, re.I)
if match == None:
self.speak("Checklist not completed")
sys.exit(0)
if fo_conf == 1:
self.speak(item_value)
self.speak(checklist_title + " checklist completed")
@intent_handler(IntentBuilder('FlightControlCheckIntent').require('flight.control.check'))
def handle_securing_check_intent(self, message):
self.speak("Full up")
sleep(2)
self.speak("Full down")
sleep(2)
self.speak("Neutral")
sleep(2)
self.speak("Full left")
sleep(2)
self.speak("Full right")
sleep(2)
self.speak("Neutral")
sleep(2)
self.speak("Full left")
sleep(4)
self.speak("Full right")
sleep(4)
self.speak("Neutral")
sleep(2)
self.speak("Flight controls checked")
#################################################################
# #
# Configuration #
# #
#################################################################
@intent_handler(IntentBuilder('FlightGearPortIntent').require('conf.flightgear.port'))
def handle_flight_gear_port_intent(self, message):
port = normalize(message.data['utterance'])
port = re.sub('\D', '', port)
if int(port) < 0 or int(port) > 65535:
self.speak("Port '" + str(port) + "' out of range")
sys.exit(0)
self.settings['port'] = int(port)
self.speak("I will use now port " + str(port) + " to connect to flightgear")
@intent_handler(IntentBuilder('AddToProfileIntent').require('conf.add.to.profile'))
def handle_add_to_profile_intent(self, message):
tn = self.connect()
# re to get the profile name
match = re.search("profile .*$", message.data['utterance'], re.I)
if match == None:
self.speak("I didn't understand a profile name")
self.exit(tn)
# remove 'profile '
profile = re.sub("profile ", '', match.group(0))
profile_id = 0
# check if profile exists
for i_profile in self.settings['profiles']:
match = re.search(i_profile['name'], profile, re.I)
if match != None:
break
profile_id = profile_id + 1
# get acid
acid = self.get_prop(tn, "/sim/aircraft")
# check if acid exists in the profile
for i_acid in i_profile['acid']:
if i_acid == acid:
self.speak("The Aircraft ID does already exsists in this profile")
self.exit(tn)
# add acid to profile
self.settings['profiles'][profile_id]['acid'].append(acid)
self.speak("Added the aircraft to the profile")
@intent_handler(IntentBuilder('CreateProfileIntent').require('conf.create.profile'))
def handle_create_profile_intent(self, message):
tn = self.connect()
profile = {}
# re to get the profile name
match = re.search("profile .*$", message.data['utterance'], re.I)
if match == None:
self.speak("I didn't understand a profile name")
self.exit(tn)
# remove 'profile '
profile['name'] = re.sub("profile ", "", match.group(0))
# get acid
profile['acid'] = []
profile['acid'].append(self.get_prop(tn, "/sim/aircraft"))
# ask user if the gear is retractable
self.speak("Has this aircraft a retractable gear?")
wait_while_speaking()
response = self.get_response("dummy")
if response != None:
match = re.search("yes|affirm|ok", response, re.I)
if match != None:
profile['gear_retractable'] = "true"
else:
profile['gear_retractable'] = "false"
# TODO find flaps path
profile['flaps-path'] = "/controls/flight/flap-lever"
profile['flaps'] = []
flaps = {}
flaps['value'] = "Never occuring value"
# make sure the flaps are fully up
for i in range(0, 10):
self.nasal_exec(tn, "controls.flapsDown(-1);")
# scan a maximum of 10 flaps positions... should be enough for every plane
for i in range(0,10):
# if the flaps pos is the same as with the previous step, then break
if self.get_prop(tn, profile['flaps-path']) == flaps['value']:
break
flaps = {}
# ask the user how to name the flaps positions
self.speak("How to name the current flaps position?")
while True:
wait_while_speaking()
response = self.get_response("dummy")
# extracting the flaps setting from the response
match = re.match(r'.*(up|full|down|\d{1,2}).*', response, re.I)
if match != None:
flaps['id'] = match.group(1)
break
self.speak_dialog("no.valid.flaps")
self.speak("Please repeat")
flaps['value'] = self.get_prop(tn, profile['flaps-path'])
flaps['min-spd'] = 0
flaps['max-spd'] = 999
profile['flaps'].append(flaps)
self.nasal_exec(tn, "controls.flapsDown(1);")
# ask if the user wants to add speeds for the flaps settings
self.speak("Do you want to add speeds for the flaps settings?")
wait_while_speaking()
response = self.get_response("dummy")
if response != None:
match = re.search("yes|affirm|ok", response, re.I)
if match != None:
for flaps in profile['flaps']:
self.speak("What's the maximum speed with flaps " + str(flaps['id']) + "?")
while True:
wait_while_speaking()
response = self.get_response("dummy")
if response != None:
match = re.search("\d{1,4}", response)
if match != None:
flaps['max-spd'] = match.group(0)
break
self.speak("I didn't understand a valid speed, please repeat")
self.speak("What's the minimum speed with flaps " + str(flaps['id']) + "?")
while True:
wait_while_speaking()
response = self.get_response("dummy")
if response != None:
match = re.search("\d{1,4}", response)
if match != None:
flaps['min-spd'] = match.group(0)
break
self.speak("I didn't understand a valid speed, please repeat")
self.settings['profiles'].append(profile)
self.speak("Successfully added profile " + profile['name'])
@intent_handler(IntentBuilder('FindFlightGearIntent').require('conf.find.fg'))
def handle_find_flight_gear_intent(self, message):
self.speak("Okay, I'm looking for a running flightgear on port " + str(self.settings['port']) + ". This can take a while.")
# check localhost first
ip = "127.0.0.1"
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
result = sock.connect_ex((ip, self.settings['port']))
if result == 0:
self.speak("Found an instance on my self, do you want to use this computer?")
wait_while_speaking()
response = self.get_response("dummy")
if response != None:
match = re.search("yes|affirm|ok", response, re.I)
if match != None:
self.settings['host'] = "localhost"
self.speak("New host " + self.settings['host'] + " is set")
sys.exit()
self.speak("Okay, I continue to search")
sock.close()
except socket.error:
pass
# get network
net = self.get_ip()
net = re.sub("\d[1-3]$", '', net)
# scan network
for host_part in range(1, 254):
ip = net + str(host_part)
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
result = sock.connect_ex((ip, self.settings['port']))
if result == 0:
# if a connection is found, ask user if it's the correct one
try:
(host, dummy, dummy) = socket.gethostbyaddr(ip)
host = socket.getfqdn(host)
except socket.gaierror:
pass
self.speak("Found an instance on " + host + ", do you want to use this computer?")
wait_while_speaking()
response = self.get_response("dummy")
if response != None:
match = re.search("yes|affirm|ok", response, re.I)
if match != None:
self.settings['host'] = host
self.speak("New host " + self.settings['host'] + " is set")
sys.exit()
self.speak("Okay, I continue to search")
sock.close()
except socket.error:
pass
self.speak("I haven't found any running flightgear on port " + str(self.settings['port']))
# load the default profile config
@intent_handler(IntentBuilder('Load DefaultProfilesIntent').require('load.default.profile'))
def handle_load_default_profile_intent(self, message):
self.speak("This will remove all profiles what you have added. Do you want to continue?")
wait_while_speaking()
response = self.get_response("dummy")
if response != None:
match = re.search("yes|affirm|ok", response, re.I)
if match != None:
self.write_default_profiles()
self.speak("Profiles reseted")
#################################################################
# #
# Help functions #
# #
#################################################################
# connect to fg
def connect(self):
try:
tn = Telnet(self.settings['host'], self.settings['port'])
except:
self.speak_dialog("no.telnet.con")
sys.exit(0)
# switch to data mode
tn.write("data\r\n")
return tn
# running a nasal command
def nasal_exec(self, tn, call):
tn.write("nasal\r\n")
tn.write(call + "\r\n")
tn.write("##EOF##\r\n")
# get a prop from the property tree
def get_prop(self, tn, prop):
tn.write("get " + prop + "\r\n")
ret = tn.read_until("\r")
tn.read_until("\n")
return ret[:-1]
# count number of items in a prop tree dir
def get_item_count(self, tn, directory):
tn.write("ls " + directory + "\r\n")
ret = 0
result = " "
while result != "":
result = tn.read_until("\n", 1)
ret = ret + 1
ret = ret - 1
return ret
# checks if a property exists at a given path
def prop_exist(self, tn, prop, path):
tn.write("ls " + path + "\r\n")
found = 0
result = " "
while result != "":
result = tn.read_until("\n", 1)
match = re.search(prop, result, re.I)
if match != None:
found = 1
break
return found
# get ip address
def get_ip(self):
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
# doesn't even have to be reachable
s.connect(('10.255.255.255', 1))
IP = s.getsockname()[0]
except:
IP = '127.0.0.1'
finally:
s.close()
return IP
# write the defaults to the settings.json file
def write_default_profiles(self):
# DEFINITION of the settings['profiles'] structure
# [
# {
# "name": "<profile-name>",
# "acid":
# [
# "<Aircraft-id> can be found in /???",
# ...
# ],
# flaps:
# [
# {
# "id": "<flaps-name> can be up|down|full|number",
# "min-spd": "<minimum speed for save retraction>",
# "max-spd": "<maximum speed for save extention>",
# "value": "<value in the prop tree>"
# },
# ...
# ]
# "flaps-path": "<path to current flaps-position>"
# "gear-retractable": "<true|false>"
# },
# ...
# ]
profiles = []
#################
# Airbus A320 #
#################
profile = {}
profile['name'] = "Airbus A320"
profile['gear-retractable'] = "true"
profile['flaps-path'] = "/controls/flight/flap-lever"
# ACIDs
profile['acid'] = []
profile['acid'].append("A320-200-CFM")
profile['acid'].append("A320-200-IAE")
profile['acid'].append("A320-100-CFM")
profile['acid'].append("A320neo-CFM")
profile['acid'].append("A320neo-PW")
# Flaps settings
profile['flaps'] = []
# Flaps up
flaps = {}
flaps['id'] = "up"
flaps['min-spd'] = 210
flaps['max-spd'] = 350
flaps['value'] = 0
profile['flaps'].append(flaps)
# Flaps 1
flaps = {}
flaps['id'] = 1
flaps['min-spd'] = 180
flaps['max-spd'] = 230
flaps['value'] = 0
profile['flaps'].append(flaps)
# Flaps 2
flaps = {}
flaps['id'] = 2
flaps['min-spd'] = 165
flaps['max-spd'] = 200
flaps['value'] = 0
profile['flaps'].append(flaps)
# Flaps 3
flaps = {}
flaps['id'] = 3
flaps['min-spd'] = 150
flaps['max-spd'] = 185
flaps['value'] = 0
profile['flaps'].append(flaps)
# Flaps full
flaps = {}
flaps['id'] = "full"
flaps['min-spd'] = 140
flaps['max-spd'] = 177
flaps['value'] = 0
profile['flaps'].append(flaps)
profiles.append(profile)
#################
# c172p #
#################
profile = {}
profile['name'] = "c172p"
profile['gear-retractable'] = "false"
profile['flaps-path'] = "/controls/flight/flaps"
# ACIDs
profile['acid'] = []
profile['acid'].append("c172p")
# Flaps settings
profile['flaps'] = []
# Flaps up
flaps = {}
flaps['id'] = "up"
flaps['min-spd'] = 54
flaps['max-spd'] = 300
flaps['value'] = 0
profile['flaps'].append(flaps)
# Flaps 10
flaps = {}
flaps['id'] = 10
flaps['min-spd'] = 48
flaps['max-spd'] = 110
flaps['value'] = 0.3333334
profile['flaps'].append(flaps)
# Flaps 20
flaps = {}
flaps['id'] = "20"
flaps['min-spd'] = 43
flaps['max-spd'] = 85
flaps['value'] = 0.6666668
profile['flaps'].append(flaps)
# Flaps full
flaps = {}
flaps['id'] = "full"
flaps['min-spd'] = 43
flaps['max-spd'] = 85
flaps['value'] = 1
profile['flaps'].append(flaps)
profiles.append(profile)
self.settings['profiles'] = profiles
pass
# expands adverbations to full words
def expand_adverbations(self, text):
# TODO reduce collisions when used with other a/c
# TODO add more adverbations
text = re.sub("\sALT\s?", "altitude", text, flags=re.I)
text = re.sub("L/G", "landing gear", text, flags=re.I)
text = re.sub("SPLRS", "spoilers", text, flags=re.I)
text = re.sub("PREP", "preperation", text, flags=re.I)
text = re.sub("^TO ", "take off ", text)
text = re.sub(" TO$", " take off", text) # have it twice to prevent collisions with words containing 'to'
text = re.sub("REF", "reference", text, flags=re.I)
text = re.sub("A/SKID", "anti skid", text, flags=re.I)
text = re.sub("N/W", "nose weel", text, flags=re.I)
text = re.sub("A/THR", "auto thrust", text, flags=re.I)
text = re.sub("THR ", "thrust", text, flags=re.I)
text = re.sub("Eng ", "engine ", text, flags=re.I)
text = re.sub("Mstr", "master", text, flags=re.I)
text = re.sub("PB", "push button", text, flags=re.I)
text = re.sub("man", "manual", text, flags=re.I)
text = re.sub("FLT", "flight", text, flags=re.I)
text = re.sub("INST", "instruments", text, flags=re.I)
text = re.sub("TEMP", "temperature", text, flags=re.I)
text = re.sub(" LT", " light", text, flags=re.I)
text = re.sub("SEL", "selector", text, flags=re.I)
text = re.sub("LDG", "landing", text, flags=re.I)
text = re.sub("MDA", "minimum decent altitude", text, flags=re.I)
text = re.sub("DH", "decision height", text, flags=re.I)
text = re.sub("EXT", "external", text, flags=re.I)
text = re.sub("FLX", "flex", text, flags=re.I)
text = re.sub("EMER ", "emergency ", text, flags=re.I)
text = re.sub("BRK", "break", text, flags=re.I)
text = re.sub("KG", "kilogram", text, flags=re.I)
text = re.sub("LB", "pounds", text, flags=re.I)
text = re.sub("LBS", "pounds", text, flags=re.I)
text = re.sub("AS RQRD", "on|set|off|normal", text, flags=re.I) # since we don't know, what is needed in the current situation
text = re.sub("AS REQUIRED", "on|set|off|normal", text, flags=re.I) # since we don't know, what is needed in the current situation
text = re.sub("CONF", "config", text, flags=re.I)
text = re.sub(" OR ", "|", text, flags=re.I)
text = re.sub("0", "zero", text, flags=re.I)
text = re.sub("GEN ", "generator", text, flags=re.I)
text = re.sub("POS ", "position", text, flags=re.I)
text = re.sub(" IAS", "indicated air speed", text, flags=re.I)
text = re.sub("RWY", "runway", text, flags=re.I)
text = re.sub("HDG", "heading", text, flags=re.I)
text = re.sub("HDG", "heading", text, flags=re.I)
text = re.sub("GND ", "ground", text, flags=re.I)
text = re.sub("APPR ", "approach", text, flags=re.I)
text = re.sub("FREQ ", "frequency", text, flags=re.I)
text = re.sub("STD ", "standart", text, flags=re.I)
text = re.sub("AFT ", "after", text, flags=re.I)
text = re.sub("FWD ", "foreward", text, flags=re.I)
return text
# exit routine to properly close the tn con
def exit(self, tn):
tn.close
sys.exit(0)
def stop(self):
pass
def create_skill():
return FlightGearCopilotSkill()