Skip to content

DX Cluster Implementation Plan

For Claude: REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.

Goal: Display real-time DX cluster spots as markers on the waterfall, with click-to-tune.

Architecture: Asyncio telnet client connects to DX cluster server, parses spot lines into DXSpot dataclasses, broadcasts to web clients via existing WebSocket control channel. Frontend renders spot markers on waterfall canvas overlay and provides click-to-tune. Feature is opt-in via --dx-cluster CLI flag.

Tech Stack: Python asyncio (stdlib telnet), existing WebSocket server, Canvas2D overlay

Task 1: DXSpot dataclass + spot parser

File: src/icom_lan/web/dx_cluster.py (NEW)

  1. Write test file tests/test_dx_cluster.py: 

    def test_parse_dxspider_spot():
    line = "DX de K1ABC:     14074.0  JA1XYZ       FT8 +05dB from PM95   1234Z"
    spot = parse_spot(line)
    assert spot.spotter == "K1ABC"
    assert spot.freq == 14074000  # Hz
    assert spot.call == "JA1XYZ"
    assert spot.comment == "FT8 +05dB from PM95"
    Add 10+ test cases: DXSpider, AR-Cluster, CC Cluster formats, edge cases (no comment, weird spacing, non-spot lines return None).

  2. Run tests — verify they fail (RED).

  3. Implement in dx_cluster.py: 

    @dataclass(frozen=True, slots=True)
class DXSpot:
    spotter: str
    freq: int          # Hz
    call: str
    comment: str = ""
    time_utc: str = ""
    timestamp: float = field(default_factory=time.monotonic)

_SPOT_RE = re.compile(r"^DX de\s+(\S+):\s+(\d+\.?\d*)\s+(\S+)\s+(.*?)\s+(\d{4}Z)?\s*$")

def parse_spot(line: str) -> DXSpot | None:
    ...

  4. Run tests — verify they pass (GREEN).

  5. Commit: feat(dx): DXSpot dataclass + spot parser with 10+ test cases

Task 2: DXClusterClient — asyncio telnet

File: src/icom_lan/web/dx_cluster.py

  1. Write tests in tests/test_dx_cluster.py:
  2. Test client connects and sends callsign login
  3. Test client calls on_spot callback for each parsed line
  4. Test client ignores non-spot lines
  5. Test client reconnects after disconnect (exponential backoff)
  6. Test client stop/cleanup

  7. Use asyncio mock streams (asyncio.StreamReader/Writer)

  8. Run tests — RED.

  9. Implement: 

    class DXClusterClient:
    def __init__(self, host: str, port: int, callsign: str,
                 on_spot: Callable[[DXSpot], None]):
        ...

    async def start(self) -> None:
        """Connect and read spots in a loop. Auto-reconnects."""
        ...

    async def stop(self) -> None:
        """Disconnect and cancel tasks."""
        ...

  10. asyncio.open_connection(host, port)
  11. Read lines, parse each with parse_spot()
  12. On disconnect: log, wait min(2**attempt, 60) seconds, retry

  13. Login: send callsign\n after connect

  14. Run tests — GREEN.

  15. Commit: feat(dx): DXClusterClient with auto-reconnect

Task 3: Spot buffer + REST API

File: src/icom_lan/web/dx_cluster.py + src/icom_lan/web/server.py

  1. Write tests:
  2. SpotBuffer: max 200 spots, oldest dropped on overflow
  3. SpotBuffer: get_spots(band=None) returns filtered list
  4. SpotBuffer: to_json() serialization

  5. REST endpoint: GET /api/v1/dx/spots returns current spots

  6. Run tests — RED.

  7. Implement: 

    class SpotBuffer:
    def __init__(self, maxlen: int = 200):
        self._spots: deque[DXSpot] = deque(maxlen=maxlen)

    def add(self, spot: DXSpot) -> None: ...
    def get_spots(self, band: str | None = None) -> list[dict]: ...
    def expire(self, max_age_s: float = 1800) -> None: ...

In server.py: - Add SpotBuffer instance - Add GET /api/v1/dx/spots route - On spot callback: add to buffer + broadcast via control WS

  1. Run tests — GREEN.

  2. Commit: feat(dx): SpotBuffer + REST endpoint

Task 4: WebSocket broadcast

File: src/icom_lan/web/server.py + src/icom_lan/web/handlers.py

  1. Write tests:
  2. New spot → JSON message {"type": "dx_spot", "spot": {...}} sent to all control WS clients

  3. Client connect → receives current spot buffer as {"type": "dx_spots", "spots": [...]}

  4. Run tests — RED.

  5. Implement:

  6. server._broadcast_dx_spot(spot) — sends to all ControlHandler clients
  7. ControlHandler._send_state_snapshot() — include current spots if DX cluster active

  8. Message format: {"type": "dx_spot", "spot": {"call": "JA1XYZ", "freq": 14074000, "spotter": "K1ABC", ...}}

  9. Run tests — GREEN.

  10. Commit: feat(dx): WebSocket spot broadcast

Task 5: CLI flags

File: src/icom_lan/cli.py

  1. Write tests:
  2. --dx-cluster flag parsed correctly (host:port)
  3. --callsign flag parsed
  4. Without --dx-cluster → no DX client started

  5. Invalid format → clear error

  6. Run tests — RED.

  7. Implement:

  8. web command: add --dx-cluster HOST:PORT and --callsign CALL
  9. Pass to WebServer → start DXClusterClient if configured

  10. Graceful shutdown: stop DX client on server stop

  11. Run tests — GREEN.

  12. Commit: feat(dx): CLI --dx-cluster and --callsign flags

Task 6: Frontend — spot overlay on waterfall

File: src/icom_lan/web/static/index.html

  1. No automated tests (canvas rendering). Manual testing.

  2. Implement:

  3. Listen for dx_spot / dx_spots WS messages
  4. Store spots in state.dxSpots[]
  5. In renderLoop(): draw spot markers on spectrum canvas
    • Only spots within current scope freq range
    • Marker: small triangle ▼ + callsign text
    • Color by age: bright cyan (fresh) → dim gray (>15 min) → remove (>30 min)
  6. Click handler on spectrum canvas: if click near spot marker → sendCommand('set_freq', {freq: spot.freq})

  7. DX toggle button in toolbar (shows/hides overlay + connects/disconnects)

  8. Commit: feat(dx): frontend spot overlay + click-to-tune

Task 7: Integration test + docs

  1. Manual integration test:
  2. Start server with --dx-cluster dxc.nc7j.com:7373 --callsign KN4KYD
  3. Verify spots appear on waterfall
  4. Click spot → radio tunes to frequency
  5. Toggle DX off → spots disappear, telnet disconnects

  6. Toggle DX on → reconnects, spots reappear

  7. Update docs:

  8. docs/guide/cli.md — add --dx-cluster and --callsign flags
  9. docs/guide/web-ui.md — DX cluster section

  10. README.md — mention DX cluster feature

  11. Update issue #108 with implementation status

  12. Commit: docs: DX cluster setup guide

  13. Run full test suite: uv run python -m pytest tests/ -q --tb=short

Agent Assignment (tmux)

Agent Tasks Window
claude:dx-backend Tasks 1-3 (parser, client, buffer) agents:backend
claude:dx-frontend Task 6 (overlay, click-to-tune) agents:frontend
sequential Tasks 4-5, 7 (integration, CLI, docs) after backend done

Backend and frontend can run in parallel (no file conflicts).