107 lines
4.3 KiB
Python
107 lines
4.3 KiB
Python
import asyncio
|
|
import aiohttp
|
|
|
|
# Character sets for decoding WSPR telemetry fields, based on the provided documentation.
|
|
WSPR_GRID_12_CHARS = "ABCDEFGHIJKLMNOPQR"
|
|
WSPR_GRID_34_CHARS = "0123456789"
|
|
|
|
|
|
def decode_telemetry(record):
|
|
"""
|
|
Decodes telemetry from a WSPR API record.
|
|
|
|
This is a partial implementation based on the general principles outlined at
|
|
https://traquito.github.io/pro/telemetry/. The specific fields used for
|
|
encoding, their order, and the telemetry data they represent are not
|
|
fully specified in the provided document. This function will need to be
|
|
completed with that "shared knowledge".
|
|
|
|
Args:
|
|
record: A list representing a single WSPR spot from the API.
|
|
|
|
Returns:
|
|
A dictionary with decoded telemetry data, or None if decoding fails.
|
|
"""
|
|
callsign = record[2]
|
|
grid = record[3]
|
|
|
|
# The documentation indicates telemetry is encoded into non-standard callsigns
|
|
# and grid squares. The callsign 'VE6AZX' in the query appears to be a
|
|
# standard callsign and may not contain telemetry data of this type.
|
|
# Telemetry callsigns often follow a pattern like '0A9BCD'.
|
|
#
|
|
# The following is an EXAMPLE of how to decode, assuming telemetry is in
|
|
# Grid characters 2 and 3, as per the documentation's example.
|
|
|
|
if len(grid) != 4:
|
|
print(f"Warning: grid '{grid}' is not 4 characters. Skipping decode.")
|
|
return None
|
|
|
|
try:
|
|
# Step 1: Extract field characters and convert to numeric indices.
|
|
# Example: using Grid 2 and Grid 3 characters.
|
|
g2_char = grid[1]
|
|
g3_char = grid[2]
|
|
|
|
g2_val = WSPR_GRID_12_CHARS.index(g2_char)
|
|
g3_val = WSPR_GRID_34_CHARS.index(g3_char)
|
|
|
|
# Step 2: Reconstruct the "big number" from field indices.
|
|
# The order of operations is the reverse of the packing process.
|
|
# The example packs Grid 2 then Grid 3, so we unpack Grid 3 then Grid 2.
|
|
big_number = 0
|
|
big_number = big_number * len(WSPR_GRID_12_CHARS) + g2_val
|
|
big_number = big_number * len(WSPR_GRID_34_CHARS) + g3_val
|
|
|
|
# The above logic is slightly simplified. Based on the example's math:
|
|
# C = G2 + 18 * G3
|
|
# big_number = g2_val + g3_val * len(WSPR_GRID_12_CHARS)
|
|
# Let's use the iterative approach from the docs for clarity.
|
|
# CC = 0; CC = CC * 10 + G3; CC = CC * 18 + G2 => CC = G3 * 18 + G2
|
|
big_number = 0
|
|
big_number = big_number * 10 + g3_val
|
|
big_number = big_number * 18 + g2_val
|
|
|
|
# Step 3: Unpack telemetry values from the "big number".
|
|
# This requires knowing the telemetry fields, their ranges, and order.
|
|
# Since this is unknown, we will just return the reconstructed big number.
|
|
return {"big_number": big_number}
|
|
|
|
except ValueError as e:
|
|
print(f"Could not decode grid '{grid}' for callsign '{callsign}': {e}")
|
|
return None
|
|
|
|
|
|
async def poll_wspr_api(url):
|
|
"""
|
|
Polls the wspr.live API and decodes telemetry from the response.
|
|
"""
|
|
async with aiohttp.ClientSession() as session:
|
|
print(f"Polling {url}")
|
|
try:
|
|
async with session.get(url) as response:
|
|
response.raise_for_status()
|
|
resp_json = await response.json()
|
|
data = resp_json.get("data", [])
|
|
|
|
if not data:
|
|
print("No data received.")
|
|
return
|
|
|
|
print(f"Received {len(data)} records.")
|
|
for record in data:
|
|
telemetry = decode_telemetry(record)
|
|
if telemetry:
|
|
print(f"Record: {record} -> Decoded: {telemetry}")
|
|
except aiohttp.ClientError as e:
|
|
print(f"An error occurred: {e}")
|
|
|
|
|
|
async def main():
|
|
"""Main function to run the polling."""
|
|
url = "https://db1.wspr.live/?query= select time , toMinute(time) % 10 as min , tx_sign as callsign , substring(tx_loc, 1, 4) as grid4 , tx_loc as gridRaw , power as powerDbm , rx_sign as rxCallsign , rx_loc as rxGrid , frequency from wspr.rx where time >= '2026-01-06 07:00:00' and band = 14 /* 20m */ and min = 0 and callsign = 'VE6AZX' /* order by (time, rxCallsign) asc */ FORMAT JSONCompact"
|
|
await poll_wspr_api(url)
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|