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

Promote bridge, demo, craft_client to core birdcage package

Browse source 
Move bridge.py, demo.py, craft_client.py from tui/src/birdcage_tui/ to
src/birdcage/ so both TUI and MCP server can share the device layer
without a circular dependency on textual.
This commit is contained in:
Ryan Malloy 2026-02-17 16:01:38 -07:00
parent 3013eeee4c
commit 16ca4892b3
7 changed files with 38 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

9
tui/src/birdcage_tui/app.py
View file

@ -102,15 +102,14 @@ class BirdcageApp(App):
def _setup_device(self) -> None:
"""Create device (demo or real) and hand it to each screen."""
if self.demo_mode:
from birdcage_tui.demo import DemoDevice
from birdcage.demo import DemoDevice
self.device = DemoDevice()
self.device.connect()
else:
from birdcage.bridge import SerialBridge
from birdcage.protocol import get_protocol
from birdcage_tui.bridge import SerialBridge
protocol = get_protocol(self.firmware_name)
self.device = SerialBridge(protocol)
self.device.connect(self.serial_port)
@ -142,11 +141,11 @@ class BirdcageApp(App):
def _setup_craft_client(self) -> None:
"""Create a Craft API client and hand it to the control screen."""
if self.demo_mode:
from birdcage_tui.demo import DemoCraftClient
from birdcage.demo import DemoCraftClient
client = DemoCraftClient()
else:
from birdcage_tui.craft_client import CraftClient
from birdcage.craft_client import CraftClient
client = CraftClient(base_url=self.craft_url)
try:

641
tui/src/birdcage_tui/bridge.py
View file

@ -1,641 +0,0 @@
"""Thread-safe bridge between Birdcage TUI and CarryoutG2Protocol.
Wraps all serial I/O in a threading.Lock so the TUI's worker threads
don't stomp on each other. Tracks the current firmware submenu to
minimize unnecessary q-then-reenter transitions.
"""
import contextlib
import logging
import re
import threading
from enum import Enum, auto
from birdcage.protocol import CarryoutG2Protocol
logger = logging.getLogger(__name__)
class Menu(Enum):
"""Firmware submenu states."""
ROOT = auto()
MOT = auto()
DVB = auto()
NVS = auto()
A3981 = auto()
ADC = auto()
OS = auto()
STEP = auto()
PEAK = auto()
EEPROM = auto()
GPIO = auto()
LATLON = auto()
DIPSWITCH = auto()
UNKNOWN = auto()
# Map Menu enum to the command that enters it from root.
_MENU_COMMANDS: dict[Menu, str] = {
Menu.MOT: "mot",
Menu.DVB: "dvb",
Menu.NVS: "nvs",
Menu.A3981: "a3981",
Menu.ADC: "adc",
Menu.OS: "os",
Menu.STEP: "step",
Menu.PEAK: "peak",
Menu.EEPROM: "eeprom",
Menu.GPIO: "gpio",
Menu.LATLON: "latlon",
Menu.DIPSWITCH: "dipswitch",
}
class SerialBridge:
"""Thread-safe wrapper around CarryoutG2Protocol for TUI consumption.
All public methods acquire a lock before touching the serial port.
The bridge tracks the current firmware submenu so it can skip
redundant quit-and-reenter cycles.
"""
def __init__(self, protocol: CarryoutG2Protocol) -> None:
self._proto = protocol
self._lock = threading.Lock()
self._cancel = threading.Event()
self._menu = Menu.UNKNOWN
self._connected = False
# ------------------------------------------------------------------
# Menu prompt → string mapping for status display
# ------------------------------------------------------------------
_MENU_PROMPTS: dict[Menu, str] = {
Menu.ROOT: "TRK>",
Menu.MOT: "MOT>",
Menu.DVB: "DVB>",
Menu.NVS: "NVS>",
Menu.A3981: "A3981>",
Menu.ADC: "ADC>",
Menu.OS: "OS>",
Menu.STEP: "STEP>",
Menu.PEAK: "PEAK>",
Menu.EEPROM: "EE>",
Menu.GPIO: "GPIO>",
Menu.LATLON: "LATLON>",
Menu.DIPSWITCH: "DIPSWITCH>",
Menu.UNKNOWN: "???",
}
@property
def current_menu(self) -> str:
"""Current firmware prompt string for status display."""
return self._MENU_PROMPTS.get(self._menu, "???")
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
def _send(self, cmd: str) -> str:
"""Send a command via the protocol's prompt-terminated send.
Caller must hold ``_lock``.
"""
return self._proto.send_raw(cmd)
def _go_to_root(self) -> None:
"""Return to TRK> root menu. Caller must hold ``_lock``."""
self._proto.reset_to_root()
self._menu = Menu.ROOT
def _detect_menu(self) -> Menu:
"""Probe firmware prompt and return the corresponding Menu enum.
Caller must hold ``_lock``.
"""
prompt = self._proto._probe_prompt()
upper = prompt.upper()
# Check against known prompt strings (handles EE> vs EEPROM>, etc.)
for menu, prompt_str in self._MENU_PROMPTS.items():
if menu == Menu.UNKNOWN:
continue
if prompt_str.upper() in upper:
return menu
return Menu.UNKNOWN
def _ensure_menu(self, target: Menu) -> None:
"""Navigate to *target* submenu if not already there.
Caller must hold ``_lock``.
"""
if self._menu == target:
return
# Always go back to root first — we don't know how to go
# directly between arbitrary submenus.
if self._menu != Menu.ROOT:
self._go_to_root()
if target == Menu.ROOT:
return
cmd = _MENU_COMMANDS.get(target)
if cmd is None:
raise ValueError(f"No entry command for menu {target!r}")
self._send(cmd)
self._menu = target
# ------------------------------------------------------------------
# Connection
# ------------------------------------------------------------------
def connect(self, port: str, baudrate: int = 115200) -> None:
"""Open the RS-422 serial connection."""
with self._lock:
self._proto.connect(port, baudrate)
self._connected = True
self._menu = self._detect_menu()
def cancel_operation(self) -> None:
"""Signal any in-progress long-running operation to abort.
Safe to call from any thread. The cancel event is checked every
~2 seconds by ``send_with_timeout``.
"""
self._cancel.set()
def clear_cancel(self) -> None:
"""Reset the cancel event so future operations proceed normally."""
self._cancel.clear()
def disconnect(self) -> None:
"""Close the serial connection.
Signals cancellation first to unblock any long-running serial
reads (e.g. firmware sweep), then acquires the lock to close
the port cleanly.
"""
self._cancel.set()
if not self._lock.acquire(timeout=5):
# Lock held by dead/stuck worker — force-close the port
# so the blocked serial read raises an exception.
logger.warning("Lock acquisition timed out, force-closing port")
with contextlib.suppress(Exception):
self._proto.disconnect()
self._connected = False
self._menu = Menu.UNKNOWN
self._cancel.clear()
return
try:
with contextlib.suppress(Exception):
self._go_to_root()
self._proto.disconnect()
self._connected = False
self._menu = Menu.UNKNOWN
self._cancel.clear()
finally:
self._lock.release()
@property
def is_connected(self) -> bool:
return self._connected and self._proto.is_connected
def initialize(self, skip_init: bool = False) -> None:
"""Prepare the dish for motor commands.
Args:
skip_init: If True, skip the protocol initialize step
(useful when re-connecting to an already-running dish).
"""
with self._lock:
if not skip_init:
self._proto.initialize()
self._menu = Menu.MOT # initialize() ends in MOT>
# else: leave _menu as whatever connect() detected
# ------------------------------------------------------------------
# Motor (MOT>)
# ------------------------------------------------------------------
def get_position(self) -> dict[str, float]:
"""Query current AZ/EL position.
Returns:
``{"azimuth": float, "elevation": float}``
"""
with self._lock:
self._ensure_menu(Menu.MOT)
response = self._send("a")
az_m = re.search(r"Angle\[0\]\s*=\s*(-?\d+\.?\d*)", response)
el_m = re.search(r"Angle\[1\]\s*=\s*(-?\d+\.?\d*)", response)
if not az_m or not el_m:
raise ValueError(f"Could not parse position: {response!r}")
return {
"azimuth": float(az_m.group(1)),
"elevation": float(el_m.group(1)),
}
def move_to(self, az: float, el: float) -> None:
"""Move the dish to an absolute AZ/EL position."""
with self._lock:
self._ensure_menu(Menu.MOT)
self._send(f"a 0 {az}")
self._send(f"a 1 {el}")
def move_motor(self, motor_id: int, degrees: float) -> None:
"""Move a single motor to an absolute position."""
with self._lock:
self._ensure_menu(Menu.MOT)
self._send(f"a {motor_id} {degrees}")
def home_motor(self, motor_id: int) -> None:
"""Home a motor to its reference position."""
with self._lock:
self._ensure_menu(Menu.MOT)
self._send(f"h {motor_id}")
def engage(self) -> None:
"""Engage (energize) the stepper motors."""
with self._lock:
self._ensure_menu(Menu.MOT)
self._send("e")
def release(self) -> None:
"""Release (de-energize) the stepper motors."""
with self._lock:
self._ensure_menu(Menu.MOT)
self._send("r")
def get_motor_list(self) -> str:
"""List motors and their state."""
with self._lock:
self._ensure_menu(Menu.MOT)
return self._send("l")
def get_motor_dynamics(self) -> dict[str, float]:
"""Read max velocity and acceleration for both axes.
Returns:
``{"az_max_vel": float, "el_max_vel": float,
"az_accel": float, "el_accel": float}``
"""
with self._lock:
self._ensure_menu(Menu.MOT)
mv_resp = self._send("mv")
ma_resp = self._send("ma")
result: dict[str, float] = {
"az_max_vel": 0.0,
"el_max_vel": 0.0,
"az_accel": 0.0,
"el_accel": 0.0,
}
# mv → "Max Vel [0] = 65.0 Max Vel [1] = 45.0" or similar
vel_matches = re.findall(r"Max Vel\s*\[(\d)\]\s*=\s*(-?\d+\.?\d*)", mv_resp)
for motor_id, val in vel_matches:
if motor_id == "0":
result["az_max_vel"] = float(val)
elif motor_id == "1":
result["el_max_vel"] = float(val)
# ma → "Accel[0] = 400.0 Accel[1] = 400.0"
acc_matches = re.findall(r"Accel\[(\d)\]\s*=\s*(-?\d+\.?\d*)", ma_resp)
for motor_id, val in acc_matches:
if motor_id == "0":
result["az_accel"] = float(val)
elif motor_id == "1":
result["el_accel"] = float(val)
return result
def set_max_velocity(self, motor_id: int, deg_per_sec: float) -> None:
"""Set max velocity for a motor axis (°/s). Firmware: MOT> mv [motor] [vel]."""
with self._lock:
self._ensure_menu(Menu.MOT)
self._send(f"mv {motor_id} {deg_per_sec:.1f}")
def set_max_acceleration(self, motor_id: int, accel: float) -> None:
"""Set max acceleration for a motor axis.
Firmware: MOT> ma [motor] [accel] (°/s²).
"""
with self._lock:
self._ensure_menu(Menu.MOT)
self._send(f"ma {motor_id} {accel:.1f}")
def get_motor_life(self) -> str:
"""Read motor lifetime / usage statistics."""
with self._lock:
self._ensure_menu(Menu.MOT)
return self._send("life")
def get_el_limits(self) -> dict[str, float]:
"""Read elevation min, max, and home angles.
Firmware returns centidegrees: ``Min: 1800 Max: 6500 Home: 6500``
Returns:
``{"min": 18.0, "max": 65.0, "home": 65.0}``
"""
with self._lock:
self._ensure_menu(Menu.MOT)
response = self._send("elminmaxhome")
result: dict[str, float] = {"min": 0.0, "max": 0.0, "home": 0.0}
min_m = re.search(r"Min:\s*(\d+)", response)
max_m = re.search(r"Max:\s*(\d+)", response)
home_m = re.search(r"Home:\s*(\d+)", response)
if min_m:
result["min"] = int(min_m.group(1)) / 100.0
if max_m:
result["max"] = int(max_m.group(1)) / 100.0
if home_m:
result["home"] = int(home_m.group(1)) / 100.0
return result
def get_step_positions(self) -> dict[str, int]:
"""Read raw step positions for both axes.
Firmware returns: ``Position[0] = 19998 Position[1] = 3116``
Returns:
``{"az_steps": int, "el_steps": int}``
"""
with self._lock:
self._ensure_menu(Menu.MOT)
response = self._send("p")
result: dict[str, int] = {"az_steps": 0, "el_steps": 0}
matches = re.findall(r"Position\[(\d)\]\s*=\s*(-?\d+)", response)
for motor_id, val in matches:
if motor_id == "0":
result["az_steps"] = int(val)
elif motor_id == "1":
result["el_steps"] = int(val)
return result
def get_pid_gains(self) -> dict[str, dict[str, float]]:
"""Read PID gains for both motor axes.
Firmware returns: ``Kp=600 Kv=60 Ki=1`` per motor.
Returns:
``{"az": {"kp": 600, "kv": 60, "ki": 1},
"el": {"kp": 250, "kv": 50, "ki": 1}}``
"""
with self._lock:
self._ensure_menu(Menu.MOT)
response = self._send("pid")
# Parse "Kp=600 Kv=60 Ki=1" patterns. The pid command without args
# shows both motors. We look for two sets of Kp/Kv/Ki values.
kp_matches = re.findall(r"Kp[=:]?\s*(\d+)", response)
kv_matches = re.findall(r"Kv[=:]?\s*(\d+)", response)
ki_matches = re.findall(r"Ki[=:]?\s*(\d+)", response)
result = {
"az": {"kp": 600.0, "kv": 60.0, "ki": 1.0},
"el": {"kp": 250.0, "kv": 50.0, "ki": 1.0},
}
if len(kp_matches) >= 2:
result["az"]["kp"] = float(kp_matches[0])
result["el"]["kp"] = float(kp_matches[1])
elif len(kp_matches) == 1:
result["az"]["kp"] = float(kp_matches[0])
if len(kv_matches) >= 2:
result["az"]["kv"] = float(kv_matches[0])
result["el"]["kv"] = float(kv_matches[1])
elif len(kv_matches) == 1:
result["az"]["kv"] = float(kv_matches[0])
if len(ki_matches) >= 2:
result["az"]["ki"] = float(ki_matches[0])
result["el"]["ki"] = float(ki_matches[1])
elif len(ki_matches) == 1:
result["az"]["ki"] = float(ki_matches[0])
return result
def set_pid_gains(self, motor_id: int, kp: float, kv: float, ki: float) -> None:
"""Write PID gains for a single motor axis.
Args:
motor_id: 0 for AZ, 1 for EL.
kp: Proportional gain.
kv: Velocity gain.
ki: Integral gain.
"""
with self._lock:
self._ensure_menu(Menu.MOT)
self._send(f"pid {motor_id} {int(kp)} {int(kv)} {int(ki)}")
def az_sweep_firmware(
self,
start_az: float,
span: float,
step_cdeg: int,
num_xponders: int,
timeout: float = 120,
) -> list[dict[str, float]]:
"""Execute a firmware-accelerated AZ sweep via azscanwxp.
Moves to *start_az* first, then runs the firmware sweep command which
handles motor movement and RSSI measurement atomically — no per-point
serial round-trips.
Args:
start_az: Starting azimuth in degrees.
span: Total sweep width in degrees.
step_cdeg: Step size in centidegrees (100 = 1.00°).
num_xponders: Number of transponders to cycle per position.
timeout: Serial read timeout for the long-running command.
Returns:
List of dicts with keys: az, rssi, lock, snr.
"""
with self._lock:
self._ensure_menu(Menu.MOT)
# Move to start position and wait for prompt.
self._send(f"a 0 {start_az}")
# Execute firmware sweep with extended timeout.
# Pass cancel event so disconnect() can interrupt the read.
response = self._proto.send_with_timeout(
f"azscanwxp 0 {span} {step_cdeg} {num_xponders}",
timeout=timeout,
cancel=self._cancel,
)
# Parse streaming output lines.
# Motor:<id> Angle:<cdeg> RSSI:<adc> Lock:<0/1> SNR:<dB>
results: list[dict[str, float]] = []
for match in re.finditer(
r"Angle:(-?\d+)\s+RSSI:(\d+)\s+Lock:(\d)\s+SNR:(-?\d+\.?\d*)",
response,
):
results.append(
{
"az": int(match.group(1)) / 100.0,
"rssi": float(match.group(2)),
"lock": float(match.group(3)),
"snr": float(match.group(4)),
}
)
logger.info("Firmware sweep returned %d points", len(results))
return results
# ------------------------------------------------------------------
# Signal (DVB>)
# ------------------------------------------------------------------
def get_rssi(self, iterations: int = 10) -> dict[str, int]:
"""Read averaged RSSI signal strength.
Returns:
``{"reads": int, "average": int, "current": int}``
"""
with self._lock:
self._ensure_menu(Menu.DVB)
response = self._send(f"rssi {iterations}")
match = re.search(
r"Reads:(\d+)\s+RSSI\[avg:\s*(\d+)\s+cur:\s*(\d+)\]",
response,
)
if match:
return {
"reads": int(match.group(1)),
"average": int(match.group(2)),
"current": int(match.group(3)),
}
raise ValueError(f"Could not parse RSSI: {response!r}")
def enable_lna(self) -> None:
"""Enable LNA in ODU mode (sets LNB to 13V)."""
with self._lock:
self._ensure_menu(Menu.DVB)
self._send("lnbdc odu")
def get_lock_status(self) -> str:
"""Read quick lock status (single-shot)."""
with self._lock:
self._ensure_menu(Menu.DVB)
return self._send("qls")
def get_dvb_config(self) -> str:
"""Read BCM hardware/firmware version."""
with self._lock:
self._ensure_menu(Menu.DVB)
return self._send("config")
def get_channel_params(self) -> str:
"""Read current channel parameters."""
with self._lock:
self._ensure_menu(Menu.DVB)
return self._send("dis")
# ------------------------------------------------------------------
# A3981
# ------------------------------------------------------------------
def get_a3981_diag(self) -> str:
"""Read A3981 diagnostic/fault status."""
with self._lock:
self._ensure_menu(Menu.A3981)
return self._send("diag")
def get_a3981_modes(self) -> dict[str, str]:
"""Read A3981 step mode, current mode, and step size.
Returns:
``{"step_mode": str, "current_mode": str, "step_size": str}``
"""
with self._lock:
self._ensure_menu(Menu.A3981)
sm_resp = self._send("sm")
cm_resp = self._send("cm")
ss_resp = self._send("ss")
return {
"step_mode": sm_resp,
"current_mode": cm_resp,
"step_size": ss_resp,
}
def get_a3981_torque(self) -> str:
"""Read A3981 torque levels."""
with self._lock:
self._ensure_menu(Menu.A3981)
return self._send("st")
# ------------------------------------------------------------------
# NVS
# ------------------------------------------------------------------
def nvs_dump(self) -> str:
"""Dump all NVS values."""
with self._lock:
self._ensure_menu(Menu.NVS)
return self._send("d")
def nvs_read(self, index: int) -> str:
"""Read a single NVS value by index."""
with self._lock:
self._ensure_menu(Menu.NVS)
return self._send(f"e {index}")
# ------------------------------------------------------------------
# ADC
# ------------------------------------------------------------------
def get_adc_rssi(self) -> str:
"""Read single-shot ADC RSSI value."""
with self._lock:
self._ensure_menu(Menu.ADC)
return self._send("rssi")
def get_board_id(self) -> str:
"""Read board identification string."""
with self._lock:
self._ensure_menu(Menu.ADC)
return self._send("bdid")
# ------------------------------------------------------------------
# OS
# ------------------------------------------------------------------
def get_firmware_id(self) -> str:
"""Read full MCU and firmware identification."""
with self._lock:
self._ensure_menu(Menu.OS)
return self._send("id")
# ------------------------------------------------------------------
# Raw / Console
# ------------------------------------------------------------------
def send_raw(self, cmd: str) -> str:
"""Send an arbitrary command and return the raw response.
After a raw command, the menu state is marked UNKNOWN because
the user may have navigated to a different submenu.
"""
with self._lock:
response = self._send(cmd)
self._menu = Menu.UNKNOWN
return response

238
tui/src/birdcage_tui/craft_client.py
View file

@ -1,238 +0,0 @@
"""Craft API client — stdlib-only HTTP client for space.warehack.ing.
Provides satellite search, pass predictions, and real-time sky positions
from the Craft orbital mechanics API. All methods are blocking — designed
to be called from @work(thread=True) workers in the TUI.
Uses urllib.request + json only (no requests/httpx dependency).
"""
import json
import logging
import time
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass, field
log = logging.getLogger(__name__)
@dataclass
class SearchResult:
"""A single result from the Craft search API."""
name: str
target_type: str
target_id: str # NORAD int for satellites, string for planets/stars/comets
score: float = 0.0
groups: list[str] = field(default_factory=list)
altitude_deg: float | None = None
@dataclass
class PassPrediction:
"""A single satellite pass prediction."""
satellite_name: str
norad_id: int
aos_time: str
aos_az: float
tca_time: str
tca_alt: float
tca_az: float
los_time: str
los_az: float
max_elevation: float
duration_seconds: int
is_visible: bool = False
magnitude: float | None = None
@dataclass
class TargetPosition:
"""Real-time position of an above-horizon object."""
name: str
target_type: str
target_id: str # NORAD int for satellites, string for planets/stars/comets
azimuth: float
altitude: float
distance_km: float = 0.0
range_rate: float = 0.0
is_above_horizon: bool = True
@dataclass
class CraftTrackingState:
"""Tracking loop state mirroring TrackingState from rotctld_server."""
status: str = "IDLE"
target_name: str = ""
azimuth: float = 0.0
elevation: float = 0.0
distance_km: float = 0.0
range_rate: float = 0.0
moves: int = 0
rate: float = 0.0
error: str = ""
_move_timestamps: list[float] = field(default_factory=list, repr=False)
def record_move(self) -> None:
self.moves += 1
now = time.monotonic()
self._move_timestamps.append(now)
cutoff = now - 30.0
self._move_timestamps = [t for t in self._move_timestamps if t > cutoff]
elapsed = now - self._move_timestamps[0]
if elapsed > 0:
self.rate = len(self._move_timestamps) / elapsed
else:
self.rate = 0.0
class CraftClient:
"""HTTP client for the Craft orbital mechanics API.
Args:
base_url: Craft API base URL.
timeout: HTTP request timeout in seconds.
"""
def __init__(
self,
base_url: str = "https://space.warehack.ing",
timeout: float = 10.0,
) -> None:
self._base_url = base_url.rstrip("/")
self._timeout = timeout
def _get(self, path: str, params: dict | None = None) -> dict:
"""Issue a GET request and return parsed JSON."""
url = f"{self._base_url}{path}"
if params:
url += "?" + urllib.parse.urlencode(params)
req = urllib.request.Request(url, headers={"Accept": "application/json"})
with urllib.request.urlopen(req, timeout=self._timeout) as resp:
return json.loads(resp.read().decode())
def health(self) -> bool:
"""Check if the Craft API is reachable."""
try:
self._get("/api/search", {"q": "ISS", "limit": "1"})
return True
except Exception:
log.debug("Craft API health check failed", exc_info=True)
return False
def search(self, query: str, limit: int = 20) -> list[SearchResult]:
"""Search the Craft object catalog.
Returns satellites, planets, stars, comets matching the query.
"""
try:
data = self._get("/api/search", {"q": query, "limit": str(limit)})
except Exception:
log.warning("Craft search failed for %r", query, exc_info=True)
return []
results = []
for item in data.get("results", []):
results.append(
SearchResult(
name=item.get("name", ""),
target_type=item.get("target_type", ""),
target_id=str(item.get("target_id", "")),
score=float(item.get("score", 0)),
groups=item.get("groups", []),
altitude_deg=item.get("altitude_deg"),
)
)
return results
def get_passes(self, norad_id: int, hours: int = 24) -> list[PassPrediction]:
"""Get pass predictions for a satellite."""
try:
data = self._get("/api/passes", {"sat": str(norad_id), "hours": str(hours)})
except Exception:
log.warning("Craft passes failed for NORAD %d", norad_id, exc_info=True)
return []
passes = []
for p in data.get("passes", []):
passes.append(
PassPrediction(
satellite_name=p.get("satellite_name", ""),
norad_id=int(p.get("norad_id", norad_id)),
aos_time=p.get("aos_time", ""),
aos_az=float(p.get("aos_az", 0)),
tca_time=p.get("tca_time", ""),
tca_alt=float(p.get("tca_alt", 0)),
tca_az=float(p.get("tca_az", 0)),
los_time=p.get("los_time", ""),
los_az=float(p.get("los_az", 0)),
max_elevation=float(p.get("max_elevation", 0)),
duration_seconds=int(p.get("duration_seconds", 0)),
is_visible=bool(p.get("is_visible", False)),
magnitude=p.get("magnitude"),
)
)
return passes
def get_next_pass(self, norad_id: int) -> PassPrediction | None:
"""Get the next upcoming pass for a satellite."""
try:
data = self._get("/api/passes/next", {"sat": str(norad_id)})
except Exception:
log.warning("Craft next pass failed for NORAD %d", norad_id, exc_info=True)
return None
passes = data.get("passes", [])
if not passes:
return None
p = passes[0]
return PassPrediction(
satellite_name=p.get("satellite_name", ""),
norad_id=int(p.get("norad_id", norad_id)),
aos_time=p.get("aos_time", ""),
aos_az=float(p.get("aos_az", 0)),
tca_time=p.get("tca_time", ""),
tca_alt=float(p.get("tca_alt", 0)),
tca_az=float(p.get("tca_az", 0)),
los_time=p.get("los_time", ""),
los_az=float(p.get("los_az", 0)),
max_elevation=float(p.get("max_elevation", 0)),
duration_seconds=int(p.get("duration_seconds", 0)),
is_visible=bool(p.get("is_visible", False)),
magnitude=p.get("magnitude"),
)
def get_visible_targets(self, min_alt: float = 0.0) -> list[TargetPosition]:
"""Get all above-horizon objects with real-time AZ/EL positions.
This is the primary tracking endpoint — returns 1000+ objects
(satellites, planets, stars, comets) with positions computed
server-side via Postgres SGP4.
"""
try:
data = self._get("/api/sky/up", {"min_alt": str(min_alt)})
except Exception:
log.warning("Craft sky/up failed", exc_info=True)
return []
targets = []
for obj in data.get("objects", []):
targets.append(
TargetPosition(
name=obj.get("name", ""),
target_type=obj.get("target_type", ""),
target_id=str(obj.get("target_id", "")),
azimuth=float(obj.get("azimuth_deg", 0)),
altitude=float(obj.get("altitude_deg", 0)),
distance_km=float(obj.get("distance_km", 0)),
range_rate=float(obj.get("range_rate_km_s", 0)),
is_above_horizon=bool(obj.get("is_above_horizon", True)),
)
)
return targets

950
tui/src/birdcage_tui/demo.py
View file

@ -1,950 +0,0 @@
"""Synthetic demo device for the Birdcage TUI.
Drop-in replacement for SerialBridge that simulates a Winegard Carryout G2
dish with motor movement, RSSI signal modeling, and canned firmware responses.
No serial hardware required.
Also provides DemoCraftClient — a duck-typed replacement for CraftClient that
returns synthetic satellite data with zero HTTP calls. Supports time-varying
LEO arcs so the tracking loop drives real pass events to the camera overlay.
"""
import contextlib
import datetime as _dt
import math
import random
import time
from enum import Enum, auto
from birdcage_tui.craft_client import PassPrediction, SearchResult, TargetPosition
class _DemoMenu(Enum):
"""Simulated firmware submenu states."""
ROOT = auto()
MOT = auto()
DVB = auto()
NVS = auto()
A3981 = auto()
ADC = auto()
OS = auto()
STEP = auto()
PEAK = auto()
EEPROM = auto()
GPIO = auto()
LATLON = auto()
DIPSWITCH = auto()
# Complete NVS dump text from firmware 02.02.48 (captured 2026-02-12).
_NVS_DUMP_TEXT = """\
Num Name Current Saved Default
---- -------------------------- ---------- ---------- ----------
0) Log ID's 0x00000007 0x00000007 0x00000007
1) Log Device 0x00000001 0x00000001 0x00000001
2) Debug 2nd Console Port 0 0 0
3) Debug 2nd Packet Port 0 0 0
4) Debug Port Connection 0 0 0
16) Pitch Deadband 0.00 0.00 0.00
17) Roll Deadband 0.00 0.00 0.00
18) Yaw Deadband 0.00 0.00 0.00
20) Disable Tracker Proc? TRUE TRUE FALSE
21) Tracker Proc Run Mode 0 0 0
22) Conical Alpha Az 200 200 200
23) Conical Alpha El 200 200 200
24) Conical Radius 1.00 1.00 1.00
25) Conical Count Max 20 20 20
26) Conical Test Drift +0 +0 +0
27) Circle RPM 120 120 120
28) Circle Pts/Rev 6 6 6
32) Conical Az Clamp 8.00 8.00 8.00
33) Conical El Clamp 8.00 8.00 8.00
35) Motor Pts/Rev 72 72 72
36) Circle Az Radius 1.00 1.00 1.00
37) Circle El Radius 1.00 1.00 1.00
38) Sleep Mode Timer Secs 420 420 420
40) Motor Type 0 0 0
41) Satellite Scan Velocity 55.00 55.00 55.00
48) Motor Spiral Velocity 55.00 55.00 55.00
49) Motor Gear Ratio 0x00000000 0x00000000 0x00000000
63) GPS Heading Threshold 1.00 1.00 1.00
64) GPS Moving Threshold 5.00 MPH 5.00 MPH 5.00 MPH
66) Spiral Signal In A Row Min +3 +3 +3
67) Spiral Signal In A Row Max +20 +20 +20
68) Signal Odd to Even Offset +0 +0 +0
69) Signal Offset 80 80 80
70) Signal Baseline Angle 65.00 65.00 65.00
71) Signal Re-Peak Degrade Percent 25 25 25
72) Gyro Sensitivity +1110 +1110 +1110
73) Gyro Filter Size +1 +1 +1
74) Gyro Calib Readings 100 100 100
75) Gyro Mount Type 1 1 1
76) Gyro Velocity Offset 4 4 4
77) Gyro Max Accel 600 600 600
80) AZ Max Vel 65.00 65.00 65.00
81) AZ Max Accel 400.00 400.00 400.00
82) AZ Home Velocity 55.00 55.00 55.00
83) AZ Steps/Rev 40000 40000 40000
84) AZ Direction +1 +1 +1
85) EL Max Vel 45.00 45.00 45.00
86) EL Max Accel 400.00 400.00 400.00
87) EL Home Velocity 45.00 45.00 45.00
88) EL Steps/Rev 24960 24960 24960
89) EL Direction +1 +1 +1
95) AZ Low current limit 0x0000ff0c 0x0000ff0c 0x0000ff0c
96) AZ High current limit 0x0000ff30 0x0000ff30 0x0000ff30
97) EL Low current limit 0x0000ff0c 0x0000ff0c 0x0000ff0c
98) EL High current limit 0x0000ff40 0x0000ff40 0x0000ff40
101) Minimum Elevation Angle 18.00 18.00 18.00
102) Maximum Elevation Angle 65.00 65.00 65.00
103) Elevation Home Angle 65.00 65.00 65.00
106) Az Stall Detect 78 78 78
107) El Stall Detect 75 75 75
108) Az Stall Samples 100 100 100
109) El Stall Samples 100 100 100
110) EL Home Current Limit 0x0000ff28 0x0000ff28 0x0000ff28
111) AZ Home Current Limit 0x0000ff40 0x0000ff40 0x0000ff40
112) Disable Dipswitch? FALSE FALSE FALSE
113) Dipswitch Value 101 101 101
114) Dipswitch Front/Rear Mount 0 0 0
115) Mount Offset Angle +0 +0 +0
118) Signal Use LNB Clamp FALSE FALSE FALSE
128) AZ PID Kp +600 +600 +600
129) AZ PID Kv +60 +60 +60
130) AZ PID Ki +1 +1 +1
131) EL PID Kp +250 +250 +250
132) EL PID Kv +50 +50 +50
133) EL PID Ki +1 +1 +1
136) AZ PWM Stall Cnt 6 6 6
137) EL PWM Stall Cnt 5 5 5
143) Tracking Number 0 0 0"""
# Parse NVS lines into a dict keyed by index for nvs_read().
_NVS_LINES: dict[int, str] = {}
for _line in _NVS_DUMP_TEXT.splitlines():
_line_stripped = _line.strip()
if _line_stripped and _line_stripped[0].isdigit():
_idx_str = _line_stripped.split(")")[0].strip()
with contextlib.suppress(ValueError):
_NVS_LINES[int(_idx_str)] = _line_stripped
# Firmware identification text matching ``os > id`` output.
_FIRMWARE_ID = """\
NVS Version: 1.02.13
System ID: TWELINCH
K60-144pin
Silicon Rev 2.4
Mask Set 4N22D
512 kBytes of P-flash
P-flash only
128 kBytes of RAM
Board Rev ID: A
Board ID: STATIONARY
Ant ID: 12-IN G2
Software version: 02.02.48
CCLK: 96000000
BCLK: 48000000
Flash Base Address: 65536
Flash Size: 458752"""
_DVB_CONFIG = """\
BCM Hardware= ID: 0x4515 VER: 0xB0
BCM Firmware= MAJOR VER: 0x71 (113) MINOR VER: 0x25 (37)
BCM Strap Config: 0x25018"""
_CHANNEL_PARAMS = """\
Power Mode: ON
Search Transponders: ON
Auto Search Mode: 1
Shuffle Mode: ON
Frequency List: Non-Stacked
Num Parameter Current Default
1 Frequency 1090640 (kHz) 974000 (kHz)
2 Symbol Rate 0 (PeakScanEnabled) 20000 (ksps)
3 Trans_Mod_CRate blind_scan blind_scan
4 Blind Scan Mode ___trb_dvb_dss_____ ___trb_dvb_dss_____
5 LNB Polarity ODU:13V ---
6 LNB Tone (ODU) off off
7 Roll-off 0.35 0.35
8 LPF Cutoff 0 (auto) 0 (MHz)
9 Carrier Offset 0 (kHz) 0 (kHz)
10 FreqSearchRange 5000 (kHz) 5000 (kHz)
11 DCII Mode dcii_qpsk_comb dcii_qpsk_comb
12 Spectral Inv scan scan
13 PScnSymRtRngMin 18000 (ksps) 18000 (ksps)
14 PScnSymRtRngMax 24000 (ksps) 24000 (ksps)
15 SignalDetectMode off off"""
_MOTOR_LIFE = """\
AZ total moves: 847
AZ total degrees: 52340.50
EL total moves: 423
EL total degrees: 18920.75
Uptime hours: 312.4"""
# Simulated satellite at AZ=200, EL=38 for RSSI modeling.
_SAT_AZ = 200.0
_SAT_EL = 38.0
_RSSI_NOISE_FLOOR = 500
_RSSI_PEAK = 2000
_RSSI_BEAM_WIDTH = 50.0 # Gaussian denominator (degrees squared)
# Motor simulation speed (degrees per second).
_MOTOR_SPEED = 10.0
class DemoDevice:
"""Synthetic demo device implementing the same interface as SerialBridge.
Simulates a Carryout G2 dish with motor movement, RSSI signal modeling,
and canned firmware responses. No serial hardware required.
"""
def __init__(self) -> None:
self._connected = False
self._engaged = True
# Current position and movement targets.
self._az = 180.0
self._el = 45.0
self._target_az = 180.0
self._target_el = 45.0
self._last_move_time = time.monotonic()
# Motor dynamics (mutable in demo for velocity/accel controls).
self._az_max_vel = 65.0
self._el_max_vel = 45.0
self._az_accel = 400.0
self._el_accel = 400.0
# Submenu tracking for console simulation.
self._menu = _DemoMenu.ROOT
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
def _update_position(self) -> None:
"""Interpolate position toward target at ~10 deg/s."""
now = time.monotonic()
dt = now - self._last_move_time
self._last_move_time = now
max_step = _MOTOR_SPEED * dt
for axis in ("az", "el"):
current = getattr(self, f"_{axis}")
target = getattr(self, f"_target_{axis}")
delta = target - current
if abs(delta) < 0.001:
continue
if abs(delta) <= max_step:
# Arrived — add a tiny settling noise.
noise = random.gauss(0.0, 0.02)
setattr(self, f"_{axis}", target + noise)
else:
direction = 1.0 if delta > 0 else -1.0
noise = random.gauss(0.0, 0.02)
setattr(self, f"_{axis}", current + direction * max_step + noise)
def _compute_rssi(self) -> float:
"""Gaussian signal model centered on the simulated satellite."""
self._update_position()
dist_sq = (self._az - _SAT_AZ) ** 2 + (self._el - _SAT_EL) ** 2
signal = _RSSI_PEAK * math.exp(-dist_sq / _RSSI_BEAM_WIDTH)
drift = math.sin(time.monotonic() / 60.0) * 50.0
return _RSSI_NOISE_FLOOR + signal + drift
@property
def _is_moving(self) -> bool:
return (
abs(self._az - self._target_az) > 0.05
or abs(self._el - self._target_el) > 0.05
)
# ------------------------------------------------------------------
# Connection
# ------------------------------------------------------------------
def connect(self, port: str = "/dev/demo", baudrate: int = 115200) -> None:
self._connected = True
self._menu = _DemoMenu.ROOT
def cancel_operation(self) -> None:
pass # Demo operations are instant, nothing to cancel.
def clear_cancel(self) -> None:
pass # No cancel state in demo mode.
def disconnect(self) -> None:
self._connected = False
self._menu = _DemoMenu.ROOT
@property
def is_connected(self) -> bool:
return self._connected
def initialize(self, skip_init: bool = False) -> None:
self._connected = True
self._menu = _DemoMenu.MOT
# ------------------------------------------------------------------
# Motor (MOT>)
# ------------------------------------------------------------------
def get_position(self) -> dict[str, float]:
self._update_position()
return {
"azimuth": round(self._az, 2),
"elevation": round(self._el, 2),
}
def move_to(self, az: float, el: float) -> None:
self._target_az = az
self._target_el = el
self._last_move_time = time.monotonic()
def move_motor(self, motor_id: int, degrees: float) -> None:
if motor_id == 0:
self._target_az = degrees
elif motor_id == 1:
self._target_el = degrees
self._last_move_time = time.monotonic()
def home_motor(self, motor_id: int) -> None:
if motor_id == 0:
self._target_az = 0.0
elif motor_id == 1:
self._target_el = 65.0
self._last_move_time = time.monotonic()
def engage(self) -> None:
self._engaged = True
def release(self) -> None:
self._engaged = False
def get_motor_list(self) -> str:
return "Motors:\n 0 - AZIMUTH: local\n 1 - ELEVATION: local"
def get_motor_dynamics(self) -> dict[str, float]:
return {
"az_max_vel": self._az_max_vel,
"el_max_vel": self._el_max_vel,
"az_accel": self._az_accel,
"el_accel": self._el_accel,
}
def set_max_velocity(self, motor_id: int, deg_per_sec: float) -> None:
if motor_id == 0:
self._az_max_vel = deg_per_sec
elif motor_id == 1:
self._el_max_vel = deg_per_sec
def set_max_acceleration(self, motor_id: int, accel: float) -> None:
if motor_id == 0:
self._az_accel = accel
elif motor_id == 1:
self._el_accel = accel
def get_motor_life(self) -> str:
return _MOTOR_LIFE
def get_el_limits(self) -> dict[str, float]:
return {"min": 18.0, "max": 65.0, "home": 65.0}
def get_step_positions(self) -> dict[str, int]:
self._update_position()
return {
"az_steps": int(self._az * 40000 / 360),
"el_steps": int(self._el * 24960 / 360),
}
def get_pid_gains(self) -> dict[str, dict[str, float]]:
return {
"az": {"kp": 600.0, "kv": 60.0, "ki": 1.0},
"el": {"kp": 250.0, "kv": 50.0, "ki": 1.0},
}
def set_pid_gains(self, motor_id: int, kp: float, kv: float, ki: float) -> None:
pass # No-op in demo mode.
def az_sweep_firmware(
self,
start_az: float,
span: float,
step_cdeg: int,
num_xponders: int,
timeout: float = 120,
) -> list[dict[str, float]]:
"""Simulate a firmware azscanwxp sweep with Gaussian signal peak."""
# Snap EL to target so 2D scans compute correct per-row signal.
# Without this, move_motor(1, el) only sets _target_el — _el stays
# stale because the position interpolator never runs mid-sweep.
self._el = self._target_el
step_deg = step_cdeg / 100.0
if step_deg <= 0:
step_deg = 1.0
results: list[dict[str, float]] = []
az = start_az
end_az = start_az + span
while az <= end_az + 1e-9:
dist_sq = (az - _SAT_AZ) ** 2 + (self._el - _SAT_EL) ** 2
signal = _RSSI_PEAK * math.exp(-dist_sq / _RSSI_BEAM_WIDTH)
rssi = _RSSI_NOISE_FLOOR + signal + random.gauss(0.0, 30.0)
locked = 1 if rssi > 1500 else 0
snr = max(0.0, (rssi - _RSSI_NOISE_FLOOR) / 50.0) + random.gauss(0.0, 0.5)
results.append(
{
"az": round(az, 2),
"rssi": round(rssi),
"lock": float(locked),
"snr": round(max(0.0, snr), 1),
}
)
az += step_deg
# Brief delay to simulate firmware execution time.
time.sleep(0.5)
self._target_az = end_az
self._last_move_time = time.monotonic()
return results
# ------------------------------------------------------------------
# Signal (DVB>)
# ------------------------------------------------------------------
def get_rssi(self, iterations: int = 10) -> dict[str, int]:
rssi = self._compute_rssi()
noise = random.gauss(0.0, 30.0)
return {
"reads": iterations,
"average": int(rssi),
"current": int(rssi + noise),
}
def enable_lna(self) -> None:
pass # No-op in demo mode.
def get_lock_status(self) -> str:
rssi = int(self._compute_rssi())
locked = 1 if rssi > 1500 else 0
return f"Lock:{locked} rssi:{rssi} cnt:0"
def get_dvb_config(self) -> str:
return _DVB_CONFIG
def get_channel_params(self) -> str:
return _CHANNEL_PARAMS
# ------------------------------------------------------------------
# A3981
# ------------------------------------------------------------------
def get_a3981_diag(self) -> str:
return "AZ DIAG: OK\nEL DIAG: OK"
def get_a3981_modes(self) -> dict[str, str]:
return {
"step_mode": "AZ Step Size Mode = AUTO\nEL Step Size Mode = AUTO",
"current_mode": "AZ: Mode = AUTO\nEL: Mode = AUTO",
"step_size": (
"KEY: FULL-16, HALF-8, QTR-4, EIGHTH-2, SIXTEENTH-1\n"
"AZ Step Size:1\n"
"EL Step Size:1"
),
}
def get_a3981_torque(self) -> str:
if self._is_moving:
return "AZ Torq:HIGH\nEL Torq:HIGH"
return "AZ Torq:LOW\nEL Torq:LOW"
# ------------------------------------------------------------------
# NVS
# ------------------------------------------------------------------
def nvs_dump(self) -> str:
return _NVS_DUMP_TEXT
def nvs_read(self, index: int) -> str:
line = _NVS_LINES.get(index)
if line:
return line
return f"NVS index {index} not found"
# ------------------------------------------------------------------
# ADC
# ------------------------------------------------------------------
def get_adc_rssi(self) -> str:
rssi = self._compute_rssi()
return str(int(rssi))
def get_board_id(self) -> str:
return "STATIONARY"
# ------------------------------------------------------------------
# OS
# ------------------------------------------------------------------
def get_firmware_id(self) -> str:
return _FIRMWARE_ID
# ------------------------------------------------------------------
# Raw / Console
# ------------------------------------------------------------------
def send_raw(self, cmd: str) -> str:
"""Simulate firmware console with basic submenu tracking."""
cmd_stripped = cmd.strip().lower()
# Submenu navigation.
if cmd_stripped == "q":
self._menu = _DemoMenu.ROOT
return "TRK>"
_enter_map: dict[str, _DemoMenu] = {
"mot": _DemoMenu.MOT,
"dvb": _DemoMenu.DVB,
"nvs": _DemoMenu.NVS,
"a3981": _DemoMenu.A3981,
"adc": _DemoMenu.ADC,
"os": _DemoMenu.OS,
"step": _DemoMenu.STEP,
"peak": _DemoMenu.PEAK,
"eeprom": _DemoMenu.EEPROM,
"gpio": _DemoMenu.GPIO,
"latlon": _DemoMenu.LATLON,
"dipswitch": _DemoMenu.DIPSWITCH,
}
if cmd_stripped in _enter_map and self._menu == _DemoMenu.ROOT:
self._menu = _enter_map[cmd_stripped]
prompt = cmd_stripped.upper() + ">"
return prompt
# Context-dependent responses.
if self._menu == _DemoMenu.MOT:
return self._handle_mot(cmd_stripped)
if self._menu == _DemoMenu.DVB:
return self._handle_dvb(cmd_stripped)
if self._menu == _DemoMenu.NVS:
return self._handle_nvs(cmd_stripped)
if self._menu == _DemoMenu.A3981:
return self._handle_a3981(cmd_stripped)
if self._menu == _DemoMenu.ADC:
return self._handle_adc(cmd_stripped)
if self._menu == _DemoMenu.OS:
return self._handle_os(cmd_stripped)
if self._menu == _DemoMenu.ROOT:
return self._handle_root(cmd_stripped)
return f"Unknown command: {cmd}\nTRK>"
def _handle_root(self, cmd: str) -> str:
if cmd in ("?", "help"):
return (
"Available commands:\n"
" a3981 adc dipswitch dvb eeprom gpio\n"
" latlon mot nvs os peak step\n"
" q reboot stow\n"
"TRK>"
)
if cmd == "reboot":
return "Rebooting...\nApplication Starting Kinetis PCB...\nTRK>"
return f"Unknown command: {cmd}\nTRK>"
def _handle_mot(self, cmd: str) -> str:
if cmd in ("?", "help"):
return (
"Available commands:\n"
" a azscan azscanwxp e ela2s elminmaxhome\n"
" els2a g h l life ma motorboth motorlife\n"
" mv p pid r sd sp sw v vms w\n"
"MOT>"
)
if cmd == "a":
self._update_position()
return f" Angle[0] = {self._az:.2f}\n Angle[1] = {self._el:.2f}\nMOT>"
if cmd == "l":
return "Motors:\n 0 - AZIMUTH: local\n 1 - ELEVATION: local\nMOT>"
if cmd == "e":
self._engaged = True
return "Motors engaged\nMOT>"
if cmd == "r":
self._engaged = False
return "Motors released\nMOT>"
if cmd == "elminmaxhome":
return "Min: 1800 Max: 6500 Home: 6500\nMOT>"
if cmd == "life":
return _MOTOR_LIFE + "\nMOT>"
if cmd.startswith("a "):
parts = cmd.split()
if len(parts) >= 3:
motor_id = int(parts[1])
degrees = float(parts[2])
if motor_id == 0:
self._target_az = degrees
elif motor_id == 1:
self._target_el = degrees
self._last_move_time = time.monotonic()
return f" Angle = {degrees:.2f}\nMOT>"
return "Invalid parameters\nMOT>"
if cmd.startswith("h "):
parts = cmd.split()
if len(parts) >= 2:
motor_id = int(parts[1])
self.home_motor(motor_id)
return f"Homing motor {motor_id}\nMOT>"
return "Invalid parameters\nMOT>"
if cmd == "mv" or cmd.startswith("mv "):
parts = cmd.split()
if len(parts) >= 3:
self.set_max_velocity(int(parts[1]), float(parts[2]))
return f"Max Vel [{parts[1]}] = {float(parts[2]):.1f}\nMOT>"
return (
f"Max Vel [0] = {self._az_max_vel:.1f}"
f" Max Vel [1] = {self._el_max_vel:.1f}\nMOT>"
)
if cmd == "ma" or cmd.startswith("ma "):
parts = cmd.split()
if len(parts) >= 3:
self.set_max_acceleration(int(parts[1]), float(parts[2]))
return f"Accel[{parts[1]}] = {float(parts[2]):.1f}\nMOT>"
return (
f"Accel[0] = {self._az_accel:.1f}"
f" Accel[1] = {self._el_accel:.1f}\nMOT>"
)
if cmd == "p":
self._update_position()
az_steps = int(self._az * 40000 / 360)
el_steps = int(self._el * 24960 / 360)
return f"Position[0] = {az_steps} Position[1] = {el_steps}\nMOT>"
if cmd == "pid" or cmd.startswith("pid "):
parts = cmd.split()
if len(parts) == 1:
return (
"Motor 0: Kp=600 Kv=60 Ki=1\nMotor 1: Kp=250 Kv=50 Ki=1\nMOT>"
)
elif len(parts) >= 4:
motor_id = parts[1]
return f"PID set for motor {motor_id}\nMOT>"
return "Usage: pid [motor] [Kp] [Kv] [Ki]\nMOT>"
return f"Unknown command: {cmd}\nMOT>"
def _handle_dvb(self, cmd: str) -> str:
if cmd in ("?", "help"):
return (
"Available commands:\n"
" agc config def diag dis e freqs\n"
" lnbdc lnbv ls man msw nid pwr\n"
" qls range rssi shuf snr srch srch_mode\n"
" stats t table tablex tabto to\n"
"DVB>"
)
if cmd.startswith("rssi"):
rssi_val = int(self._compute_rssi())
parts = cmd.split()
iters = int(parts[1]) if len(parts) > 1 else 10
noise = random.gauss(0.0, 30.0)
cur = int(rssi_val + noise)
return (
f"iterations:{iters} interval(msec):20\n"
f" Reads:{iters} RSSI[avg: {rssi_val} cur: {cur}]\n"
"DVB>"
)
if cmd == "config":
return _DVB_CONFIG + "\nDVB>"
if cmd == "dis":
return _CHANNEL_PARAMS + "\nDVB>"
if cmd == "lnbdc odu":
return "Enabled LNB ODU 13V\nDVB>"
if cmd == "qls":
rssi_val = int(self._compute_rssi())
locked = 1 if rssi_val > 1500 else 0
return f"Lock:{locked} rssi:{rssi_val} cnt:0\nDVB>"
return f"Unknown command: {cmd}\nDVB>"
def _handle_nvs(self, cmd: str) -> str:
if cmd in ("?", "help"):
return "Available commands:\n d e s\nNVS>"
if cmd == "d":
return _NVS_DUMP_TEXT + "\nNVS>"
if cmd == "s":
return "NVS saved\nNVS>"
if cmd.startswith("e "):
parts = cmd.split()
if len(parts) >= 2:
try:
idx = int(parts[1])
line = _NVS_LINES.get(idx)
if line:
return line + "\nNVS>"
return f"NVS index {idx} not found\nNVS>"
except ValueError:
pass
return "Invalid parameters\nNVS>"
return f"Unknown command: {cmd}\nNVS>"
def _handle_a3981(self, cmd: str) -> str:
if cmd in ("?", "help"):
return "Available commands:\n cm diag reset sm ss st\nA3981>"
if cmd == "diag":
return "AZ DIAG: OK\nEL DIAG: OK\nA3981>"
if cmd == "sm":
return "AZ Step Size Mode = AUTO\nEL Step Size Mode = AUTO\nA3981>"
if cmd == "cm":
return "AZ: Mode = AUTO\nEL: Mode = AUTO\nA3981>"
if cmd == "ss":
return (
"KEY: FULL-16, HALF-8, QTR-4, EIGHTH-2, SIXTEENTH-1\n"
"AZ Step Size:1\n"
"EL Step Size:1\n"
"A3981>"
)
if cmd == "st":
if self._is_moving:
return "AZ Torq:HIGH\nEL Torq:HIGH\nA3981>"
return "AZ Torq:LOW\nEL Torq:LOW\nA3981>"
if cmd == "reset":
return "Az/El A3981 Faults Reset.\nA3981>"
return f"Unknown command: {cmd}\nA3981>"
def _handle_adc(self, cmd: str) -> str:
if cmd in ("?", "help"):
return "Available commands:\n bdid bdrevid m rssi scan\nADC>"
if cmd == "rssi":
return str(int(self._compute_rssi())) + "\nADC>"
if cmd == "bdid":
return "STATIONARY\nADC>"
if cmd == "bdrevid":
return "A\nADC>"
return f"Unknown command: {cmd}\nADC>"
def _handle_os(self, cmd: str) -> str:
if cmd in ("?", "help"):
return "Available commands:\n id reboot\nOS>"
if cmd == "id":
return _FIRMWARE_ID + "\nOS>"
if cmd == "reboot":
return "Rebooting...\nApplication Starting Kinetis PCB...\nTRK>"
return f"Unknown command: {cmd}\nOS>"
# ------------------------------------------------------------------
# DemoCraftClient — offline Craft API replacement
# ------------------------------------------------------------------
# Canned satellite catalog for search results.
_DEMO_CATALOG: list[dict] = [
{
"name": "ISS (ZARYA)",
"type": "satellite",
"id": "25544",
"groups": ["stations"],
},
{
"name": "NOAA 19",
"type": "satellite",
"id": "33591",
"groups": ["weather"],
},
{
"name": "SO-50 (SAUDISAT 1C)",
"type": "satellite",
"id": "27607",
"groups": ["amateur"],
},
{
"name": "TEVEL-2",
"type": "satellite",
"id": "50988",
"groups": ["amateur"],
},
{
"name": "AO-91 (FOX-1B)",
"type": "satellite",
"id": "43017",
"groups": ["amateur"],
},
{
"name": "Moon",
"type": "celestial",
"id": "moon",
"groups": ["solar-system"],
},
{
"name": "Sun",
"type": "celestial",
"id": "sun",
"groups": ["solar-system"],
},
{
"name": "Jupiter",
"type": "celestial",
"id": "jupiter",
"groups": ["solar-system"],
},
]
# LEO arc parameters: (phase_offset_minutes, period_minutes, max_el)
_LEO_ARCS: dict[str, tuple[float, float, float]] = {
"25544": (0.0, 10.0, 55.0), # ISS — primary demo target
"33591": (3.0, 9.0, 42.0), # NOAA 19
"27607": (5.0, 8.5, 38.0), # SO-50
"50988": (7.0, 11.0, 48.0), # TEVEL-2
"43017": (2.0, 9.5, 35.0), # AO-91
}
def _leo_position(target_id: str, t: float) -> tuple[float, float]:
"""Compute time-varying AZ/EL for a simulated LEO satellite.
The arc traces: rise east (AZ ~90, EL 0) -> TCA (~180, max_el)
-> set west (AZ ~270, EL 0) over one period, then resets.
Returns (az, el) where el < 0 means below horizon.
"""
offset, period, max_el = _LEO_ARCS.get(target_id, (0.0, 10.0, 40.0))
period_sec = period * 60.0
phase = ((t + offset * 60.0) % period_sec) / period_sec # 0.0 → 1.0
# Visible window: phase 0.0-0.5 = above horizon, 0.5-1.0 = below
if phase > 0.5:
return 0.0, -10.0 # Below horizon
# Map 0→0.5 to AZ 90→270, EL 0→max→0 (sine arc)
arc_phase = phase / 0.5 # 0.0 → 1.0 through the visible pass
az = 90.0 + 180.0 * arc_phase
el = max_el * math.sin(math.pi * arc_phase)
return az, el
def _celestial_position(target_id: str, t: float) -> tuple[float, float]:
"""Slow-drift positions for celestial bodies."""
if target_id == "moon":
az = 145.0 + 0.5 * math.sin(t / 600.0) * (t / 60.0 % 10)
el = 32.0 + 3.0 * math.sin(t / 900.0)
return az, max(el, 5.0)
if target_id == "sun":
az = 210.0 + 0.3 * (t / 60.0 % 15)
el = 45.0 + 5.0 * math.sin(t / 1200.0)
return az, max(el, 10.0)
# Jupiter — nearly fixed
return 255.0, 28.0
class DemoCraftClient:
"""Offline replacement for CraftClient returning synthetic orbital data.
Duck-typed to match CraftClient's interface. No HTTP calls are made.
LEO satellites trace realistic arcs using time.monotonic() so the
tracking loop sees genuine AOS/TCA/LOS transitions.
"""
def __init__(self) -> None:
self._t0 = time.monotonic()
def health(self) -> bool:
return True
def search(self, query: str, limit: int = 20) -> list[SearchResult]:
q_lower = query.lower()
results = []
for entry in _DEMO_CATALOG:
if q_lower in entry["name"].lower():
results.append(
SearchResult(
name=entry["name"],
target_type=entry["type"],
target_id=entry["id"],
score=1.0,
groups=entry["groups"],
)
)
if len(results) >= limit:
break
return results
def get_passes(self, norad_id: int, hours: int = 24) -> list[PassPrediction]:
now = _dt.datetime.now(tz=_dt.UTC)
name = "Unknown"
for entry in _DEMO_CATALOG:
if entry["id"] == str(norad_id):
name = entry["name"]
break
arc_params = _LEO_ARCS.get(str(norad_id), (0.0, 10.0, 40.0))
_, period, max_el = arc_params
passes = []
for i in range(4):
aos = now + _dt.timedelta(minutes=30 * (i + 1))
tca = aos + _dt.timedelta(minutes=period / 2)
los = aos + _dt.timedelta(minutes=period)
duration = int(period * 60)
passes.append(
PassPrediction(
satellite_name=name,
norad_id=norad_id,
aos_time=aos.strftime("%Y-%m-%dT%H:%M:%S"),
aos_az=90.0 + random.uniform(-10, 10),
tca_time=tca.strftime("%Y-%m-%dT%H:%M:%S"),
tca_alt=max_el,
tca_az=180.0 + random.uniform(-15, 15),
los_time=los.strftime("%Y-%m-%dT%H:%M:%S"),
los_az=270.0 + random.uniform(-10, 10),
max_elevation=max_el,
duration_seconds=duration,
is_visible=i < 2,
)
)
return passes
def get_next_pass(self, norad_id: int) -> PassPrediction | None:
passes = self.get_passes(norad_id)
return passes[0] if passes else None
def get_visible_targets(self, min_alt: float = 0.0) -> list[TargetPosition]:
t = time.monotonic() - self._t0
targets = []
for entry in _DEMO_CATALOG:
tid = entry["id"]
ttype = entry["type"]
if ttype == "satellite" and tid in _LEO_ARCS:
az, el = _leo_position(tid, t)
elif ttype == "celestial":
az, el = _celestial_position(tid, t)
else:
continue
if el < min_alt:
continue
# Synthetic distance/range-rate for LEO targets
if ttype == "satellite":
dist = 400.0 + 200.0 * math.cos(math.pi * el / 90.0)
rr = -2.0 + 4.0 * math.sin(t / 120.0)
else:
dist = 384400.0 if tid == "moon" else 0.0
rr = 0.0
targets.append(
TargetPosition(
name=entry["name"],
target_type=ttype,
target_id=tid,
azimuth=round(az, 2),
altitude=round(el, 2),
distance_km=round(dist, 1),
range_rate=round(rr, 3),
)
)
return targets

2
tui/src/birdcage_tui/screens/control.py
View file

@ -9,6 +9,7 @@ import contextlib
import logging
import threading
from birdcage.craft_client import CraftClient, CraftTrackingState
from textual import work
from textual.app import ComposeResult
from textual.binding import Binding
@ -16,7 +17,6 @@ from textual.containers import Container, Horizontal, Vertical
from textual.widgets import Button, ContentSwitcher, Input, Static
from textual.worker import Worker
from birdcage_tui.craft_client import CraftClient, CraftTrackingState
from birdcage_tui.rotctld_server import TrackingState, TuiRotctldServer
from birdcage_tui.widgets.compass_rose import CompassRose
from birdcage_tui.widgets.craft_panel import CraftPanel

2
tui/tests/test_craft_mode.py
View file

@ -13,7 +13,7 @@ from unittest.mock import MagicMock
import pytest
from birdcage_tui.app import BirdcageApp
from birdcage_tui.craft_client import PassPrediction, SearchResult, TargetPosition
from birdcage.craft_client import PassPrediction, SearchResult, TargetPosition
from birdcage_tui.screens.control import ControlScreen
from birdcage_tui.widgets.craft_panel import CraftPanel, CraftTrackingStatus