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

Wire firmware-accelerated AZ sweep via azscanwxp

Browse source 
Adds send_with_timeout() to CarryoutG2Protocol for long-running
commands, and az_sweep_firmware() to both SerialBridge and DemoDevice.
Sweep and Sky Map modes now try the firmware path first (single
azscanwxp command, streaming results) and fall back to software
step-dwell-measure on error or when "Software mode" checkbox is
checked. Software sweep fixed to set EL once and move AZ only.
This commit is contained in:
Ryan Malloy 2026-02-14 16:40:53 -07:00
parent ba8859cc31
commit 3cd6424168
4 changed files with 973 additions and 69 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -66,6 +66,32 @@ class SerialBridge:
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
# ------------------------------------------------------------------
@ -82,6 +108,21 @@ class SerialBridge:
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.
@ -114,7 +155,7 @@ class SerialBridge:
with self._lock:
self._proto.connect(port, baudrate)
self._connected = True
self._menu = Menu.UNKNOWN
self._menu = self._detect_menu()
def disconnect(self) -> None:
"""Close the serial connection."""
@ -139,7 +180,8 @@ class SerialBridge:
with self._lock:
if not skip_init:
self._proto.initialize()
self._menu = Menu.MOT # initialize() ends in MOT>
self._menu = Menu.MOT # initialize() ends in MOT>
# else: leave _menu as whatever connect() detected
# ------------------------------------------------------------------
# Motor (MOT>)
@ -296,6 +338,116 @@ class SerialBridge:
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.
response = self._proto.send_with_timeout(
f"azscanwxp 0 {span} {step_cdeg} {num_xponders}",
timeout=timeout,
)
# 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>)
# ------------------------------------------------------------------