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