mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-11-03 19:42:22 +00:00
agessaman
7aba432b54
S4. BaseCommand and the 40 bundled overrides that took no argument now take `message: MeshMessage | None = None`, matching joke and rain, which already use the message. None of the 40 bodies read it, so their help text is unchanged. help and command_manager keep their TypeError retry, since local plugins may still implement get_help_text(self). For bundled commands the retry no longer runs: `help <command>` calls get_help_text(message) once instead of failing and calling it again. A new test pins the signature for every bundled command class. The offline smoke run against the real config produces the same help text for all 46 commands.
842 lines
32 KiB
Python
842 lines
32 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Airplanes command for the MeshCore Bot
|
|
Provides aircraft tracking using ADS-B data from adsb.lol or compatible APIs
|
|
"""
|
|
|
|
import asyncio
|
|
import math
|
|
import re
|
|
from typing import Any, Optional
|
|
from urllib.parse import urlparse
|
|
|
|
import requests
|
|
|
|
from ..location import get_bot_lat_lon, latest_contact_position_rows
|
|
from ..models import MeshMessage
|
|
from ..security_utils import sanitize_name
|
|
from ..utils import calculate_distance
|
|
from .base_command import BaseCommand
|
|
|
|
DEFAULT_API_URL = "https://api.adsb.lol/v2/"
|
|
USER_AGENT = "MeshCoreBot (https://github.com/agessaman/meshcore-bot)"
|
|
# Public airplanes.live v2 now returns 403 for unregistered clients.
|
|
_DEPRECATED_PUBLIC_API_HOSTS = frozenset({"api.airplanes.live"})
|
|
|
|
|
|
def normalize_api_url(url: str) -> str:
|
|
"""Strip whitespace and ensure a trailing slash on a non-empty API base URL."""
|
|
url = (url or "").strip()
|
|
if url and not url.endswith("/"):
|
|
url += "/"
|
|
return url
|
|
|
|
|
|
def is_deprecated_public_airplanes_live_url(url: str) -> bool:
|
|
"""Return True for the old public airplanes.live host (not the Pro REST host)."""
|
|
host = (urlparse(url).hostname or "").lower()
|
|
return host in _DEPRECATED_PUBLIC_API_HOSTS
|
|
|
|
|
|
def _response_body_snippet(response: requests.Response, max_length: int = 200) -> str:
|
|
"""Return a log-safe snippet of an HTTP response body."""
|
|
text = response.text or response.reason or ""
|
|
return sanitize_name(text, max_length=max_length) or "(empty body)"
|
|
|
|
|
|
class AirplanesCommand(BaseCommand):
|
|
"""Handles aircraft tracking commands using ADS-B data.
|
|
|
|
Provides aircraft information overhead at companion location, bot location,
|
|
or specified coordinates. Supports filtering and detailed single-aircraft display.
|
|
Uses adsb.lol by default; any readsb/ADSBExchange v2 endpoint can be configured.
|
|
"""
|
|
|
|
# Plugin metadata
|
|
# Read-only informational output; safe for scheduled {cmd:...} rendering.
|
|
render_safe = True
|
|
name = "airplanes"
|
|
honors_skip_channel_check = False
|
|
enabled_attr = "airplanes_enabled"
|
|
keywords = ['airplanes', 'aircraft', 'planes', 'adsb', 'overhead']
|
|
description = "Get aircraft overhead (usage: airplanes [location] [options] or overhead [lat,lon])"
|
|
category = "general"
|
|
cooldown_seconds = 2 # Respect API rate limit of 1 req/sec with buffer
|
|
requires_internet = True
|
|
|
|
# Documentation
|
|
short_description = "Get aircraft overhead using ADS-B data"
|
|
usage = "airplanes [lat,lon|here] [radius=N] [options]"
|
|
examples = ["airplanes", "overhead 47.6,-122.3"]
|
|
parameters = [
|
|
{"name": "location", "description": "Coordinates or here for your companion's location if advertised"},
|
|
{"name": "radius", "description": "Search radius in nautical miles (default: 25)"},
|
|
{"name": "filters", "description": "alt=, type=, military, closest, etc."}
|
|
]
|
|
|
|
# Web-viewer settings schema (see modules/settings_schema.py)
|
|
settings_schema = [
|
|
{"key": "api_url", "label": "API URL", "type": "str",
|
|
"default": DEFAULT_API_URL,
|
|
"help": "ADS-B API endpoint (readsb/ADSBExchange v2 format). Default is adsb.lol."},
|
|
{"key": "default_radius", "label": "Default radius", "type": "float",
|
|
"min": 1, "max": 250, "default": 25, "unit": "nm",
|
|
"help": "Default search radius in nautical miles (max 250)."},
|
|
{"key": "max_results", "label": "Max results", "type": "int",
|
|
"min": 1, "max": 10, "default": 3,
|
|
"help": "Maximum aircraft to return in default list results."},
|
|
{"key": "url_timeout", "label": "Request timeout", "type": "int",
|
|
"min": 1, "default": 10, "unit": "s",
|
|
"help": "API request timeout in seconds."},
|
|
]
|
|
|
|
def __init__(self, bot):
|
|
super().__init__(bot)
|
|
self.airplanes_enabled = self.get_config_value('Airplanes_Command', 'enabled', fallback=True, value_type='bool')
|
|
configured_url = self.get_config_value(
|
|
'Airplanes_Command', 'api_url', fallback=DEFAULT_API_URL, value_type='str'
|
|
)
|
|
self.api_url = self._resolve_api_url(configured_url)
|
|
self.default_radius = self.get_config_value('Airplanes_Command', 'default_radius', fallback=25, value_type='float')
|
|
# Default chosen to fit single-message constraints on the smallest channel budget.
|
|
# Channel payload can be as low as 130 bytes; three compact lines are reliable.
|
|
self.max_results = self.get_config_value('Airplanes_Command', 'max_results', fallback=3, value_type='int')
|
|
self.url_timeout = self.get_config_value('Airplanes_Command', 'url_timeout', fallback=10, value_type='int')
|
|
|
|
def get_help_text(self, message: MeshMessage | None = None) -> str:
|
|
"""Get help text for this command.
|
|
|
|
Returns:
|
|
str: The help text for this command.
|
|
"""
|
|
return self.translate('commands.airplanes.description')
|
|
|
|
def _resolve_api_url(self, configured: Optional[str]) -> str:
|
|
"""Normalize api_url and remap the blocked public airplanes.live host.
|
|
|
|
Custom URLs (local readsb, airplanes.live Pro, other v2 hosts) are left
|
|
unchanged so a private feeder is never redirected to a public API.
|
|
"""
|
|
url = normalize_api_url(configured or "")
|
|
if not url:
|
|
return DEFAULT_API_URL
|
|
if is_deprecated_public_airplanes_live_url(url):
|
|
self.logger.warning(
|
|
"Airplanes_Command api_url %s is no longer publicly accessible; "
|
|
"using %s instead. Point api_url at a local readsb instance or "
|
|
"another compatible ADS-B API to override.",
|
|
url,
|
|
DEFAULT_API_URL,
|
|
)
|
|
return DEFAULT_API_URL
|
|
return url
|
|
|
|
def _calculate_bearing(self, lat1: float, lon1: float, lat2: float, lon2: float) -> float:
|
|
"""Calculate bearing from point 1 to point 2 in degrees.
|
|
|
|
Args:
|
|
lat1: Latitude of point 1 in degrees.
|
|
lon1: Longitude of point 1 in degrees.
|
|
lat2: Latitude of point 2 in degrees.
|
|
lon2: Longitude of point 2 in degrees.
|
|
|
|
Returns:
|
|
float: Bearing in degrees (0-360, where 0 is North).
|
|
"""
|
|
lat1_rad = math.radians(lat1)
|
|
lat2_rad = math.radians(lat2)
|
|
dlon_rad = math.radians(lon2 - lon1)
|
|
|
|
y = math.sin(dlon_rad) * math.cos(lat2_rad)
|
|
x = math.cos(lat1_rad) * math.sin(lat2_rad) - \
|
|
math.sin(lat1_rad) * math.cos(lat2_rad) * math.cos(dlon_rad)
|
|
bearing_rad = math.atan2(y, x)
|
|
bearing_deg = (math.degrees(bearing_rad) + 360) % 360
|
|
|
|
return bearing_deg
|
|
|
|
def _bearing_to_cardinal(self, bearing: float) -> str:
|
|
"""Convert bearing in degrees to cardinal direction.
|
|
|
|
Args:
|
|
bearing: Bearing in degrees (0-360).
|
|
|
|
Returns:
|
|
str: Cardinal direction (N, NE, E, SE, S, SW, W, NW).
|
|
"""
|
|
directions = ['N', 'NE', 'E', 'SE', 'S', 'SW', 'W', 'NW']
|
|
index = round(bearing / 45) % 8
|
|
return directions[index]
|
|
|
|
def _get_companion_location(self, message: MeshMessage) -> Optional[tuple[float, float]]:
|
|
"""Get companion/sender location from database.
|
|
|
|
Args:
|
|
message: The message object.
|
|
|
|
Returns:
|
|
Optional[Tuple[float, float]]: Tuple of (latitude, longitude) or None.
|
|
"""
|
|
try:
|
|
sender_pubkey = message.sender_pubkey
|
|
if not sender_pubkey:
|
|
return None
|
|
|
|
results = latest_contact_position_rows(self.bot, sender_pubkey, zero_rule="either")
|
|
|
|
if results:
|
|
row = results[0]
|
|
return (row['latitude'], row['longitude'])
|
|
return None
|
|
except Exception as e:
|
|
self.logger.debug(f"Error getting companion location: {e}")
|
|
return None
|
|
|
|
def _get_bot_location(self) -> Optional[tuple[float, float]]:
|
|
"""Get bot location from config.
|
|
|
|
Returns:
|
|
Optional[Tuple[float, float]]: Tuple of (latitude, longitude) or None.
|
|
"""
|
|
return get_bot_lat_lon(self.bot, self.logger)
|
|
|
|
def _parse_coordinates(self, args: str) -> Optional[tuple[float, float]]:
|
|
"""Parse latitude and longitude from command arguments.
|
|
|
|
Args:
|
|
args: Command arguments string.
|
|
|
|
Returns:
|
|
Optional[Tuple[float, float]]: Tuple of (latitude, longitude) or None.
|
|
"""
|
|
# Handle formats: "47.6,-122.3", "47.6 -122.3", "47.6, -122.3"
|
|
pattern = r'^\s*(-?\d+\.?\d*)\s*[, ]\s*(-?\d+\.?\d*)\s*$'
|
|
match = re.match(pattern, args)
|
|
|
|
if match:
|
|
try:
|
|
lat = float(match.group(1))
|
|
lon = float(match.group(2))
|
|
|
|
# Validate ranges
|
|
if -90 <= lat <= 90 and -180 <= lon <= 180:
|
|
return (lat, lon)
|
|
except ValueError:
|
|
pass
|
|
|
|
return None
|
|
|
|
def _parse_filters(self, args: list[str]) -> dict[str, Any]:
|
|
"""Parse filter options from command arguments.
|
|
|
|
Args:
|
|
args: List of command argument strings.
|
|
|
|
Returns:
|
|
Dict[str, Any]: Dictionary of filter options.
|
|
"""
|
|
filters = {
|
|
'radius': self.default_radius,
|
|
'alt_min': None,
|
|
'alt_max': None,
|
|
'speed_min': None,
|
|
'speed_max': None,
|
|
'aircraft_type': None,
|
|
'callsign': None,
|
|
'military': False,
|
|
'ladd': False,
|
|
'pia': False,
|
|
'squawk': None,
|
|
'limit': self.max_results, # 0 = no configured cap (still single-message bounded at send time)
|
|
'sort': 'distance' # distance, altitude, speed
|
|
}
|
|
|
|
for arg in args:
|
|
arg_lower = arg.lower()
|
|
|
|
# Radius
|
|
if arg_lower.startswith('radius='):
|
|
try:
|
|
filters['radius'] = float(arg_lower.split('=')[1])
|
|
filters['radius'] = min(250, max(1, filters['radius'])) # Clamp 1-250nm
|
|
except (ValueError, IndexError):
|
|
pass
|
|
|
|
# Altitude range
|
|
elif arg_lower.startswith('alt='):
|
|
try:
|
|
alt_str = arg_lower.split('=')[1]
|
|
if '-' in alt_str:
|
|
parts = alt_str.split('-')
|
|
filters['alt_min'] = float(parts[0])
|
|
filters['alt_max'] = float(parts[1])
|
|
else:
|
|
filters['alt_min'] = float(alt_str)
|
|
except (ValueError, IndexError):
|
|
pass
|
|
|
|
# Speed range
|
|
elif arg_lower.startswith('speed='):
|
|
try:
|
|
speed_str = arg_lower.split('=')[1]
|
|
if '-' in speed_str:
|
|
parts = speed_str.split('-')
|
|
filters['speed_min'] = float(parts[0])
|
|
filters['speed_max'] = float(parts[1])
|
|
else:
|
|
filters['speed_min'] = float(speed_str)
|
|
except (ValueError, IndexError):
|
|
pass
|
|
|
|
# Aircraft type
|
|
elif arg_lower.startswith('type='):
|
|
filters['aircraft_type'] = arg_lower.split('=')[1].upper()
|
|
|
|
# Callsign
|
|
elif arg_lower.startswith('callsign='):
|
|
filters['callsign'] = arg_lower.split('=')[1].upper()
|
|
|
|
# Squawk
|
|
elif arg_lower.startswith('squawk='):
|
|
filters['squawk'] = arg_lower.split('=')[1]
|
|
|
|
# Limit
|
|
elif arg_lower.startswith('limit='):
|
|
try:
|
|
filters['limit'] = int(arg_lower.split('=')[1])
|
|
filters['limit'] = min(50, max(1, filters['limit'])) # Clamp 1-50
|
|
except (ValueError, IndexError):
|
|
pass
|
|
|
|
# Flags
|
|
elif arg_lower == 'military':
|
|
filters['military'] = True
|
|
elif arg_lower == 'ladd':
|
|
filters['ladd'] = True
|
|
elif arg_lower == 'pia':
|
|
filters['pia'] = True
|
|
|
|
# Sort options
|
|
elif arg_lower in ['closest', 'distance']:
|
|
filters['sort'] = 'distance'
|
|
elif arg_lower in ['highest', 'altitude']:
|
|
filters['sort'] = 'altitude'
|
|
elif arg_lower == 'fastest':
|
|
filters['sort'] = 'speed'
|
|
|
|
return filters
|
|
|
|
def _fetch_aircraft_data(self, lat: float, lon: float, radius: float) -> Optional[dict[str, Any]]:
|
|
"""Fetch aircraft data from API.
|
|
|
|
Args:
|
|
lat: Latitude in degrees.
|
|
lon: Longitude in degrees.
|
|
radius: Search radius in nautical miles.
|
|
|
|
Returns:
|
|
Optional[Dict[str, Any]]: API response JSON or None on error.
|
|
"""
|
|
url = f"{self.api_url}point/{lat}/{lon}/{radius}"
|
|
headers = {"User-Agent": USER_AGENT, "Accept": "application/json"}
|
|
self.logger.debug("Fetching aircraft data from %s", url)
|
|
|
|
try:
|
|
response = requests.get(url, headers=headers, timeout=self.url_timeout)
|
|
except requests.exceptions.Timeout:
|
|
self.logger.warning("Aircraft API request timed out: %s", url)
|
|
return None
|
|
except requests.exceptions.RequestException as e:
|
|
self.logger.warning("Aircraft API request failed: %s", e)
|
|
return None
|
|
|
|
if not response.ok:
|
|
self.logger.warning(
|
|
"Aircraft API request failed: HTTP %s from %s: %s",
|
|
response.status_code,
|
|
url,
|
|
_response_body_snippet(response),
|
|
)
|
|
return None
|
|
|
|
try:
|
|
data = response.json()
|
|
except ValueError as e:
|
|
self.logger.warning(
|
|
"Invalid JSON from aircraft API %s: %s (%s)",
|
|
url,
|
|
e,
|
|
_response_body_snippet(response),
|
|
)
|
|
return None
|
|
return data
|
|
|
|
async def _fetch_aircraft_data_async(
|
|
self,
|
|
lat: float,
|
|
lon: float,
|
|
radius: float,
|
|
) -> Optional[dict[str, Any]]:
|
|
"""Fetch aircraft data without blocking the bot's event loop."""
|
|
return await asyncio.to_thread(self._fetch_aircraft_data, lat, lon, radius)
|
|
|
|
def _filter_aircraft(self, aircraft_list: list[dict[str, Any]], filters: dict[str, Any], query_lat: float, query_lon: float) -> list[dict[str, Any]]:
|
|
"""Filter and sort aircraft based on criteria.
|
|
|
|
Args:
|
|
aircraft_list: List of aircraft dictionaries.
|
|
filters: Filter criteria dictionary.
|
|
query_lat: Query latitude for distance calculation.
|
|
query_lon: Query longitude for distance calculation.
|
|
|
|
Returns:
|
|
List[Dict[str, Any]]: Filtered and sorted aircraft list.
|
|
"""
|
|
filtered = []
|
|
|
|
for aircraft in aircraft_list:
|
|
# Skip if no position
|
|
if 'lat' not in aircraft or 'lon' not in aircraft:
|
|
continue
|
|
|
|
# Calculate distance
|
|
distance_km = calculate_distance(query_lat, query_lon, aircraft['lat'], aircraft['lon'])
|
|
distance_nm = distance_km / 1.852 # Convert km to nautical miles
|
|
aircraft['_distance_nm'] = distance_nm
|
|
aircraft['_distance_km'] = distance_km
|
|
|
|
# Calculate bearing
|
|
bearing = self._calculate_bearing(query_lat, query_lon, aircraft['lat'], aircraft['lon'])
|
|
aircraft['_bearing'] = bearing
|
|
aircraft['_bearing_cardinal'] = self._bearing_to_cardinal(bearing)
|
|
|
|
# Apply filters
|
|
# Altitude filtering
|
|
if filters['alt_min'] is not None or filters['alt_max'] is not None:
|
|
alt = aircraft.get('alt_baro') or aircraft.get('alt_geom')
|
|
if alt is None or (isinstance(alt, str) and alt.lower() == 'ground'):
|
|
continue
|
|
if isinstance(alt, str):
|
|
continue
|
|
if filters['alt_min'] is not None and alt < filters['alt_min']:
|
|
continue
|
|
if filters['alt_max'] is not None and alt > filters['alt_max']:
|
|
continue
|
|
|
|
if filters['speed_min'] is not None:
|
|
gs = aircraft.get('gs')
|
|
if gs is None or gs < filters['speed_min']:
|
|
continue
|
|
|
|
if filters['speed_max'] is not None:
|
|
gs = aircraft.get('gs')
|
|
if gs is None or gs > filters['speed_max']:
|
|
continue
|
|
|
|
if filters['aircraft_type']:
|
|
ac_type = aircraft.get('t', '').upper()
|
|
if filters['aircraft_type'] not in ac_type:
|
|
continue
|
|
|
|
if filters['callsign']:
|
|
callsign = aircraft.get('flight', '').upper()
|
|
if filters['callsign'] not in callsign:
|
|
continue
|
|
|
|
if filters['squawk']:
|
|
squawk = aircraft.get('squawk', '')
|
|
if str(filters['squawk']) != str(squawk):
|
|
continue
|
|
|
|
if filters['military']:
|
|
db_flags = aircraft.get('dbFlags', 0)
|
|
if not (db_flags & 1): # Military flag
|
|
continue
|
|
|
|
if filters['ladd']:
|
|
db_flags = aircraft.get('dbFlags', 0)
|
|
if not (db_flags & 8): # LADD flag
|
|
continue
|
|
|
|
if filters['pia']:
|
|
db_flags = aircraft.get('dbFlags', 0)
|
|
if not (db_flags & 4): # PIA flag
|
|
continue
|
|
|
|
# Check radius (already calculated distance)
|
|
if distance_nm > filters['radius']:
|
|
continue
|
|
|
|
filtered.append(aircraft)
|
|
|
|
# Sort
|
|
if filters['sort'] == 'distance':
|
|
filtered.sort(key=lambda x: x.get('_distance_nm', float('inf')))
|
|
elif filters['sort'] == 'altitude':
|
|
filtered.sort(key=lambda x: (x.get('alt_baro') or x.get('alt_geom') or 0), reverse=True)
|
|
elif filters['sort'] == 'speed':
|
|
filtered.sort(key=lambda x: (x.get('gs') or 0), reverse=True)
|
|
|
|
# Apply limit (0 = no limit)
|
|
if filters['limit'] > 0:
|
|
return filtered[:filters['limit']]
|
|
return filtered
|
|
|
|
def _format_single_aircraft(self, aircraft: dict[str, Any], query_lat: float, query_lon: float, max_length: int = 130) -> str:
|
|
"""Format detailed single aircraft response (~130 characters).
|
|
|
|
More user-friendly format: puts distance/bearing first, uses commas in numbers,
|
|
and formats in a more readable way.
|
|
|
|
Args:
|
|
aircraft: Aircraft data dictionary.
|
|
query_lat: Query latitude.
|
|
query_lon: Query longitude.
|
|
max_length: Maximum message length (default 130).
|
|
|
|
Returns:
|
|
str: Formatted aircraft string.
|
|
"""
|
|
# Get basic info
|
|
callsign = aircraft.get('flight', '').strip() or aircraft.get('r', 'N/A')
|
|
ac_type = aircraft.get('t', '')
|
|
operator = aircraft.get('ownOp', '').strip()
|
|
|
|
# Distance and bearing (most important for overhead command - put first)
|
|
distance_nm = aircraft.get('_distance_nm', 0)
|
|
bearing_cardinal = aircraft.get('_bearing_cardinal', 'N')
|
|
|
|
# Altitude with comma formatting
|
|
alt = aircraft.get('alt_baro') or aircraft.get('alt_geom')
|
|
if isinstance(alt, str):
|
|
alt_str = alt
|
|
elif alt is not None:
|
|
alt_str = f"{int(alt):,}ft" # Add comma separator for readability
|
|
else:
|
|
alt_str = "N/A"
|
|
|
|
# Speed
|
|
gs = aircraft.get('gs')
|
|
speed_str = f"{int(gs)}kt" if gs is not None else "N/A"
|
|
|
|
# Track
|
|
track = aircraft.get('track')
|
|
track_str = f"{int(track)}°" if track is not None else "N/A"
|
|
|
|
# Vertical rate
|
|
baro_rate = aircraft.get('baro_rate')
|
|
geom_rate = aircraft.get('geom_rate')
|
|
vs = baro_rate or geom_rate
|
|
vs_str = ""
|
|
if vs is not None:
|
|
vs_str = f"{'+' if vs > 0 else ''}{int(vs)}fpm"
|
|
|
|
# Build user-friendly response: "Callsign (Type) Operator distance bearing: altitude @ speed, heading, vertical_rate"
|
|
# Example: "QXE2307 (E75L) Horizon Air 7.5nm NW: 21,675ft @ 366kt, 354°, +1600fpm"
|
|
response_parts = []
|
|
|
|
# Callsign and type
|
|
if ac_type:
|
|
response_parts.append(f"{callsign} ({ac_type})")
|
|
else:
|
|
response_parts.append(callsign)
|
|
|
|
# Add operator/airline if available
|
|
if operator:
|
|
# Abbreviate long operator names to fit within limits
|
|
# Common abbreviations for major airlines
|
|
operator_abbrev = {
|
|
'ALASKA AIRLINES INC': 'Alaska',
|
|
'ALASKA AIRLINES': 'Alaska',
|
|
'HORIZON AIR': 'Horizon Air',
|
|
'HORIZON AIR INDUSTRIES': 'Horizon Air',
|
|
'DELTA AIR LINES INC': 'Delta',
|
|
'DELTA AIR LINES': 'Delta',
|
|
'AMERICAN AIRLINES INC': 'American',
|
|
'AMERICAN AIRLINES': 'American',
|
|
'UNITED AIR LINES INC': 'United',
|
|
'UNITED AIR LINES': 'United',
|
|
'SOUTHWEST AIRLINES CO': 'Southwest',
|
|
'SOUTHWEST AIRLINES': 'Southwest',
|
|
'JETBLUE AIRWAYS CORP': 'JetBlue',
|
|
'JETBLUE AIRWAYS': 'JetBlue',
|
|
'SPIRIT AIRLINES INC': 'Spirit',
|
|
'SPIRIT AIRLINES': 'Spirit',
|
|
'FRONTIER AIRLINES INC': 'Frontier',
|
|
'FRONTIER AIRLINES': 'Frontier',
|
|
'ALLEGIANT AIR LLC': 'Allegiant',
|
|
'ALLEGIANT AIR': 'Allegiant',
|
|
}
|
|
|
|
# Try to find abbreviation, otherwise use original (truncated if too long)
|
|
operator_display = operator_abbrev.get(operator.upper(), operator)
|
|
|
|
# If still too long, truncate intelligently (keep first part)
|
|
if len(operator_display) > 20:
|
|
# Try to truncate at a word boundary
|
|
words = operator_display.split()
|
|
operator_display = words[0]
|
|
if len(operator_display) > 20:
|
|
operator_display = operator_display[:17] + "..."
|
|
|
|
response_parts.append(operator_display)
|
|
|
|
# Distance and bearing (most important)
|
|
response_parts.append(f"{distance_nm:.1f}nm {bearing_cardinal}")
|
|
|
|
# Main info with colon separator
|
|
main_info = f"{alt_str} @ {speed_str}, {track_str}"
|
|
if vs_str:
|
|
main_info += f", {vs_str}"
|
|
|
|
response = f"{' '.join(response_parts)}: {main_info}"
|
|
|
|
# Truncate if too long (shouldn't happen, but safety check)
|
|
if len(response) > max_length:
|
|
response = response[:max_length-3] + "..."
|
|
|
|
return response
|
|
|
|
def _format_aircraft_list(
|
|
self,
|
|
aircraft_list: list[dict[str, Any]],
|
|
query_lat: float,
|
|
query_lon: float,
|
|
max_length: Optional[int] = None,
|
|
) -> str:
|
|
"""Format compact list for multiple aircraft.
|
|
|
|
Args:
|
|
aircraft_list: List of aircraft dictionaries.
|
|
query_lat: Query latitude.
|
|
query_lon: Query longitude.
|
|
|
|
Returns:
|
|
str: Formatted aircraft list.
|
|
"""
|
|
lines = []
|
|
for aircraft in aircraft_list:
|
|
callsign = aircraft.get('flight', '').strip() or aircraft.get('r', 'N/A')
|
|
|
|
alt = aircraft.get('alt_baro') or aircraft.get('alt_geom')
|
|
if isinstance(alt, str):
|
|
alt_str = alt
|
|
elif alt is not None:
|
|
alt_str = f"{int(alt)}ft"
|
|
else:
|
|
alt_str = "N/A"
|
|
|
|
gs = aircraft.get('gs')
|
|
speed_str = f"{int(gs)}kt" if gs is not None else "N/A"
|
|
|
|
distance_nm = aircraft.get('_distance_nm', 0)
|
|
bearing_cardinal = aircraft.get('_bearing_cardinal', 'N')
|
|
|
|
line = f"{callsign} {alt_str} {speed_str} {distance_nm:.1f}nm {bearing_cardinal}"
|
|
lines.append(line)
|
|
|
|
if max_length is None:
|
|
return '\n'.join(lines)
|
|
|
|
# Pack whole lines into the available single-message budget.
|
|
fitted_lines: list[str] = []
|
|
used = 0
|
|
for line in lines:
|
|
line_len = len(line)
|
|
projected = line_len if not fitted_lines else used + 1 + line_len
|
|
if projected > max_length:
|
|
break
|
|
fitted_lines.append(line)
|
|
used = projected
|
|
|
|
if not fitted_lines:
|
|
# Fall back to a hard-truncated first line for pathological edge cases.
|
|
first = lines[0] if lines else ""
|
|
return first[: max(0, max_length - 3)].rstrip() + "..." if len(first) > max_length else first
|
|
|
|
omitted = len(lines) - len(fitted_lines)
|
|
if omitted > 0:
|
|
suffix = f"\n...+{omitted} more"
|
|
if used + len(suffix) <= max_length:
|
|
return ''.join(['\n'.join(fitted_lines), suffix])
|
|
|
|
return '\n'.join(fitted_lines)
|
|
|
|
async def execute(self, message: MeshMessage) -> bool:
|
|
"""Execute the airplanes command.
|
|
|
|
Args:
|
|
message: The message triggering the command.
|
|
|
|
Returns:
|
|
bool: True if executed successfully, False otherwise.
|
|
"""
|
|
try:
|
|
content = message.content.strip()
|
|
parts = content.split()
|
|
|
|
if len(parts) < 1:
|
|
help_text = self.translate('commands.airplanes.usage')
|
|
await self.send_response(message, help_text)
|
|
return True
|
|
|
|
# Check if this is the "overhead" command
|
|
is_overhead_command = False
|
|
command_word = parts[0].lower()
|
|
|
|
# Check if command is "overhead" or "airplanes overhead" / "aircraft overhead" etc.
|
|
if command_word == 'overhead':
|
|
is_overhead_command = True
|
|
args = parts[1:] if len(parts) > 1 else []
|
|
elif len(parts) > 1 and parts[1].lower() == 'overhead':
|
|
# Handle "airplanes overhead" or "aircraft overhead" etc.
|
|
is_overhead_command = True
|
|
args = parts[2:] if len(parts) > 2 else []
|
|
else:
|
|
args = parts[1:] if len(parts) > 1 else []
|
|
|
|
# Parse location and filters
|
|
location = None
|
|
|
|
# For overhead command, only use companion location or specified coordinates
|
|
if is_overhead_command:
|
|
# Check if coordinates are provided
|
|
if args:
|
|
# Try joining first two args (handles "47.444356, -122.309483")
|
|
coords_str = ' '.join(args[:2])
|
|
coords = self._parse_coordinates(coords_str)
|
|
if coords:
|
|
location = coords
|
|
args = args[2:]
|
|
else:
|
|
# Try parsing just first arg as coordinates (lat,lon format like "47.444356,-122.309483")
|
|
coords = self._parse_coordinates(args[0])
|
|
if coords:
|
|
location = coords
|
|
args = args[1:]
|
|
elif len(args) >= 2:
|
|
# Handle case where first arg has trailing comma: "47.444356," + "-122.309483"
|
|
first_arg = args[0].rstrip(',')
|
|
second_arg = args[1]
|
|
coords_str = f"{first_arg},{second_arg}"
|
|
coords = self._parse_coordinates(coords_str)
|
|
if coords:
|
|
location = coords
|
|
args = args[2:]
|
|
|
|
# If no coordinates, try companion location only
|
|
if location is None:
|
|
location = self._get_companion_location(message)
|
|
if location:
|
|
pass
|
|
|
|
# If still no location, show specific error
|
|
if location is None:
|
|
error_msg = self.translate('commands.airplanes.overhead_no_location')
|
|
await self.send_response(message, error_msg)
|
|
return True
|
|
else:
|
|
# Regular airplanes command - check for "here" keyword
|
|
if args and args[0].lower() == 'here':
|
|
location = self._get_bot_location()
|
|
args = args[1:]
|
|
# Check if first arg is coordinates
|
|
elif args:
|
|
# Try joining first two args (handles "47.444356, -122.309483")
|
|
coords_str = ' '.join(args[:2])
|
|
coords = self._parse_coordinates(coords_str)
|
|
if coords:
|
|
location = coords
|
|
args = args[2:]
|
|
else:
|
|
# Try parsing just first arg as coordinates (lat,lon format like "47.444356,-122.309483")
|
|
coords = self._parse_coordinates(args[0])
|
|
if coords:
|
|
location = coords
|
|
args = args[1:]
|
|
elif len(args) >= 2:
|
|
# Handle case where first arg has trailing comma: "47.444356," + "-122.309483"
|
|
first_arg = args[0].rstrip(',')
|
|
second_arg = args[1]
|
|
coords_str = f"{first_arg},{second_arg}"
|
|
coords = self._parse_coordinates(coords_str)
|
|
if coords:
|
|
location = coords
|
|
args = args[2:]
|
|
|
|
# If no location specified, try companion then bot
|
|
if location is None:
|
|
location = self._get_companion_location(message)
|
|
if location:
|
|
pass
|
|
else:
|
|
location = self._get_bot_location()
|
|
if location:
|
|
pass
|
|
|
|
if location is None:
|
|
error_msg = self.translate('commands.airplanes.no_location')
|
|
await self.send_response(message, error_msg)
|
|
return True
|
|
|
|
# Parse filters
|
|
filters = self._parse_filters(args)
|
|
|
|
# For overhead command, force single closest aircraft
|
|
if is_overhead_command:
|
|
filters['limit'] = 1
|
|
filters['sort'] = 'distance'
|
|
|
|
# Fetch aircraft data
|
|
api_data = await self._fetch_aircraft_data_async(
|
|
location[0], location[1], filters['radius']
|
|
)
|
|
|
|
if api_data is None:
|
|
error_msg = self.translate('commands.airplanes.api_error')
|
|
await self.send_response(message, error_msg)
|
|
return True
|
|
|
|
# Extract aircraft list (API uses 'ac' key, not 'aircraft')
|
|
aircraft_list = api_data.get('ac', api_data.get('aircraft', []))
|
|
|
|
if not aircraft_list:
|
|
error_msg = self.translate('commands.airplanes.no_aircraft', radius=filters['radius'])
|
|
await self.send_response(message, error_msg)
|
|
return True
|
|
|
|
# Filter and sort
|
|
filtered = self._filter_aircraft(aircraft_list, filters, location[0], location[1])
|
|
|
|
if not filtered:
|
|
error_msg = self.translate('commands.airplanes.no_aircraft_filtered', radius=filters['radius'])
|
|
await self.send_response(message, error_msg)
|
|
return True
|
|
|
|
# Get max message length for this message type
|
|
max_length = self.get_max_message_length(message)
|
|
|
|
# Format and send response
|
|
# For overhead command or single result, send as one message
|
|
if is_overhead_command or len(filtered) == 1:
|
|
response = self._format_single_aircraft(filtered[0], location[0], location[1], max_length)
|
|
await self.send_response(message, response)
|
|
else:
|
|
# Keep list output to a single frame to avoid multi-message flood traffic.
|
|
response = self._format_aircraft_list(
|
|
filtered,
|
|
location[0],
|
|
location[1],
|
|
max_length=max_length,
|
|
)
|
|
await self.send_response(message, response)
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error executing airplanes command: {e}")
|
|
error_msg = self.translate('commands.airplanes.error', error=str(e))
|
|
await self.send_response(message, error_msg)
|
|
return False
|
|
|