radio-copilot

Original source / Raw SKILL.md

---
name: radio-copilot
description: Predict satellite passes (NOAA APT, METEOR LRPT, ISS) for a configured latitude/longitude and send WhatsApp alerts with manual dish alignment info (AOS/LOS azimuth+elevation, track direction, inclination). Use when setting up or operating a zero-AI pass scheduler/orchestrator for SDR satellite reception, including configuring NORAD IDs, minimum elevation, alert lead time, and optional remote capture/decode hooks (Pi RTL-SDR capture + Jetson SatDump decode).
---

# radio-copilot

Plan and alert on satellite passes for SDR reception.

## Quick start

1) Copy the example config:

```bash
mkdir -p ~/.clawdbot/radio-copilot
cp config.example.json ~/.clawdbot/radio-copilot/config.json
chmod 600 ~/.clawdbot/radio-copilot/config.json
```

2) Run the orchestrator once:

```bash
python3 scripts/orchestrator.py
```

3) Run periodically (system cron):

```bash
*/5 * * * * /usr/bin/python3 /path/to/radio-copilot/scripts/orchestrator.py >> ~/.clawdbot/radio-copilot/orchestrator.log 2>&1
```

## What it sends

Alerts include:
- pass start/max/end timestamps
- **AOS** (start) azimuth/elevation
- **LOS** (end) azimuth/elevation
- track direction (AOS→LOS compass)
- orbit inclination

## Notes

- Runtime is **zero-AI** by default.
- Capture/decode hooks are included but disabled until you configure your Pi/Jetson commands.


---

## Referenced Files

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

### scripts/orchestrator.py

```python
#!/usr/bin/env python3
"""Radio/Satellite Copilot Orchestrator (skeleton).

Purpose:
- Plan satellite pass jobs (NOAA / METEOR / ISS) based on pass predictions.
- Create per-pass run folders with metadata.
- Trigger remote capture on a Raspberry Pi (RTL-SDR USB) via SSH.
- Trigger remote decode on a Jetson (SatDump) via SSH.
- Send WhatsApp alerts (start/end/summary) via Clawdbot CLI.

This is a *skeleton*: it supports the job lifecycle and state/dedupe now.
TODO: Once we have Pi/Jetson details + exact commands, we plug in capture/decode.

Safety:
- Does nothing unless config.enabled=true
- Capture/decode commands are opt-in per satellite
- One job per pass (dedupe by passStart)
- Timeouts for remote commands
"""

from __future__ import annotations

import json
import os
import shlex
import subprocess
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional


CONFIG = Path.home() / ".clawdbot" / "radio-copilot" / "config.json"


def utc_now_iso() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def load_json(path: Path, default: Any) -> Any:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return default


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


def find_clawdbot() -> str:
    # 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
    return "clawdbot"


def send_whatsapp(channel: str, target: str, message: str) -> None:
    claw = find_clawdbot()
    subprocess.run([claw, "message", "send", "--channel", channel, "--target", target, "--message", message], capture_output=True)


def run(cmd: List[str], timeout: int = 30) -> subprocess.CompletedProcess:
    return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)


def ssh(host: str, remote_cmd: str, timeout: int = 60, user: Optional[str] = None) -> subprocess.CompletedProcess:
    target = f"{user}@{host}" if user else host
    return run(["ssh", "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=accept-new", target, remote_cmd], timeout=timeout)


def predict_passes(norad: int, lat: float, lon: float, height_m: float, min_el: float, lookahead_min: float) -> List[dict]:
    # Avoid shell wrappers (this host prints a banner on login shells).
    # We can safely assume node is on PATH for this box, so just call "node".
    node = "node"
    script = Path(__file__).with_name("pass_predictor.mjs")
    p = run(
        [
            node,
            str(script),
            "--norad",
            str(norad),
            "--lat",
            str(lat),
            "--lon",
            str(lon),
            "--height-m",
            str(height_m),
            "--min-el",
            str(min_el),
            "--look-ahead-min",
            str(lookahead_min),
        ],
        timeout=120,
    )
    if p.returncode != 0:
        raise RuntimeError(p.stderr.strip() or "pass predictor failed")
    out = []
    for line in (p.stdout or "").splitlines():
        line = line.strip()
        if line:
            out.append(json.loads(line))
    return out


def main() -> int:
    cfg = load_json(CONFIG, {})
    if not cfg.get("enabled"):
        return 0

    obs = cfg.get("observer") or {}
    lat = float(obs.get("lat"))
    lon = float(obs.get("lon"))
    height_m = float(obs.get("heightM", 0))

    schedule = cfg.get("schedule") or {}
    lookahead = float(schedule.get("lookAheadMinutes", 180))
    lead_min = float(schedule.get("alertLeadMinutes", 10))
    min_repeat_min = float(schedule.get("minRepeatMinutes", 30))

    storage = cfg.get("storage") or {}
    root = Path(str(storage.get("root") or (Path.home() / ".clawdbot" / "radio-copilot"))).expanduser()
    runs_dir = root / str(storage.get("runsDir") or "runs")
    state_file = root / str(storage.get("stateFile") or "state.json")

    state = load_json(state_file, {"lastAlert": {}, "lastAlertEpoch": {}, "lastCapture": {}, "lastDecode": {}})

    notify = cfg.get("notify") or {}
    channel = notify.get("channel", "whatsapp")
    target = notify.get("target")

    now = time.time()

    for sat_cfg in cfg.get("satellites") or []:
        norad = int(sat_cfg.get("norad"))
        name = sat_cfg.get("name") or str(norad)
        min_el = float(sat_cfg.get("minElevationDeg", 20))

        passes = predict_passes(norad, lat, lon, height_m, min_el, lookahead)
        for p in passes:
            start_ts = datetime.fromisoformat(p["passStart"].replace("Z", "+00:00")).timestamp()
            end_ts = datetime.fromisoformat(p["passEnd"].replace("Z", "+00:00")).timestamp()
            # Dedup key: passStart can drift by a few seconds between runs depending on
            # TLE freshness + step size. Normalise to the nearest minute bucket.
            pass_bucket = int(start_ts // 60) * 60
            pass_id = f"{norad}:{pass_bucket}"

            # Only notify before the pass starts (within lead window).
            # Avoid repeated notifications while the pass is already ongoing.
            if now < start_ts - lead_min * 60 or now >= start_ts:
                continue

            run_dir = runs_dir / f"{norad}" / p["passStart"].replace(":", "_")
            run_dir.mkdir(parents=True, exist_ok=True)
            (run_dir / "pass.json").write_text(json.dumps(p, indent=2), encoding="utf-8")

            # Notify once per pass, with a safety minimum interval per satellite.
            last_epoch = float(state.get("lastAlertEpoch", {}).get(str(norad), 0) or 0)
            if (now - last_epoch) < min_repeat_min * 60:
                continue

            if target and state.get("lastAlert", {}).get(str(norad)) != pass_id:
                def dir8(az: float) -> str:
                    dirs = ["N","NE","E","SE","S","SW","W","NW"]
                    return dirs[int(((az % 360) / 45) + 0.5) % 8]

                aosAz = p.get('aosAzDeg')
                aosEl = p.get('aosElDeg')
                losAz = p.get('losAzDeg')
                losEl = p.get('losElDeg')

                aos = f"AOS Az/El: {aosAz}°/{aosEl}°" if aosAz is not None else "AOS Az/El: ?/?"
                los = f"LOS Az/El: {losAz}°/{losEl}°" if losAz is not None else "LOS Az/El: ?/?"
                inc = f"Inclination: {p.get('inclinationDeg','?')}°" if p.get('inclinationDeg') is not None else "Inclination: ?"

                track = "Track: ?"
                if aosAz is not None and losAz is not None:
                    track = f"Track: {dir8(float(aosAz))}→{dir8(float(losAz))}"

                send_whatsapp(
                    channel,
                    target,
                    f"SAT PASS SOON: {name} ({norad})\n"
                    f"Start: {p['passStart']} ({aos})\n"
                    f"Max: {p['passMax']} ({p['maxElevationDeg']}°)\n"
                    f"End: {p['passEnd']} ({los})\n"
                    f"{track}\n"
                    f"{inc}",
                )
                state.setdefault("lastAlert", {})[str(norad)] = pass_id
                state.setdefault("lastAlertEpoch", {})[str(norad)] = now

            # CAPTURE (placeholder)
            cap = sat_cfg.get("capture") or {}
            if cap.get("enabled") and state.get("lastCapture", {}).get(str(norad)) != pass_id:
                # Tomorrow: fill in host/user and actual capture command.
                (run_dir / "capture.todo").write_text("TODO: configure Pi host + capture command\n")
                state.setdefault("lastCapture", {})[str(norad)] = pass_id

            # DECODE (placeholder)
            dec = sat_cfg.get("decode") or {}
            if dec.get("enabled") and state.get("lastDecode", {}).get(str(norad)) != pass_id:
                (run_dir / "decode.todo").write_text("TODO: configure Jetson host + SatDump decode command\n")
                state.setdefault("lastDecode", {})[str(norad)] = pass_id

            # Save state and stop after first actionable pass per run
            save_json(state_file, state)
            return 0

    save_json(state_file, state)
    return 0


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

```



---

## Skill Companion Files

> Additional files collected from the skill directory layout.

### README.md

```markdown
# moltbot-satellite-copilot (Moltbot/Clawdbot add-on)

A *zero-AI* satellite pass planner + orchestrator that can:
- predict upcoming satellite passes over a given lat/lon (NORAD + TLE)
- notify you on WhatsApp with **manual dish alignment** info (AOS/LOS az+el, track direction, inclination)
- (optional) trigger remote capture on a Raspberry Pi and remote decode on a Jetson (disabled by default)

This repo is the **skeleton/orchestration layer**. It’s designed to be safe and production-friendly: nothing transmits or captures unless you enable it.

## Example alert

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

## What you need

- Node.js (for pass prediction)
- Python 3 (for the orchestrator)
- A Moltbot/Clawdbot instance configured for WhatsApp notifications

Optional (when you’re ready to automate captures):
- Raspberry Pi with RTL-SDR dongle (USB)
- Jetson (or any Linux host) to run SatDump decode jobs

## Data flow

1) **Pass prediction**
- `scripts/pass_predictor.mjs` fetches a TLE for a satellite by NORAD ID and predicts passes above a minimum elevation.
- Output includes:
  - start/max/end times
  - AOS/LOS azimuth/elevation (for manual pointing)
  - a simple compass track direction (AOS→LOS)

2) **Orchestrator**
- `scripts/orchestrator.py` reads config, calls the predictor, dedupes alerts, and sends a WhatsApp message before the pass.
- It can also (optionally) run capture and decode hooks (disabled by default).

3) **State + run folders**
- State is persisted so you don’t get spammy repeats.
- Run folders are created per pass under `~/.clawdbot/radio-copilot/runs/…` (for future capture/decode artifacts).

## Configuration

Copy the example config:

```bash
mkdir -p ~/.clawdbot/radio-copilot
cp config.example.json ~/.clawdbot/radio-copilot/config.json
chmod 600 ~/.clawdbot/radio-copilot/config.json
```

Edit `~/.clawdbot/radio-copilot/config.json`:
- `observer.lat` / `observer.lon`
- `satellites[]` NORAD IDs (e.g. NOAA 19 = 33591)
- schedule:
  - `alertLeadMinutes`
  - `minRepeatMinutes` (anti-spam backstop)

## Install & run (system cron)

Run every 5 minutes:

```bash
*/5 * * * * /usr/bin/python3 /path/to/radio-copilot/scripts/orchestrator.py \
  >> ~/.clawdbot/radio-copilot/orchestrator.log 2>&1
```

## SatDump Pi + Jetson architecture (planned / in development)

This repo currently focuses on **pass prediction + orchestration + alerts**.

The end-to-end SDR capture/decode pipeline (Pi capture + Jetson SatDump decode) is **still in planning and development**.

Target architecture (Option B: capture first, decode after):

- **Raspberry Pi (USB RTL‑SDR dongle)**
  - role: capture appliance
  - job: record raw IQ (or baseband/audio depending on mode) during the pass window
  - output: time-bounded capture files + a small status/metadata file per pass

- **Jetson (or any Linux box with enough CPU/GPU)**
  - role: decode worker
  - job: run **SatDump** to decode NOAA APT / METEOR LRPT (and later ISS SSTV workflows)
  - output: decoded images/products + a preview thumbnail

- **Clawdbot/Moltbot host**
  - role: orchestrator
  - job: predict passes, schedule jobs, trigger remote capture/decode via SSH, and send WhatsApp summaries

Why this split:
- capture during a pass is timing-sensitive → keep it simple and robust on the Pi
- decoding can be retried/queued → run it on the Jetson where you have headroom

## License

MIT

```

### _meta.json

```json
{
  "owner": "davestarling",
  "slug": "moltbot-satellite-copilot",
  "displayName": "Satellite Copilot",
  "latest": {
    "version": "0.1.0",
    "publishedAt": 1769960707097,
    "commit": "https://github.com/clawdbot/skills/commit/9407c2cd7757d9ca237dfdef0d2525686d62fcbc"
  },
  "history": []
}

```

### scripts/radio_scheduler.py

```python
#!/usr/bin/env python3
"""Radio/Satellite Copilot scheduler (zero-AI core).

- Reads ~/.clawdbot/radio-copilot/config.json
- Predicts upcoming passes for configured satellites
- Dedupes per pass (by passStart)
- Optionally:
  - emits alerts (via clawdbot message send)
  - runs a capture command during a pass window (guarded by timeout)

This is intentionally conservative and safe:
- Does nothing unless config.enabled=true
- Capture commands are opt-in per satellite
- Logs to stderr for cron

AI layer will come later (classify capture outputs, summarise daily results).
"""

from __future__ import annotations

import json
import os
import subprocess
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List

CONFIG = Path.home() / ".clawdbot" / "radio-copilot" / "config.json"


def find_clawdbot() -> str:
    # 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
    return "clawdbot"


def load_json(path: Path, default: Any) -> Any:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return default


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


def utc_now_iso() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def run_pass_predictor(norad: int, lat: float, lon: float, height_m: float, min_el: float, lookahead_min: float) -> List[dict]:
    node = subprocess.check_output(["command", "-v", "node"], text=True).strip() or "node"
    script = Path(__file__).with_name("pass_predictor.mjs")
    cmd = [
        node,
        str(script),
        "--norad",
        str(norad),
        "--lat",
        str(lat),
        "--lon",
        str(lon),
        "--height-m",
        str(height_m),
        "--min-el",
        str(min_el),
        "--look-ahead-min",
        str(lookahead_min),
    ]
    p = subprocess.run(cmd, capture_output=True, text=True)
    if p.returncode != 0:
        raise RuntimeError(p.stderr.strip() or "pass predictor failed")
    out = []
    for line in (p.stdout or "").splitlines():
        line = line.strip()
        if not line:
            continue
        out.append(json.loads(line))
    return out


def send_notify(channel: str, target: str, message: str) -> None:
    claw = find_clawdbot()
    subprocess.run([claw, "message", "send", "--channel", channel, "--target", target, "--message", message], capture_output=True)


def main() -> int:
    cfg = load_json(CONFIG, {})
    if not cfg.get("enabled"):
        return 0

    storage = cfg.get("storage") or {}
    root = Path(str(storage.get("root") or (Path.home() / ".clawdbot" / "radio-copilot"))).expanduser()
    runs_dir = root / str(storage.get("runsDir") or "runs")
    state_file = root / str(storage.get("stateFile") or "state.json")

    state = load_json(state_file, {"lastAlert": {}, "lastCapture": {}})

    obs = cfg.get("observer") or {}
    lat = float(obs.get("lat"))
    lon = float(obs.get("lon"))
    height_m = float(obs.get("heightM", 0))

    schedule = cfg.get("schedule") or {}
    lookahead = float(schedule.get("lookAheadMinutes", 180))
    lead = float(schedule.get("alertLeadMinutes", 10))

    notify = cfg.get("notify") or {}
    channel = notify.get("channel", "whatsapp")
    target = notify.get("target")

    now = time.time()

    for sat_cfg in cfg.get("satellites") or []:
        norad = int(sat_cfg.get("norad"))
        name = sat_cfg.get("name") or str(norad)
        min_el = float(sat_cfg.get("minElevationDeg", 20))

        passes = run_pass_predictor(norad, lat, lon, height_m, min_el, lookahead)
        for p in passes:
            # Alert when within lead window
            start_ts = datetime.fromisoformat(p["passStart"].replace("Z", "+00:00")).timestamp()
            end_ts = datetime.fromisoformat(p["passEnd"].replace("Z", "+00:00")).timestamp()
            pass_id = p["passStart"]

            if now < start_ts - lead * 60 or now > end_ts:
                continue

            # Dedup alert per pass
            if state.get("lastAlert", {}).get(str(norad)) == pass_id:
                continue

            msg = (
                f"SAT PASS: {name} ({norad})\n"
                f"Start: {p['passStart']}\n"
                f"Max: {p['passMax']} ({p['maxElevationDeg']}°)\n"
                f"End: {p['passEnd']}\n"
            )
            if target:
                send_notify(channel, target, msg)

            state.setdefault("lastAlert", {})[str(norad)] = pass_id

            # Optional capture
            cap = sat_cfg.get("capture") or {}
            if cap.get("enabled"):
                # one capture per pass
                if state.get("lastCapture", {}).get(str(norad)) == pass_id:
                    continue

                cmd = cap.get("command")
                timeout = int(cap.get("timeoutSeconds", 900))
                if cmd:
                    run_dir = runs_dir / f"{norad}" / pass_id.replace(":", "_")
                    run_dir.mkdir(parents=True, exist_ok=True)
                    # store pass metadata
                    (run_dir / "pass.json").write_text(json.dumps(p, indent=2), encoding="utf-8")

                    # Run capture command in shell, with timeout
                    env = os.environ.copy()
                    env["RADIO_RUN_DIR"] = str(run_dir)
                    env["SAT_NORAD"] = str(norad)
                    env["SAT_NAME"] = name
                    env["PASS_START"] = p["passStart"]
                    env["PASS_END"] = p["passEnd"]
                    env["PASS_MAX"] = p["passMax"]
                    try:
                        subprocess.run(cmd, shell=True, cwd=str(run_dir), env=env, timeout=timeout)
                    except Exception:
                        pass

                    state.setdefault("lastCapture", {})[str(norad)] = pass_id

            break

    save_json(state_file, state)
    return 0


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

```