Add script to control attenuation of Mini-Circuits RC*DAT
This commit is contained in:
Robert Schmidt
@@ -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)
|
||||
Reference in New Issue
Block a user