#!/usr/bin/env python3 """Forward direct observations from an existing local ADS-B decoder to Skylark.""" from __future__ import annotations import argparse import json import math import os from pathlib import Path import random import stat import sys import time import uuid from urllib import request, error, parse VERSION = '0.1.0' MAX_FILE = 4_000_000 FIELDS = ('hex', 'type', 'flight', 'lat', 'lon', 'alt_baro', 'alt_geom', 'gs', 'track', 'baro_rate', 'seen_pos', 'squawk') class FeedError(Exception): pass class NoRedirect(request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): raise FeedError('Upload redirects are not followed; check the configured endpoint.') def finite(value): return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value) def config(path: Path, allow_local: bool = False) -> dict: if os.name != 'nt' and stat.S_IMODE(path.stat().st_mode) & 0o077: raise FeedError('The config contains a credential. Set its permissions to 600 or 400.') data = json.loads(path.read_text()) url = parse.urlsplit(data.get('endpoint', '')) local = allow_local and url.hostname in ('127.0.0.1', 'localhost', '::1') and url.scheme == 'http' if (url.scheme != 'https' and not local) or not url.netloc or url.username or url.password or url.query or url.fragment or url.path != '/api/network/ingest': raise FeedError('Use an HTTPS endpoint ending in /api/network/ingest.') key = data.get('feedKey', '') if not isinstance(key, str) or not key.startswith('sk_feed_') or len(key) != 51: raise FeedError('A valid station feed credential is required.') source = Path(data.get('aircraftFile', '/run/readsb/aircraft.json')) if not source.is_absolute(): raise FeedError('aircraftFile must be an absolute local filename.') interval = data.get('intervalSeconds', 10) if not finite(interval) or not 5 <= interval <= 300: raise FeedError('intervalSeconds must be between 5 and 300.') return {'endpoint': data['endpoint'], 'feedKey': key, 'aircraftFile': source, 'intervalSeconds': interval} def make_batch(document: dict, now: float | None = None) -> dict: now = time.time() if now is None else now captured = document.get('now') if not finite(captured) or captured > now + 5 or now - captured > 30: raise FeedError('Decoder snapshot is stale or its clock is incorrect. No data sent.') aircraft = document.get('aircraft') if not isinstance(aircraft, list): raise FeedError('Decoder file does not contain an aircraft array.') rows = [] for item in aircraft: if not isinstance(item, dict) or item.get('type') != 'adsb_icao' or item.get('mlat') or item.get('tisb'): continue age = item.get('seen_pos') if not finite(age) or age < 0 or now - captured + age > 45: continue if not finite(item.get('lat')) or not finite(item.get('lon')): continue rows.append({key: item[key] for key in FIELDS if key in item}) rows.sort(key=lambda row: row['seen_pos']) return {'schemaVersion': 1, 'batchId': uuid.uuid4().hex, 'capturedAt': round(captured * 1000), 'kind': 'adsb-direct', 'aircraft': rows[:500]} def read_snapshot(path: Path) -> dict: with path.open('rb') as handle: data = handle.read(MAX_FILE + 1) if len(data) > MAX_FILE: raise FeedError('Decoder file exceeds the 4 MB safety limit.') value = json.loads(data) if not isinstance(value, dict): raise FeedError('Expected a decoder JSON object.') return value def upload(settings: dict, payload: bytes) -> dict: req = request.Request(settings['endpoint'], data=payload, method='POST', headers={'Authorization': 'Bearer ' + settings['feedKey'], 'Content-Type': 'application/json', 'User-Agent': 'SkylarkFeeder/' + VERSION}) opener = request.build_opener(NoRedirect()) with opener.open(req, timeout=12) as response: data = response.read(65537) if len(data) > 65536: raise FeedError('Unexpectedly large ingestion response.') result = json.loads(data) if not result.get('ok'): raise FeedError('Ingestion did not acknowledge the batch.') return result def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--config', type=Path, required=True) parser.add_argument('--check', action='store_true', help='Validate the local decoder file without transmitting.') parser.add_argument('--once', action='store_true', help='Send one current snapshot and stop.') parser.add_argument('--allow-local-http', action='store_true', help='Localhost integration testing only.') args = parser.parse_args() try: settings = config(args.config, args.allow_local_http) except (OSError, ValueError, FeedError): print('Configuration invalid. Check private file permissions, endpoint and credential.', file=sys.stderr) return 2 pending = None delay = settings['intervalSeconds'] while True: try: if pending is None or time.time() * 1000 - pending['capturedAt'] > 25000: pending = make_batch(read_snapshot(settings['aircraftFile'])) if args.check: print(json.dumps({'ok': True, 'directPositions': len(pending['aircraft']), 'transmitted': False})) return 0 result = upload(settings, json.dumps(pending, separators=(',', ':'), allow_nan=False).encode()) print(json.dumps({key: result.get(key) for key in ('accepted', 'duplicate', 'rejected', 'publication', 'replayed')}), flush=True) pending = None delay = settings['intervalSeconds'] if args.once: return 0 except error.HTTPError as exc: print('Receiver service returned HTTP ' + str(exc.code) + '. Credentials are not logged.', file=sys.stderr, flush=True) if exc.code in (401, 403) or args.once: return 3 if exc.code in (400, 409, 413, 422): pending = None delay = min(60, max(10, delay * 2)) except (OSError, ValueError, FeedError): print('No data sent or acknowledged. Check the local decoder, clock and connection.', file=sys.stderr, flush=True) if args.once or args.check: return 4 delay = min(60, max(10, delay * 2)) time.sleep(delay + random.uniform(0, 1)) if __name__ == '__main__': try: sys.exit(main()) except KeyboardInterrupt: sys.exit(0)