Perform Inference Using the AirStack Edge API¶

This tutorial walks through using the AirStack Edge REST API to upload a model, open a receive stream, and run inference on samples from the device. The Client UI is the easiest way to validate a device manually, but the REST API is the recommended interface for scripts, production systems, and applications that need to integrate AirStack Edge into a larger workflow.

In this example, we use the same basic PyTorch FFT model from the Client UI tutorial. The model accepts complex samples represented as interleaved float32 I/Q values and returns a shifted FFT result. For information on integrating more complex pipelines, see the ensemble model section here.

Note: this tutorial requires that you have configured your AirStack Edge instance to support inference. If you have not done this you can find more information on how to enable it in the installation guide here.

Note: this tutorial focuses on one working inference flow. For the complete endpoint schema, request parameters, response formats, and additional API options, see the AirStack Edge API documentation. When AirStack Edge is running, the API documentation is also available from the device at https://hostname:port/docs/.

Prerequisites¶

You will need the following before running the example:

An AirStack Edge device with inference enabled.
Network access from your Python environment to the Edge API host and port.
A bearer token for authenticating API requests.
The client certificate file if your Edge instance is using TLS with a custom or self-signed certificate.
Python 3 with the packages used by this example.

If you do not have the bearer token or certificate details yet, start with the configuration documentation here.

Setup¶

Install the Python dependencies on the computer that will run the API script:

python -m pip install requests numpy torch

The snippets below can be run interactively or combined into a single Python script. Replace the placeholder values before running:

YOUR_BEARER_TOKEN: the token used to authorize requests to AirStack Edge, see the configuration documentation for more information
https://YOUR_HOSTNAME:PORT: the URL for your Edge API. Use http:// only if TLS security was disabled.
PATH_TO_CERT_FILE: the path to the certificate used to verify HTTPS requests. Set this to False only for a trusted test environment where certificate verification is intentionally disabled.

Windows and macOS Notes¶

This tutorial can be run from Linux, Windows, or macOS, but keep the following platform differences in mind:

Windows and macOS firewall or certificate trust settings may block access to an Edge device using a self-signed TLS certificate. If HTTPS verification fails, pass the certificate file explicitly with --certificate-path.
PyTorch installation varies by operating system and CPU architecture, especially on Apple Silicon Macs. If python -m pip install torch fails, use the install command recommended by the PyTorch project for your platform.
A TorchScript model created with a newer local PyTorch version may not be compatible with the PyTorch backend on the Edge device. If you only want to validate the API workflow, use the prepackaged simple_fft model linked above.

Connect to the Device¶

All AirStack Edge API requests need a configured HTTP session. The session reuses the connection for later requests, and the Authorization header sends the bearer token with each request.

The first workflow call is device/setup. This gives your application exclusive access to the radio so another application cannot change the device state while your script is configuring streams and running inference.

import requests

token = "YOUR_BEARER_TOKEN"
device_addr = "https://YOUR_HOSTNAME:PORT"  # use http if TLS security was disabled
certificate_path = "PATH_TO_CERT_FILE"  # set to False to disable

session = requests.Session()
headers = {
    "Authorization": f"Bearer {token}",
    "Accept-Charset": "UTF-8",
}

url = f'{device_addr}/airstack-edge/v1/device/setup'
resp = session.post(url, headers=headers, verify=certificate_path, timeout=10)
assert resp.status_code == 200

Save and Upload a Model¶

Once you have established that Python can authenticate with the device, you need a model package to upload. AirStack Edge uses the Triton model repository layout rules, so a model upload archive must contain the contents of one model directory.

At a minimum, the upload package should include:

A config.pbtxt file that describes the model name, backend, inputs, and outputs.
A folder named with the model version number, such as 0.
A model file in the model version folder, such as 0/model.pt.

You can find further information on how to set up an inference model here.

For this example, we create a simple PyTorch FFT model. The input is a real-valued float32 array containing complex float32 samples as interleaved I/Q values: I0, Q0, I1, Q1, .... The model converts those values into a complex tensor, performs an FFT, shifts the FFT output so DC is centered, and returns the real and imaginary FFT values as a [2, L] float32 array.

If you do not want to rebuild the model, you can download a prepackaged version here.

from pathlib import Path

import torch


class IQFFT(torch.nn.Module):
    def forward(self, x):
        # x shape: [2L] interleaved: I0, Q0, I1, Q1, ...
        I = x[0::2]   # even indices
        Q = x[1::2]   # odd indices

        c = torch.complex(I, Q)         # shape [L]
        fft_out = torch.fft.fft(c)      # complex FFT [L]

        # Apply FFT shift
        fft_out = torch.fft.fftshift(fft_out)

        # return real/imag as [2, L]
        return torch.stack([fft_out.real, fft_out.imag], dim=0)


# Save TorchScript model for Triton
if __name__ == "__main__":
    model = IQFFT()
    model.eval()

    model_dir = Path("tmp/simple_fft")
    version_dir = model_dir / "0"
    version_dir.mkdir(parents=True, exist_ok=True)

    # Save the model file to this directory.
    example = torch.zeros(4096, dtype=torch.float32)  # 2048 complex samples
    traced = torch.jit.trace(model, example)
    traced.save(str(version_dir / "model.pt"))

    # Create and write to the config.pbtxt file.
    with open(model_dir / "config.pbtxt", "w") as f:
        f.write("""name: "simple_fft"
platform: "pytorch_libtorch"
max_batch_size: 0

input [
{
    name: "INPUT__0"
    data_type: TYPE_FP32
    dims: [-1]     # [2L] interleaved IQ samples
}
]

output [
{
    name: "OUTPUT__0"
    data_type: TYPE_FP32
    dims: [2, -1]     # [2, L]
}
]
""")

Create Model Upload File¶

Now that the model directory is created, compress it for upload. The archive should contain config.pbtxt and the version folder at the archive root. Do not include the parent simple_fft directory inside the tar file.

Allowed file extensions for upload include: "tar", "bz2" , "tbz2" , "tbz" , "gz" , "tgz" , "zst" , " tzst"

import os
import tarfile

source_dir = "./tmp/simple_fft/"
with tarfile.open("./tmp/simple_fft.tar.gz", "w:gz") as tar:
    for fn in os.listdir(source_dir):
        p = os.path.join(source_dir, fn)
        tar.add(p, arcname=fn)

After compressing the files, upload the package to the Edge instance. The upload endpoint stores the model archive on the device so it can be enabled in the next step. For large uploads, this can take longer than a normal API request, so the timeout should be adjusted accordingly.

file_path = "./tmp/simple_fft.tar.gz"
url = f'{device_addr}/airstack-edge/v1/apps/config/upload'
with open(file_path, 'rb') as file:
    resp = session.post(url, headers=headers, files={'upload': file},
                        verify=certificate_path, timeout=120)
print(resp.content)

Load the Model into Memory¶

Once the model is uploaded, enable it. This loads the model into the inference service and makes it available for apps/run requests. You can optionally override supported model configuration parameters when enabling, but this example uses the configuration from config.pbtxt.

data = {"model_name": "simple_fft"}
url = f'{device_addr}/airstack-edge/v1/apps/config/enable'
resp = session.post(url, json=data, headers=headers, verify=certificate_path, timeout=40)
print(resp.content)

Open the Stream¶

After the model is ready, open an RX stream using the radio parameters for the samples you want to process. The stream controls how samples are acquired from the radio before they are sent to the model.

The inference request in this tutorial expects complex float32 input, so data_type must be set to c_f32. The example below uses channel 0, an internal clock source, AGC, and a sample rate of 31.25 MSPS. Adjust the center frequency, gain, and sample rate for your signal of interest.

data = {"channel": [0],
        "data_type": "c_f32",
        "stream_opts": {
            "frequency": 2350e6,
            "sample_rate": 31.25e6,
            "gain": 0,
            "use_agc": True
        },
        "time_opts": {
            "clock_source": "internal",
            "master_clock_rate": 125000000,
            "time_source": "internal"
        }
    }
url = f'{device_addr}/airstack-edge/v1/receive/open'
resp = session.post(url, json=data, headers=headers, verify=certificate_path, timeout=40)
print(resp.content)

Request a Prediction¶

Once the RX stream is open, request predictions with apps/run. The num_samples field controls how many complex samples are read from the stream for one inference request. For this FFT model, 2048 complex samples become 4096 interleaved float32 input values.

This example uses single stream mode, so the API reads one block of samples, runs one inference request, and returns the result directly in the HTTP response. The response is requested as raw bytes for efficient transfer, then converted back into float32 values in Python. The model output shape is [2, 2048]: row 0 is the real FFT output and row 1 is the imaginary FFT output.

import numpy as np

data = {
  "model_name": "simple_fft",
  "model_version": "0",
  "batch_size": 1,
  "num_samples": 2048,
  "stream_mode": {
    "single": {
      "return_type": "byte",
      "return_input": False,
      "start_time": None
    }
  }
}
url = f'{device_addr}/airstack-edge/v1/apps/run'
resp = session.post(url, json=data, headers=headers, verify=certificate_path, timeout=40)
if resp.status_code == 200:
    u8_data = np.frombuffer(resp.content, dtype=np.uint8)
    f32_data = u8_data.view(dtype=np.float32)
    fft = f32_data.reshape(2, -1)
    fft_complex = fft[0] + 1j * fft[1]
    print(fft_complex)
else:
    print(resp.content)

Close the Stream¶

Close the RX stream when inference is complete. This stops sample acquisition and returns the receive path to an idle state.

url = f'{device_addr}/airstack-edge/v1/receive/close'
resp = session.post(url, headers=headers, verify=certificate_path, timeout=20)
print(resp.content)

Release the Device¶

Finally, release the device so other applications can use it as needed. If your script exits early during development, run the close and release calls again before starting another workflow.

url = f'{device_addr}/airstack-edge/v1/device/release'
resp = session.post(url, headers=headers, verify=certificate_path, timeout=20)
print(resp.content)

Full Example Script¶

The full script below combines the steps above into one runnable example. Update the token, device address, certificate path, and stream settings before running it. For more information on fields not shown here, see the AirStack Edge API documentation.

import argparse
import os
import tarfile
from pathlib import Path

import numpy as np
import requests
import torch


class IQFFT(torch.nn.Module):
    def forward(self, x):
        # x shape: [2L] interleaved: I0, Q0, I1, Q1, ...
        i_values = x[0::2]
        q_values = x[1::2]

        c = torch.complex(i_values, q_values)
        fft_out = torch.fft.fft(c)
        fft_out = torch.fft.fftshift(fft_out)

        return torch.stack([fft_out.real, fft_out.imag], dim=0)


def create_simple_fft_model(work_dir):
    model = IQFFT()
    model.eval()

    model_dir = work_dir / "simple_fft"
    version_dir = model_dir / "0"
    version_dir.mkdir(parents=True, exist_ok=True)

    example = torch.zeros(4096, dtype=torch.float32)  # 2048 complex samples
    traced = torch.jit.trace(model, example)
    traced.save(str(version_dir / "model.pt"))

    with open(model_dir / "config.pbtxt", "w") as f:
        f.write("""name: "simple_fft"
platform: "pytorch_libtorch"
max_batch_size: 0

input [
{
    name: "INPUT__0"
    data_type: TYPE_FP32
    dims: [-1]
}
]

output [
{
    name: "OUTPUT__0"
    data_type: TYPE_FP32
    dims: [2, -1]
}
]
""")

    archive_path = work_dir / "simple_fft.tar.gz"
    with tarfile.open(archive_path, "w:gz") as tar:
        for fn in os.listdir(model_dir):
            p = model_dir / fn
            tar.add(p, arcname=fn)

    return archive_path


def post(session, device_addr, path, headers, certificate_path, **kwargs):
    url = f"{device_addr}/airstack-edge/v1/{path}"
    resp = session.post(url, headers=headers, verify=certificate_path, **kwargs)
    if resp.status_code >= 400:
        raise RuntimeError(f"{path} failed: {resp.status_code} {resp.content!r}")
    return resp


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--token", required=True)
    parser.add_argument("--device-addr", required=True)
    parser.add_argument("--certificate-path", default=False)
    parser.add_argument("--frequency", type=float, default=2350e6)
    parser.add_argument("--sample-rate", type=float, default=31.25e6)
    parser.add_argument("--gain", type=float, default=0)
    parser.add_argument("--num-samples", type=int, default=2048)
    parser.add_argument("--work-dir", default="tmp")
    args = parser.parse_args()

    certificate_path = args.certificate_path
    if isinstance(certificate_path, str) and certificate_path.lower() == "false":
        certificate_path = False

    session = requests.Session()
    headers = {
        "Authorization": f"Bearer {args.token}",
        "Accept-Charset": "UTF-8",
    }

    work_dir = Path(args.work_dir)
    work_dir.mkdir(parents=True, exist_ok=True)
    archive_path = create_simple_fft_model(work_dir)

    stream_open = False
    device_setup = False

    try:
        post(
            session,
            args.device_addr,
            "device/setup",
            headers,
            certificate_path,
            timeout=10,
        )
        device_setup = True

        with open(archive_path, "rb") as file:
            post(
                session,
                args.device_addr,
                "apps/config/upload",
                headers,
                certificate_path,
                files={"upload": file},
                timeout=120,
            )

        post(
            session,
            args.device_addr,
            "apps/config/enable",
            headers,
            certificate_path,
            json={"model_name": "simple_fft"},
            timeout=40,
        )

        stream_config = {
            "channel": [0],
            "data_type": "c_f32",
            "stream_opts": {
                "frequency": args.frequency,
                "sample_rate": args.sample_rate,
                "gain": args.gain,
                "use_agc": True,
            },
            "time_opts": {
                "clock_source": "internal",
                "master_clock_rate": 125000000,
                "time_source": "internal",
            },
        }
        post(
            session,
            args.device_addr,
            "receive/open",
            headers,
            certificate_path,
            json=stream_config,
            timeout=40,
        )
        stream_open = True

        run_config = {
            "model_name": "simple_fft",
            "model_version": "0",
            "batch_size": 1,
            "num_samples": args.num_samples,
            "stream_mode": {
                "single": {
                    "return_type": "byte",
                    "return_input": False,
                    "start_time": None,
                }
            },
        }
        resp = post(
            session,
            args.device_addr,
            "apps/run",
            headers,
            certificate_path,
            json=run_config,
            timeout=40,
        )

        f32_data = np.frombuffer(resp.content, dtype=np.uint8).view(dtype=np.float32)
        fft = f32_data.reshape(2, -1)
        fft_complex = fft[0] + 1j * fft[1]
        print(fft_complex)

    finally:
        if stream_open:
            post(
                session,
                args.device_addr,
                "receive/close",
                headers,
                certificate_path,
                timeout=20,
            )
        if device_setup:
            post(
                session,
                args.device_addr,
                "device/release",
                headers,
                certificate_path,
                timeout=20,
            )


if __name__ == "__main__":
    main()

Run the script with your device details:

python python_api_inference.py --token YOUR_BEARER_TOKEN --device-addr https://YOUR_HOSTNAME:PORT --certificate-path PATH_TO_CERT_FILE