SimTX API v24/Quick start

This page gets a program talking to SimTX with nothing but a standard Python 3 installation. The first example receives IQ anonymously; the second transmits a tone with an account. Both use only the standard library, so that every byte on the wire is visible. The same steps translate directly to any language with TLS sockets.

Receiving

The program below connects to the public server, completes an anonymous handshake, requests a receiver on 7.1 MHz in Budapest, and prints the power of the first ten IQ blocks.

import socket
import ssl
import struct

HOST, PORT = "argon.simtx.net", 5556


def read_exact(sock, n):
    buf = bytearray()
    while len(buf) < n:
        chunk = sock.recv(n - len(buf))
        if not chunk:
            raise ConnectionError("server closed the connection")
        buf += chunk
    return bytes(buf)


def write_utf(s):
    data = s.encode("utf-8")  # fine for ASCII; see Data types for modified UTF-8
    return struct.pack(">H", len(data)) + data


def read_utf(sock):
    (n,) = struct.unpack(">H", read_exact(sock, 2))
    return read_exact(sock, n).decode("utf-8", "replace")


raw = socket.create_connection((HOST, PORT), timeout=10)
raw.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
sock = ssl.create_default_context().wrap_socket(raw, server_hostname=HOST)

# Handshake: magic, version, role 1 (RECEIVER), authKind 0 (ANONYMOUS), empty credential
sock.sendall(struct.pack(">iibb", 0x53494D54, 24, 1, 0) + write_utf(""))

packet_id, status = struct.unpack(">BB", read_exact(sock, 2))
assert packet_id == 30, packet_id
message = read_utf(sock)
if read_exact(sock, 1)[0]:
    read_exact(sock, 16)  # account UUID
read_utf(sock)  # handle
read_utf(sock)  # display name
for _ in range(read_exact(sock, 1)[0]):
    read_utf(sock)  # permission
read_exact(sock, 16)  # authorizationRevision, leaseExpiresAtEpochMillis
if status != 0:
    raise SystemExit(f"handshake rejected ({status}): {message}")

# ReceiverSessionRequest
antenna = write_utf("DIPOLE") + struct.pack(">ffffi", 10.0, 0.0, 0.0, 0.0, 0)
request = (
    write_utf("N0CALL")
    + struct.pack(">ddd", 47.4979, 19.0402, 120.0)
    + antenna
    + struct.pack(">fffii", 5.0, 7_100_000.0, 48_000.0, 48_000, 4_800)
    + write_utf("")
)
sock.sendall(b"\x01" + request)

packet_id = read_exact(sock, 1)[0]
assert packet_id == 1, packet_id
rate, center, chunk, fmt = struct.unpack(">ifib", read_exact(sock, 13))
print(f"stream: {rate} S/s at {center:.0f} Hz, {chunk} samples/block, format {fmt}")

width = 2 if fmt == 1 else 1
full_scale = 32767 if fmt == 1 else 127
for _ in range(10):
    packet_id = read_exact(sock, 1)[0]
    assert packet_id == 2, packet_id
    (length,) = struct.unpack(">i", read_exact(sock, 4))
    payload = read_exact(sock, length)
    (scale,) = struct.unpack(">f", payload[:4])
    code = ">%dh" % (chunk * 2) if fmt == 1 else ">%db" % (chunk * 2)
    raw_iq = struct.unpack(code, payload[4:])
    power = sum((v * scale / full_scale) ** 2 for v in raw_iq) / chunk
    print(f"block: {length} bytes, scale {scale:.4g}, mean power {power:.3e}")

sock.close()

Typical output:

stream: 48000 S/s at 7100000 Hz, 1920 samples/block, format 1
block: 7684 bytes, scale 0.0009043, mean power 1.337e-07
block: 7684 bytes, scale 0.0009922, mean power 1.366e-07
...

Things to notice:

• The program asked for 4,800 samples per block and received 1,920. The server decides; the IqStreamDescriptor says what it decided. See Negotiation.
• A mean power of about 1.3 × 10⁻⁷ is −12 + 10·log₁₀(1.3 × 10⁻⁷) ≈ −81 dBm: mostly the 40 m band's background noise in a 48 kHz bandwidth. Tune to a frequency where someone is transmitting and it rises.
• HOST and PORT are hard-coded for brevity. Real software should take them from the server directory or from its configuration.

To feed the samples to a demodulator, convert each block into complex values (I + jQ, scaled as shown) and pass them on. To change frequency, send b"\x0d" + struct.pack(">f", new_frequency_hz) at any time.

Transmitting

Transmitting requires an account with the game.transmit permission. For a script, the simplest credential is a personal credential created on the account website. Put it in the SIMTX_TOKEN environment variable. Interactive software should use device login instead.

The program below obtains a game ticket, opens a transmitter session on 14.230 MHz USB at 100 W, waits for admission, transmits a 1 kHz tone for ten seconds while printing telemetry, and goes off air.

import json
import math
import os
import socket
import ssl
import struct
import threading
import time
import urllib.request

AUTH = "https://auth.simtx.net"
TOKEN = os.environ["SIMTX_TOKEN"]  # a personal credential (stx_pc1_...)
SERVER_ID = "argon"
HOST, PORT = "argon.simtx.net", 5556
RATE = 48_000


def post_json(path, body):
    request = urllib.request.Request(
        AUTH + path,
        data=json.dumps(body).encode(),
        headers={"Authorization": "Bearer " + TOKEN, "Content-Type": "application/json"},
    )
    with urllib.request.urlopen(request, timeout=10) as response:
        return json.load(response)


def read_exact(sock, n):
    buf = bytearray()
    while len(buf) < n:
        chunk = sock.recv(n - len(buf))
        if not chunk:
            raise ConnectionError("server closed the connection")
        buf += chunk
    return bytes(buf)


def write_utf(s):
    data = s.encode("utf-8")
    return struct.pack(">H", len(data)) + data


def read_utf(sock):
    (n,) = struct.unpack(">H", read_exact(sock, 2))
    return read_exact(sock, n).decode("utf-8", "replace")


def field(field_id, payload):
    return struct.pack(">Bi", field_id, len(payload)) + payload


def f32(value):
    return struct.pack(">f", value)


def transmitter_request():
    antenna = write_utf("DIPOLE") + struct.pack(">ffffi", 10.0, 0.0, 0.0, 20.0, 0)
    fields = [
        field(1, b"N0CALL"),                                    # callsign
        field(2, struct.pack(">ddd", 47.4979, 19.0402, 120.0)),  # location
        field(3, antenna),                                      # antenna
        field(4, b"USB"),                                       # modulation
        field(5, f32(14_230_000.0)),                            # frequencyHz
        field(6, f32(100.0)),                                   # powerWatts
        field(7, b"AUDIO_STREAM"),                              # sourceType
        field(8, struct.pack(">i", RATE)),                      # sampleRate
        field(9, f32(1.0)),                                     # audioGain
        field(10, f32(2_700.0)),                                # txBandwidthHz
        field(11, f32(300.0)),                                  # audioLowCutHz
        field(12, f32(0.0)),                                    # toneHz
        field(13, f32(1.0)),                                    # dutyCycle
    ]
    return b"\x02" + struct.pack(">B", len(fields)) + b"".join(fields)


ticket = post_json("/api/v1/game-tickets", {"serverId": SERVER_ID, "role": "tx-audio"})["ticket"]

raw = socket.create_connection((HOST, PORT), timeout=10)
raw.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
sock = ssl.create_default_context().wrap_socket(raw, server_hostname=HOST)

# Handshake: role 2 (TX_AUDIO), authKind 1 (GAME_TICKET)
sock.sendall(struct.pack(">iibb", 0x53494D54, 24, 2, 1) + write_utf(ticket))
packet_id, status = struct.unpack(">BB", read_exact(sock, 2))
message = read_utf(sock)
if read_exact(sock, 1)[0]:
    read_exact(sock, 16)
handle = read_utf(sock)
read_utf(sock)
permissions = [read_utf(sock) for _ in range(read_exact(sock, 1)[0])]
read_exact(sock, 16)
if status != 0:
    raise SystemExit(f"handshake rejected ({status}): {message}")
print("signed in as", handle, "with", permissions)

sock.sendall(transmitter_request())

# Admission: TxQueue (4) until TxAccept (3), or TxReject (5)
while True:
    packet_id = read_exact(sock, 1)[0]
    if packet_id == 3:
        print("on air")
        break
    if packet_id == 4:
        ahead, total = struct.unpack(">ii", read_exact(sock, 8))
        print(f"queued: {ahead} ahead of {total}")
    elif packet_id == 5:
        raise SystemExit("rejected: " + read_utf(sock))
    else:
        raise SystemExit(f"unexpected packet {packet_id}")

sock.settimeout(None)


def read_telemetry():
    try:
        while True:
            packet_id = read_exact(sock, 1)[0]
            if packet_id == 7:
                values = struct.unpack(">qq16f?", read_exact(sock, 8 + 8 + 16 * 4 + 1))
                forward, vswr, buffer_s = values[4], values[7], values[12]
                print(f"forward {forward:6.1f} W  SWR {vswr:4.2f}  buffer {buffer_s:4.2f} s")
            elif packet_id == 6:
                print("kicked:", read_utf(sock))
                return
            else:
                print("unexpected packet", packet_id)
                return
    except (ConnectionError, OSError):
        return


threading.Thread(target=read_telemetry, daemon=True).start()

# Ten seconds of a 1 kHz tone, sent in real time in 20 ms blocks
block = RATE // 50
start = time.monotonic()
for n in range(500):
    samples = [0.5 * math.sin(2 * math.pi * 1000 * (n * block + i) / RATE) for i in range(block)]
    sock.sendall(b"\x03" + struct.pack(">i%df" % block, block, *samples))
    time.sleep(max(0.0, start + (n + 1) * 0.02 - time.monotonic()))

sock.shutdown(socket.SHUT_WR)
sock.close()
print("off air")

Things to notice:

• Only the 13 required fields of the TransmitterSessionRequest are sent. The server uses defaults for the transmitter imperfections (fields 14 to 28).
• The sender paces itself to real time with a deadline per block. Sending as fast as possible would fill the server's buffer, and anything beyond the allowed burst would be discarded. The buffer value in the telemetry should stay roughly constant.
• Telemetry is read on a separate thread. A client that stops reading is disconnected after five seconds.
• There is no "stop transmitting" packet. Closing the socket takes the station off air.

Next steps

• SimTX API v24/Connection and handshake explains the handshake and its failure codes.
• SimTX API v24/Receiver session and SimTX API v24/Transmitter session cover negotiation, flow control and live updates.
• SimTX API v24/Control session adds the station list, so that a receiver can show who is on the air.
• SimTX API v24/Limits and errors explains what to do when the server simply hangs up.