adsb-overhead

Original source / Raw SKILL.md

---
name: adsb-overhead
description: Notify when aircraft are overhead within a configurable radius using a local ADS-B SBS/BaseStation feed (readsb port 30003). Use when setting up or troubleshooting plane-overhead alerts, configuring radius/home coordinates/cooldowns, or creating a Clawdbot cron watcher that sends WhatsApp notifications for nearby aircraft.
---

# adsb-overhead

Detect aircraft overhead (within a radius) from a **local readsb SBS/BaseStation TCP feed** and notify via Clawdbot messaging.

This skill is designed for a periodic checker (cron) rather than a long-running daemon.

## Quick start (manual test)

1) Run the checker for a few seconds to see if it detects aircraft near you:

```bash
python3 skills/public/adsb-overhead/scripts/sbs_overhead_check.py \
  --host <SBS_HOST> --port 30003 \
  --home-lat <LAT> --home-lon <LON> \
  --radius-km 2 \
  --listen-seconds 5 \
  --cooldown-min 15
```

- If it prints lines, those are *new* alerts (not in cooldown).
- If it prints nothing, there were no new overhead aircraft during the sample window.

## How it works

- Connect to the SBS feed (TCP) for `--listen-seconds`.
- Track latest lat/lon per ICAO hex.
- Compute distance to `--home-lat/--home-lon` (Haversine).
- Emit alerts for aircraft within `--radius-km` **only if** not alerted within `--cooldown-min`.
- Persist state to a JSON file (default: `~/.clawdbot/adsb-overhead/state.json`).

SBS parsing assumptions are documented in: `references/sbs-fields.md`.

## Create a Clawdbot watcher (cron)

Use a Clawdbot cron job to run periodically. The cron job should:
1) `exec` the script
2) If stdout is non-empty, `message.send` it via WhatsApp

Pseudocode for the agent:

- Run:
  - `python3 .../sbs_overhead_check.py ...`
- If stdout trimmed is not empty:
  - send a WhatsApp message with that text

Suggested polling intervals:
- 30–60 seconds is usually enough (given cooldowns)
- Use `--listen-seconds 3..8` so each run can gather a few position frames

## Tuning knobs

- Increase `--radius-km` if you want fewer misses.
- Increase `--listen-seconds` if your feed is busy but you’re missing position updates.
- Use `--cooldown-min` to prevent spam (15–60 minutes recommended).


---

## Referenced Files

> The following files are referenced in this skill and included for context.

### references/sbs-fields.md

```markdown
# SBS/BaseStation MSG fields (quick ref)

This skill assumes the common SBS/BaseStation CSV format emitted on TCP port **30003**.

We only parse `MSG` lines.

Typical indices (0-based) for `MSG`:
- `0` message type: `MSG`
- `1` transmission type: `1..8`
- `4` ICAO hex (e.g. `406BCA`)
- `10` callsign
- `11` altitude (ft)
- `12` ground speed (kt)
- `13` track (deg)
- `14` latitude
- `15` longitude

Notes:
- Not every message carries lat/lon; position often appears intermittently.
- Callsign may be blank.

```



---

## Skill Companion Files

> Additional files collected from the skill directory layout.

### README.md

```markdown
# adsb-overhead (Moltbot/Clawdbot skill)

Get WhatsApp alerts when aircraft are within a configurable radius of your home using a **local ADS-B SBS/BaseStation feed** (e.g. `readsb` on port **30003**) and optional enrichment from **tar1090**.

This project is designed to be **production-friendly**:
- **Zero-AI runtime** (no model calls for the watcher)
- Rate-limited per aircraft (cooldowns) with persistent state
- Optional photo attachment via Planespotters (public API)

## Example alert

![Example WhatsApp alert](assets/example-alert.jpg)

## What you need

- An ADS-B receiver feeding **SBS/BaseStation** on TCP (commonly `:30003`)
- (Optional) `tar1090`/`readsb` HTTP endpoint for enrichment, e.g.:
  - `http://TAR1090_HOST/tar1090/data/aircraft.json`
- Clawdbot (aka Moltbot) running with WhatsApp configured

## Data flow (how it works)

1) **SBS/BaseStation feed (TCP 30003)**
- Source of truth for live aircraft positions.
- `sbs_overhead_check.py` connects for a short window, tracks latest position per ICAO hex, computes distance to your home coordinates, and applies per-aircraft cooldowns.

2) **tar1090 aircraft.json (optional enrichment)**
- Used only to enrich aircraft metadata by ICAO hex (e.g. callsign/flight, emitter category, sometimes registration/type depending on your tar1090/readsb DB).
- Does *not* drive the distance check.

3) **Planespotters photo API (optional)**
- If enabled, looks up a thumbnail by ICAO hex and can download it locally for attaching to WhatsApp.

4) **Zero-AI notifier**
- `adsb_overhead_notify.py` runs the checker in JSONL mode and sends one WhatsApp message per aircraft using `clawdbot message send`.

## Files

- `scripts/sbs_overhead_check.py`
  - Connects to the SBS feed for a short window and detects aircraft within radius.
  - Can output `text` or `jsonl`.
  - Can look up aircraft photos from Planespotters (hex endpoint) and optionally download thumbnails.

- `scripts/adsb_overhead_notify.py`
  - **Zero-AI notifier**.
  - Reads a config file, runs the checker, and sends **one WhatsApp per aircraft** using `clawdbot message send`.

- `scripts/adsb_config.py`
  - Helper for safely editing the config file.

## Quick start (manual test)

Run the checker directly:

```bash
python3 skills/public/adsb-overhead/scripts/sbs_overhead_check.py \
  --host SBS_HOST --port 30003 \
  --home-lat LAT --home-lon LON \
  --radius-km 2 \
  --listen-seconds 6 \
  --cooldown-min 15 \
  --aircraft-json-url http://TAR1090_HOST/tar1090/data/aircraft.json \
  --photo --photo-mode download --photo-size large \
  --output jsonl
```

## WhatsApp controls (how configuration changes work)

This repo **does not** include a dedicated “WhatsApp command parser” by itself.

In the original Moltbot/Clawdbot setup, configuration changes were made by:
- editing `~/.clawdbot/adsb-overhead/config.json`, or
- running the helper script `scripts/adsb_config.py`, and
- letting the watcher pick up the new settings on its next cron run.

If you want to control it via WhatsApp (e.g. “radius 5km”, “adsb off”), implement that in **your bot’s chat handler** and have it call `adsb_config.py`.

### Supported config operations via `adsb_config.py`

Examples:

```bash
# Show current settings
python3 scripts/adsb_config.py --config ~/.clawdbot/adsb-overhead/config.json status

# Enable/disable
python3 scripts/adsb_config.py --config ~/.clawdbot/adsb-overhead/config.json enable --on
python3 scripts/adsb_config.py --config ~/.clawdbot/adsb-overhead/config.json enable --off

# Change radius (km)
python3 scripts/adsb_config.py --config ~/.clawdbot/adsb-overhead/config.json set-radius 5

# Change home coordinates
python3 scripts/adsb_config.py --config ~/.clawdbot/adsb-overhead/config.json set-home 51.5007 -0.1246

# Quiet hours
python3 scripts/adsb_config.py --config ~/.clawdbot/adsb-overhead/config.json set-quiet 23:00 07:00
python3 scripts/adsb_config.py --config ~/.clawdbot/adsb-overhead/config.json set-quiet --off
```

## Install & run as a watcher (system cron)

1) Copy the example config and edit it:

```bash
mkdir -p ~/.clawdbot/adsb-overhead
cp skills/public/adsb-overhead/config.example.json ~/.clawdbot/adsb-overhead/config.json
nano ~/.clawdbot/adsb-overhead/config.json
chmod 600 ~/.clawdbot/adsb-overhead/config.json
```

2) Add a user crontab entry (runs every minute):

```bash
* * * * * /usr/bin/python3 /path/to/adsb_overhead_notify.py \
  --config ~/.clawdbot/adsb-overhead/config.json \
  >> ~/.clawdbot/adsb-overhead/notifier.log 2>&1
```

## Notes on Flightradar24 links

The alert text can include a Flightradar24 link using callsign (`https://www.flightradar24.com/CALLSIGN`). This is best-effort. A reliable fallback tracking link is ADSBexchange by hex.

## License

MIT

```

### _meta.json

```json
{
  "owner": "davestarling",
  "slug": "moltbot-adsb-overhead",
  "displayName": "ADS-B Overhead",
  "latest": {
    "version": "1.0.0",
    "publishedAt": 1769960653130,
    "commit": "https://github.com/clawdbot/skills/commit/78960c6c11a3138d597ed0e00a716ad5667b3058"
  },
  "history": [
    {
      "version": "1.0.0",
      "publishedAt": 1769766377663,
      "commit": "https://github.com/clawdbot/skills/commit/341285a9a1d329a117a9d5df05939de270f2ca67"
    }
  ]
}

```

### scripts/adsb_config.py

```python
#!/usr/bin/env python3
"""Manage adsb-overhead config.json (safe edits).

This exists so the assistant can update settings without hand-editing JSON.
"""

from __future__ import annotations

import argparse
import json
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict

DEFAULT_CONFIG = str(Path.home() / ".clawdbot" / "adsb-overhead" / "config.json")


def load(path: str) -> Dict[str, Any]:
    p = Path(path).expanduser()
    return json.loads(p.read_text(encoding="utf-8"))


def save(path: str, cfg: Dict[str, Any]) -> None:
    p = Path(path).expanduser()
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_text(json.dumps(cfg, indent=2, sort_keys=True), encoding="utf-8")


def set_path(cfg: Dict[str, Any], dotted: str, value: Any) -> None:
    parts = dotted.split(".")
    cur: Dict[str, Any] = cfg
    for k in parts[:-1]:
        nxt = cur.get(k)
        if not isinstance(nxt, dict):
            nxt = {}
            cur[k] = nxt
        cur = nxt
    cur[parts[-1]] = value


def main() -> int:
    ap = argparse.ArgumentParser(description="Update adsb-overhead config")
    ap.add_argument("--config", default=DEFAULT_CONFIG)

    sub = ap.add_subparsers(dest="cmd", required=True)

    sub.add_parser("status")

    p = sub.add_parser("enable")
    p.add_argument("--on", action="store_true")
    p.add_argument("--off", action="store_true")

    p = sub.add_parser("set-radius")
    p.add_argument("km", type=float)

    p = sub.add_parser("set-home")
    p.add_argument("lat", type=float)
    p.add_argument("lon", type=float)

    p = sub.add_parser("set-quiet")
    p.add_argument("--off", action="store_true", help="Disable quiet hours")
    p.add_argument("start", nargs="?", help="HH:MM")
    p.add_argument("end", nargs="?", help="HH:MM")

    args = ap.parse_args()
    cfg = load(args.config)

    if args.cmd == "status":
        out = {
            "enabled": cfg.get("enabled", True),
            "tz": cfg.get("tz"),
            "quietHours": cfg.get("quietHours"),
            "home": cfg.get("home"),
            "radiusKm": cfg.get("radiusKm"),
            "notify": cfg.get("notify"),
        }
        print(json.dumps(out, indent=2, sort_keys=True))
        return 0

    if args.cmd == "enable":
        if args.on == args.off:
            raise SystemExit("Specify exactly one of --on or --off")
        set_path(cfg, "enabled", bool(args.on))

    if args.cmd == "set-radius":
        if args.km <= 0:
            raise SystemExit("radius must be > 0")
        set_path(cfg, "radiusKm", float(args.km))

    if args.cmd == "set-home":
        set_path(cfg, "home.lat", float(args.lat))
        set_path(cfg, "home.lon", float(args.lon))

    if args.cmd == "set-quiet":
        if args.off:
            set_path(cfg, "quietHours.enabled", False)
        else:
            if not args.start or not args.end:
                raise SystemExit("Provide start and end (HH:MM HH:MM) or use --off")
            set_path(cfg, "quietHours.enabled", True)
            set_path(cfg, "quietHours.start", args.start)
            set_path(cfg, "quietHours.end", args.end)

    save(args.config, cfg)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

```

### scripts/adsb_overhead_notify.py

```python
#!/usr/bin/env python3
"""Zero-AI ADS-B overhead notifier.

Runs the sbs_overhead_check.py checker, then sends one WhatsApp message per alert
via the Clawdbot CLI (no model invocation).

- Runtime config is loaded from JSON (default: ~/.clawdbot/adsb-overhead/config.json)
- A simple quiet-hours gate can suppress notifications overnight.

Intended to be called by *system cron* or systemd timers.
"""

from __future__ import annotations

import argparse
import os
import json
import shutil
import subprocess
import sys
from datetime import datetime
from pathlib import Path
from typing import List, Optional

from zoneinfo import ZoneInfo


DEFAULT_CONFIG = str(Path.home() / ".clawdbot" / "adsb-overhead" / "config.json")


def acquire_lock(lock_path: str) -> Optional[int]:
    """Return an open fd if lock acquired, else None."""
    try:
        import fcntl

        p = Path(lock_path).expanduser()
        p.parent.mkdir(parents=True, exist_ok=True)
        fd = os.open(str(p), os.O_RDWR | os.O_CREAT, 0o600)
        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        return fd
    except Exception:
        return None


def _expand(path: str) -> str:
    return str(Path(path).expanduser())


def load_config(path: str) -> dict:
    p = Path(_expand(path))
    return json.loads(p.read_text(encoding="utf-8"))


def in_quiet_hours(cfg: dict, now: Optional[datetime] = None) -> bool:
    q = (cfg.get("quietHours") or {})
    if not q.get("enabled"):
        return False

    tz = cfg.get("tz") or "Europe/London"
    z = ZoneInfo(tz)
    now = now or datetime.now(tz=z)

    def parse_hhmm(s: str) -> tuple[int, int]:
        hh, mm = s.strip().split(":", 1)
        return int(hh), int(mm)

    start_s = q.get("start") or "23:00"
    end_s = q.get("end") or "07:00"
    sh, sm = parse_hhmm(start_s)
    eh, em = parse_hhmm(end_s)

    start = now.replace(hour=sh, minute=sm, second=0, microsecond=0)
    end = now.replace(hour=eh, minute=em, second=0, microsecond=0)

    if start <= end:
        return start <= now < end
    # crosses midnight
    return now >= start or now < end


def quiet_mode(cfg: dict, now: Optional[datetime] = None) -> str:
    """Return 'off' | 'suppress' | 'priority'."""
    q = (cfg.get("quietHours") or {})
    if not q.get("enabled"):
        return 'off'
    if not in_quiet_hours(cfg, now=now):
        return 'off'
    mode = (q.get('mode') or 'suppress').lower()
    if mode in ('priority', 'priority-only', 'military-only'):
        return 'priority'
    return 'suppress'


def run_checker(cfg: dict) -> List[dict]:
    sbs = cfg.get("sbs") or {}
    home = cfg.get("home") or {}
    photo_cfg = cfg.get("photo") or {}

    cmd = [
        sys.executable,
        str(Path(__file__).with_name("sbs_overhead_check.py")),
        "--host",
        str(sbs.get("host")),
        "--port",
        str(int(sbs.get("port", 30003))),
        "--home-lat",
        str(home.get("lat")),
        "--home-lon",
        str(home.get("lon")),
        "--radius-km",
        str(cfg.get("radiusKm", 2)),
        "--listen-seconds",
        str(cfg.get("listenSeconds", 6)),
        "--cooldown-min",
        str(cfg.get("cooldownMin", 15)),
        "--state-file",
        _expand(cfg.get("stateFile") or str(Path.home() / ".clawdbot" / "adsb-overhead" / "state.json")),
        "--aircraft-json-url",
        str(cfg.get("aircraftJsonUrl")),
        "--output",
        "jsonl",
    ]

    if photo_cfg.get("enabled"):
        cmd += [
            "--photo",
            "--photo-mode",
            "download",
            "--photo-size",
            str(photo_cfg.get("size", "large")),
            "--photo-cache-hours",
            str(photo_cfg.get("cacheHours", 24)),
            "--photo-dir",
            _expand(photo_cfg.get("dir") or str(Path.home() / ".clawdbot" / "adsb-overhead" / "photos")),
        ]

    p = subprocess.run(cmd, capture_output=True, text=True)
    if p.returncode != 0:
        raise RuntimeError(f"checker failed ({p.returncode}): {p.stderr.strip()}")

    out = p.stdout.strip()
    if not out:
        return []

    alerts: List[dict] = []
    for line in out.splitlines():
        line = line.strip()
        if not line:
            continue
        alerts.append(json.loads(line))
    return alerts


def find_clawdbot() -> str:
    """Locate the Clawdbot CLI.

    Cron often runs with a minimal PATH, so we check a few common locations.
    We avoid hard-coding any specific user's home directory.
    """

    candidates = [
        str(Path.home() / ".npm-global" / "bin" / "clawdbot"),
        "/usr/local/bin/clawdbot",
        "/usr/bin/clawdbot",
    ]
    for p in candidates:
        if Path(p).exists():
            return p

    w = shutil.which("clawdbot")
    if w:
        return w

    raise FileNotFoundError("clawdbot executable not found; check PATH for cron")


def send_message(channel: str, target: str, caption: str, media: Optional[str]) -> None:
    claw = find_clawdbot()
    cmd = [claw, "message", "send", "--channel", channel, "--target", target]

    if media:
        cmd += ["--media", media]
        if caption:
            cmd += ["--message", caption]
    else:
        cmd += ["--message", caption]

    p = subprocess.run(cmd, capture_output=True, text=True)
    if p.returncode != 0:
        raise RuntimeError(f"send failed ({p.returncode}): {p.stderr.strip() or p.stdout.strip()}")


def main() -> int:
    ap = argparse.ArgumentParser(description="Zero-AI overhead notifier (config-driven)")
    ap.add_argument("--config", default=DEFAULT_CONFIG, help=f"Path to config JSON (default: {DEFAULT_CONFIG})")
    args = ap.parse_args()

    cfg = load_config(args.config)

    lock_fd = acquire_lock((cfg.get('lockFile') or str(Path.home()/'.clawdbot'/'adsb-overhead'/'notifier.lock')))
    if lock_fd is None:
        # Another run is in progress; avoid overlap.
        return 0

    if not cfg.get("enabled", True):
        return 0

    qm = quiet_mode(cfg)
    if qm == 'suppress':
        return 0

    notify = cfg.get("notify") or {}
    channel = notify.get("channel") or "whatsapp"
    target = notify.get("target")
    if not target:
        raise RuntimeError("config.notify.target missing")

    alerts = run_checker(cfg)

    # Quiet-hours priority mode: only allow military when operation is known.
    if qm == 'priority':
        filtered = []
        for a in alerts:
            cap = (a.get('caption') or '')
            # crude: look for "Op: military" line
            if 'Op: military' in cap:
                filtered.append(a)
        alerts = filtered

    for a in alerts:
        caption = a.get("caption") or ""
        media = a.get("photoFile") or None
        if media:
            p = Path(str(media))
            if not p.exists():
                media = None

        try:
            send_message(channel=channel, target=target, caption=caption, media=media)
        except Exception as e:
            # Don't crash the whole run; just log to stderr so cron captures it.
            print(f"[adsb-overhead] send failed: {e}", file=sys.stderr)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

```

### scripts/sbs_overhead_check.py

```python
#!/usr/bin/env python3
"""Connect to an SBS/BaseStation TCP feed (port 30003) for a short window,
collect latest positions, and print *new* "overhead" aircraft within a radius.

Designed to be run periodically (cron) rather than as a long-lived daemon.

Output:
  - text (default): human readable alerts (optionally includes photo link)
  - jsonl: one JSON object per alert (optionally includes photoUrl / photoFile)

State:
  A small JSON file tracks last-alert timestamps per ICAO hex to avoid spam.
  Photo lookups are cached (by ICAO hex) when enabled.
"""

from __future__ import annotations

import argparse
import json
import math
import os
import socket
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, Optional, Tuple
from urllib.request import Request, urlopen


@dataclass
class Aircraft:
    hex: str
    callsign: Optional[str] = None
    lat: Optional[float] = None
    lon: Optional[float] = None
    alt_ft: Optional[int] = None
    gs_kt: Optional[float] = None
    track_deg: Optional[float] = None
    last_seen_ts: float = 0.0

    # Optional enrichment (e.g., from readsb HTTP JSON or external DB)
    reg: Optional[str] = None
    ac_type: Optional[str] = None  # e.g., A320, B738
    operation: Optional[str] = None  # military|commercial|private|unknown
    emitter_category: Optional[str] = None  # ADS-B emitter category code e.g. A3
    photo_url: Optional[str] = None
    photo_file: Optional[str] = None


def haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    r = 6371.0088
    phi1 = math.radians(lat1)
    phi2 = math.radians(lat2)
    dphi = math.radians(lat2 - lat1)
    dlambda = math.radians(lon2 - lon1)
    a = math.sin(dphi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlambda / 2) ** 2
    c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
    return r * c


def load_state(path: Path) -> dict:
    if not path.exists():
        return {"lastAlert": {}}
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return {"lastAlert": {}}


def save_state(path: Path, state: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(state, indent=2, sort_keys=True), encoding="utf-8")


def parse_sbs_line(line: str) -> Optional[Tuple[str, Dict[str, str]]]:
    parts = [p.strip() for p in line.split(",")]
    if not parts or parts[0] != "MSG":
        return None
    if len(parts) < 16:
        return None

    hex_ = parts[4].upper() if parts[4] else ""
    if not hex_:
        return None

    d: Dict[str, str] = {
        "tx_type": parts[1],
        "callsign": parts[10],
        "altitude": parts[11],
        "gs": parts[12],
        "track": parts[13],
        "lat": parts[14],
        "lon": parts[15],
    }
    return hex_, d


def coerce_float(s: str) -> Optional[float]:
    try:
        return float(s) if s != "" else None
    except Exception:
        return None


def coerce_int(s: str) -> Optional[int]:
    try:
        return int(float(s)) if s != "" else None
    except Exception:
        return None


def fetch_json(url: str, timeout_s: float = 2.5) -> dict:
    req = Request(url, headers={"User-Agent": "clawdbot-adsb-overhead/1.0"})
    with urlopen(req, timeout=timeout_s) as resp:
        return json.loads(resp.read().decode("utf-8", errors="ignore"))


def fetch_readsb_aircraft_index(url: str, timeout_s: float = 2.5) -> Dict[str, dict]:
    """Fetch readsb/tar1090 aircraft.json and index by hex."""
    try:
        data = fetch_json(url, timeout_s=timeout_s)
        ac = data.get("aircraft") or []
        out: Dict[str, dict] = {}
        for item in ac:
            h = (item.get("hex") or item.get("icao") or "").upper()
            if h:
                out[h] = item
        return out
    except Exception:
        return {}


def _planespotters_pick_photo_url_sized(data: dict, size: str) -> Optional[str]:
    photos = data.get("photos") or []
    if not photos:
        return None
    p0 = photos[0]
    if size == "small":
        return (p0.get("thumbnail") or {}).get("src") or (p0.get("thumbnail_large") or {}).get("src")
    return (p0.get("thumbnail_large") or {}).get("src") or (p0.get("thumbnail") or {}).get("src")


def fetch_planespotters_photo_by_hex(hex_: str, size: str = "large", timeout_s: float = 3.5) -> Optional[str]:
    h = hex_.strip().upper()
    if not h:
        return None
    url = f"https://api.planespotters.net/pub/photos/hex/{h}"
    try:
        data = fetch_json(url, timeout_s=timeout_s)
        return _planespotters_pick_photo_url_sized(data, size=size)
    except Exception:
        return None


def download_image(url: str, dest: Path, timeout_s: float = 5.0, max_bytes: int = 3_000_000) -> Optional[Path]:
    try:
        # Basic allowlist: only fetch HTTPS thumbnails from Planespotters CDN.
        if not url.startswith("https://"):
            return None
        if "plnspttrs.net" not in url:
            return None

        req = Request(url, headers={"User-Agent": "clawdbot-adsb-overhead/1.0"})
        with urlopen(req, timeout=timeout_s) as resp:
            ctype = (resp.headers.get("Content-Type") or "").lower()
            if "image" not in ctype:
                return None

            # Respect Content-Length if present.
            clen = resp.headers.get("Content-Length")
            if clen:
                try:
                    if int(clen) > max_bytes:
                        return None
                except Exception:
                    pass

            data = resp.read(max_bytes + 1)
            if len(data) > max_bytes:
                return None

        dest.parent.mkdir(parents=True, exist_ok=True)
        dest.write_bytes(data)
        return dest
    except Exception:
        return None


def guess_operation(enriched: dict, callsign: Optional[str], reg: Optional[str]) -> str:
    # Some forks include boolean-ish flags.
    for k in ("mil", "military"):
        v = enriched.get(k)
        if isinstance(v, bool) and v:
            return "military"
        if isinstance(v, str) and v.lower() in ("1", "true", "yes"):
            return "military"

    cs = (callsign or "").strip().upper()

    if cs:
        military_prefixes = ("RFR", "RRR", "RAF", "NAVY", "EAGLE", "TYPHO", "ASCOT")
        if cs.startswith(military_prefixes):
            return "military"

    import re

    if cs and re.match(r"^[A-Z]{3}\d", cs):
        return "commercial"

    r = (reg or "").strip().upper()
    if r and r != "?":
        return "private"

    return "unknown"


def main() -> int:
    ap = argparse.ArgumentParser(description="Detect overhead aircraft from an SBS/BaseStation feed")
    ap.add_argument("--host", required=True, help="SBS host (e.g. 192.168.1.10)")
    ap.add_argument("--port", type=int, default=30003, help="SBS port (default: 30003)")
    ap.add_argument("--home-lat", type=float, required=True)
    ap.add_argument("--home-lon", type=float, required=True)
    ap.add_argument("--radius-km", type=float, default=2.0)
    ap.add_argument("--listen-seconds", type=float, default=5.0, help="How long to listen for SBS messages")
    ap.add_argument("--cooldown-min", type=float, default=15.0, help="Per-aircraft alert cooldown")
    ap.add_argument("--state-file", default=str(Path.home() / ".clawdbot" / "adsb-overhead" / "state.json"))
    ap.add_argument("--max-aircraft", type=int, default=5000)

    ap.add_argument(
        "--aircraft-json-url",
        default="",
        help="Optional URL to readsb/tar1090 aircraft.json (e.g. http://192.168.1.10/tar1090/data/aircraft.json)",
    )

    ap.add_argument("--photo", action="store_true", help="Enable Planespotters photo lookup")
    ap.add_argument(
        "--photo-mode",
        choices=["link", "download"],
        default="link",
        help="If --photo: include link (text/jsonl) or download (jsonl emits photoFile)",
    )
    ap.add_argument(
        "--photo-size",
        choices=["large", "small"],
        default="large",
        help="Planespotters photo size to use (default: large thumbnail)",
    )
    ap.add_argument(
        "--photo-cache-hours",
        type=float,
        default=24.0,
        help="Cache Planespotters photo lookups per ICAO hex for N hours (default: 24)",
    )
    ap.add_argument(
        "--photo-dir",
        default=str(Path.home() / ".clawdbot" / "adsb-overhead" / "photos"),
        help="Directory for downloaded photos when --photo-mode=download",
    )

    ap.add_argument(
        "--output",
        choices=["text", "jsonl"],
        default="text",
        help="Output format: text (default) or jsonl (one JSON object per alert)",
    )

    args = ap.parse_args()

    state_path = Path(args.state_file).expanduser()
    state = load_state(state_path)
    last_alert: dict = state.setdefault("lastAlert", {})
    photo_cache: dict = state.setdefault("photoCache", {})  # key: hex -> {url, ts}

    enrich_index: Dict[str, dict] = {}
    if args.aircraft_json_url:
        enrich_index = fetch_readsb_aircraft_index(args.aircraft_json_url)

    aircraft: Dict[str, Aircraft] = {}
    end = time.time() + args.listen_seconds

    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(2.0)
    try:
        s.connect((args.host, args.port))
        s.settimeout(1.0)
        buf = b""
        while time.time() < end:
            try:
                chunk = s.recv(4096)
            except socket.timeout:
                continue
            if not chunk:
                break
            buf += chunk
            while b"\n" in buf:
                raw, buf = buf.split(b"\n", 1)
                line = raw.decode("utf-8", errors="ignore").strip()
                parsed = parse_sbs_line(line)
                if not parsed:
                    continue
                hex_, d = parsed

                a = aircraft.get(hex_)
                if not a:
                    a = Aircraft(hex=hex_)
                    aircraft[hex_] = a

                if d.get("callsign"):
                    cs = d["callsign"].strip()
                    if cs:
                        a.callsign = cs

                alt = coerce_int(d.get("altitude", ""))
                if alt is not None:
                    a.alt_ft = alt

                gs = coerce_float(d.get("gs", ""))
                if gs is not None:
                    a.gs_kt = gs

                trk = coerce_float(d.get("track", ""))
                if trk is not None:
                    a.track_deg = trk

                lat = coerce_float(d.get("lat", ""))
                lon = coerce_float(d.get("lon", ""))
                if lat is not None and lon is not None:
                    a.lat = lat
                    a.lon = lon

                a.last_seen_ts = time.time()

                if len(aircraft) > args.max_aircraft:
                    oldest = sorted(aircraft.values(), key=lambda x: x.last_seen_ts)[: len(aircraft) // 10]
                    for o in oldest:
                        aircraft.pop(o.hex, None)

    finally:
        try:
            s.close()
        except Exception:
            pass

    now = time.time()
    cooldown_s = args.cooldown_min * 60.0

    alerts_text = []
    alerts_jsonl = []

    for a in aircraft.values():
        if a.lat is None or a.lon is None:
            continue

        dist_km = haversine_km(args.home_lat, args.home_lon, a.lat, a.lon)
        if dist_km > args.radius_km:
            continue

        last = float(last_alert.get(a.hex, 0) or 0)
        if now - last < cooldown_s:
            continue

        last_alert[a.hex] = now

        enriched = enrich_index.get(a.hex, {})
        if enriched:
            if not a.callsign:
                f = (enriched.get("flight") or enriched.get("callsign") or "").strip()
                if f:
                    a.callsign = f

            a.reg = (enriched.get("r") or enriched.get("reg") or a.reg)
            a.ac_type = (enriched.get("t") or enriched.get("ac_type") or a.ac_type)

            ec = enriched.get("category") or enriched.get("cat")
            if isinstance(ec, str) and ec.strip():
                a.emitter_category = ec.strip()

            a.operation = guess_operation(enriched, a.callsign, a.reg)

        cs = a.callsign or "(no callsign)"
        alt = f"{a.alt_ft}ft" if a.alt_ft is not None else "?ft"
        gs = f"{a.gs_kt:.0f}kt" if a.gs_kt is not None else "?kt"
        typ = a.ac_type or "?"
        reg = a.reg or "?"
        op = a.operation or "unknown"
        ec = a.emitter_category or "?"
        # Prefer per-aircraft tracking links over generic map coordinates.
        # Flightradar24 supports lookup by callsign in the URL in many cases.
        cs_for_url = (cs or "").strip().replace(" ", "")
        fr24 = f"https://www.flightradar24.com/{cs_for_url}" if cs_for_url and cs_for_url != "(nocallsign)" else ""
        # Reliable fallback: ADSBexchange by ICAO hex.
        adsbx = f"https://globe.adsbexchange.com/?icao={a.hex}"

        photo_url = None
        if args.photo:
            entry = photo_cache.get(a.hex)
            cached_url = None
            cached_ts = None
            if isinstance(entry, dict):
                cached_url = entry.get("url")
                cached_ts = entry.get("ts")

            if cached_url and isinstance(cached_ts, (int, float)) and (now - float(cached_ts) < args.photo_cache_hours * 3600):
                photo_url = cached_url
            else:
                photo_url = fetch_planespotters_photo_by_hex(a.hex, size=args.photo_size)
                if photo_url:
                    photo_cache[a.hex] = {"url": photo_url, "ts": now}

        photo_file = None
        if args.photo and args.photo_mode == "download" and photo_url:
            # store as <photo-dir>/<hex>.jpg (overwrite OK)
            photo_dir = Path(args.photo_dir).expanduser()
            dest = photo_dir / f"{a.hex}.jpg"
            got = download_image(photo_url, dest)
            if got:
                photo_file = str(got)

        track_link = fr24 or adsbx

        caption = (
            f"✈️ Overhead ({dist_km:.2f}km ≤ {args.radius_km:g}km)\n"
            f"Callsign: {cs}\n"
            f"Type: {typ} · Reg: {reg} · Op: {op} · EmitterCat: {ec}\n"
            f"Alt/Speed: {alt} · {gs}\n"
            f"Hex: {a.hex}\n"
            f"Track: {track_link}"
        )

        if args.output == "text":
            if photo_url and args.photo_mode == "link":
                alerts_text.append(caption + f"\nPhoto: {photo_url}")
            else:
                alerts_text.append(caption)
        else:
            alerts_jsonl.append(
                {
                    "hex": a.hex,
                    "callsign": cs,
                    "lat": a.lat,
                    "lon": a.lon,
                    "distanceKm": round(dist_km, 3),
                    "caption": caption,
                    "photoUrl": photo_url,
                    "photoFile": photo_file,
                }
            )

    save_state(state_path, state)

    if args.output == "text":
        if alerts_text:
            print("\n\n".join(alerts_text))
    else:
        for obj in alerts_jsonl:
            print(json.dumps(obj, ensure_ascii=False))

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

```