wsprctrl

#!/usr/bin/env python
#
# Copyright 2018 Alexander Fasching
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#

import sys
import ctypes
import time
import argparse
import struct
import contextlib
import datetime as dt
import enum

import usb

class Request(enum.IntEnum):
    ENABLE_LED      = 0
    DISABLE_LED     = 1
    SET_FREQ        = 2
    SET_DRIVE       = 3
    SET_CALL        = 4
    SET_FREQ_CORR   = 5
    TRANSMIT_TONE   = 6
    TRANSMIT_WSPR   = 7
    SET_LAT         = 8
    SET_LON         = 9


class BeaconError(Exception):
    pass


class Beacon(object):
    def __init__(self, dev : usb.Device):
        self.dev = dev

    @staticmethod
    def get_devices():
        device_filter = dict(
            idVendor=0x16c0,
            idProduct=0x05dc,
            serial_number='oe5tkm.net:wspr-beacon'
        )
        return list(usb.core.find(**device_filter, find_all=True))

    def generic_cmd(self, request, index, value):
        buf = self.dev.ctrl_transfer(0xC0, request, value, index, 4)
        value, status = struct.unpack('HH', buf)
        return value, status

    def enable_led(self):
        _, status = self.generic_cmd(Request.ENABLE_LED, 0, 0)
        if status != 0:
            raise BeaconError('Returned status code %d' % status)

    def disable_led(self):
        _, status = self.generic_cmd(Request.DISABLE_LED, 0, 0)
        if status != 0:
            raise BeaconError('Returned status code %d' % status)

    def set_frequency(self, output, value):
        # The output is passed in the upper 4 bits of the double word.
        req_index = (output << 12) | (value >> 16)
        req_value = value & 0xFFFF

        assert 0 <= req_index < 2**16
        assert 0 <= req_value < 2**16

        _, status = self.generic_cmd(Request.SET_FREQ, req_index, req_value)
        if status != 0:
            raise BeaconError('Returned status code %d' % status)

    def set_drive_strength(self, output, value):
        _, status = self.generic_cmd(Request.SET_DRIVE, output, value)
        if status != 0:
            raise BeaconError('Returned status code %d' % status)

    def set_callsign(self, call):
        for i, c in enumerate(call):
            _, status = self.generic_cmd(Request.SET_CALL, i, ord(c))
            if status != 0:
                raise BeaconError('Returned status code %d' % status)

    def set_freq_correction(self, ppm):
        value = ctypes.c_uint32(round(ppm * 10))
        req_index = value.value >> 16
        req_value = value.value & 0xFFFF
        _, status = self.generic_cmd(Request.SET_FREQ_CORR, req_index, req_value)
        if status != 0:
            raise BeaconError('Returned status code %d' % status)

    def transmit_tone(self, output, msecs):
        _, status = self.generic_cmd(Request.TRANSMIT_TONE, output, msecs)
        if status != 0:
            raise BeaconError('Returned status code %d' % status)

    def transmit_wspr(self, output):
        _, status = self.generic_cmd(Request.TRANSMIT_WSPR, output, 0)
        if status != 0:
            raise BeaconError('Returned status code %d' % status)

    def set_latitude(self, lat):
        value = ctypes.c_uint32(round(lat * 2**16))
        req_index = value.value >> 16
        req_value = value.value & 0xFFFF
        _, status = self.generic_cmd(Request.SET_LAT, req_index, req_value)
        if status != 0:
            raise BeaconError('Returned status code %d' % status)

    def set_longitude(self, lon):
        value = ctypes.c_uint32(round(lon * 2**16))
        req_index = value.value >> 16
        req_value = value.value & 0xFFFF
        _, status = self.generic_cmd(Request.SET_LON, req_index, req_value)
        if status != 0:
            raise BeaconError('Returned status code %d' % status)


def cmd_blink(beacon):
    with contextlib.suppress(KeyboardInterrupt):
        while True:
            beacon.enable_led()
            time.sleep(0.05)
            beacon.disable_led()
            time.sleep(0.95)
    beacon.disable_led()


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--list', '-l', action='store_true',
                        help='show available devices')
    parser.add_argument('--device', '-d', type=int, default=0,
                        help='device index')
    parser.add_argument('--blink', '-b', action='store_true',
                        help='continuously flash LED')
    parser.add_argument('--tone', '-t', action='store_true',
                        help='play a tone for 1 second')
    parser.add_argument('--wspr', '-w', action='store_true',
                        help='transmit WSPR message')
    parser.add_argument('--output', '-o', type=int, default=0,
            help='clock output')

    parser.add_argument('--frequency', '-f', type=int, help='frequency')
    parser.add_argument('--lat', type=float, help='latitude')
    parser.add_argument('--lon', type=float, help='longitude')
    parser.add_argument('--drive', type=int, help='drive strength')
    parser.add_argument('--correction', type=float,
            help='frequency correction in PPM')
    parser.add_argument('--call', type=str, help='callsign in AABCCC format')


    args = parser.parse_args()

    devices = Beacon.get_devices()
    if len(devices) == 0:
        print('Error: no WSPR devices found', file=sys.stderr)
        return 1

    if args.list:
        for i, dev in enumerate(devices):
            name = usb.util.get_string(dev, dev.iProduct)
            print('%d: %s' % (i, name))
        return

    if args.device < 0 or args.device >= len(devices):
        print('Error: invalid device index', file=sys.stderr)
        return 1

    dev = devices[args.device]
    dev.set_configuration()
    beacon = Beacon(dev)

    # Set the options

    if args.frequency:
        beacon.set_frequency(args.output, args.frequency)

    if args.lat:
        beacon.set_latitude(args.lat)

    if args.lon:
        beacon.set_longitude(args.lon)

    if args.drive:
        beacon.set_drive_strength(args.output, args.drive)

    if args.correction:
        beacon.set_freq_correction(args.correction)

    if args.call:
        beacon.set_callsign(args.call)


    # Run a command

    if args.blink:
        cmd_blink(beacon)

    elif args.tone:
        beacon.transmit_tone(args.output, 1000)

    elif args.wspr:
        # Get the time when the transmission starts
        now = dt.datetime.now()
        mdelta = dt.timedelta(minutes=(2 if (now.minute % 2 == 0) else 1))
        start = now.replace(second=1, microsecond=0) + mdelta
        print('Starting transmission at %s' % start.strftime('%H:%M'))

        # Wait until the transmission starts
        diff = (start - now).total_seconds()
        time.sleep(diff - 0.5)

        beacon.transmit_wspr(args.output)


if __name__ == '__main__':
    sys.exit(main() or 0)