From 8c85acbba66e3023a416dd82f1a6d0c82e6c1925 Mon Sep 17 00:00:00 2001 From: Robert Schmidt Date: Tue, 23 Sep 2025 11:59:48 +0200 Subject: [PATCH] Add script to control attenuation of Mini-Circuits RC*DAT --- ci-scripts/attenuatorctl.py | 175 ++++++++++++++++++++++++++++++++++++ 1 file changed, 175 insertions(+) create mode 100644 ci-scripts/attenuatorctl.py diff --git a/ci-scripts/attenuatorctl.py b/ci-scripts/attenuatorctl.py new file mode 100644 index 0000000000..e6f8c01837 --- /dev/null +++ b/ci-scripts/attenuatorctl.py @@ -0,0 +1,175 @@ +#!/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)