ADS TechのRDX-155-EFっていうUSBラジオチューナーを発掘したんだけど、Windows 11なPCに接続したらドライバはインストールされるものの、ツールがなかったのでChatGPTにPythonツールを作ってもらってみた。
USBで接続すると普通にオーディオデバイスとして認識する。(VID:06E1 PID:A155)
録音デバイスとして認識するのでこのデバイスを聴くを有効にするとサーっていうノイズが聞こえる感じ。ラジオの周波数を合わせるのに他のアプリが必要な感じ。
Linuxだとツールがあるっぽいので、ChatGPTでPythonツールを作ってみた。デバイス自体はHIDのコマンドで制御できるっぽいし、使用しているチップがSi470xって言うことがわかっているのでLinux用のツールからコマンドがわかってるっぽいのでそんなに大変ではないらしい。
import tkinter as tk
from tkinter import ttk, messagebox
import hid
import threading
import time
# ============================================================
# RDX-155 / ADS InstantFM Music
#
# USB:
# VID = 0x06E1
# PID = 0xA155
#
# Si470x:
# Report 1..16 = registers 0x00..0x0F
# Report 18 = STATUSRSSI + READCHAN + RDSA..RDSD
#
# Windows
# Python + hidapi + Tkinter
# ============================================================
VID = 0x06E1
PID = 0xA155
# Si470x register numbers
REG_DEVICEID = 0x00
REG_CHIPID = 0x01
REG_POWERCFG = 0x02
REG_CHANNEL = 0x03
REG_SYSCONFIG1 = 0x04
REG_SYSCONFIG2 = 0x05
REG_SYSCONFIG3 = 0x06
REG_TEST1 = 0x07
REG_TEST2 = 0x08
REG_BOOTCONFIG = 0x09
REG_STATUSRSSI = 0x0A
REG_READCHAN = 0x0B
REG_RDSA = 0x0C
REG_RDSB = 0x0D
REG_RDSC = 0x0E
REG_RDSD = 0x0F
# HID
RDS_REPORT = 18
# STATUSRSSI bits
STATUS_RDSR = 1 << 15
STATUS_STC = 1 << 14
STATUS_SFBL = 1 << 13
STATUS_AFCRL = 1 << 12
STATUS_RDSS = 1 << 11
# CHANNEL bits
CHANNEL_TUNE = 1 << 15
CHANNEL_MASK = 0x03FF
# Si470x default band:
# 76.0 - 108.0 MHz
# 100 kHz channel spacing
BAND_MIN = 76.0
BAND_MAX = 108.0
CHANNEL_SPACE = 0.1
class RDX155:
def __init__(self):
self.dev = None
self.path = None
self.lock = threading.Lock()
# --------------------------------------------------------
# Find device
# --------------------------------------------------------
def find(self):
devices = hid.enumerate(VID, PID)
if not devices:
return None
# Prefer interface 2, which was found on this RDX-155
for d in devices:
if d.get("interface_number") == 2:
return d
return devices[0]
# --------------------------------------------------------
# Open
# --------------------------------------------------------
def open(self):
info = self.find()
if info is None:
raise RuntimeError(
"RDX-155 (VID=06E1 PID=A155) が見つかりません。"
)
self.path = info["path"]
self.dev = hid.device()
# Open the exact HID interface/path.
self.dev.open_path(self.path)
try:
self.dev.set_nonblocking(False)
except Exception:
pass
return info
# --------------------------------------------------------
# Close
# --------------------------------------------------------
def close(self):
if self.dev is not None:
try:
self.dev.close()
except Exception:
pass
self.dev = None
# --------------------------------------------------------
# HID register read
# --------------------------------------------------------
def read_reg(self, reg):
if self.dev is None:
raise RuntimeError("デバイスが接続されていません")
report_id = reg + 1
with self.lock:
data = self.dev.get_feature_report(report_id, 3)
if data is None or len(data) < 3:
raise RuntimeError(
f"Report {report_id} の読み取りに失敗しました: {data}"
)
value = (data[1] << 8) | data[2]
return value
# --------------------------------------------------------
# HID register write
# --------------------------------------------------------
def write_reg(self, reg, value):
if self.dev is None:
raise RuntimeError("デバイスが接続されていません")
report_id = reg + 1
value &= 0xFFFF
data = [
report_id,
(value >> 8) & 0xFF,
value & 0xFF
]
with self.lock:
result = self.dev.send_feature_report(data)
return result
# --------------------------------------------------------
# Read all registers
# --------------------------------------------------------
def read_all(self):
regs = {}
for r in range(16):
regs[r] = self.read_reg(r)
return regs
# --------------------------------------------------------
# Frequency
#
# CHANNEL = (frequency - 76.0) / 0.1
#
# Example:
# 76.0 MHz -> channel 0
# 94.9 MHz -> channel 189
# 108.0 MHz -> channel 320
# --------------------------------------------------------
def frequency_to_channel(self, freq):
if freq < BAND_MIN or freq > BAND_MAX:
raise ValueError(
f"周波数は {BAND_MIN:.1f}～{BAND_MAX:.1f} MHz です"
)
channel = round((freq - BAND_MIN) / CHANNEL_SPACE)
if channel < 0 or channel > CHANNEL_MASK:
raise ValueError("CHANNEL範囲外です")
return channel
def channel_to_frequency(self, channel):
return BAND_MIN + channel * CHANNEL_SPACE
# --------------------------------------------------------
# Tune
# --------------------------------------------------------
def set_frequency(self, freq):
channel = self.frequency_to_channel(freq)
# Read current CHANNEL register so we preserve
# unrelated bits.
current = self.read_reg(REG_CHANNEL)
# Remove old channel and TUNE
value = current & ~(CHANNEL_MASK | CHANNEL_TUNE)
# Set channel + TUNE
value |= channel
value |= CHANNEL_TUNE
# Start tuning
self.write_reg(REG_CHANNEL, value)
# Wait for STC
start = time.monotonic()
while True:
status = self.read_reg(REG_STATUSRSSI)
if status & STATUS_STC:
break
if time.monotonic() - start > 2.0:
raise TimeoutError(
"TUNE完了(STC)を2秒以内に確認できませんでした"
)
time.sleep(0.02)
# Clear TUNE
value &= ~CHANNEL_TUNE
self.write_reg(REG_CHANNEL, value)
# Give the tuner a little time to settle
time.sleep(0.05)
return self.get_frequency()
# --------------------------------------------------------
# Current frequency
# --------------------------------------------------------
def get_frequency(self):
channel = self.read_reg(REG_READCHAN) & CHANNEL_MASK
return self.channel_to_frequency(channel)
# --------------------------------------------------------
# RSSI / stereo / status
# --------------------------------------------------------
def get_status(self):
status = self.read_reg(REG_STATUSRSSI)
rssi = status & 0xFF
stereo = bool(status & (1 << 8))
rds_ready = bool(status & STATUS_RDSR)
rds_sync = bool(status & STATUS_RDSS)
stc = bool(status & STATUS_STC)
return {
"raw": status,
"rssi": rssi,
"stereo": stereo,
"rds_ready": rds_ready,
"rds_sync": rds_sync,
"stc": stc,
}
# --------------------------------------------------------
# Volume
#
# SYSCONFIG2:
# bits 3:0 = volume
# --------------------------------------------------------
def get_volume(self):
value = self.read_reg(REG_SYSCONFIG2)
return value & 0x0F
def set_volume(self, volume):
volume = max(0, min(15, int(volume)))
value = self.read_reg(REG_SYSCONFIG2)
value &= ~0x0F
value |= volume
self.write_reg(REG_SYSCONFIG2, value)
# --------------------------------------------------------
# Mute
#
# POWERCFG bit 9 = DMUTE
# --------------------------------------------------------
def get_mute(self):
value = self.read_reg(REG_POWERCFG)
# DMUTE=0 means muted
return not bool(value & (1 << 14))
def set_mute(self, mute):
value = self.read_reg(REG_POWERCFG)
if mute:
value &= ~(1 << 14)
else:
value |= (1 << 14)
self.write_reg(REG_POWERCFG, value)
# --------------------------------------------------------
# Seek
#
# POWERCFG:
# SEEK bit 8
# SEEKUP bit 9
# SKMODE bit 10
#
# --------------------------------------------------------
def seek(self, up=True):
value = self.read_reg(REG_POWERCFG)
# SEEK mode
value |= (1 << 8)
# SEEKUP
if up:
value |= (1 << 9)
else:
value &= ~(1 << 9)
# SKMODE=0 -> wrap around
value &= ~(1 << 10)
self.write_reg(REG_POWERCFG, value)
start = time.monotonic()
while True:
status = self.read_reg(REG_STATUSRSSI)
if status & STATUS_STC:
break
if time.monotonic() - start > 10.0:
raise TimeoutError("Seek timeout")
time.sleep(0.05)
# Clear SEEK
value &= ~(1 << 8)
self.write_reg(REG_POWERCFG, value)
time.sleep(0.05)
# READCHAN contains actual channel
return self.get_frequency()
# --------------------------------------------------------
# RDS report
#
# Report 18:
#
# byte 0 = report ID
# byte 1/2 = STATUSRSSI
# byte 3/4 = READCHAN
# byte 5/6 = RDSA
# byte 7/8 = RDSB
# byte 9/10 = RDSC
# byte 11/12 = RDSD
# --------------------------------------------------------
def read_rds_report(self):
if self.dev is None:
raise RuntimeError("デバイスが接続されていません")
with self.lock:
data = self.dev.get_feature_report(RDS_REPORT, 13)
if data is None or len(data) < 13:
return None
status = (data[1] << 8) | data[2]
readchan = (data[3] << 8) | data[4]
rdsa = (data[5] << 8) | data[6]
rdsb = (data[7] << 8) | data[8]
rdsc = (data[9] << 8) | data[10]
rdsd = (data[11] << 8) | data[12]
return {
"status": status,
"readchan": readchan,
"a": rdsa,
"b": rdsb,
"c": rdsc,
"d": rdsd,
}
class RDSDecoder:
def __init__(self):
self.pi = None
self.ps_chars = [" "] * 8
self.ps_seen = set()
self.last_group = None
def reset(self):
self.pi = None
self.ps_chars = [" "] * 8
self.ps_seen.clear()
self.last_group = None
def process(self, report):
if report is None:
return
status = report["status"]
# RDSR
if not (status & STATUS_RDSR):
return
# RDSS
if not (status & STATUS_RDSS):
return
a = report["a"]
b = report["b"]
c = report["c"]
d = report["d"]
self.pi = a
group_type = (b >> 12) & 0x0F
version_b = bool(b & (1 << 11))
group_key = (a, b, c, d)
if group_key == self.last_group:
return
self.last_group = group_key
# Group 0A / 0B = Program Service Name
if group_type == 0:
segment = b & 0x03
chars = [
(d >> 8) & 0xFF,
d & 0xFF
]
pos = segment * 2
if pos < 8:
self.ps_chars[pos] = (
chr(chars[0])
if 32 <= chars[0] <= 126
else " "
)
if pos + 1 < 8:
self.ps_chars[pos + 1] = (
chr(chars[1])
if 32 <= chars[1] <= 126
else " "
)
ps = "".join(self.ps_chars).strip()
return {
"pi": self.pi,
"ps": ps,
"group": f"{group_type}{'B' if version_b else 'A'}",
"a": a,
"b": b,
"c": c,
"d": d,
}
class RDX155GUI:
def __init__(self, root):
self.root = root
self.root.title("RDX-155 FM Radio")
self.root.geometry("720x620")
self.root.minsize(680, 560)
self.radio = RDX155()
self.rds_decoder = RDSDecoder()
self.running = True
self.connected = False
self.freq_var = tk.StringVar(value="94.9")
self.volume_var = tk.IntVar(value=15)
self.status_var = tk.StringVar(value="未接続")
self.rssi_var = tk.StringVar(value="RSSI: --")
self.stereo_var = tk.StringVar(value="--")
self.rds_var = tk.StringVar(value="RDS: --")
self.pi_var = tk.StringVar(value="PI: ----")
self.ps_var = tk.StringVar(value="PS: --------")
self._build_gui()
self.root.protocol("WM_DELETE_WINDOW", self.on_close)
# Start status updater
self.root.after(500, self.update_status)
# Try automatic connection
self.root.after(200, self.auto_connect)
# ========================================================
# GUI
# ========================================================
def _build_gui(self):
main = ttk.Frame(self.root, padding=12)
main.pack(fill="both", expand=True)
# ----------------------------------------------------
# Device
# ----------------------------------------------------
device_frame = ttk.LabelFrame(
main,
text="RDX-155",
padding=10
)
device_frame.pack(fill="x", pady=(0, 10))
ttk.Label(
device_frame,
text="VID / PID:"
).grid(row=0, column=0, sticky="w")
ttk.Label(
device_frame,
text="06E1 / A155"
).grid(row=0, column=1, sticky="w", padx=8)
self.connect_button = ttk.Button(
device_frame,
text="接続",
command=self.connect
)
self.connect_button.grid(
row=0,
column=2,
padx=10
)
self.disconnect_button = ttk.Button(
device_frame,
text="切断",
command=self.disconnect,
state="disabled"
)
self.disconnect_button.grid(
row=0,
column=3
)
ttk.Label(
device_frame,
textvariable=self.status_var
).grid(
row=1,
column=0,
columnspan=4,
sticky="w",
pady=(8, 0)
)
# ----------------------------------------------------
# Frequency
# ----------------------------------------------------
freq_frame = ttk.LabelFrame(
main,
text="FM周波数",
padding=10
)
freq_frame.pack(fill="x", pady=(0, 10))
self.freq_entry = ttk.Entry(
freq_frame,
textvariable=self.freq_var,
width=10,
justify="center",
font=("Segoe UI", 18)
)
self.freq_entry.grid(
row=0,
column=0,
padx=(0, 8)
)
ttk.Label(
freq_frame,
text="MHz"
).grid(
row=0,
column=1,
sticky="w"
)
ttk.Button(
freq_frame,
text="選局",
command=self.tune
).grid(
row=0,
column=2,
padx=10
)
ttk.Button(
freq_frame,
text="◀ Seek",
command=lambda: self.do_seek(False)
).grid(
row=0,
column=3,
padx=3
)
ttk.Button(
freq_frame,
text="Seek ▶",
command=lambda: self.do_seek(True)
).grid(
row=0,
column=4,
padx=3
)
# Frequency slider
self.freq_scale = tk.Scale(
freq_frame,
from_=BAND_MIN,
to=BAND_MAX,
resolution=0.1,
orient="horizontal",
showvalue=True,
command=self.scale_changed,
length=560
)
self.freq_scale.set(94.9)
self.freq_scale.grid(
row=1,
column=0,
columnspan=5,
sticky="ew",
pady=(10, 0)
)
# ----------------------------------------------------
# Status
# ----------------------------------------------------
status_frame = ttk.LabelFrame(
main,
text="受信状態",
padding=10
)
status_frame.pack(fill="x", pady=(0, 10))
ttk.Label(
status_frame,
textvariable=self.rssi_var,
font=("Segoe UI", 14)
).grid(
row=0,
column=0,
padx=10
)
ttk.Label(
status_frame,
textvariable=self.stereo_var,
font=("Segoe UI", 14)
).grid(
row=0,
column=1,
padx=10
)
ttk.Label(
status_frame,
textvariable=self.rds_var,
font=("Segoe UI", 14)
).grid(
row=0,
column=2,
padx=10
)
# ----------------------------------------------------
# RDS
# ----------------------------------------------------
rds_frame = ttk.LabelFrame(
main,
text="RDS",
padding=10
)
rds_frame.pack(fill="x", pady=(0, 10))
ttk.Label(
rds_frame,
textvariable=self.pi_var,
font=("Consolas", 14)
).grid(
row=0,
column=0,
sticky="w",
padx=5
)
ttk.Label(
rds_frame,
textvariable=self.ps_var,
font=("Consolas", 20)
).grid(
row=1,
column=0,
sticky="w",
padx=5,
pady=(8, 0)
)
# ----------------------------------------------------
# Audio
# ----------------------------------------------------
audio_frame = ttk.LabelFrame(
main,
text="音量",
padding=10
)
audio_frame.pack(fill="x", pady=(0, 10))
self.volume_scale = tk.Scale(
audio_frame,
from_=0,
to=15,
orient="horizontal",
variable=self.volume_var,
showvalue=True,
command=self.volume_changed,
length=400
)
self.volume_scale.pack(
side="left",
padx=5
)
self.mute_button = ttk.Button(
audio_frame,
text="Mute",
command=self.toggle_mute
)
self.mute_button.pack(
side="left",
padx=10
)
# ----------------------------------------------------
# Registers
# ----------------------------------------------------
reg_frame = ttk.LabelFrame(
main,
text="Si470x Registers",
padding=5
)
reg_frame.pack(
fill="both",
expand=True
)
self.reg_text = tk.Text(
reg_frame,
height=9,
width=70,
font=("Consolas", 10)
)
self.reg_text.pack(
fill="both",
expand=True
)
# ----------------------------------------------------
# Bottom
# ----------------------------------------------------
bottom = ttk.Frame(main)
bottom.pack(fill="x", pady=(8, 0))
ttk.Button(
bottom,
text="レジスタ更新",
command=self.refresh_registers
).pack(side="left")
ttk.Label(
bottom,
text="RDX-155 / Si470x"
).pack(side="right")
# ========================================================
# Connect
# ========================================================
def auto_connect(self):
try:
if self.radio.find() is not None:
self.connect()
except Exception:
pass
def connect(self):
try:
info = self.radio.open()
self.connected = True
self.status_var.set(
f"接続: {info.get('product_string', 'RDX-155')}"
)
self.connect_button.config(
state="disabled"
)
self.disconnect_button.config(
state="normal"
)
# Read current state
try:
freq = self.radio.get_frequency()
self.freq_var.set(
f"{freq:.1f}"
)
self.freq_scale.set(freq)
except Exception:
pass
try:
volume = self.radio.get_volume()
self.volume_var.set(volume)
except Exception:
pass
self.refresh_registers()
except Exception as e:
self.connected = False
messagebox.showerror(
"RDX-155",
f"接続できませんでした。\n\n{e}"
)
# ========================================================
# Disconnect
# ========================================================
def disconnect(self):
self.connected = False
self.radio.close()
self.status_var.set("未接続")
self.connect_button.config(
state="normal"
)
self.disconnect_button.config(
state="disabled"
)
# ========================================================
# Tune
# ========================================================
def tune(self):
if not self.connected:
return
try:
freq = float(
self.freq_var.get()
)
freq = round(freq * 10) / 10
if freq < BAND_MIN or freq > BAND_MAX:
raise ValueError(
f"{BAND_MIN:.1f}～{BAND_MAX:.1f} MHz"
)
self.status_var.set(
f"{freq:.1f} MHz に選局中..."
)
# Do USB operation in worker thread
threading.Thread(
target=self._tune_worker,
args=(freq,),
daemon=True
).start()
except Exception as e:
messagebox.showerror(
"選局エラー",
str(e)
)
def _tune_worker(self, freq):
try:
actual = self.radio.set_frequency(freq)
self.root.after(
0,
lambda: self._tune_done(actual)
)
except Exception as e:
self.root.after(
0,
lambda: messagebox.showerror(
"選局エラー",
str(e)
)
)
def _tune_done(self, actual):
self.freq_var.set(
f"{actual:.1f}"
)
self.freq_scale.set(actual)
self.rds_decoder.reset()
self.status_var.set(
f"受信中: {actual:.1f} MHz"
)
# ========================================================
# Frequency scale
# ========================================================
def scale_changed(self, value):
if not self.connected:
return
# Don't automatically tune on every slider movement.
self.freq_var.set(
f"{float(value):.1f}"
)
# ========================================================
# Seek
# ========================================================
def do_seek(self, up):
if not self.connected:
return
self.status_var.set(
"Seek中..."
)
threading.Thread(
target=self._seek_worker,
args=(up,),
daemon=True
).start()
def _seek_worker(self, up):
try:
freq = self.radio.seek(up)
self.root.after(
0,
lambda: self._seek_done(freq)
)
except Exception as e:
self.root.after(
0,
lambda: messagebox.showerror(
"Seekエラー",
str(e)
)
)
def _seek_done(self, freq):
self.freq_var.set(
f"{freq:.1f}"
)
self.freq_scale.set(freq)
self.rds_decoder.reset()
self.status_var.set(
f"受信中: {freq:.1f} MHz"
)
# ========================================================
# Volume
# ========================================================
def volume_changed(self, value):
if not self.connected:
return
try:
self.radio.set_volume(
int(float(value))
)
except Exception:
pass
# ========================================================
# Mute
# ========================================================
def toggle_mute(self):
if not self.connected:
return
try:
current = self.radio.get_mute()
self.radio.set_mute(
not current
)
except Exception as e:
messagebox.showerror(
"Mute",
str(e)
)
# ========================================================
# Periodic status
# ========================================================
def update_status(self):
if not self.running:
return
if self.connected:
try:
status = self.radio.get_status()
self.rssi_var.set(
f"RSSI: {status['rssi']}"
)
if status["stereo"]:
self.stereo_var.set(
"STEREO"
)
else:
self.stereo_var.set(
"MONO"
)
if status["rds_sync"]:
self.rds_var.set(
"RDS: SYNC"
)
elif status["rds_ready"]:
self.rds_var.set(
"RDS: READY"
)
else:
self.rds_var.set(
"RDS: --"
)
# RDS
report = self.radio.read_rds_report()
if report is not None:
decoded = self.rds_decoder.process(
report
)
if decoded:
pi = decoded.get("pi")
if pi is not None:
self.pi_var.set(
f"PI: {pi:04X}"
)
ps = decoded.get("ps", "")
if ps:
self.ps_var.set(
f"PS: {ps}"
)
except Exception:
pass
self.root.after(
300,
self.update_status
)
# ========================================================
# Register display
# ========================================================
def refresh_registers(self):
if not self.connected:
return
try:
regs = self.radio.read_all()
names = [
"DEVICEID",
"CHIPID",
"POWERCFG",
"CHANNEL",
"SYSCONFIG1",
"SYSCONFIG2",
"SYSCONFIG3",
"TEST1",
"TEST2",
"BOOTCONFIG",
"STATUSRSSI",
"READCHAN",
"RDSA",
"RDSB",
"RDSC",
"RDSD",
]
self.reg_text.delete(
"1.0",
"end"
)
for i in range(16):
value = regs[i]
self.reg_text.insert(
"end",
f"0x{i:02X} "
f"{names[i]:11s} "
f"0x{value:04X} "
f"{value:016b}\n"
)
except Exception as e:
self.reg_text.delete(
"1.0",
"end"
)
self.reg_text.insert(
"end",
f"ERROR: {e}"
)
# ========================================================
# Close
# ========================================================
def on_close(self):
self.running = False
try:
self.radio.close()
except Exception:
pass
self.root.destroy()
# ============================================================
# Main
# ============================================================
if __name__ == "__main__":
root = tk.Tk()
app = RDX155GUI(root)
root.mainloop()
これを普通にThonnyで実行したら普通に使えた。(pipでhidライブラリをインストールしないといけないかも)
0 件のコメント:
コメントを投稿