tessl/pypi-astropy
Astronomy and astrophysics core library providing comprehensive tools for astronomical computations and data handling
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
samp.mddocs/
SAMP (Simple Application Messaging Protocol)
SAMP client and hub functionality for interoperability with other astronomical software applications.
Core Imports
from astropy.samp import SAMPHubServer, SAMPIntegratedClient
from astropy.samp import SAMPClientCapabilities
SAMP Hub Server
SAMP hub server for coordinating communication between astronomical applications.
class SAMPHubServer:
"""
SAMP hub server implementation.
Parameters:
- secret: hub secret key
- addr: server address
- port: server port (0 for automatic)
- lockfile: path to lockfile
- timeout: client timeout
- client_timeout: timeout for client operations
- mode: hub mode ('single' or 'multiple')
"""
def __init__(self, secret=None, addr=None, port=0, lockfile=None, timeout=10, client_timeout=120, mode='single'): ...
def start(self, wait=False):
"""
Start the hub server.
Parameters:
- wait: whether to wait for hub to be ready
"""
def stop(self):
"""Stop the hub server."""
def is_running(self):
"""Check if hub is running."""
@property
def clients(self):
"""Dictionary of connected clients."""
class SAMPHubProxy:
"""
Proxy for communicating with SAMP hub.
Parameters:
- hub_params: hub connection parameters
"""
def __init__(self, hub_params=None): ...
def connect(self):
"""Connect to hub."""
def disconnect(self):
"""Disconnect from hub."""
def is_connected(self):
"""Check if connected to hub."""SAMP Clients
Client classes for connecting applications to SAMP hubs.
class SAMPClient:
"""
Basic SAMP client.
Parameters:
- hub: hub proxy instance
- name: client name
- description: client description
- metadata: client metadata dictionary
"""
def __init__(self, hub=None, name=None, description=None, metadata=None): ...
def connect(self, hub=None, pool_size=20):
"""
Connect to SAMP hub.
Parameters:
- hub: hub proxy to connect to
- pool_size: thread pool size
"""
def disconnect(self):
"""Disconnect from hub."""
def is_connected(self):
"""Check if connected."""
def bind_receive_message(self, mtype, function):
"""
Bind function to receive messages of given type.
Parameters:
- mtype: message type to bind
- function: callback function
"""
def bind_receive_call(self, mtype, function):
"""
Bind function to receive calls of given type.
Parameters:
- mtype: message type to bind
- function: callback function
"""
def bind_receive_notification(self, mtype, function):
"""
Bind function to receive notifications of given type.
Parameters:
- mtype: message type to bind
- function: callback function
"""
class SAMPIntegratedClient(SAMPClient):
"""
SAMP client with integrated hub functionality.
This client can start its own hub if none is available.
Parameters:
- name: client name
- description: client description
- metadata: client metadata
"""
def __init__(self, name=None, description=None, metadata=None): ...
def connect(self, hub=None, pool_size=20):
"""Connect to hub, starting one if necessary."""Message Operations
Functions for sending and receiving SAMP messages.
def notify(client_id, message):
"""
Send notification message.
Parameters:
- client_id: target client ID
- message: message dictionary
"""
def notify_all(message):
"""
Send notification to all clients.
Parameters:
- message: message dictionary
"""
def call(client_id, message, timeout=10):
"""
Send synchronous call message.
Parameters:
- client_id: target client ID
- message: message dictionary
- timeout: response timeout
Returns:
dict: response message
"""
def call_all(message, timeout=10):
"""
Send call to all clients.
Parameters:
- message: message dictionary
- timeout: response timeout
Returns:
dict: responses from all clients
"""
def call_and_wait(client_id, message, timeout=10):
"""
Send asynchronous call and wait for response.
Parameters:
- client_id: target client ID
- message: message dictionary
- timeout: response timeout
Returns:
dict: response message
"""Usage Examples
Starting a SAMP Hub
from astropy.samp import SAMPHubServer
# Start a SAMP hub
hub = SAMPHubServer()
hub.start(wait=True)
print(f"Hub running on port {hub.port}")
print(f"Hub secret: {hub.secret}")
# Hub will run until stopped
# hub.stop()Connecting a Client
from astropy.samp import SAMPIntegratedClient
# Create and connect client
client = SAMPIntegratedClient(
name="My Astronomy App",
description="Example SAMP client",
metadata={
"samp.name": "My App",
"samp.description.text": "Example application",
"author.name": "Astronomer"
}
)
# Connect to hub (will start one if needed)
client.connect()
print(f"Connected as client ID: {client.get_public_id()}")
print(f"Connected clients: {list(client.get_subscribed_clients('*').keys())}")Sending Messages
# Send table to all clients that can receive tables
message = {
"samp.mtype": "table.load.votable",
"samp.params": {
"url": "file:///path/to/table.xml",
"table-id": "example-table",
"name": "Example Data"
}
}
# Send to all subscribed clients
responses = client.call_all(message, timeout=30)
print(f"Sent table to {len(responses)} clients")Receiving Messages
def receive_table(private_key, sender_id, msg_id, mtype, params, extra):
"""Handle incoming table load requests."""
print(f"Received table from {sender_id}: {params['name']}")
print(f"Table URL: {params['url']}")
# Return success response
return {"samp.status": "samp.ok"}
# Bind handler for table messages
client.bind_receive_call("table.load.*", receive_table)
print("Client ready to receive table messages")Hub Discovery
from astropy.samp import SAMPHubProxy
# Find existing hub
try:
hub = SAMPHubProxy()
hub.connect()
print("Connected to existing hub")
except:
print("No hub found")Types
# SAMP types
SAMPHubServer = astropy.samp.SAMPHubServer
SAMPClient = astropy.samp.SAMPClient
SAMPIntegratedClient = astropy.samp.SAMPIntegratedClient
SAMPHubProxy = astropy.samp.SAMPHubProxy