Using the Python Interface

While the web interface, the XRA-31 Command-line Interface and the XRA-31 Python Interface provide the same options for interaction with the XRA-31, the latter introduces the power of Python to allow for more complex scenarios.

After the basics have been introduced, the tutorial will move on to a typical debugging scenario (Full example).

Storing and loading configurations

The same JSON files that can be stored and loaded using the command-line interface (Storing and loading configurations), can also be created or loaded from the Python interface.

Channel configuration

Assuming the XRA-31 can be accessed at http://xra31_hostname, the following script will store the active configuration to a.json, and load b.json for a new test:

import json
from excentis import xra31

# Connect to the XRA-31
with xra31.connect(address="xra31_hostname") as client:

    # Store the active configuration
    description = client.configuration.describe()
    json.dump(description, open("a.json", "w"), indent=4)

    # Load the desired configuration
    description = json.load(open("b.json", "r"))
    client.configuration.apply(description)

Capture configuration

The same can be done for capture configurations:

import json
from excentis import xra31

# Connect to the XRA-31
with xra31.connect(address="xra31_hostname") as client:

    # Store the active capture settings
    description = client.capture.describe()
    json.dump(description, open("a.json", "w"), indent=4)

    # Load the desired capture settings
    description = json.load(open("b.json", "r"))
    client.capture.apply(description)

Fine-grained configuration

For fine-grained channel configuration, a minimal example can be found in the Introduction. This also covers configuring, starting and stopping a capture, as well as downloading the result.

Full example

As a full example, this script configures the XRA-31 channels using a previously stored JSON file, starts an unlimited rolling file capture and waits for the moment a given cable modem goes offline.

We’re only interested in the final minute before, and the first minute after the cable modem goes offline, so using 3 files with a per-file duration limit of 1 minute will provide the necessary time window.

While we choose to stop the capture, it can also be kept alive and the procedure can then be repeated to observe multiple occurrences.

This script uses the OS-specific system ping to avoid permission issues.

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
#!/usr/bin/env python3
"""Capture the final minute before, and the first after a CM goes offline."""

import argparse
import datetime
import json
import logging
import pathlib
import subprocess
import sys
import time

from excentis import xra31

# Arguments
parser = argparse.ArgumentParser(
    description="Capture while waiting for a modem to go offline")
parser.add_argument("address",
                    metavar="XRA-31",
                    help="the XRA-31 hostname or IP address")
parser.add_argument("cm",
                    metavar="cable-modem",
                    help="the IP address of the cable modem",
                    type=str)
parser.add_argument("--load",
                    metavar="configuration.json",
                    help="load a JSON configuration to the XRA-31",
                    type=argparse.FileType('r'))

if __name__ == "__main__":
    args = parser.parse_args()

    # Logging
    logger = logging.getLogger(pathlib.Path(__file__).name)
    logging.basicConfig(format="[%(name)s][%(levelname)s]: %(message)s")
    logger.setLevel(logging.INFO)

    # Create a day-specific file for the output
    path = "cm-offline/cm-offline-{}".format(
        datetime.datetime.now().strftime("%Y-%m-%d"))

    # Connect to the XRA-31, force full access mode
    with xra31.connect(address=args.address, full_access=True,
                       force=True) as client:

        # Stop capturing if needed
        logger.info("Stop capturing")
        client.capture.stop()

        # Optionally change its channels
        if args.load:
            logger.info("Load configuration")
            description = json.load(args.load)
            client.configuration.apply(description)

        # Select all channels
        logger.info("Update channel selection")
        client.capture.channels = client.configuration.channels

        # Select all packet types, OFDM streams and OFDM profiles
        logger.info("Update filtering")
        client.capture.filtering.packet_types = xra31.capture.PacketType
        client.capture.filtering.ofdm_streams = xra31.capture.OfdmStream
        client.capture.filtering.ofdm_profiles = xra31.capture.OfdmProfile
        # Remove NCP
        logger.info("Ignore NCP")
        client.capture.filtering.remove_ofdm_stream(
            xra31.capture.OfdmStream.NCP)

        # Update the capture output settings
        logger.info("Update capture output settings")
        client.capture.output.path = path
        client.capture.output.size = None  # unlimited
        client.capture.output.duration = None  # unlimited
        client.capture.output.number_of_files = 3
        client.capture.output.file_size = None  # unlimited
        client.capture.output.file_duration = 60

        # Start the capture
        logger.info("Start capturing")
        client.capture.start()

        # Check if the CM is online by pinging with a 2s timeout
        ping_command = ["ping", "/n", "1", "/w", "2000", args.cm
                        ] if sys.platform == "win32" else [
                            "ping", "-c", "1", "-W", "2", args.cm
                        ]
        # Ping about every 10s
        # until non-zero exit code causes CalledProcessError (check=True)
        try:
            while True:
                logger.info("ping")
                subprocess.run(ping_command,
                               stdout=subprocess.DEVNULL,
                               stderr=subprocess.DEVNULL,
                               check=True)
                time.sleep(10)
        except subprocess.CalledProcessError:
            pass

        logger.info("No ping response, continue capturing for 1 minute")
        time.sleep(60)
        logger.info("Stop capturing")
        client.capture.stop()
        logger.info("Download the captured files")
        client.analysis.download(path,
                                 rolling=True,
                                 compress=True,
                                 verbose=True)
        logger.info("Remove the captured files from the XRA-31")
        client.analysis.delete(path, rolling=True)