openairinterface5g/ci-scripts/attenuatorctl.py

# SPDX-License-Identifier: LicenseRef-CSSL-1.0

#!/usr/bin/env python3

import argparse
import logging
import itertools
import usb.core
import usb.util
import sys
import time

logging.basicConfig(
    level=logging.DEBUG,
    stream=sys.stdout,
    format="%(message)s"
)

# helper class parsing command line options for channels and attenuators
class ValidateChAtt(argparse.Action):
    def _parse_channels(s):
        if len(s.split("-")) == 2:
            s, e = [int(x) for x in s.split("-")]
            chs = [x for x in range(s, e+1)]
            if len(chs) == 0:
                raise Exception(f"empty range")
        elif len(s.split(",")) > 1:
            chs = [int(x) for x in s.split(",")]
        elif s.isdigit():
            chs = [int(s)]
        else:
            raise Exception(f"could not parse channel expression")
        for c in chs:
            if not 1 <= c <= 4:
                raise Exception(f"channel number must be within [1,4], but have {c}")
        return chs

    def __call__(self, parse, args, values, option_string=None):
        ch, att = values
        try:
            chs = ValidateChAtt._parse_channels(ch)
        except Exception as e:
            parse.exit(1, f"while parsing channel expression {ch}: {e}\n")
        if not att.isdigit():
            parse.exit(1, f'expected number for attenuation, but got {att}\n')
        if not 0 <= int(att) <= 63:
            parse.exit(1, f'attenuation must be within [0,63], but is {att}\n')
        opts = getattr(args, self.dest) or {}
        for c in chs:
            opts[c] = int(att)
        setattr(args, self.dest, opts)

def _parse_args() -> argparse.Namespace:
    epilog = '''
[CHs] can have the following forms:
  (1) single channel number w (within [1-4])
  (2) a range of channels in the form x-y (x <= y)
  (3) a list of channels in the form a,b,c
[ATT] must be within [0,63] (dB)

--set and --reach are mutually exclusive

Examples:
- reset all channels to 0: -s 1-4 0
- set channels 3 and 4 to attenuation 60: -s 3,4 60
- perform attenuation sweep during 10s, reaching ch. 1+2 to 60: -r 1,2 60 -d 10
'''
    parser = argparse.ArgumentParser(description='Mini-Circuit RC*DAT attenuator controller',
                                     epilog=epilog,
                                     formatter_class=argparse.RawDescriptionHelpFormatter)

    parser.add_argument('--info', '-i', action='store_true', default=False,
                        help='Get Infos for all connected Mini-Circuits RC*DAT',)
    parser.add_argument('--set', '-s', nargs=2, metavar=("[CHs]","[ATT]"),
                        action=ValidateChAtt, help="Set a fixed attenuation ATT on channels CHs")
    parser.add_argument('--reach', '-r', nargs=2, metavar=("[CHs]","[ATT]"),
                        action=ValidateChAtt, help="Perform an \"attenuation sweep\": reach attenuation ATT on channels CHs")
    parser.add_argument('--duration', '-d', action='store', type=float, default=5.0,
                        help='DURATION for attenuation sweep to --reach given attenuation (Default: 5)',)
    parser.add_argument('--progress', '-p', action='store_true', default=False,
                        help='If provided, will show progress during attenuation sweep',)
    return parser.parse_args()

def _exec(dev, cmd):
    dev.write(1, cmd)
    ans = dev.read(0x81,64)
    # take from ans(wer) while valid bytes
    sn = itertools.takewhile(lambda x: x < 255 and x > 0, ans)
    # turn bytes into characters and accumulate into string
    return "".join(map(chr, sn))

def devId(dev):
    return f"{dev.idVendor:04x}:{dev.idProduct:04x}"

def _info(dev):
    ident = devId(dev)
    SerialN = _exec(dev, "*:SN?")
    ModelN = _exec(dev, "*:MN?")
    Fw = _exec(dev, "*:FIRMWARE?")

    logging.info(f'{ident}: model is {ModelN}')
    logging.info(f'{ident}: serial number is {SerialN}')
    logging.info(f'{ident}: FW version is {Fw}')

def _get_attenuation(dev):
    ident = devId(dev)
    att = _exec(dev, "*:ATT?")
    # remove leading "*" and split by channels
    return [float(n) for n in att[1:].split(" ")]

def _set_attenuation(dev, ch_att):
    ident = devId(dev)
    for ch, att in ch_att:
        #logging.debug(f"{ident}: Set attenuation channel {ch}: {att} dB")
        att = round(4 * att) / 4 # round to closest 0.25dB
        resp = _exec(dev, f"*:CHAN:{ch}:SETATT:{att};" )

def _continuous_set_attenuation(dev, ch_att_end, duration, show_progress=False, step_duration=0.25):
    ident = devId(dev)

    current = _get_attenuation(dev)
    steps = duration / step_duration
    ch_att_step = [(ch, (att_end - current[ch - 1]) / steps) for ch, att_end in ch_att_end]
    logging.info(f"{ident}: sweep attenuation during {duration} s in {steps} steps: increments {ch_att_step}")
    for i in range(1, int(steps)+1):
        next_att = [(ch, current[ch - 1] + i * att_step) for ch, att_step in ch_att_step]
        _set_attenuation(dev, next_att)
        if show_progress:
            logging.debug(f"{ident}: attenuation for channel(s): {next_att}")
        time.sleep(step_duration)
    _set_attenuation(dev, ch_att_end)

def _get_usb_devs(idVendor, idProduct):
    devs = []
    for dev in usb.core.find(idVendor=idVendor, idProduct=idProduct, find_all=True):
        for configuration in dev:
            for interface in configuration:
                ifnum = interface.bInterfaceNumber
                if not dev.is_kernel_driver_active(ifnum):
                    continue
                try:
                    dev.detach_kernel_driver(ifnum)
                except e:
                    pass
        devs.append(dev)
    return devs

if __name__ == '__main__':
    # Parse the arguments
    args = _parse_args()
    if args.set and args.reach:
        logging.error("--set and --reach are mutually exclusive")
        sys.exit(1)

    # 20ce:0023 is Mini-Circuit RC*DAT device
    devs = _get_usb_devs(idVendor=0x20ce, idProduct=0x0023)
    if len(devs) == 0:
        logging.error('no Mini-Circuits RC*DAT device detected')
        sys.exit(-1)
    if len(devs) > 1:
        logging.error('more than one Mini-Circuits RC*DAT device detected')
        sys.exit(-1)
    dev = devs[0]

    if args.info:
        _info(dev)

    if args.set:
        _set_attenuation(dev, [(k,v) for k,v in args.set.items()])

    if args.reach:
        _continuous_set_attenuation(dev, [(k,v) for k,v in args.reach.items()], args.duration, args.progress)

    ident = devId(dev)
    for i, a in enumerate(_get_attenuation(dev)):
        logging.info(f"{ident}: Attenuation channel {i+1}: {a} dB")
    sys.exit(0)