#!/usr/bin/env python3 """ Simple Picondo API Client in Python 3 Demonstrates: 1. REST API queries (health check, query by callsign, date, mode, band) 2. WebSocket listener for real-time QSO events Usage: python3 simple_picondo_client.py YOUR-API-KEY-HERE Requirements: - Python 3.10+ - websockets library: pip install websockets """ import asyncio import json import sys import urllib.request import urllib.parse import urllib.error from datetime import datetime, date, timedelta from typing import Optional, Dict, List, Any try: import websockets except ImportError: print("Error: websockets library not found!") print("Install with: pip install websockets") sys.exit(1) class SimplePicondoClient: """Simple Python client for Picondo API""" def __init__(self, api_key: str, base_url: str = "http://127.0.0.1:7373"): self.api_key = api_key self.base_url = base_url self.ws_url = base_url.replace("http://", "ws://").replace("https://", "wss://") # ========== REST API Methods ========== def _make_request(self, endpoint: str) -> Dict[str, Any]: """Make an authenticated REST API request""" url = f"{self.base_url}{endpoint}" req = urllib.request.Request(url) req.add_header("X-API-Key", self.api_key) req.add_header("Accept", "application/json") try: with urllib.request.urlopen(req) as response: data = response.read().decode('utf-8') return json.loads(data) except urllib.error.HTTPError as e: error_msg = e.read().decode('utf-8') raise Exception(f"API Error {e.code}: {error_msg}") except urllib.error.URLError as e: raise Exception(f"Connection Error: {e.reason}") def check_health(self) -> Dict[str, Any]: """Check API health (no authentication required)""" url = f"{self.base_url}/api/health" try: with urllib.request.urlopen(url) as response: data = response.read().decode('utf-8') return json.loads(data) except Exception as e: raise Exception(f"Health check failed: {e}") def query_by_callsign(self, callsign: str, start_date: Optional[date] = None, end_date: Optional[date] = None) -> Dict[str, Any]: """Query QSOs by callsign""" endpoint = f"/api/qsos/callsign/{callsign}" params = {} if start_date: params['startDate'] = start_date.isoformat() if end_date: params['endDate'] = end_date.isoformat() if params: query_string = urllib.parse.urlencode(params) endpoint = f"{endpoint}?{query_string}" return self._make_request(endpoint) def query_by_date(self, start_date: date, end_date: Optional[date] = None) -> Dict[str, Any]: """Query QSOs by date range""" endpoint = f"/api/qsos/date/{start_date.isoformat()}" if end_date: query_string = urllib.parse.urlencode({'endDate': end_date.isoformat()}) endpoint = f"{endpoint}?{query_string}" return self._make_request(endpoint) def query_by_mode(self, mode: str, start_date: Optional[date] = None, end_date: Optional[date] = None) -> Dict[str, Any]: """Query QSOs by mode""" endpoint = f"/api/qsos/mode/{mode}" params = {} if start_date: params['startDate'] = start_date.isoformat() if end_date: params['endDate'] = end_date.isoformat() if params: query_string = urllib.parse.urlencode(params) endpoint = f"{endpoint}?{query_string}" return self._make_request(endpoint) def query_by_band(self, band: str, start_date: Optional[date] = None, end_date: Optional[date] = None) -> Dict[str, Any]: """Query QSOs by band""" endpoint = f"/api/qsos/band/{band}" params = {} if start_date: params['startDate'] = start_date.isoformat() if end_date: params['endDate'] = end_date.isoformat() if params: query_string = urllib.parse.urlencode(params) endpoint = f"{endpoint}?{query_string}" return self._make_request(endpoint) # ========== WebSocket Methods ========== async def connect_websocket(self): """Connect to WebSocket and listen for QSO events""" ws_endpoint = f"{self.ws_url}/api/events?apiKey={self.api_key}" print("\n✓ Connecting to WebSocket...") try: async with websockets.connect(ws_endpoint) as websocket: print("✓ WebSocket Connected!") print("Listening for QSO events... (Press Ctrl+C to exit)\n") async for message in websocket: try: event = json.loads(message) self._display_qso_event(event) except json.JSONDecodeError as e: print(f"Error parsing event: {e}") except Exception as e: print(f"Error handling event: {e}") except websockets.exceptions.WebSocketException as e: print(f"✗ WebSocket Error: {e}") except KeyboardInterrupt: print("\n✗ WebSocket Disconnected") except Exception as e: print(f"✗ Connection Error: {e}") def _display_qso_event(self, event: Dict[str, Any]): """Display a QSO logged event""" qso = event.get('qso', {}) print("┌─ NEW QSO EVENT ──────────────────────") print(f"│ Callsign: {qso.get('callsign', 'N/A')}") print(f"│ Date/Time: {qso.get('dateOn', 'N/A')} {qso.get('timeOn', 'N/A')}") print(f"│ Frequency: {qso.get('frequency', 'N/A')} MHz") print(f"│ Mode: {qso.get('mode', 'N/A')}") print(f"│ RST: {qso.get('rstSent', 'N/A')} / {qso.get('rstReceived', 'N/A')}") if qso.get('outdoorsRef'): print(f"│ POTA/SOTA: {qso.get('outdoorsRef')}") if qso.get('myOutdoorsRef'): print(f"│ My Ref: {qso.get('myOutdoorsRef')}") print(f"│ Profile: {qso.get('profile', 'N/A')}") print(f"│ Timestamp: {event.get('timestamp', 'N/A')}") print("└──────────────────────────────────────\n") # ========== Display Helper Methods ========== @staticmethod def _display_response(response: Dict[str, Any], title: str): """Display API response in a formatted way""" print(f"\n{title}") print("=" * 60) if not response.get('success', False): print(f"❌ Error: {response.get('message', 'Unknown error')}") return print(f"✓ Success: {response.get('message', '')}") print(f"Count: {response.get('count', 0)} QSOs") qsos = response.get('qsos', []) if qsos: print("\nQSOs:") print("-" * 60) for i, qso in enumerate(qsos, 1): callsign = qso.get('callsign', 'N/A') date_on = qso.get('dateOn', 'N/A') time_on = qso.get('timeOn', 'N/A') mode = qso.get('mode', 'N/A') freq = qso.get('frequency', 'N/A') rst_sent = qso.get('rstSent', 'N/A') rst_rcvd = qso.get('rstReceived', 'N/A') print(f"{i:3d}. {callsign:10s} @ {date_on} {time_on} - " f"{mode:6s} {freq:8.3f} MHz - {rst_sent}/{rst_rcvd}") print() def main(): """Main function demonstrating client usage""" # Check for API key argument if len(sys.argv) < 2: print("=" * 60) print(" Picondo API Client - Python 3") print("=" * 60) print() print("Usage:") print(f" {sys.argv[0]} YOUR-API-KEY") print() print("Example:") print(f" {sys.argv[0]} a1b2c3d4-e5f6-7890-abcd-ef1234567890") print() print("Get your API key from Picondo settings.") print() sys.exit(1) api_key = sys.argv[1] # Create client client = SimplePicondoClient(api_key) print("=" * 60) print(" Simple Picondo API Client - Python 3") print("=" * 60) print() try: # Example 1: Health check print("1. Checking API health...") health = client.check_health() print(f"Health: {json.dumps(health, indent=2)}") print() # Example 2: Query by callsign print("2. Querying QSOs for callsign G4XYZ...") response = client.query_by_callsign("G4XYZ") client._display_response(response, "Query by Callsign") # Example 3: Query by date range (last 7 days) print("3. Querying QSOs from last 7 days...") end_date = date.today() start_date = end_date - timedelta(days=7) response = client.query_by_date(start_date, end_date) client._display_response(response, "Query by Date Range") # Example 4: Query by mode print("4. Querying FT8 QSOs...") response = client.query_by_mode("FT8", start_date, end_date) client._display_response(response, "Query by Mode (FT8)") # Example 5: Query by band print("5. Querying 20m QSOs...") response = client.query_by_band("20M", start_date, end_date) client._display_response(response, "Query by Band (20M)") # Example 6: WebSocket listener print("6. Starting WebSocket listener...") print("(This will run until you press Ctrl+C)") print() # Run WebSocket listener asyncio.run(client.connect_websocket()) except KeyboardInterrupt: print("\n\n✓ Shutdown requested") except Exception as e: print(f"\n❌ Error: {e}") import traceback traceback.print_exc() sys.exit(1) if __name__ == "__main__": main()