Create aircraft_module.py
This commit is contained in:
1 parent
1902f2646c
commit
4f9dc8edaa
1 file changed
+194
@@ -0,0 +1,194 @@
|
||||
import os
|
||||
import math
|
||||
import time
|
||||
import requests
|
||||
import webbrowser
|
||||
import re
|
||||
from difflib import SequenceMatcher
|
||||
from dotenv import load_dotenv
|
||||
from voice_player import edge_speak
|
||||
|
||||
load_dotenv()
|
||||
|
||||
MY_LAT = float(os.getenv("LAT", 40.7908711))
|
||||
MY_LON = float(os.getenv("LON", -73.3746079))
|
||||
|
||||
OPENSKY_STATES = "https://opensky-network.org/api/states/all"
|
||||
OPENSKY_FLIGHTS = "https://opensky-network.org/api/flights/aircraft"
|
||||
|
||||
def normalize(text: str):
|
||||
text = text.lower()
|
||||
text = re.sub(r"[^a-z0-9\s]", "", text)
|
||||
text = re.sub(r"\s+", " ", text).strip()
|
||||
return text
|
||||
|
||||
|
||||
def smart_match(user_input: str, command: str):
|
||||
user = normalize(user_input)
|
||||
cmd = normalize(command)
|
||||
|
||||
if cmd in user:
|
||||
return True
|
||||
|
||||
user_words = set(user.split())
|
||||
cmd_words = set(cmd.split())
|
||||
|
||||
if cmd_words and len(user_words & cmd_words) / len(cmd_words) >= 0.6:
|
||||
return True
|
||||
|
||||
return SequenceMatcher(None, user, cmd).ratio() >= 0.6
|
||||
|
||||
|
||||
def match_commands(user_input: str, commands: list):
|
||||
return any(smart_match(user_input, cmd) for cmd in commands)
|
||||
|
||||
def haversine(lat1, lon1, lat2, lon2):
|
||||
R = 6371
|
||||
d_lat = math.radians(lat2 - lat1)
|
||||
d_lon = math.radians(lon2 - lon1)
|
||||
a = (
|
||||
math.sin(d_lat / 2) ** 2
|
||||
+ math.cos(math.radians(lat1))
|
||||
* math.cos(math.radians(lat2))
|
||||
* math.sin(d_lon / 2) ** 2
|
||||
)
|
||||
return 2 * R * math.atan2(math.sqrt(a), math.sqrt(1 - a))
|
||||
|
||||
def get_nearby_aircraft(radius_km=200):
|
||||
lamin, lamax = MY_LAT - 1, MY_LAT + 1
|
||||
lomin, lomax = MY_LON - 1, MY_LON + 1
|
||||
|
||||
try:
|
||||
r = requests.get(
|
||||
f"{OPENSKY_STATES}?lamin={lamin}&lamax={lamax}&lomin={lomin}&lomax={lomax}",
|
||||
timeout=10
|
||||
)
|
||||
data = r.json()
|
||||
except:
|
||||
return []
|
||||
|
||||
aircraft = []
|
||||
|
||||
for s in data.get("states", []):
|
||||
lat, lon = s[6], s[5]
|
||||
if lat is None or lon is None:
|
||||
continue
|
||||
|
||||
dist = haversine(MY_LAT, MY_LON, lat, lon)
|
||||
if dist > radius_km:
|
||||
continue
|
||||
|
||||
velocity = s[9] or 0
|
||||
heading = s[10] or 0
|
||||
|
||||
aircraft.append({
|
||||
"icao24": s[0],
|
||||
"callsign": (s[1] or "Unknown").strip(),
|
||||
"country": s[2] or "Unknown",
|
||||
"lat": lat,
|
||||
"lon": lon,
|
||||
"distance": dist,
|
||||
"velocity": velocity,
|
||||
"heading": heading
|
||||
})
|
||||
|
||||
return aircraft
|
||||
|
||||
|
||||
def get_flight_route(icao24):
|
||||
now = int(time.time())
|
||||
begin = now - 6 * 3600
|
||||
|
||||
try:
|
||||
r = requests.get(
|
||||
f"{OPENSKY_FLIGHTS}?icao24={icao24}&begin={begin}&end={now}",
|
||||
timeout=10
|
||||
)
|
||||
flights = r.json()
|
||||
if flights:
|
||||
f = flights[-1]
|
||||
return f.get("estDepartureAirport"), f.get("estArrivalAirport")
|
||||
except:
|
||||
pass
|
||||
|
||||
return None, None
|
||||
|
||||
def generate_jarvis_aircraft_report(planes):
|
||||
total = len(planes)
|
||||
closest = min(planes, key=lambda x: x["distance"])
|
||||
|
||||
dist = closest["distance"]
|
||||
speed = closest["velocity"] * 3.6
|
||||
heading = closest["heading"]
|
||||
|
||||
if dist < 50:
|
||||
situation = "approaching your position"
|
||||
elif dist < 120:
|
||||
situation = "passing through nearby airspace"
|
||||
else:
|
||||
situation = "operating at a safe distance"
|
||||
|
||||
dep, arr = get_flight_route(closest["icao24"])
|
||||
dep = dep or "unknown origin"
|
||||
arr = arr or "unknown destination"
|
||||
|
||||
return (
|
||||
f"Sir, radar scan complete. "
|
||||
f"I have detected {total} aircraft in your vicinity. "
|
||||
f"The nearest one is {closest['callsign']}, registered in {closest['country']}. "
|
||||
f"It is approximately {dist:.1f} kilometers away, "
|
||||
f"traveling at {speed:.0f} kilometers per hour, "
|
||||
f"heading {heading:.0f} degrees, "
|
||||
f"currently {situation}. "
|
||||
f"Estimated route is from {dep} to {arr}. "
|
||||
f"No immediate threat detected."
|
||||
)
|
||||
|
||||
def speak_detailed_aircraft_report(player):
|
||||
planes = get_nearby_aircraft()
|
||||
|
||||
if not planes:
|
||||
text = "Sir, no aircraft detected in your operational airspace."
|
||||
edge_speak(text, player)
|
||||
player.write_log(text)
|
||||
return
|
||||
|
||||
text = generate_jarvis_aircraft_report(planes)
|
||||
edge_speak(text, player)
|
||||
player.write_log(text)
|
||||
|
||||
def handle_aircraft_command(user_text, player):
|
||||
if match_commands(user_text, [
|
||||
"give me aircraft details",
|
||||
"tell me detailed report",
|
||||
"give me full detailed aircraft report",
|
||||
"give me advanced aircraft report",
|
||||
"tell me aircraft details"
|
||||
]):
|
||||
speak_detailed_aircraft_report(player)
|
||||
return True
|
||||
|
||||
if match_commands(user_text, [
|
||||
"open aircraft",
|
||||
"open flight radar",
|
||||
"show aircraft map"
|
||||
]):
|
||||
webbrowser.open(f"https://www.flightradar24.com/{MY_LAT},{MY_LON}/10")
|
||||
edge_speak("Opening aircraft radar, sir.", player)
|
||||
player.write_log("Opening aircraft radar, sir.")
|
||||
return True
|
||||
|
||||
if match_commands(user_text, [
|
||||
"aircraft summary",
|
||||
"planes nearby",
|
||||
"give me aircraft"
|
||||
]):
|
||||
planes = get_nearby_aircraft()
|
||||
text = f"Sir, I currently detect {len(planes)} aircraft nearby."
|
||||
edge_speak(text, player)
|
||||
player.write_log(text)
|
||||
return True
|
||||
|
||||
|
||||
|
||||
return False
|
||||
Reference in new issue
Block a user