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

Initial travler-rotor library scaffolding

Browse source 
Extract Gabe Emerson's Trav'ler rotor scripts into a proper Python
library with firmware protocol abstraction (HAL 2.05 + HAL 0.0.00),
Hamlib rotctld TCP server, Click CLI, and isolated leap-frog algorithm
with the elevation copy-paste bug fixed.
This commit is contained in:
Ryan Malloy 2026-02-11 04:10:17 -07:00
commit c93bbef26d
17 changed files with 1374 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

22
src/travler_rotor/__init__.py Normal file
View file

@ -0,0 +1,22 @@
"""travler-rotor: Control Winegard Trav'ler satellite dishes via RS-485."""
from travler_rotor.antenna import AntennaConfig, TravlerAntenna
from travler_rotor.leapfrog import apply_leapfrog
from travler_rotor.protocol import (
FirmwareProtocol,
HAL000Protocol,
HAL205Protocol,
Position,
)
from travler_rotor.rotctld import RotctldServer
__all__ = [
"AntennaConfig",
"FirmwareProtocol",
"HAL000Protocol",
"HAL205Protocol",
"Position",
"RotctldServer",
"TravlerAntenna",
"apply_leapfrog",
]

129
src/travler_rotor/antenna.py Normal file
View file

@ -0,0 +1,129 @@
"""High-level antenna control for the Winegard Trav'ler.
This is the main interface that consumers (CLI, rotctld server, future MCP
server) should use. It wraps the firmware protocol with position tracking,
leap-frog compensation, and motor command alternation.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from travler_rotor.leapfrog import apply_leapfrog
from travler_rotor.protocol import (
MOTOR_AZIMUTH,
MOTOR_ELEVATION,
FirmwareProtocol,
Position,
)
logger = logging.getLogger(__name__)
@dataclass
class AntennaConfig:
"""Configuration for a Trav'ler antenna connection."""
port: str = "/dev/ttyUSB0"
baudrate: int = 57600
min_elevation: float = 15.0
leapfrog_enabled: bool = True
class TravlerAntenna:
"""High-level interface to a Winegard Trav'ler dish.
Manages the full lifecycle: connect, initialize (boot + search kill),
track positions, and move with leap-frog compensation.
"""
def __init__(
self, protocol: FirmwareProtocol, config: AntennaConfig | None = None
) -> None:
self._protocol = protocol
self._config = config or AntennaConfig()
self._last_position = Position(azimuth=0.0, elevation=0.0)
self._move_count: int = 0
@property
def config(self) -> AntennaConfig:
return self._config
@property
def is_connected(self) -> bool:
return self._protocol.is_connected
def connect(self) -> None:
"""Open serial connection to the dish."""
self._protocol.connect(self._config.port, self._config.baudrate)
def disconnect(self) -> None:
"""Close serial connection."""
self._protocol.disconnect()
def initialize(self) -> None:
"""Full boot sequence: connect (if needed), wait for calibration, kill search.
After this completes the dish is ready to accept motor commands.
"""
if not self.is_connected:
self.connect()
def _boot_status(line: str) -> None:
logger.info("Boot: %s", line)
self._protocol.initialize(callback=_boot_status)
self._protocol.enter_motor_menu()
logger.info("Antenna initialized and ready")
def get_position(self) -> Position:
"""Query and return current dish position."""
pos = self._protocol.get_position()
self._last_position = pos
return pos
def move_to(self, azimuth: float, elevation: float) -> None:
"""Move dish to the specified AZ/EL with leap-frog compensation.
The elevation floor from config.min_elevation is enforced — HAL 2.05
won't reliably go below ~15 degrees with direct motor commands.
Motor commands are alternated (AZ-first on even moves, EL-first on
odd) to prevent one axis from starving the other on the shared
serial bus.
"""
# Apply leap-frog prediction if enabled
if self._config.leapfrog_enabled:
azimuth, elevation = apply_leapfrog(
azimuth,
elevation,
self._last_position.azimuth,
self._last_position.elevation,
)
# Enforce elevation floor
if elevation < self._config.min_elevation:
elevation = self._config.min_elevation
logger.debug(
"Moving to AZ=%.1f EL=%.1f (move #%d)",
azimuth,
elevation,
self._move_count,
)
# Alternate motor command order to avoid one axis starving
if self._move_count % 2 == 0:
self._protocol.move_motor(MOTOR_AZIMUTH, azimuth)
self._protocol.move_motor(MOTOR_ELEVATION, elevation)
else:
self._protocol.move_motor(MOTOR_ELEVATION, elevation)
self._protocol.move_motor(MOTOR_AZIMUTH, azimuth)
self._move_count += 1
def stop(self) -> None:
"""Stop tracking and reset move counter."""
self._move_count = 0
logger.info("Tracking stopped")

202
src/travler_rotor/cli.py Normal file
View file

@ -0,0 +1,202 @@
"""CLI entry point for travler-rotor.
Provides subcommands for initialization, position queries, manual moves,
and running a full rotctld-compatible server for Gpredict integration.
"""
from __future__ import annotations
import logging
import sys
import click
from travler_rotor.antenna import AntennaConfig, TravlerAntenna
from travler_rotor.protocol import get_protocol
from travler_rotor.rotctld import RotctldServer
def _setup_logging(verbose: bool) -> None:
level = logging.DEBUG if verbose else logging.INFO
logging.basicConfig(
level=level,
format="%(asctime)s %(levelname)-8s %(name)s: %(message)s",
datefmt="%H:%M:%S",
)
def _build_antenna(port: str, firmware: str, **config_kwargs) -> TravlerAntenna:
"""Create a TravlerAntenna from CLI options."""
protocol = get_protocol(firmware)
config = AntennaConfig(port=port, **config_kwargs)
return TravlerAntenna(protocol, config)
@click.group()
@click.version_option(package_name="travler-rotor")
@click.option("-v", "--verbose", is_flag=True, help="Enable debug logging.")
def main(verbose: bool) -> None:
"""Control a Winegard Trav'ler satellite dish via RS-485."""
_setup_logging(verbose)
@main.command()
@click.option(
"--port",
envvar="TRAVLER_PORT",
default="/dev/ttyUSB0",
show_default=True,
help="Serial port for the RS-485 adapter.",
)
@click.option(
"--firmware",
envvar="TRAVLER_FIRMWARE",
default="hal205",
show_default=True,
type=click.Choice(["hal205", "hal000"], case_sensitive=False),
help="Firmware version on the dish.",
)
def init(port: str, firmware: str) -> None:
"""Initialize the antenna: wait for boot, kill satellite search."""
antenna = _build_antenna(port, firmware)
try:
click.echo(f"Connecting to {port} (firmware: {firmware})...")
antenna.initialize()
click.echo("Antenna initialized and ready.")
except KeyboardInterrupt:
click.echo("\nInterrupted.")
finally:
antenna.disconnect()
@main.command()
@click.option(
"--port",
envvar="TRAVLER_PORT",
default="/dev/ttyUSB0",
show_default=True,
help="Serial port for the RS-485 adapter.",
)
@click.option(
"--firmware",
envvar="TRAVLER_FIRMWARE",
default="hal205",
show_default=True,
type=click.Choice(["hal205", "hal000"], case_sensitive=False),
help="Firmware version on the dish.",
)
@click.option(
"--host",
envvar="TRAVLER_LISTEN_HOST",
default="127.0.0.1",
show_default=True,
help="Address to listen on for rotctld connections.",
)
@click.option(
"--listen-port",
envvar="TRAVLER_LISTEN_PORT",
default=4533,
show_default=True,
type=int,
help="TCP port for rotctld protocol.",
)
@click.option(
"--skip-init",
is_flag=True,
help="Skip boot wait and search kill (dish already initialized).",
)
def serve(
port: str, firmware: str, host: str, listen_port: int, skip_init: bool
) -> None:
"""Run a rotctld-compatible TCP server for Gpredict."""
antenna = _build_antenna(port, firmware)
try:
if skip_init:
click.echo(f"Connecting to {port} (skipping init)...")
antenna.connect()
antenna._protocol.enter_motor_menu()
else:
click.echo(f"Initializing antenna on {port}...")
antenna.initialize()
server = RotctldServer(antenna, host=host, port=listen_port)
click.echo(f"rotctld server listening on {host}:{listen_port}")
click.echo("Ctrl-C to stop")
server.serve_forever()
except KeyboardInterrupt:
click.echo("\nShutting down...")
finally:
antenna.disconnect()
@main.command()
@click.option(
"--port",
envvar="TRAVLER_PORT",
default="/dev/ttyUSB0",
show_default=True,
help="Serial port for the RS-485 adapter.",
)
@click.option(
"--firmware",
envvar="TRAVLER_FIRMWARE",
default="hal205",
show_default=True,
type=click.Choice(["hal205", "hal000"], case_sensitive=False),
help="Firmware version on the dish.",
)
def pos(port: str, firmware: str) -> None:
"""Query and print the current dish position."""
antenna = _build_antenna(port, firmware)
try:
antenna.connect()
antenna._protocol.enter_motor_menu()
position = antenna.get_position()
click.echo(f"AZ: {position.azimuth:.1f}")
click.echo(f"EL: {position.elevation:.1f}")
if position.skew is not None:
click.echo(f"SK: {position.skew:.1f}")
except Exception as exc:
click.echo(f"Error: {exc}", err=True)
sys.exit(1)
finally:
antenna.disconnect()
@main.command()
@click.option(
"--port",
envvar="TRAVLER_PORT",
default="/dev/ttyUSB0",
show_default=True,
help="Serial port for the RS-485 adapter.",
)
@click.option(
"--firmware",
envvar="TRAVLER_FIRMWARE",
default="hal205",
show_default=True,
type=click.Choice(["hal205", "hal000"], case_sensitive=False),
help="Firmware version on the dish.",
)
@click.option("--az", required=True, type=float, help="Target azimuth (degrees).")
@click.option("--el", required=True, type=float, help="Target elevation (degrees).")
@click.option(
"--no-leapfrog",
is_flag=True,
help="Disable leap-frog compensation for this move.",
)
def move(port: str, firmware: str, az: float, el: float, no_leapfrog: bool) -> None:
"""Move the dish to a specific AZ/EL position."""
antenna = _build_antenna(port, firmware, leapfrog_enabled=not no_leapfrog)
try:
antenna.connect()
antenna._protocol.enter_motor_menu()
click.echo(f"Moving to AZ={az:.1f} EL={el:.1f}...")
antenna.move_to(az, el)
click.echo("Command sent.")
except Exception as exc:
click.echo(f"Error: {exc}", err=True)
sys.exit(1)
finally:
antenna.disconnect()

50
src/travler_rotor/leapfrog.py Normal file
View file

@ -0,0 +1,50 @@
"""Leap-frog prediction algorithm for mechanical lag compensation.
The Trav'ler dish has inherent mechanical lag — by the time the motors
reach the commanded position, a satellite in motion has already moved on.
This module applies a small predictive overshoot so the dish "leaps ahead"
of the target, reducing tracking error during a pass.
"""
def apply_leapfrog(
target_az: float,
target_el: float,
current_az: float,
current_el: float,
) -> tuple[float, float]:
"""Apply predictive overshoot to compensate for mechanical lag.
For each axis, if the delta exceeds a threshold, the target is nudged
further in the direction of travel. This keeps the dish slightly ahead
of fast-moving satellites.
Args:
target_az: Desired azimuth from tracking software (degrees).
target_el: Desired elevation from tracking software (degrees).
current_az: Last-known azimuth of the dish (degrees).
current_el: Last-known elevation of the dish (degrees).
Returns:
Adjusted (azimuth, elevation) with overshoot applied.
Note:
The original upstream code had a copy-paste bug where the elevation
delta adjustments modified target_az instead of target_el.
See docs/bugs.md for details.
"""
# Azimuth compensation
az_delta = target_az - current_az
if abs(az_delta) > 2:
target_az += 1.0 if az_delta > 0 else -1.0
elif abs(az_delta) > 1:
target_az += 0.5 if az_delta > 0 else -0.5
# Elevation compensation (bug fix: original modified target_az here)
el_delta = target_el - current_el
if abs(el_delta) > 2:
target_el += 1.0 if el_delta > 0 else -1.0
elif abs(el_delta) > 1:
target_el += 0.5 if el_delta > 0 else -0.5
return target_az, target_el

244
src/travler_rotor/protocol.py Normal file
View file

@ -0,0 +1,244 @@
"""Firmware protocol abstraction for Winegard Trav'ler RS-485 communication.
Each firmware version (HAL) uses slightly different serial commands, boot
signals, and submenu structures. This module defines an abstract protocol
and concrete implementations so the rest of the library doesn't care which
firmware is on the other end of the wire.
"""
from __future__ import annotations
import logging
import re
import time
from abc import ABC, abstractmethod
from collections.abc import Callable
from dataclasses import dataclass
import serial
logger = logging.getLogger(__name__)
# Motor IDs used in "a <id> <degrees>" commands
MOTOR_AZIMUTH = 0
MOTOR_ELEVATION = 1
@dataclass
class Position:
"""Current dish orientation."""
azimuth: float
elevation: float
skew: float | None = None
class FirmwareProtocol(ABC):
"""Abstract base for Winegard firmware communication over RS-485."""
def __init__(self) -> None:
self._serial: serial.Serial | None = None
@property
def is_connected(self) -> bool:
return self._serial is not None and self._serial.is_open
def connect(self, port: str, baudrate: int = 57600) -> None:
"""Open the RS-485 serial connection."""
self._serial = serial.Serial(
port=port,
baudrate=baudrate,
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,
bytesize=serial.EIGHTBITS,
timeout=1,
)
logger.info("Connected on %s at %d baud", port, baudrate)
def disconnect(self) -> None:
"""Close the serial connection."""
if self._serial and self._serial.is_open:
self.reset_to_root()
self._serial.close()
logger.info("Disconnected")
self._serial = None
def _write(self, cmd: str) -> None:
"""Send a command string followed by carriage return."""
if not self._serial:
raise RuntimeError("Not connected")
self._serial.write(f"{cmd}\r".encode("ascii"))
def _read(self, size: int = 200) -> str:
"""Read up to `size` bytes from serial, decode with error tolerance."""
if not self._serial:
raise RuntimeError("Not connected")
self._serial.flush()
return self._serial.read(size).decode(errors="ignore").strip()
def _readline(self) -> str:
"""Read a single line from serial."""
if not self._serial:
raise RuntimeError("Not connected")
return self._serial.readline().decode(errors="ignore").strip()
def reset_to_root(self) -> None:
"""Return to the firmware root menu."""
self._write("q")
self._write("") # clear prompt
@abstractmethod
def initialize(self, callback: Callable[[str], None] | None = None) -> None:
"""Wait for boot to complete and kill the satellite search task.
Args:
callback: Optional function called with each status line
received during boot (useful for progress display).
"""
@abstractmethod
def enter_motor_menu(self) -> None:
"""Navigate into the motor control submenu."""
def get_position(self) -> Position:
"""Query the dish for its current AZ/EL/SK position."""
self._write("a")
reply = self._read()
az_match = re.search(r"AZ =\s*(\d+\.\d+)", reply)
el_match = re.search(r"EL =\s*(\d+\.\d+)", reply)
sk_match = re.search(r"SK =\s*(\d+\.\d+)", reply)
if not az_match or not el_match:
raise ValueError(f"Could not parse position from: {reply!r}")
return Position(
azimuth=float(az_match.group(1)),
elevation=float(el_match.group(1)),
skew=float(sk_match.group(1)) if sk_match else None,
)
def move_motor(self, motor_id: int, degrees: float) -> None:
"""Command a single motor to an absolute position.
Args:
motor_id: MOTOR_AZIMUTH (0) or MOTOR_ELEVATION (1).
degrees: Target angle in degrees.
"""
self._write(f"a {motor_id} {degrees}")
@abstractmethod
def kill_search(self) -> None:
"""Cancel the firmware's automatic TV satellite search."""
class HAL205Protocol(FirmwareProtocol):
"""HAL 2.05.003 firmware.
Boot signals: "NoGPS" or "No LNB Voltage"
Motor submenu: "motor"
Search kill: ngsearch -> s -> q
"""
BOOT_SIGNALS = ("NoGPS", "No LNB Voltage")
MOTOR_COMMAND = "motor"
def initialize(self, callback: Callable[[str], None] | None = None) -> None:
logger.info("Waiting for HAL 2.05 boot (ensure IDU is powered on)...")
while True:
line = self._readline()
if not line:
continue
if callback:
callback(line)
logger.debug("Boot: %s", line)
if any(signal in line for signal in self.BOOT_SIGNALS):
logger.info("Boot complete — homing finished")
break
self.kill_search()
self.reset_to_root()
def kill_search(self) -> None:
self.reset_to_root()
self._write("ngsearch")
time.sleep(0.2)
self._write("s")
time.sleep(0.2)
self._write("q")
self._write("")
logger.info("Search task cancelled")
def enter_motor_menu(self) -> None:
self.reset_to_root()
self._write(self.MOTOR_COMMAND)
class HAL000Protocol(FirmwareProtocol):
"""HAL 0.0.00 firmware.
Uses shorter command names and a different init sequence.
Motor submenu: "mot"
"""
MOTOR_COMMAND = "mot"
def initialize(self, callback: Callable[[str], None] | None = None) -> None:
# HAL 0.0.00 has a different boot sequence — the exact signals
# are not documented in the upstream repo. This is a best-effort
# implementation that should be validated against real hardware.
logger.info("Waiting for HAL 0.0.00 boot...")
while True:
line = self._readline()
if not line:
continue
if callback:
callback(line)
logger.debug("Boot: %s", line)
# HAL 0.0.00 boot detection — adjust if hardware reveals
# different signals
if "NoGPS" in line or "ready" in line.lower():
logger.info("Boot complete")
break
self.kill_search()
self.reset_to_root()
def kill_search(self) -> None:
# HAL 0.0.00 may have a different search-kill sequence.
# Falling back to root-menu reset as a safe default.
self.reset_to_root()
logger.info("Search task cancelled (HAL 0.0.00)")
def enter_motor_menu(self) -> None:
self.reset_to_root()
self._write(self.MOTOR_COMMAND)
# Registry for firmware lookup by name
FIRMWARE_REGISTRY: dict[str, type[FirmwareProtocol]] = {
"hal205": HAL205Protocol,
"hal000": HAL000Protocol,
}
def get_protocol(name: str) -> FirmwareProtocol:
"""Instantiate a firmware protocol by short name.
Args:
name: One of "hal205", "hal000".
Raises:
KeyError: If the firmware name is not recognized.
"""
try:
return FIRMWARE_REGISTRY[name.lower()]()
except KeyError:
available = ", ".join(sorted(FIRMWARE_REGISTRY))
raise KeyError(f"Unknown firmware {name!r}. Available: {available}") from None

135
src/travler_rotor/rotctld.py Normal file
View file

@ -0,0 +1,135 @@
"""Hamlib rotctld-compatible TCP server.
Implements the subset of the rotctld protocol that Gpredict (and other
Hamlib clients) use for AZ/EL rotor control:
p — get position (returns "AZ\\nEL\\n")
P — set position ("P <az> <el>")
S — stop / disconnect
_ — get model name
q — quit connection
"""
from __future__ import annotations
import logging
import socket
from travler_rotor.antenna import TravlerAntenna
logger = logging.getLogger(__name__)
MODEL_NAME = "Winegard Trav'ler RS-485 Rotor"
class RotctldServer:
"""TCP server speaking the Hamlib rotctld wire protocol."""
def __init__(
self,
antenna: TravlerAntenna,
host: str = "127.0.0.1",
port: int = 4533,
) -> None:
self._antenna = antenna
self._host = host
self._port = port
self._server_socket: socket.socket | None = None
self._running = False
def serve_forever(self) -> None:
"""Listen for connections and handle rotctld commands.
Blocks until shutdown() is called or the process is interrupted.
"""
self._server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self._server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self._server_socket.bind((self._host, self._port))
self._server_socket.listen(1)
self._server_socket.settimeout(1.0)
self._running = True
logger.info("rotctld listening on %s:%d", self._host, self._port)
while self._running:
try:
conn, addr = self._server_socket.accept()
except TimeoutError:
continue
logger.info("Connection from %s", addr)
try:
self._handle_connection(conn)
except Exception:
logger.exception("Error handling connection from %s", addr)
finally:
conn.close()
logger.info("Connection closed from %s", addr)
def shutdown(self) -> None:
"""Signal the server to stop accepting connections."""
self._running = False
if self._server_socket:
self._server_socket.close()
self._server_socket = None
def _handle_connection(self, conn: socket.socket) -> None:
"""Process commands from a single client connection."""
while self._running:
data = conn.recv(1024)
if not data:
break
cmd_parts = data.decode("utf-8").strip().split()
if not cmd_parts:
continue
cmd = cmd_parts[0]
if cmd == "p":
self._handle_get_position(conn)
elif cmd == "P":
self._handle_set_position(conn, cmd_parts)
elif cmd == "S":
self._handle_stop(conn)
break
elif cmd == "_":
self._handle_model_name(conn)
elif cmd == "q":
break
else:
logger.warning("Unknown command: %s", cmd)
conn.sendall(b"RPRT -1\n")
def _handle_get_position(self, conn: socket.socket) -> None:
"""Respond to 'p' — return current AZ and EL."""
try:
pos = self._antenna.get_position()
response = f"{pos.azimuth}\n{pos.elevation}\n"
conn.sendall(response.encode("utf-8"))
except Exception:
logger.exception("Failed to get position")
conn.sendall(b"RPRT -1\n")
def _handle_set_position(self, conn: socket.socket, parts: list[str]) -> None:
"""Respond to 'P <az> <el>' — move dish to target."""
try:
target_az = float(parts[1])
target_el = float(parts[2])
self._antenna.move_to(target_az, target_el)
conn.sendall(b"RPRT 0\n")
except (IndexError, ValueError):
logger.error("Bad P command: %s", parts)
conn.sendall(b"RPRT -1\n")
except Exception:
logger.exception("Failed to move")
conn.sendall(b"RPRT -1\n")
def _handle_stop(self, conn: socket.socket) -> None:
"""Respond to 'S' — stop tracking."""
self._antenna.stop()
logger.info("Client sent stop")
def _handle_model_name(self, conn: socket.socket) -> None:
"""Respond to '_' — return model identification string."""
conn.sendall(f"{MODEL_NAME}\n".encode())