warehack.ing/birdcage
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Add Craft mode — direct satellite tracking via space.warehack.ing API

Browse source 
New F2 Control sub-mode that searches the Craft orbital catalog (22k+
objects), displays pass predictions, and drives the dish in real-time
using server-computed AZ/EL positions from /api/sky/up.

Tracking loop polls at ~1Hz, filters for the tracked target by
target_type + target_id (str, not int — handles satellites, planets,
stars, comets), and issues motor commands through the existing serial
bridge. Verified end-to-end with Carryout G2 hardware tracking NOAA 17.

New files:
- craft_client.py — stdlib HTTP client (urllib only, no deps)
- widgets/craft_panel.py — search table, pass info, tracking status
- tests/test_craft_mode.py — 5 unit tests with mocked API
- tests/test_craft_integration.py — 3 hardware integration tests
This commit is contained in:
Ryan Malloy 2026-02-16 02:07:42 -07:00
parent a249c98208
commit 6c1e9da773
9 changed files with 1379 additions and 85 deletions
Download patch file Download diff file Expand all files Collapse all files

222
tui/tests/test_craft_integration.py Normal file
View file

@ -0,0 +1,222 @@
"""Integration test: Craft mode with real hardware.
Exercises the full Craft -> SerialBridge -> dish code path.
Requires /dev/ttyUSB2 (Carryout G2 via RS-422).
NOT part of the normal test suite -- run explicitly:
uv run pytest tests/test_craft_integration.py -v -s
"""
import asyncio
import os
import pytest
from birdcage_tui.app import BirdcageApp
from birdcage_tui.screens.control import ControlScreen
from birdcage_tui.widgets.craft_panel import CraftPanel, CraftTrackingStatus
SERIAL_PORT = "/dev/ttyUSB2"
def _az_delta(a: float, b: float) -> float:
"""Minimum angular distance accounting for 360 wrap."""
d = abs(a - b) % 360.0
return min(d, 360.0 - d)
pytestmark = pytest.mark.skipif(
not os.path.exists(SERIAL_PORT),
reason=f"{SERIAL_PORT} not available",
)
@pytest.mark.asyncio
async def test_craft_search_with_real_api():
"""Search the live Craft API and verify results populate."""
app = BirdcageApp()
app.demo_mode = True
app.craft_url = "https://space.warehack.ing"
async with app.run_test(size=(120, 40)) as pilot:
await pilot.pause()
control = app.query_one("#control", ControlScreen)
control.switch_mode("craft")
await pilot.pause()
panel = app.query_one("#ctrl-craft-panel", CraftPanel)
panel.post_message(CraftPanel.SearchRequested("ISS"))
await asyncio.sleep(3.0)
from textual.widgets import DataTable
table = app.query_one("#craft-results-table", DataTable)
print(f" Search returned {table.row_count} results")
assert table.row_count > 0, "Craft API returned no results for 'ISS'"
@pytest.mark.asyncio
async def test_direct_motor_command():
"""Verify move_motor works through the bridge.
Moves AZ by +1 degree and back. Isolates serial bridge
independent of Craft.
"""
app = BirdcageApp()
app.demo_mode = False
app.serial_port = SERIAL_PORT
app.firmware_name = "g2"
app.skip_init = True
async with app.run_test(size=(120, 40)) as pilot:
await pilot.pause()
await asyncio.sleep(1.0)
assert app.device is not None
assert app.device.is_connected
pos = app.device.get_position()
start_az = pos["azimuth"]
start_el = pos["elevation"]
print(f" Current: AZ={start_az:.2f} EL={start_el:.2f}")
# Move AZ by +1 degree (stay within wrap range)
target_az = (start_az + 1.0) % 360.0
print(f" Moving AZ to {target_az:.2f}")
app.device.move_motor(0, target_az)
await asyncio.sleep(3.0)
pos2 = app.device.get_position()
print(f" After move: AZ={pos2['azimuth']:.2f} EL={pos2['elevation']:.2f}")
delta = _az_delta(pos2["azimuth"], target_az)
print(f" AZ delta from target: {delta:.2f} degrees")
assert delta < 2.0, f"Dish didn't move close to target (delta={delta})"
# Move back
print(f" Returning to AZ={start_az:.2f}")
app.device.move_motor(0, start_az)
await asyncio.sleep(3.0)
pos3 = app.device.get_position()
print(f" Final: AZ={pos3['azimuth']:.2f} EL={pos3['elevation']:.2f}")
@pytest.mark.asyncio
async def test_craft_tracking_moves_real_dish():
"""Track a target via Craft API and verify the dish gets commands.
Uses the real serial device. Tracks briefly then returns
to the original position.
"""
app = BirdcageApp()
app.demo_mode = False
app.serial_port = SERIAL_PORT
app.firmware_name = "g2"
app.skip_init = True
app.craft_url = "https://space.warehack.ing"
async with app.run_test(size=(120, 40)) as pilot:
await pilot.pause()
await asyncio.sleep(1.0)
control = app.query_one("#control", ControlScreen)
control.switch_mode("craft")
await pilot.pause()
assert app.device is not None
assert app.device.is_connected
print(f" Device connected on {SERIAL_PORT}")
pos_before = app.device.get_position()
print(
f" Starting position: "
f"AZ={pos_before['azimuth']:.2f} "
f"EL={pos_before['elevation']:.2f}"
)
panel = app.query_one("#ctrl-craft-panel", CraftPanel)
# Search for something likely above horizon
panel.post_message(CraftPanel.SearchRequested("Moon"))
await asyncio.sleep(3.0)
from textual.widgets import DataTable
table = app.query_one("#craft-results-table", DataTable)
print(f" Moon search: {table.row_count} results")
if table.row_count == 0:
panel.post_message(CraftPanel.SearchRequested("Jupiter"))
await asyncio.sleep(3.0)
print(f" Jupiter search: {table.row_count} results")
if table.row_count == 0:
panel.post_message(CraftPanel.SearchRequested("Sun"))
await asyncio.sleep(3.0)
print(f" Sun search: {table.row_count} results")
assert table.row_count > 0, "No search results found"
# Read the first row
first_key = list(table.rows.keys())[0]
row = table.get_row(first_key)
target_name = row[0]
target_type = row[1]
target_id = int(row[2])
print(f" Tracking: {target_name} ({target_type}:{target_id})")
# Start tracking
panel.post_message(
CraftPanel.TrackRequested(
target_type=target_type,
target_id=target_id,
name=target_name,
min_el=18.0,
)
)
# Let the tracking loop run a few cycles
await asyncio.sleep(6.0)
status = app.query_one("#craft-tracking-status", CraftTrackingStatus)
print(
f" State: {status.state}, "
f"AZ={status.azimuth:.2f}, EL={status.elevation:.2f}, "
f"moves={status.moves}, error={status.error!r}"
)
assert status.state in ("TRACKING", "WAITING"), (
f"Unexpected state: {status.state}"
)
if status.state == "TRACKING":
assert status.moves >= 1, "No motor commands issued"
pos_after = app.device.get_position()
print(
f" Position after: "
f"AZ={pos_after['azimuth']:.2f} "
f"EL={pos_after['elevation']:.2f}"
)
daz = _az_delta(pos_after["azimuth"], pos_before["azimuth"])
del_ = abs(pos_after["elevation"] - pos_before["elevation"])
print(f" Dish moved: dAZ={daz:.2f} dEL={del_:.2f}")
else:
print(" Target below horizon or min_el -- WAITING is OK")
# Stop tracking
control._stop_craft_tracking()
await pilot.pause()
# Return to starting position
print(
f" Returning to: "
f"AZ={pos_before['azimuth']:.2f} "
f"EL={pos_before['elevation']:.2f}"
)
app.device.move_to(pos_before["azimuth"], pos_before["elevation"])
await asyncio.sleep(2.0)
print(" Integration test complete")

259
tui/tests/test_craft_mode.py Normal file
View file

@ -0,0 +1,259 @@
"""Test Craft mode — API-driven satellite tracking via the F2 Control screen.
Verifies search, pass predictions, tracking loop, and stop lifecycle using
mocked CraftClient methods. No real network calls.
NOTE: Uses post_message() for reliable button activation (same pattern as
test_track_mode.py).
"""
import asyncio
from unittest.mock import MagicMock
import pytest
from birdcage_tui.app import BirdcageApp
from birdcage_tui.craft_client import PassPrediction, SearchResult, TargetPosition
from birdcage_tui.screens.control import ControlScreen
from birdcage_tui.widgets.craft_panel import CraftPanel, CraftTrackingStatus
async def _switch_to_craft(pilot, app) -> None:
"""Navigate to F2 Control > Craft sub-mode."""
await pilot.press("f2")
await pilot.pause()
control = app.query_one("#control", ControlScreen)
control.switch_mode("craft")
await pilot.pause()
def _mock_search_results() -> list[SearchResult]:
return [
SearchResult(
name="ISS (ZARYA)",
target_type="satellite",
target_id="25544",
score=1.0,
groups=["stations"],
),
SearchResult(
name="NOAA 19",
target_type="satellite",
target_id="33591",
score=0.8,
groups=["weather"],
),
]
def _mock_passes() -> list[PassPrediction]:
return [
PassPrediction(
satellite_name="ISS (ZARYA)",
norad_id=25544,
aos_time="2026-02-16T08:20:00Z",
aos_az=220.0,
tca_time="2026-02-16T08:25:00Z",
tca_alt=45.3,
tca_az=180.0,
los_time="2026-02-16T08:30:00Z",
los_az=140.0,
max_elevation=45.3,
duration_seconds=600,
is_visible=True,
),
]
def _mock_visible_target() -> list[TargetPosition]:
return [
TargetPosition(
name="ISS (ZARYA)",
target_type="satellite",
target_id="25544",
azimuth=245.30,
altitude=34.20,
distance_km=420.0,
range_rate=-2.1,
is_above_horizon=True,
),
]
def _mock_below_horizon() -> list[TargetPosition]:
"""Return an empty list — target is not in the sky."""
return []
@pytest.mark.asyncio
async def test_search_populates_table():
"""Searching should populate the DataTable with results."""
app = BirdcageApp()
app.demo_mode = True
async with app.run_test(size=(120, 40)) as pilot:
await pilot.pause()
await _switch_to_craft(pilot, app)
control = app.query_one("#control", ControlScreen)
panel = app.query_one("#ctrl-craft-panel", CraftPanel)
# Mock the client
mock_client = MagicMock()
mock_client.search.return_value = _mock_search_results()
control.set_craft_client(mock_client)
# Post search request
panel.post_message(CraftPanel.SearchRequested("ISS"))
await asyncio.sleep(0.5)
# Verify table was populated
from textual.widgets import DataTable
table = app.query_one("#craft-results-table", DataTable)
assert table.row_count == 2
mock_client.search.assert_called_once_with("ISS")
@pytest.mark.asyncio
async def test_tracking_drives_device():
"""Tracking should poll positions and move the demo device."""
app = BirdcageApp()
app.demo_mode = True
async with app.run_test(size=(120, 40)) as pilot:
await pilot.pause()
await _switch_to_craft(pilot, app)
control = app.query_one("#control", ControlScreen)
panel = app.query_one("#ctrl-craft-panel", CraftPanel)
mock_client = MagicMock()
mock_client.get_visible_targets.return_value = _mock_visible_target()
control.set_craft_client(mock_client)
# Start tracking
panel.post_message(
CraftPanel.TrackRequested(
target_type="satellite",
target_id="25544",
name="ISS (ZARYA)",
min_el=18.0,
)
)
# Let the tracking loop run a couple iterations
await asyncio.sleep(2.5)
# Device target should have been updated
assert app.device._target_az == pytest.approx(245.30, abs=0.1)
assert app.device._target_el == pytest.approx(34.20, abs=0.1)
# Status should show TRACKING
status = app.query_one("#craft-tracking-status", CraftTrackingStatus)
assert status.state == "TRACKING"
assert status.moves >= 1
# Cleanup
control._stop_craft_tracking()
await pilot.pause()
@pytest.mark.asyncio
async def test_tracking_stops_cleanly():
"""Stopping tracking should return to IDLE state."""
app = BirdcageApp()
app.demo_mode = True
async with app.run_test(size=(120, 40)) as pilot:
await pilot.pause()
await _switch_to_craft(pilot, app)
control = app.query_one("#control", ControlScreen)
panel = app.query_one("#ctrl-craft-panel", CraftPanel)
mock_client = MagicMock()
mock_client.get_visible_targets.return_value = _mock_visible_target()
control.set_craft_client(mock_client)
# Start tracking
panel.post_message(
CraftPanel.TrackRequested(
target_type="satellite",
target_id="25544",
name="ISS (ZARYA)",
min_el=18.0,
)
)
await asyncio.sleep(1.5)
# Stop
panel.post_message(CraftPanel.StopTrackingRequested())
await asyncio.sleep(0.5)
status = app.query_one("#craft-tracking-status", CraftTrackingStatus)
assert status.state == "IDLE"
assert not control._craft_tracking
@pytest.mark.asyncio
async def test_below_horizon_shows_waiting():
"""Target not in sky/up results should show WAITING status."""
app = BirdcageApp()
app.demo_mode = True
async with app.run_test(size=(120, 40)) as pilot:
await pilot.pause()
await _switch_to_craft(pilot, app)
control = app.query_one("#control", ControlScreen)
panel = app.query_one("#ctrl-craft-panel", CraftPanel)
mock_client = MagicMock()
mock_client.get_visible_targets.return_value = _mock_below_horizon()
control.set_craft_client(mock_client)
# Start tracking
panel.post_message(
CraftPanel.TrackRequested(
target_type="satellite",
target_id="25544",
name="ISS (ZARYA)",
min_el=18.0,
)
)
await asyncio.sleep(1.5)
status = app.query_one("#craft-tracking-status", CraftTrackingStatus)
assert status.state == "WAITING"
# Cleanup
control._stop_craft_tracking()
await pilot.pause()
@pytest.mark.asyncio
async def test_passes_display():
"""Pass predictions should update the CraftPassInfo widget."""
app = BirdcageApp()
app.demo_mode = True
async with app.run_test(size=(120, 40)) as pilot:
await pilot.pause()
await _switch_to_craft(pilot, app)
control = app.query_one("#control", ControlScreen)
panel = app.query_one("#ctrl-craft-panel", CraftPanel)
mock_client = MagicMock()
mock_client.get_passes.return_value = _mock_passes()
control.set_craft_client(mock_client)
# Request passes
panel.post_message(CraftPanel.PassesRequested(norad_id=25544))
await asyncio.sleep(0.5)
from birdcage_tui.widgets.craft_panel import CraftPassInfo
info = app.query_one("#craft-pass-info", CraftPassInfo)
assert "45.3" in info.passes_text
mock_client.get_passes.assert_called_once_with(25544)

56
tui/tests/test_track_mode.py
View file

@ -64,9 +64,7 @@ async def test_start_creates_listening_server():
await pilot.pause()
# Post StartRequested with a known port.
panel.post_message(
TrackingPanel.StartRequested("127.0.0.1", 14533, 18.0)
)
panel.post_message(TrackingPanel.StartRequested("127.0.0.1", 14533, 18.0))
await _wait_for_listening(app)
status = app.query_one("#tracking-status", TrackingStatus)
@ -74,9 +72,7 @@ async def test_start_creates_listening_server():
assert control._rotctld_server is not None
# Verify the TCP port is open.
reader, writer = await asyncio.open_connection(
"127.0.0.1", 14533
)
reader, writer = await asyncio.open_connection("127.0.0.1", 14533)
writer.close()
await writer.wait_closed()
@ -102,9 +98,7 @@ async def test_stop_shuts_down_server():
await pilot.pause()
# Start.
panel.post_message(
TrackingPanel.StartRequested("127.0.0.1", 14534, 18.0)
)
panel.post_message(TrackingPanel.StartRequested("127.0.0.1", 14534, 18.0))
await _wait_for_listening(app)
assert control._rotctld_server is not None
@ -119,9 +113,7 @@ async def test_stop_shuts_down_server():
# Port should no longer be accepting.
with pytest.raises(OSError):
_r, _w = await asyncio.open_connection(
"127.0.0.1", 14534
)
_r, _w = await asyncio.open_connection("127.0.0.1", 14534)
@pytest.mark.asyncio
@ -147,16 +139,12 @@ async def test_status_display_updates():
assert status.moves == 0
# Start server.
panel.post_message(
TrackingPanel.StartRequested("127.0.0.1", 14535, 18.0)
)
panel.post_message(TrackingPanel.StartRequested("127.0.0.1", 14535, 18.0))
await _wait_for_listening(app)
assert status.state == "LISTENING"
# Connect a client -- should transition to CONNECTED.
reader, writer = await asyncio.open_connection(
"127.0.0.1", 14535
)
reader, writer = await asyncio.open_connection("127.0.0.1", 14535)
# Send a command so the server accept loop processes it.
writer.write(b"_\n")
await writer.drain()
@ -195,25 +183,17 @@ async def test_rotctld_get_position():
control.switch_mode("track")
await pilot.pause()
panel.post_message(
TrackingPanel.StartRequested("127.0.0.1", 14536, 18.0)
)
panel.post_message(TrackingPanel.StartRequested("127.0.0.1", 14536, 18.0))
await _wait_for_listening(app)
reader, writer = await asyncio.open_connection(
"127.0.0.1", 14536
)
reader, writer = await asyncio.open_connection("127.0.0.1", 14536)
# Query position.
writer.write(b"p\n")
await writer.drain()
az_line = await asyncio.wait_for(
reader.readline(), timeout=3.0
)
el_line = await asyncio.wait_for(
reader.readline(), timeout=3.0
)
az_line = await asyncio.wait_for(reader.readline(), timeout=3.0)
el_line = await asyncio.wait_for(reader.readline(), timeout=3.0)
az = float(az_line.decode().strip())
el = float(el_line.decode().strip())
@ -244,14 +224,10 @@ async def test_rotctld_set_position():
control.switch_mode("track")
await pilot.pause()
panel.post_message(
TrackingPanel.StartRequested("127.0.0.1", 14537, 18.0)
)
panel.post_message(TrackingPanel.StartRequested("127.0.0.1", 14537, 18.0))
await _wait_for_listening(app)
reader, writer = await asyncio.open_connection(
"127.0.0.1", 14537
)
reader, writer = await asyncio.open_connection("127.0.0.1", 14537)
# Move to a specific position.
writer.write(b"P 200.0 55.0\n")
@ -297,14 +273,10 @@ async def test_rotctld_quit_command():
control.switch_mode("track")
await pilot.pause()
panel.post_message(
TrackingPanel.StartRequested("127.0.0.1", 14538, 18.0)
)
panel.post_message(TrackingPanel.StartRequested("127.0.0.1", 14538, 18.0))
await _wait_for_listening(app)
reader, writer = await asyncio.open_connection(
"127.0.0.1", 14538
)
reader, writer = await asyncio.open_connection("127.0.0.1", 14538)
writer.write(b"q\n")
await writer.drain()