<-[[.:start]] ====== AOR AR7030 ====== ===== - Faults/repairs/mods ===== * July 2022 - Volume encoder u/s * Bourns ECW1J-B24-AC0024L available at Farnell at £3.80 * Ordered 4 off 16/7/22 * Replaced vol encoder : July 2022 * July 2022 - VFO Tuning encoder u/s * Only changed freq. over a small portion of a revolution * Using remote control via LAN to operate until a new encoder is sourced ===== - Filters ===== ====Currently fitted:==== Before change (below) ^PCB ref^Description^Model^Bandwidth^ | X3 | AM Narrow | CKK455J | 3.9 | | X4 | AM Normal # | CFW455IT | 5.6 | | X5 | SSB # | CFJ455K14 | 2.1 | | X6 | FM # | CFU455G2 | 9.5 | | X7 | AM Wide # | CFW455HT | 6.4 | | X8 | CW Collins | MF300 | 0.5 | # = factory supplied ====Standard Fitting==== ^PCB ref^Description^Model^Bandwidth^ | X3 | AM Wide | CFW455HT | 6.4 | | X4 | AM Normal | CFW455IT | 5.6 | | X5 | SSB | CFJ455K14 | 2.1 | | X6 | FM | CFU455G2 | 9.5 | | X7 | Option 1 | | | | X8 | Option 2 | | | ====Available to fit:==== CW / Data / SSB narrow : CFJ455K8 : 1.4 === July 2022 === Removed un-needed AM-Wide 6.4kHz from X7 and installed 1.4kHz "CW/Data/SSB Narrow"? This gives : ^PCB ref^Description^Model^Bandwidth^ | X3 | AM Narrow | CKK455J | 3.9 | | X4 | AM Normal # | CFW455IT | 5.6 | | X5 | SSB # | CFJ455K14 | 2.1 | | X6 | FM # | CFU455G2 | 9.5 | | X7 | Data | CFJ455K8 | 1.4 | | X8 | CW Collins | MF300 | 0.5 | # = factory supplied In order of Bandwidth 1 : 0.5 CW 2 : 1.4 CW/Data 3 : 2.1 SSB 4 : 3.9 AM-Narrow 5 : 5.5 AM-Normal 6 : 9.5 FM ===== - Remote Control ===== The Python module used in [[https://github.com/gm4slv/netradio|netradio]] for accessing AR7030. ++++ Generic aor.py | import serial import threading #sport = "/dev/ttyS0" sport = "/dev/ttyUSB1" sbaud = 1200 lock = threading.Lock() fract = float(2 ** 24) / 44545 class Ar7030(object): def __init__(self, model): self.ser = serial.Serial(sport, sbaud, timeout=10) self.model = model def check_bit(self, byte, bit): mask = 1 << bit result = (ord(byte) & mask) if result: return 1 else: return 0 def set_bit(self, byte, bit): mask = 1 << bit result = ord(byte) | mask return result def clear_bit(self, byte, bit): mask = ~(1 << bit) result = (ord(byte) & mask) return result def get_ident(self): sendStr = '\x5f' + '\x40' ident = self.tx_rx(sendStr, False, 8) return ident def get_pre(self): sendStr = '\x50' + '\x32' + '\x48' byte = self.tx_rx(sendStr, False, 1) p = self.check_bit(byte, 4) if p: return 1 else: return 0 return def pre_on(self): #get current 8-bit rxcon byte : # bit0 = filter FS3 # bit1 = filter FS2 # bit2 = filter FS1 # bit3 = filter FS4 # bit4 = preamp enabled # bit5 = atten 0 = 20dB / 1 = 40dB # bit6 = input filter 0 = HF / 1 = LF # bit7 = attenuator enabled sendStr = '\x50' + '\x32' + '\x48' byte = self.tx_rx(sendStr, False, 1) # set bit 4 ON = preamp ON and get the new 8-bit rxcon byte pon = self.set_bit(byte, 4) # split new rxcon byte into two 4-bit nibbles, add 48/96 (\x30 and \x60) high = 48 + (pon >> 4) low = 96 + (pon & 15) sendStr = '\x81' + '\x50' + '\x32' + '\x48' + chr(high) + chr(low) + '\x29' + '\x80' self.tx_rx(sendStr, False, 0) return "Command sent" def pre_off(self): #get current 8-bit rxcon byte sendStr = '\x50' + '\x32' + '\x48' byte = self.tx_rx(sendStr, False, 1) # set bit 4 OFF = preamp OFF and get the new 8-bit rxcon byte pon = self.clear_bit(byte, 4) # split new rxcon byte into two 4-bit nibbles, add 48/96 (\x30 and \x60) high = 48 + (pon >> 4) low = 96 + (pon & 15) sendStr = '\x81' + '\x50' + '\x32' + '\x48' + chr(high) + chr(low) + '\x29' + '\x80' self.tx_rx(sendStr, False, 0) return "Command sent" def get_att(self): sendStr = '\x50' + '\x32' + '\x48' byte = self.tx_rx(sendStr, False, 1) a = self.check_bit(byte, 7) if a: return 1 else: return 0 def att_on(self): #get current 8-bit rxcon byte sendStr = '\x50' + '\x32' + '\x48' byte = self.tx_rx(sendStr, False, 1) # set bit 7 ON = ATT ON and get the new 8-bit rxcon byte pon = self.set_bit(byte, 7) # split new rxcon byte into two 4-bit nibbles, add 48/96 (\x30 and \x60) high = 48 + (pon >> 4) low = 96 + (pon & 15) sendStr = '\x81' + '\x50' + '\x32' + '\x48' + chr(high) + chr(low) + '\x29' + '\x80' self.tx_rx(sendStr, False, 0) return "Command sent" def att_off(self): #get current 8-bit rxcon byte sendStr = '\x50' + '\x32' + '\x48' byte = self.tx_rx(sendStr, False, 1) # set bit 7 OFF = att OFF and get the new 8-bit rxcon byte pon = self.clear_bit(byte, 7) # split new rxcon byte into two 4-bit nibbles, add 48/96 (\x30 and \x60) high = 48 + (pon >> 4) low = 96 + (pon & 15) sendStr = '\x81' + '\x50' + '\x32' + '\x48' + chr(high) + chr(low) + '\x29' + '\x80' self.tx_rx(sendStr, False, 0) return "Command sent" def set_freq(self, freq): fval = freq * fract #print fval b1 = 48 + int(fval / 1048576) fval = fval % 1048576 b2 = 96 + int(fval / 65536) fval = fval % 65536 b3 = 48 + int(fval / 4096) fval = fval % 4096 b4 = 96 + int(fval / 256) fval = fval % 256 b5 = 48 + int(fval / 16) b6 = 96 + int(fval % 16) f_tuple = ( b1, b2, b3, b4, b5, b6 ) freqStr = "" for byte in f_tuple: freqStr += chr(byte) sendStr = '\x81' + '\x50' + '\x31' + '\x4a' + freqStr + '\x24' + '\x80' self.tx_rx(sendStr, False, 0) return "Freq Set" def get_freq(self): sendStr = '\x50' + '\x31' + '\x4a' freqStr = self.tx_rx(sendStr, False, 3) f_val = 0 for k in freqStr: f_val = f_val * 256 + ord(k) return "%.3f" % round(float(f_val / fract),2 ) def set_mode(self, mode): if mode == "lsb": set_mode_val = "\x66" elif mode == "usb": set_mode_val = "\x67" elif mode == "am": set_mode_val = "\x61" elif mode == "cw": set_mode_val = "\x65" elif mode == "data": set_mode_val = "\x64" elif mode == "fm": set_mode_val = "\x63" elif mode == "s-am": set_mode_val = "\x62" else: return "Mode not recognized" sendStr = '\x81' + '\x50' + '\x31' + '\x4d' + set_mode_val + '\x22' + '\x80' self.tx_rx(sendStr, False, 0) return "Mode sent" def get_mode(self): sendStr = "\x50" + "\x31" + "\x4d" m = self.tx_rx(sendStr, False, 1) mode = "" if m == "\x01": mode = "am" elif m == "\x02": mode = "s-am" elif m == "\x03": mode = "fm" elif m == "\x04": mode = "data" elif m == "\x05": mode = "cw" elif m == "\x06": mode = "lsb" elif m == "\x07": mode = "usb" return mode.upper() def get_s(self): sendStr = '\x2e' sraw = self.tx_rx(sendStr, True, 0) return ord(sraw) def get_cal(self): sendStr = '\x52' + '\x3f' + '\x44' + '\x11' #print "sending getcal" cbytes = self.tx_rx(sendStr, False, 8) cal = [] for c in cbytes: cal.append(ord(c)) return cal def get_smeter(self): s = float(self.get_s()) cal = self.get_cal() s1 = s - cal[0] s2 = s1 - cal[1] s3 = s2 - cal[2] s4 = s3 - cal[3] s5 = s4 - cal[4] s6 = s5 - cal[5] s7 = s6 - cal[6] if s1 <= 0: dbm = -123 adj = s / cal[0] * 10 return str(dbm + adj) elif s2 <= 0: dbm = -113 adj = s1 / cal[1] * 10 return str(dbm + adj) elif s3 <= 0: dbm = -103 adj = s2 / cal[2] * 10 return str(dbm + adj) elif s4 <= 0: dbm = -93 adj = s3 / cal[3] * 10 return str(dbm + adj) elif s5 <= 0: dbm = -83 adj = s4 / cal[4] * 10 return str(dbm + adj) elif s6 <= 0: dbm = -73 adj = s5 / cal[5] * 10 return str(dbm + adj) elif s7 <= 0: dbm = -63 adj = s6 / cal[6] * 20 return str(dbm + adj) else: dbm = -43 adj = s7 / cal[7] * 20 return str(dbm + adj) def tx_rx(self, sendStr, reply, n): # apply thread lock lock.acquire() self.ser.write(sendStr) if reply: # for reading S-meter result = "" while not result: result = self.ser.read(1) lock.release() return result else: result = "" byte = "" while n != 0: self.ser.write('\x71') #byte = "" while not byte: byte = self.ser.read(1) result += byte byte = "" n -= 1 lock.release() return result ++++ === Update July 2022 === I've modified the server/client code to create a single-radio AR7030 control program that gives access to IF filter, and reads s-meter etc. {{:public:radio:radio_database:screenshot_2022-08-05_07.24.46.png?direct&400|}} ++++ New (July 2022) version of Tkinter GUI with S-Meter graph and IF filter Selection | # # new class-based gui from Tkinter import * import socket import threading import time class Network(object): def __init__(self): self.sock = self.make_con() def make_con(self): self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.connect(("192.168.21.107", 9999)) return self.sock def connect(self, data): try: lock.acquire() self.sock.sendall(data + "\n") self.received = self.sock.recv(2048) finally: lock.release() return self.received class Dash(object): def __init__(self, master): self.master = master self.s = (n1.sock.getsockname()) self.p = (n1.sock.getpeername()) dash_frame = Toplevel(master, borderwidth = 2, relief = GROOVE, bg = 'black') dash_frame.title("Server") dash_frame.protocol("WM_DELETE_WINDOW", self.handler) dash_frame.resizable(0, 0) dash_frame.geometry("300x150+10-10") dash_frame.grid() self.utc_time = StringVar() Label(dash_frame, textvariable = self.utc_time, fg = 'green', bg = 'black').grid(row = 0, column = 0, sticky = W) Label(dash_frame, text = self.s, bg = 'black', fg = 'white').grid(row = 1, column = 0, sticky = W) Label(dash_frame, text = " < --- > ", bg = 'black', fg = 'white').grid(row = 1, column = 1) Label(dash_frame, text = self.p, bg = 'black', fg = 'white').grid(row = 1, column = 2, sticky = E) self.server_msg_l = StringVar() Label(dash_frame, textvariable = self.server_msg_l, fg = 'yellow', bg = 'black').grid(row = 2, column = 0, columnspan = 3, sticky = W) q_button = Button(dash_frame, text = "Quit", command = lambda: close()) q_button.grid(row = 3, column = 0, sticky = E) def handler(self): pass def up_dash(self): self.server_msg = n1.connect("ident") self.server_msg_l.set(self.server_msg) self.utc_time.set(time.strftime("%d/%m/%Y %H:%M", time.gmtime(time.time()))) return class nRadio(object): def __init__(self, master, radio): self.master = master self.radio = radio self.num = "".join(("r", self.radio)) radio_frame = Frame(master, borderwidth=2, relief=GROOVE) radio_frame.grid() self.smeter_list = [] filter_frame = Frame(radio_frame) filter_frame.grid(row=5, column=0, columnspan=3) mode_frame = Frame(radio_frame) mode_frame.grid(row=3, column=0, columnspan=3, rowspan=2) control_frame = Frame(radio_frame) control_frame.grid(row=5, column=4, columnspan=3) q_button = Checkbutton(control_frame, text = "Quit", command = lambda: close()) q_button.grid(row = 1, column = 1, sticky = W) Label(radio_frame, text="Name", width=10).grid(row=0, column=0) Label(radio_frame, text="Freq/kHz", width=10).grid(row=0, column=1) Label(radio_frame, text="Mode", width=10).grid(row=0, column=2) Label(radio_frame, text="Signal/dBm", width=10).grid(row=0, column=3) Label(radio_frame, text="Max/dBm", width=10, fg='red').grid(row=0, column=4) Label(radio_frame, text="Ave/dBm", width=10, fg='green').grid(row=0, column=5) Label(radio_frame, text="Min/dBm", width=10, fg='blue').grid(row=0, column=6) self.name_l = StringVar() self.l_name = Label(radio_frame, textvariable=self.name_l, width=10) self.l_name.grid(row=1, column=0, sticky=W) self.freq_l = StringVar() self.l_freq = Label(radio_frame, fg='red', textvariable=self.freq_l, width=10) self.l_freq.grid(row=1, column=1, sticky=E) self.mode_l = StringVar() self.l_mode = Label(radio_frame, textvariable=self.mode_l, width=10) self.l_mode.grid(row=1, column=2, sticky=E) self.filt_l = IntVar() self.r_filt_1 = Radiobutton(filter_frame, variable=self.filt_l, value=1, text="0.5", command = lambda: self.set_filt(self.num)) self.r_filt_1.grid(row=5, column=2) self.r_filt_2 = Radiobutton(filter_frame, variable=self.filt_l, value=2, text="1.4", command = lambda: self.set_filt(self.num)) self.r_filt_2.grid(row=5, column=3) self.r_filt_3 = Radiobutton(filter_frame, variable=self.filt_l, value=3, text="2.1", command = lambda: self.set_filt(self.num)) self.r_filt_3.grid(row=5, column=4) self.r_filt_4 = Radiobutton(filter_frame, variable=self.filt_l, value=4, text="3.9", command = lambda: self.set_filt(self.num)) self.r_filt_4.grid(row=6, column=2) self.r_filt_5 = Radiobutton(filter_frame, variable=self.filt_l, value=5, text="5.5", command = lambda: self.set_filt(self.num)) self.r_filt_5.grid(row=6, column=3) self.r_filt_6 = Radiobutton(filter_frame, variable=self.filt_l, value=6, text="9.5", command = lambda: self.set_filt(self.num)) self.r_filt_6.grid(row=6, column=4) #Label(filter_frame, text="Filter", width=10).grid(row=4, column=2, columnspan=3, sticky=NSEW) #self.l_filt = Label(radio_frame, textvariable=self.filt_l, width=10) #self.l_filt.grid(row=5, column=0, sticky=E) #self.e_filt = Entry(radio_frame, width=10, bg = 'white', fg = 'black', insertbackground = 'blue') #self.e_filt.grid(row=6, column=0) #self.e_filt.bind('', (lambda event: self.set_filt(self.num))) #self.e_mode = StringVar() self.b_mode_usb = Radiobutton(mode_frame, text = "USB",variable=self.mode_l, value="USB", width = 4, command = lambda: self.set_mode("usb", self.num)) self.b_mode_usb.grid(row = 0, column = 0) self.b_mode_lsb = Radiobutton(mode_frame, text = "LSB",variable=self.mode_l, value="LSB", width = 4, command = lambda: self.set_mode("lsb", self.num)) self.b_mode_lsb.grid(row = 0, column = 1) self.b_mode_cw = Radiobutton(mode_frame, text = "CW",variable=self.mode_l, value="CW", width = 4, command = lambda: self.set_mode("cw", self.num)) self.b_mode_cw.grid(row = 0, column = 2) self.b_mode_am = Radiobutton(mode_frame, text = "AM",variable=self.mode_l,value = "AM", width = 4, command = lambda: self.set_mode("am", self.num)) self.b_mode_am.grid(row = 1, column = 1) self.b_mode_am = Radiobutton(mode_frame, text = "S-AM",variable=self.mode_l,value = "S-AM", width = 4, command = lambda: self.set_mode("s-am", self.num)) self.b_mode_am.grid(row = 1, column = 3) self.b_mode_data = Radiobutton(mode_frame, text = "Data",variable=self.mode_l, value="DATA", width = 4, command = lambda: self.set_mode("data", self.num)) self.b_mode_data.grid(row = 1, column = 0) self.b_mode_fm = Radiobutton(mode_frame, text = "FM",variable=self.mode_l, value="FM", width = 4, command = lambda: self.set_mode("fm", self.num)) self.b_mode_fm.grid(row = 1, column = 2) self.b_qsy_up = Button(radio_frame, text = "+1kHz", width = 6, command = lambda: self.qsy_up( self.num)) self.b_qsy_up.grid(row = 2, column = 2) self.b_qsy_down = Button(radio_frame, text = "-1kHz", width = 6, command = lambda: self.qsy_down( self.num)) self.b_qsy_down.grid(row = 2, column = 0) self.b_update = Button(radio_frame, text = "Get", width = 6, command = lambda:self.get_all()) #self.b_update.grid(row = 0, column = 3) self.smeter_l = StringVar() self.l_smeter = Label(radio_frame, textvariable=self.smeter_l, width=10) self.l_smeter.grid(row=1, column=3, sticky=E) self.max_var = StringVar() self.l_max = Label(radio_frame, textvariable=self.max_var, width=10) self.l_max.grid(row=1, column=4) self.ave_var = StringVar() self.l_average = Label(radio_frame, textvariable=self.ave_var, width=10) self.l_average.grid(row=1, column=5) self.min_var = StringVar() self.l_min = Label(radio_frame, textvariable=self.min_var, width=10) self.l_min.grid(row=1, column=6) self.e_freq = Entry(radio_frame, width=10, bg = 'white', fg = 'black', insertbackground = 'blue') self.e_freq.grid(row=2, column=1) self.e_freq.focus() self.e_freq.bind('', (lambda event: self.set_freq(self.num))) self.pre = IntVar() self.cb_pre = Checkbutton(control_frame,text="Preamp", variable=self.pre, command=lambda: self.preamp_onoff(self.num)) self.cb_pre.grid(row=0, column=0, sticky=W) self.l_pre = Label(control_frame, text="Preamp") #self.l_pre.grid(row=0, column=0, sticky=E) self.att = IntVar() self.cb_att = Checkbutton(control_frame,text="Att", variable=self.att, command=lambda: self.att_onoff(self.num)) self.cb_att.grid(row=1, column=0, sticky=W) self.l_att = Label(control_frame, text="Att") #self.l_att.grid(row=1, column=0, sticky=E) self.set_log = IntVar() self.cb_log = Checkbutton(control_frame,text = "Log to file...", variable = self.set_log) self.cb_log.grid(row = 0, column = 1, sticky = W) self.l_log = Label(control_frame, width = 10, text = "Log to file...") #self.l_log.grid(row = 2, column = 0, sticky = E) self.c1 = Canvas(radio_frame, width=260, height=100, bg='black') self.c1.grid(row=2, column=3, columnspan=4, rowspan=3) name = self.get_name(radio) def get_all(self): self.get_freq(self.num) self.get_mode(self.num) self.get_smeter(self.num) self.get_preamp(self.num) self.get_atten(self.num) self.graph_points() self.avg() self.max() self.min() self.get_filt(self.num) if self.set_log.get() == 1: if int(time.time()) % 10.0 == 0: self.write_file() return def get_filt(self,radio): self.filt = n1.connect("getfil" + " " + radio) self.filt_l.set(self.filt) return def set_filt(self,radio): self.filt = str(self.filt_l.get()) self.newfilt = n1.connect("setfil" + " " + self.filt + " " + radio) #self.e_filt.delete(0,END) self.get_all() return def get_freq(self, radio): self.freq = n1.connect("getfreq" + " " + radio) self.freq_l.set(self.freq) return def set_freq(self, radio): self.freq = str(self.e_freq.get()) self.newfreq = n1.connect("setfreq" + " " + self.freq + " " + radio) self.e_freq.delete(0, END) self.get_all() return def qsy_up(self,radio): oldf = n1.connect("getfreq" + " " + radio) newf = str(float(oldf) + 1) freq = n1.connect("setfreq" + " " + newf + " " + radio) self.get_all() return def qsy_down(self,radio): oldf = n1.connect("getfreq" + " " + radio) newf = str(float(oldf) - 1) freq = n1.connect("setfreq" + " " + newf + " " + radio) self.get_all() return def get_mode(self, radio): self.mode = n1.connect("getmode" + " " + radio) self.mode_l.set(self.mode) return def set_mode(self, mode, radio): #self.mode = str(self.e_mode.get()) self.newmode = n1.connect("setmode" + " " + mode + " " + radio) #self.e_mode.delete(0, END) self.get_all() return def get_smeter(self, radio): self.smeter = n1.connect("getsmeter" + " " + radio) self.smeter_l.set(self.smeter) self.smeter_list.append(float(self.smeter)) if len(self.smeter_list) > 129: self.smeter_list.pop(0) return def avg(self): s = self.smeter_list total = 0 for i in s: total += i self.av = round((total / len(s)), 1) self.ave_var.set(str(self.av)) self.c1.create_line(0, 100 - (self.av + 123 + 5), 310, 100 - (self.av + 123 + 5), fill="green") return self.av def min(self): s = self.smeter_list self.mn = s[0] for i in s: if i < self.mn: self.mn = i self.min_var.set(str(self.mn)) self.c1.create_line(0, 100 - (self.mn + 123 + 5), 310, 100 - (self.mn + 123 + 5), fill="blue") return self.mn def max(self): s = self.smeter_list self.mx = s[0] for i in s: if i > self.mx: self.mx = i self.max_var.set(str(self.mx)) self.c1.create_line(0, 100 - (self.mx + 123 + 5), 310, 100 - (self.mx + 123 + 5), fill="red") return self.mx def get_preamp(self, radio): self.preamp = n1.connect("getpreamp" + " " + radio) if self.preamp == "1": self.cb_pre.select() elif self.preamp == "0": self.cb_pre.deselect() return def preamp_onoff(self, radio): self.prestate = self.pre.get() if self.prestate: n1.connect("preampon" + " " + radio) else: n1.connect("preampoff" + " " + radio) self.get_all() return def att_onoff(self, radio): self.attstate = self.att.get() if self.attstate: n1.connect("atton" + " " + radio) else: n1.connect("attoff" + " " + radio) self.get_all() return def get_atten(self, radio): self.atten = n1.connect("getatt" + " " + radio) if self.atten == "1": self.cb_att.select() elif self.atten == "0": self.cb_att.deselect() return def get_name(self, radio): print radio self.all_names = n1.connect("listradios") self.radios = self.all_names.split() print self.radios self.name = self.radios[int(radio) - 1] self.name_l.set(self.name) return def graph_points(self): seq = self.smeter_list y_stretch = 1 y_gap = 5 x_stretch = 0 x_width = 2 x_gap = 2 height = 100 self.c1.delete(ALL) self.c1.create_line(0, 100 - (-73 + 123 + 5), 310, 100 - (-73 + 123 + 5), fill='white') for x, y in enumerate(seq): yd = y + 123 x0 = x * x_stretch + x * x_width + x_gap y0 = height - (yd * y_stretch + y_gap) x1 = x * x_stretch + x * x_width + x_width + x_gap y1 = height - y_gap self.c1.create_rectangle(x0, y0, x1, y1, fill="yellow") def write_file(self): self.filename = self.name+".txt" self.f = open(self.filename, 'a+') timenow = time.strftime("%d/%m/%Y %H:%M:%S", time.gmtime(time.time())) log = " ".join((timenow, self.name, self.mode, self.freq, self.smeter, self.preamp, self.atten, "\r\n")) self.f.write(log) self.f.close() def update(): #loops = 0 try: nRadio1.get_all() except: pass try: nRadio2.get_all() except: pass try: nRadio3.get_all() except: pass try: nRadio4.get_all() except: pass #d1.up_dash() return def main(): #loops = 0 while True: try: nRadio1.get_all() except: pass try: nRadio2.get_all() except: pass try: nRadio3.get_all() except: pass try: nRadio4.get_all() except: pass #d1.up_dash() #loops += 1 #print threading.currentThread().name, loops time.sleep(1) def close(): n1.connect("quit") root.destroy() if __name__ == "__main__": version = "v0.1" lock = threading.Lock() root = Tk() #w, h = root.winfo_screenwidth(), root.winfo_screenheight() #root.geometry("%dx%d+0+0" % (w, h)) #root.geometry("300x520+0+0") root.title("AR7030 " + version) #root.withdraw() n1 = Network() #radio_count = (n1.connect("count")) #radio_count = 3 radio_count = int(n1.connect("count")) #if radio_count > 0: # nRadio1 = nRadio(root, "1") #if radio_count > 1: # nRadio2 = nRadio(root, "2") #if radio_count > 2: # nRadio3 = nRadio(root, "3") #if radio_count > 3: # nRadio4 = nRadio(root, "4") nRadio1 = nRadio(root, "1") #d1 = Dash(root) #print threading.currentThread().name m1 = threading.Thread(target = main) m1.setDaemon(True) m1.start() update() root.mainloop() ++++ ++++ New (July 2022) version of aor module, with access to IF filter selection | import serial import threading #sport = "/dev/ttyS0" sport = "/dev/ttyS0" sbaud = 1200 lock = threading.Lock() fract = float(2 ** 24) / 44545 class Ar7030(object): def __init__(self, model): self.ser = serial.Serial(sport, sbaud, timeout=10) self.model = model self.filter = self.get_filter() def check_bit(self, byte, bit): mask = 1 << bit result = (ord(byte) & mask) if result: return 1 else: return 0 def set_bit(self, byte, bit): mask = 1 << bit result = ord(byte) | mask return result def clear_bit(self, byte, bit): mask = ~(1 << bit) result = (ord(byte) & mask) return result def get_ident(self): sendStr = '\x5f' + '\x40' ident = self.tx_rx(sendStr, False, 8) return ident def get_pre(self): sendStr = '\x50' + '\x32' + '\x48' byte = self.tx_rx(sendStr, False, 1) p = self.check_bit(byte, 4) if p: return 1 else: return 0 return def get_filter(self): sendStr= '\x50' + '\x33' + '\x44' byte = self.tx_rx(sendStr, False, 1) #print ord(byte) return ord(byte) def set_filter(self,filt_no): #print "In set_filter with ", filt_no try: filt_no = int(filt_no) except ValueError: return "Not a number" if filt_no > 6: return "Out of range" high = 48 + (filt_no >> 4) low = 96 + (filt_no & 15) sendStr= '\x81' + '\x50' + '\x33' + '\x44' + chr(high) + chr(low) + '\x23' + '\x80' self.tx_rx(sendStr, False,0) #self.get_filter() return "Command Sent" def pre_on(self): #get current 8-bit rxcon byte : # bit0 = filter FS3 # bit1 = filter FS2 # bit2 = filter FS1 # bit3 = filter FS4 # bit4 = preamp enabled # bit5 = atten 0 = 20dB / 1 = 40dB # bit6 = input filter 0 = HF / 1 = LF # bit7 = attenuator enabled sendStr = '\x50' + '\x32' + '\x48' byte = self.tx_rx(sendStr, False, 1) # set bit 4 ON = preamp ON and get the new 8-bit rxcon byte pon = self.set_bit(byte, 4) # split new rxcon byte into two 4-bit nibbles, add 48/96 (\x30 and \x60) high = 48 + (pon >> 4) low = 96 + (pon & 15) sendStr = '\x81' + '\x50' + '\x32' + '\x48' + chr(high) + chr(low) + '\x29' + '\x80' self.tx_rx(sendStr, False, 0) return "Command sent" def pre_off(self): #get current 8-bit rxcon byte sendStr = '\x50' + '\x32' + '\x48' byte = self.tx_rx(sendStr, False, 1) # set bit 4 OFF = preamp OFF and get the new 8-bit rxcon byte pon = self.clear_bit(byte, 4) # split new rxcon byte into two 4-bit nibbles, add 48/96 (\x30 and \x60) high = 48 + (pon >> 4) low = 96 + (pon & 15) sendStr = '\x81' + '\x50' + '\x32' + '\x48' + chr(high) + chr(low) + '\x29' + '\x80' self.tx_rx(sendStr, False, 0) return "Command sent" def get_att(self): sendStr = '\x50' + '\x32' + '\x48' byte = self.tx_rx(sendStr, False, 1) a = self.check_bit(byte, 7) if a: return 1 else: return 0 def att_on(self): #get current 8-bit rxcon byte sendStr = '\x50' + '\x32' + '\x48' byte = self.tx_rx(sendStr, False, 1) # set bit 7 ON = ATT ON and get the new 8-bit rxcon byte pon = self.set_bit(byte, 7) # split new rxcon byte into two 4-bit nibbles, add 48/96 (\x30 and \x60) high = 48 + (pon >> 4) low = 96 + (pon & 15) sendStr = '\x81' + '\x50' + '\x32' + '\x48' + chr(high) + chr(low) + '\x29' + '\x80' self.tx_rx(sendStr, False, 0) return "Command sent" def att_off(self): #get current 8-bit rxcon byte sendStr = '\x50' + '\x32' + '\x48' byte = self.tx_rx(sendStr, False, 1) # set bit 7 OFF = att OFF and get the new 8-bit rxcon byte pon = self.clear_bit(byte, 7) # split new rxcon byte into two 4-bit nibbles, add 48/96 (\x30 and \x60) high = 48 + (pon >> 4) low = 96 + (pon & 15) sendStr = '\x81' + '\x50' + '\x32' + '\x48' + chr(high) + chr(low) + '\x29' + '\x80' self.tx_rx(sendStr, False, 0) return "Command sent" def set_freq(self, freq): fval = freq * fract #print fval b1 = 48 + int(fval / 1048576) fval = fval % 1048576 b2 = 96 + int(fval / 65536) fval = fval % 65536 b3 = 48 + int(fval / 4096) fval = fval % 4096 b4 = 96 + int(fval / 256) fval = fval % 256 b5 = 48 + int(fval / 16) b6 = 96 + int(fval % 16) f_tuple = ( b1, b2, b3, b4, b5, b6 ) freqStr = "" for byte in f_tuple: freqStr += chr(byte) sendStr = '\x81' + '\x50' + '\x31' + '\x4a' + freqStr + '\x24' + '\x80' self.tx_rx(sendStr, False, 0) return "Freq Set" def get_freq(self): sendStr = '\x50' + '\x31' + '\x4a' freqStr = self.tx_rx(sendStr, False, 3) f_val = 0 for k in freqStr: f_val = f_val * 256 + ord(k) return "%.3f" % round(float(f_val / fract),2 ) def set_mode(self, mode): if mode == "lsb": set_mode_val = "\x66" elif mode == "usb": set_mode_val = "\x67" elif mode == "am": set_mode_val = "\x61" elif mode == "cw": set_mode_val = "\x65" elif mode == "data": set_mode_val = "\x64" elif mode == "fm": set_mode_val = "\x63" elif mode == "s-am": set_mode_val = "\x62" else: return "Mode not recognized" sendStr = '\x81' + '\x50' + '\x31' + '\x4d' + set_mode_val + '\x22' + '\x80' self.tx_rx(sendStr, False, 0) self.filter = self.get_filter() return "Mode sent" def get_mode(self): sendStr = "\x50" + "\x31" + "\x4d" m = self.tx_rx(sendStr, False, 1) mode = "" if m == "\x01": mode = "am" elif m == "\x02": mode = "s-am" elif m == "\x03": mode = "fm" elif m == "\x04": mode = "data" elif m == "\x05": mode = "cw" elif m == "\x06": mode = "lsb" elif m == "\x07": mode = "usb" return mode.upper() def get_s(self): sendStr = '\x2e' sraw = self.tx_rx(sendStr, True, 0) return ord(sraw) def get_cal(self): sendStr = '\x52' + '\x3f' + '\x44' + '\x11' #print "sending getcal" cbytes = self.tx_rx(sendStr, False, 8) cal = [] for c in cbytes: cal.append(ord(c)) #print cal return cal def get_smeter(self): s = float(self.get_s()) cal = self.get_cal() s1 = s - cal[0] s2 = s1 - cal[1] s3 = s2 - cal[2] s4 = s3 - cal[3] s5 = s4 - cal[4] s6 = s5 - cal[5] s7 = s6 - cal[6] if s1 <= 0: dbm = -123 adj = s / cal[0] * 10 return str(dbm + adj) elif s2 <= 0: dbm = -113 adj = s1 / cal[1] * 10 return str(dbm + adj) elif s3 <= 0: dbm = -103 adj = s2 / cal[2] * 10 return str(dbm + adj) elif s4 <= 0: dbm = -93 adj = s3 / cal[3] * 10 return str(dbm + adj) elif s5 <= 0: dbm = -83 adj = s4 / cal[4] * 10 return str(dbm + adj) elif s6 <= 0: dbm = -73 adj = s5 / cal[5] * 10 return str(dbm + adj) elif s7 <= 0: dbm = -63 adj = s6 / cal[6] * 20 return str(dbm + adj) else: dbm = -43 adj = s7 / cal[7] * 20 return str(dbm + adj) def tx_rx(self, sendStr, reply, n): # apply thread lock lock.acquire() self.ser.write(sendStr) if reply: # for reading S-meter result = "" while not result: result = self.ser.read(1) lock.release() return result else: result = "" byte = "" while n != 0: self.ser.write('\x71') #byte = "" while not byte: byte = self.ser.read(1) result += byte byte = "" n -= 1 lock.release() return result ++++ ++++ New (July 2022) server with AR7030 filter selection | #from aor import * from aor_filter import * #from icom import * from conf import * #from m710 import * import SocketServer import time try: import readline except: pass lock = threading.Lock() radios = [] #r1 = Icom(n1, a1, cal1) #radios.append(n1) #r2 = m710(n4) #radios.append(n4) #r2 = Icom(n2, a2, cal2) #radios.append(n2) #r3 = Ar7030(n3) #radios.append(n3) r1 = Ar7030(n3) radios.append(n3) print radios #print r1.digi_off() #print r2.remote_on() def count_radios(): count = len(radios) return count def list_radios(): radiolist = "" for n in range(0, len(radios)): r = radios[n] radiolist += (r + " ") return radiolist def write_file(text): filename = 'commandlog.txt' f = open(filename, 'a+') # a+ is "append to file, create it if it doesn't exist" timenow = time.strftime("%d/%m/%Y %H:%M:%S", time.gmtime(time.time())) log = " ".join((timenow, text)) # make an entry for the log by joining the timestamp with the text passed in f.write(log) f.close() def write_con(text): filename = 'conlog.txt' f = open(filename, 'a+') # a+ is "append to file, create it if it doesn't exist" timenow = time.strftime("%d/%m/%Y %H:%M:%S", time.gmtime(time.time())) log = " ".join((timenow, text)) # make an entry for the log by joining the timestamp with the text passed in f.write(log) f.close() # The Server class ThreadedRequestHandler(SocketServer.StreamRequestHandler): def handle(self): # we find the current thread for the client connection just set up, to # use in the log file cur_thread = threading.currentThread() # log the new connection details write_con("Connect from %s using %s \n" % ((self.client_address[0]), cur_thread.getName())) # print to the server's console the new connection IP address/port print self.client_address # loop to handle client requests.... while True: # using StreamRequestHandler means our input from the client # is "file-like" and can be read with "file-like" commands # we read a line at a time, using readline() cmd = self.rfile.readline().lower() # to keep things clean, we remove any characters that aren't # "printable" simple ASCII # these are between 32 and 127 in the ASCII table # we look at each character, and then make a new word by # .join()ing each accepted character with no space in between asccmd = "".join(x for x in cmd if ord(x) < 128 and ord(x) > 31) # we make a list called "words" holding the received words which # will be inspected by various functions words = asccmd.split() # If a client uses sock.close() itself, to disconnect, it appears that # we read a continuous stream of "" on the dead # socket, which puts CPU to 100%. # # The "While" loop is probably responsible, but I can't see another # way to keep the connection up for multiple commands. # # Further connection are accepted due to the Threaded nature of the server. # The CPU load is unacceptable though # HACK ?>>>>> # Looking for "" and then breaking # the connection from the server end (even though the client has # gone) cures this. if cmd == "": break else: pass # if the words list is empty, go back and get more input if not words: continue # we have input.... # filter based on the first word - these are the # pre-set commands the server will accept # the client wants to know the currently available # radio names - held in the variable "rnames" elif words[0] == "getnames": self.wfile.write(rnames) # words[-1] (the last word in the list) will always be the # radio name. We give the variable "my_radio" this value, for # identifying which radio object to apply the method to elif words[0] == "count": count = count_radios() self.wfile.write(count) elif words[0] == "ident": ident_text = "GM4SLV Radio Server" radio_list = list_radios() self.wfile.write(ident_text + "\rAvailable : \n" + radio_list + "\r\n") elif words[0] == "setfil": my_radio = eval(words[-1]) filt_no = words[1] newfilt = my_radio.set_filter(filt_no) self.wfile.write(newfilt) elif words[0] == "getfil": my_radio = eval(words[1]) filt = my_radio.get_filter() self.wfile.write(filt) elif words[0] == "getmode": my_radio = eval(words[-1]) mode = my_radio.get_mode() self.wfile.write(mode) elif words[0] == "getfreq": my_radio = eval(words[-1]) freq = words[1] freq = my_radio.get_freq() self.wfile.write(freq) elif words[0] == "setmode": my_radio = eval(words[-1]) mode = words[1] newmode = my_radio.set_mode(mode) self.wfile.write(newmode) elif words[0] == "setfreq": my_radio = eval(words[-1]) try: freq = float(words[1]) newfreq = my_radio.set_freq(freq) self.wfile.write(newfreq) except ValueError: #freq = float(my_radio.get_freq()) self.wfile.write("Error in freq. %s No change\r\n" % words[1]) elif words[0] == "getsmeter": my_radio = eval(words[-1]) smeter = round(float(my_radio.get_smeter()), 1) self.wfile.write(smeter) elif words[0] == "gets": my_radio = eval(words[-1]) s = my_radio.get_s() self.wfile.write(s) elif words[0] == "listradios": radios = list_radios() self.wfile.write(radios) elif words[0] == "getpreamp": my_radio = eval(words[-1]) preamp = my_radio.get_pre() self.wfile.write(preamp) elif words[0] == "preampon": my_radio = eval(words[-1]) preamp = my_radio.pre_on() self.wfile.write(preamp) elif words[0] == "preamp2on": my_radio = eval(words[-1]) preamp = my_radio.pre_2_on() self.wfile.write(preamp) elif words[0] == "preampoff": my_radio = eval(words[-1]) preamp = my_radio.pre_off() self.wfile.write(preamp) elif words[0] == "getatt": my_radio = eval(words[-1]) att = my_radio.get_att() self.wfile.write(att) elif words[0] == "atton": my_radio = eval(words[-1]) att = my_radio.att_on() self.wfile.write(att) elif words[0] == "attoff": my_radio = eval(words[-1]) att = my_radio.att_off() self.wfile.write(att) elif words[0] == "tune": my_radio = eval(words[-1]) tune = my_radio.tune() self.wfile.write(tune) elif words[0] == "getpwr": my_radio = eval(words[-1]) pwr = my_radio.get_pwr() self.wfile.write(pwr) elif words[0] == "setpwr": my_radio = eval(words[-1]) spwr = words[1] pwr = my_radio.set_pwr(spwr) self.wfile.write(pwr) elif words[0] == "quit": write_con("Got quit from {}\n".format(self.client_address[0])) # log it self.wfile.write("Goodbye! \r\n") # say Goodbye break else: # nothing in words[0] matches a pre-set command.... write_file("Received %s\n" % words) # log it, it's unusual self.wfile.write("Command not recognized\r\n") # inform the client class ThreadedIcomServer(SocketServer.ThreadingMixIn, SocketServer.TCPServer): pass if __name__ == '__main__': # define the lock to be used on the serial port access lock = threading.Lock() # address ('' = all available interfaces) to listen on, and port number address = ('', 9999) server = ThreadedIcomServer(address, ThreadedRequestHandler) server.allow_reuse_address = True # define that the server will be threaded, and will serve "forever" ie. not quit after the client disconnects t = threading.Thread(target=server.serve_forever) # start the server thread t.start() write_con( "Server loop running in thread: %s\n" % "".join(t.getName())) ++++ ===== Documentation ===== * {{ :public:radio:radio_database:7030_-_corruption_agc_table_-_preset_test.pdf |}} * {{ :public:radio:radio_database:7030serv.pdf |}} ==== Page Info ==== Page created Wed May 25 00:04:50 2022 by John Pumford-Green Page last updated: ~~LASTMOD~~ ==== tags ==== {{tag>radio ar7030}}