Python package for interacting with Major Tom's Gateway API. This package provides a WebSocket-based client for bidirectional communication with Major Tom, supporting command execution, telemetry transmission, file transfers, and more.
pip install majortom-gatewayimport asyncio
from majortom_gateway import GatewayAPI
async def command_callback(command, gateway):
"""Handle incoming commands from Major Tom"""
print(f"Received command: {command.type}")
# Process command...
await gateway.complete_command(command.id, output="Command executed successfully")
async def main():
# Initialize the gateway
gateway = GatewayAPI(
host="your-majortom-instance.com",
gateway_token="your-gateway-token",
command_callback=command_callback
)
# Connect with automatic retries
await gateway.connect_with_retries()
if __name__ == "__main__":
asyncio.run(main())- WebSocket-based Communication: Real-time bidirectional messaging with Major Tom
- Automatic Reconnection: Built-in retry logic with exponential backoff
- Async/Sync Callbacks: Support for both async and synchronous callback functions
- File Transfers: Upload and download files to/from Major Tom
- Comprehensive Validation: Input validation with clear error messages
- Custom Exceptions: Structured exception hierarchy for better error handling
- SSL/TLS Support: Configurable certificate verification
from majortom_gateway import GatewayAPI
gateway = GatewayAPI(
host="your-instance.com", # Required: Major Tom hostname
gateway_token="your-token", # Required: Gateway authentication token
ssl_verify=False, # Optional: Verify SSL certificates (default: False)
basic_auth="username:password", # Optional: HTTP Basic Auth credentials
http=False, # Optional: Use ws:// instead of wss:// (default: False)
ssl_ca_bundle="/path/to/cacert.pem", # Optional: Path to CA certificate bundle
command_callback=handle_command, # Optional: Handler for command messages
error_callback=handle_error, # Optional: Handler for error messages
rate_limit_callback=handle_rate_limit, # Optional: Handler for rate limit messages
cancel_callback=handle_cancel, # Optional: Handler for command cancellation
transit_callback=handle_transit, # Optional: Handler for ground station transits
received_blob_callback=handle_blob, # Optional: Handler for received binary data
max_queue_size=100, # Optional: Max queued payloads when disconnected (default: 100)
)# Send metrics to Major Tom
await gateway.transmit_metrics([
{
"system": "spacecraft",
"subsystem": "power",
"metric": "battery_voltage",
"value": 28.5,
"timestamp": int(time.time() * 1000) # Milliseconds since epoch
}
])# Send events to Major Tom
await gateway.transmit_events([
{
"system": "spacecraft",
"type": "System Status",
"level": "nominal", # Can be: "debug", "nominal", "warning", or "error"
"message": "System operating normally",
"timestamp": int(time.time() * 1000)
}
])async def command_callback(command, gateway):
"""Process commands from Major Tom"""
try:
if command.type == "ping":
await gateway.complete_command(command.id, output="pong")
elif command.type == "get_status":
status = get_system_status()
await gateway.complete_command(command.id, output=status)
else:
await gateway.fail_command(
command.id,
errors=[f"Unknown command type: {command.type}"]
)
except Exception as e:
await gateway.fail_command(command.id, errors=[str(e)])# Download a staged file from Major Tom
filename, content = gateway.download_staged_file("/gateway/download/path/file.bin")
with open(filename, 'wb') as f:
f.write(content)
# Upload a file to Major Tom
gateway.upload_downlinked_file(
filename="telemetry_log.csv",
filepath="/path/to/telemetry_log.csv",
system="spacecraft",
content_type="text/csv",
command_id=123, # Optional: associate with a command
metadata={"duration": 3600} # Optional: custom metadata
)# Disconnect gracefully (note: disconnect() is async)
await gateway.disconnect()The package provides a structured exception hierarchy for better error handling:
from majortom_gateway import (
GatewayAPIError, # Base exception for all gateway errors
ValidationError, # Invalid parameters
FileDownloadError, # File download failures
FileUploadError, # File upload failures
)
try:
gateway = GatewayAPI(
host="", # Invalid: empty host
gateway_token="token"
)
except ValidationError as e:
print(f"Invalid configuration: {e}")
try:
filename, content = gateway.download_staged_file("/invalid/path")
except FileDownloadError as e:
print(f"Download failed: {e}")
# Catch all gateway-related errors
try:
await gateway.transmit_metrics(metrics)
except GatewayAPIError as e:
print(f"Gateway error: {e}")Breaking Change: Python 3.13+ is now required. The disconnect() method is now async and must be awaited.
Before:
gateway.disconnect() # Old synchronous versionAfter:
await gateway.disconnect() # New async versionOther Changes:
- File operation errors now raise specific exceptions (
FileDownloadError,FileUploadError) instead ofRuntimeError - Input parameters are now validated at initialization with clear error messages via
ValidationError - The
dictparameter intransmit_command_update()has been renamed toextra_fields - The local message queue now defaults to a maximum of 100 items (previously unlimited). Payloads queued while disconnected will be dropped when the limit is reached. Set
max_queue_sizeto a higher value if needed.
- Python 3.13+
- websockets >= 13.0, < 14.0
- requests
- asgiref
The Gateway API Package is currently in Beta. Please submit an issue or come talk to us if you have any comments/questions/feedback.
A test script is included to quickly verify your connection to Major Tom and test basic gateway functionality:
# Set up virtual environment
python3 -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
pip install -r requirements.txt
# Run the test gateway
python test_gateway.py <host> <gateway_token> [options]Command-line options:
--system NAME- System name for command definitions (default: test_gateway)--http- Use HTTP/WS instead of HTTPS/WSS (for local testing)--ssl-verify- Verify SSL certificates--ca-bundle PATH- Path to CA certificate bundle--basic-auth USER:PASS- HTTP Basic Auth credentials--debug- Enable debug logging
Example usage:
# Connect to Major Tom production
python test_gateway.py majortom.example.com abc123token --system my_satellite
# Local development
python test_gateway.py localhost:3001 test_token --http --system test_sat
# With SSL verification
python test_gateway.py majortom.example.com token123 --ssl-verify --ca-bundle /path/to/ca.pemThe test gateway automatically registers these commands with Major Tom:
- ping - Simple connectivity test, responds with "pong"
- echo - Echo back a message (accepts a "message" field)
- test_telemetry - Sends sample telemetry data
- test_event - Sends a test event
After connecting, you can send these commands from Major Tom's UI to verify bidirectional communication.
To run all tests:
# Using Docker (recommended)
./dockertest.sh
# Or using a virtual environment
python3 -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
pip install -r requirements.txt -r test_requirements.txt
pytest tests/ -v- Demo Gateway - Complete example implementation
- Major Tom Documentation - Platform documentation
- API Reference - Gateway API specification
See LICENSE file for details.