ax_radio.py

# Python for controlling AX radio
# Copyright (C) 2016  Richard Meadows <richardeoin>

# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:

# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.

# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
# LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
# WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

from _ax_radio import lib,ffi
from enum import Enum
import time

class AxRadio:
    Modulations = Enum('Modulation', 'FSK MSK GFSK GMSK PSK AFSK CW')
    VcoTypes = Enum('VcoType', 'Undefined Internal Inductor External')
    RadioStates = Enum('RadioState', 'Off Transmit Receive')

    def __init__(self,
                 spi=0, vco_type=VcoTypes.Undefined,
                 frequency_MHz=434.6, modu=Modulations.FSK,
                 bitrate=20000, fec=False, power=0.1, cont=True,
                 accept_crc_failures=False):

        self.config = ffi.new('ax_config*')
        self.mod = ffi.new('ax_modulation*')
        self.state = self.RadioStates.Off
        self.autotune_batch = []

        # attempt to open the SPI port
        spi_status = lib.ax_set_spi_transfer(self.config, spi)

        if spi_status == lib.AX_SET_SPI_TRANSFER_FAILED:
            raise RuntimeError('Failed to open SPI port.')
        elif spi_status == lib.AX_SET_SPI_TRANSFER_BAD_SPI:
            raise ValueError('Bad spi number. Try 0 or 1')

        # default configuration for our hardware
        self.config.clock_source = lib.AX_CLOCK_SOURCE_TCXO
        self.config.f_xtal = 16369000

        # set frequencies
        self.config.synthesiser.A.frequency = int(frequency_MHz * 1e6)
        self.config.synthesiser.B.frequency = int(frequency_MHz * 1e6)
        if vco_type == self.VcoTypes.Undefined: # guess VCO type
            if frequency_MHz > 400:
                vco_type = self.VcoTypes.Internal
            else:
                vco_type = self.VcoTypes.Inductor
        if vco_type == self.VcoTypes.Inductor:
            self.config.synthesiser.vco_type = lib.AX_VCO_INTERNAL_EXTERNAL_INDUCTOR
        elif vco_type == self.VcoTypes.External:
            self.config.synthesiser.vco_type = lib.AX_VCO_EXTERNAL

        # report rssi and rf frequency offset
        self.config.pkt_store_flags = lib.AX_PKT_STORE_RSSI | \
                                lib.AX_PKT_STORE_RF_OFFSET

        # maybe accept CRC failures
        if accept_crc_failures:
            self.config.pkt_accept_flags = lib.AX_PKT_ACCEPT_CRC_FAILURES

        # actually initialise the radio
        init_status = lib.ax_init(self.config)

        if init_status == lib.AX_INIT_BAD_SCRATCH or \
           init_status == lib.AX_INIT_BAD_REVISION:
            raise RuntimeError('Read bad scratch or revision value. '
                               'Is the radio plugged in correctly?')
        elif init_status == lib.AX_INIT_VCO_RANGING_FAILED:
            raise RuntimeError('VCO ranging failed. Try a different frequency '
                               'or changing `vco_type` and `rf_div`')

        # platform init (rpi/chip/...)
        lib.ax_platform_init(self.config)

        self.in_transmit_mode = False

        # set modulation parameters
        self.modulation(bitrate, modu, fec, power, cont)

        # calculate tweakable parameters
        lib.ax_default_params(self.config, self.mod)


    def modulation(self, bitrate, modu, fec, power, cont):
        if modu == self.Modulations.FSK or modu == self.Modulations.GFSK:
            self.mod.modulation = lib.AX_MODULATION_FSK
        if modu == self.Modulations.MSK or modu == self.Modulations.GMSK:
            self.mod.modulation = lib.AX_MODULATION_MSK
        if modu == self.Modulations.PSK:
            self.mod.modulation = lib.AX_MODULATION_PSK
        if modu == self.Modulations.AFSK:
            self.mod.modulation = lib.AX_MODULATION_AFSK
        if modu == self.Modulations.CW:
            self.mod.modulation = lib.AX_MODULATION_CW

        # fec and framing
        if fec:                 # forward error correction
            self.mod.fec = 1
            self.mod.encoding = lib.AX_ENC_NRZ | lib.AX_ENC_SCRAM
            # HDLC required for FEC
            self.mod.framing = lib.AX_FRAMING_MODE_HDLC | \
                               lib.AX_FRAMING_CRCMODE_CCITT

        else:                   # nrzi with pattern match
            self.mod.fec = 0
            self.mod.encoding = lib.AX_ENC_NRZI
            self.mod.framing = lib.AX_FRAMING_MODE_RAW_PATTERN_MATCH | \
                               lib.AX_FRAMING_CRCMODE_CCITT

        # gaussian shaping
        if modu == self.Modulations.GFSK or modu == self.Modulations.GMSK:
            self.mod.shaping = lib.AX_MODCFGF_FREQSHAPE_GAUSSIAN_BT_0_5

        self.mod.bitrate = bitrate

        if cont:
            self.mod.continuous = 1
        else:
            self.mod.continuous = 0

        if (power <= 1) and (power >= 0):
            self.mod.power = power
        else:
            self.mod.power = 0.1

        # fsk: modulation index
        if modu == self.Modulations.FSK or modu == self.Modulations.GFSK:
            self.mod.parameters.fsk.modulation_index = 2.0/3

        # afsk: deviation, mark, space
        if modu == self.Modulations.AFSK:
            # set to reasonable APRS values
            self.mod.parameters.afsk.deviation = 3000
            self.mod.parameters.afsk.space = 2200 # bell 202
            self.mod.parameters.afsk.mark  = 1200


    def transmit(self, bytes_to_transmit): # transmit
        if self.state != self.RadioStates.Transmit:
            self.off()          # need to turn off firstn
            lib.ax_tx_on(self.config, self.mod)
            self.state = self.RadioStates.Transmit

        lib.ax_tx_packet(self.config, self.mod,
                         bytes_to_transmit, len(bytes_to_transmit))


    def receive(self, rx_func, timeout=0): # receive
        pkt = ffi.new('ax_packet*')

        if self.state != self.RadioStates.Receive:
            self.off()          # need to turn off first
            lib.ax_rx_on(self.config, self.mod)
            self.state = self.RadioStates.Receive

        start_time = time.time()

        while (self.state == self.RadioStates.Receive):
            while lib.ax_rx_packet(self.config, pkt): # empty the fifo
                data_c = ffi.cast('char*', pkt.data)
                data = ffi.unpack(data_c[0:pkt.length], pkt.length)
                metadata = {
                    'rssi': pkt.rssi,
                    'rffreqoffs': pkt.rffreqoffs,
                }
                if rx_func:
                    rx_func(data, pkt.length, metadata)

            time.sleep(0.025)         # 25ms sleep

            if (timeout > 0) and ((time.time() - start_time) > timeout):
                return          # timeout

    # averages 10 rf freq offsets and autotunes to them
    # only call with known good offsets (passed CRC etc.)
    def autotune(self, rffreqoffs):
        self.autotune_batch.append(rffreqoffs)

        if len(self.autotune_batch) >= 10:
            # autotune
            offs   = self.autotune_batch
            # mean and std. dev
            mean   = sum(offs) / len(offs)
            stddev = sum([(e-mean)**2 for e in offs]) / len(offs)
            # filter outliers
            offs   = [e for e in offs if abs(e - mean) < 3*stddev] # 3 std. devs
            # new mean
            offset = int(sum(offs) / len(offs))
            # limit
            if (offset >  200): offset =  200
            if (offset < -200): offset = -200
            # update
            if (abs(offset) >= 5): # don't bother with < 5Hz
                current_freq = self.config.synthesiser.A.frequency
                updated_freq = current_freq - offset
                lib.ax_force_quick_adjust_frequency(self.config, updated_freq)
                print("Updated frequency by {}Hz...".format(-offset))
            # clear batch
            self.autotune_batch = []

    def off(self):              # off
        if self.state != self.RadioStates.Off:
            lib.ax_off(self.config)
            self.state = self.RadioStates.Off

    def get_modulation(self):       # getter
        return self.mod


"""
GMSK-{X,Y,Z} modes
"""
class AxRadioGMSK(AxRadio):
    def __init__(self,
                 spi=0, vco_type=AxRadio.VcoTypes.Undefined,
                 frequency_MHz=434.6, mode='X', power=0.1,
                 accept_crc_failures=False, cont=True):

        if mode == 'X' or mode == 'x':
            bitrate = 12000
        elif mode == 'Y' or mode == 'y':
            bitrate = 24000
        elif mode == 'Z' or mode == 'z':
            bitrate = 115000
        else:
            raise RuntimeError('Unknown mode \'{}\'!'.format(mode))

        # configure radio
        AxRadio.__init__(self, spi, vco_type, frequency_MHz,
                         modu=AxRadio.Modulations.GMSK,
                         bitrate=bitrate, fec=True, power=power,
                         cont=cont,
                         accept_crc_failures=accept_crc_failures)

"""
APRS
"""
class AxRadioAPRS(AxRadio):
    def __init__(self,
                 spi=0, vco_type=AxRadio.VcoTypes.Undefined,
                 frequency_MHz=434.6, power=0.1, deviation=3000):

        # configure radio
        AxRadio.__init__(self, spi, vco_type, frequency_MHz,
                         modu=AxRadio.Modulations.AFSK,
                         bitrate=1200, fec=False, power=power, cont=False)

        # HDLC but with no FEC
        self.mod.fec = 0
        self.mod.encoding = lib.AX_ENC_NRZI
        self.mod.framing = lib.AX_FRAMING_MODE_HDLC | \
                           lib.AX_FRAMING_CRCMODE_CCITT

        # set new deviation
        self.mod.parameters.afsk.deviation = deviation

        # re-calculate tweakable parameters
        lib.ax_default_params(self.config, self.mod)


if __name__ == "__main__":

    def rx_callback(data, length, metadata):
        print(length)
        print(data[:-2].decode('utf-8'))

    radio = AxRadio()

    #radio.transmit("Hello World de Q0QQQ "*10)

    radio.receive(rx_callback)