2026年9月21日月曜日

USBラジオチューナーをPythonで制御してみた。

 ADS TechのRDX-155-EFっていうUSBラジオチューナーを発掘したんだけど、Windows 11なPCに接続したらドライバはインストールされるものの、ツールがなかったのでChatGPTにPythonツールを作ってもらってみた。

ADS TechのInstant FM Musicってやつ。2012年ぐらいにeBayで6.75ドル+送料11.46ドルで購入したやつ。当時のレートで合計で1,583円という。
裏面には品番。
アンテナは巻取り式が付属しているけど、MCXコネクタが付いてるので変換アダプタ次第でいろんなアンテナが付けられそう。

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ライブラリをインストールしないといけないかも)

2026年9月6日日曜日

ESP32-P4でRaspberry Pi Cameraを使ってみた。

 昨年末に購入していたESP32-P4-Picoなんだけど、Arduino core for the ESP32のバージョンが3.3.11に上がってIDF v5.5.4 with ESP-Videoとなっていたので、Arduinoでもカメラモジュールが簡単に使えるようになってる!ということで試してみた。

ESP32-P4-PicoはRaspberry Pi Zero 2Wなどと同じ、0.5mmピッチの22Pinタイプのフラットケーブルが刺さるようになっている。カメラ側は1.0mmピッチの15Pinのものが多いので変換ケーブルが必要かも。

ESP-VideoではOV5647をサポートしているので、Raspberry Pi Camera v1.3がそのまま使えそう。昔購入したラズパイカメラの互換品と、Raspberry Pi Zero用に購入したカメラ変換ケーブルを使用してテストしてみることに。

フラットケーブルはカメラによって反対だったりするので注意かも。黒い可動部分と反対側に端子面が接触するように差し込む感じ。

とりあえずテストでOV5647で写真を撮影して、ESP32-P4のハードウェアエンコーダでJPEGに変換してmicroSDに保存してみた。

#include <Arduino.h>
#include <FS.h>
#include <SD_MMC.h>
#include <ESP_Video.h>
#include "driver/jpeg_encode.h"

#define CAMERA_SCCB_PORT      0
#define CAMERA_SCCB_SCL       8
#define CAMERA_SCCB_SDA       7
#define CAMERA_I2C_FREQ       400000
#define CAMERA_BUFFERS        2

#define SD_CLK                43
#define SD_CMD                44
#define SD_D0                 39
#define SD_D1                 40
#define SD_D2                 41
#define SD_D3                 42

#define JPEG_QUALITY          80
#define JPEG_OUTPUT_BUFFER_SIZE (4UL * 1024UL * 1024UL)
#define CAPTURE_INTERVAL_MS   5000

ESPVideoClass video;
ESPVideoCaptureDevClass capture_dev;
jpeg_encoder_handle_t jpeg_encoder = nullptr;
uint8_t *jpeg_buffer = nullptr;
size_t jpeg_buffer_size = 0;
uint32_t photo_number = 1;
unsigned long last_capture = 0;

bool initSD() {
    return SD_MMC.setPins(SD_CLK, SD_CMD, SD_D0, SD_D1, SD_D2, SD_D3) && SD_MMC.begin("/sdcard", false);
}

bool initCamera() {
    ESPVideoCamConfigClass cam_cfg;
    ESPVideoCSIConfigClass csi_cfg;
    return cam_cfg.begin(CAMERA_SCCB_PORT, CAMERA_SCCB_SCL, CAMERA_SCCB_SDA, CAMERA_I2C_FREQ) &&
           csi_cfg.begin(cam_cfg) &&
           video.begin(csi_cfg) &&
           capture_dev.begin(ESP_VIDEO_MIPI_CSI_DEVICE_NAME, CAMERA_BUFFERS) &&
           capture_dev.setFormat(ESP_VIDEO_FORMAT_RGB565) &&
           capture_dev.startCapture();
}

bool initJPEG() {
    jpeg_encode_engine_cfg_t eng_cfg = {.intr_priority = 0, .timeout_ms = 100};
    if (jpeg_new_encoder_engine(&eng_cfg, &jpeg_encoder) != ESP_OK) return false;

    jpeg_encode_memory_alloc_cfg_t mem_cfg = {.buffer_direction = JPEG_ENC_ALLOC_OUTPUT_BUFFER};
    jpeg_buffer = (uint8_t *)jpeg_alloc_encoder_mem(JPEG_OUTPUT_BUFFER_SIZE, &mem_cfg, &jpeg_buffer_size);
    if (!jpeg_buffer) {
        jpeg_del_encoder_engine(jpeg_encoder);
        jpeg_encoder = nullptr;
        return false;
    }
    return true;
}

String nextFilename() {
    char filename[32];
    while (true) {
        snprintf(filename, sizeof(filename), "/photo_%04lu.jpg", (unsigned long)photo_number);
        if (!SD_MMC.exists(filename)) {
            photo_number = (photo_number >= 999999) ? 1 : photo_number + 1;
            return String(filename);
        }
        photo_number = (photo_number >= 999999) ? 1 : photo_number + 1;
    }
}

bool capturePhoto() {
    if (!capture_dev.isOpened() || !capture_dev.isCaptureStarted()) return false;

    ESPVideoBufferClass buffer = capture_dev.captureBuffer();
    if (!buffer.valid() || buffer.formatType() != ESP_VIDEO_FORMAT_RGB565) return false;

    uint32_t width = buffer.getWidth();
    uint32_t height = buffer.getHeight();
    size_t input_size = buffer.size();
    uint8_t *input = buffer.data();
    if (!input || width == 0 || height == 0) return false;

    jpeg_encode_cfg_t enc_cfg = {};
    enc_cfg.src_type      = JPEG_ENCODE_IN_FORMAT_RGB565;
    enc_cfg.sub_sample    = JPEG_DOWN_SAMPLING_YUV420;
    enc_cfg.image_quality = JPEG_QUALITY;
    enc_cfg.width         = width;
    enc_cfg.height        = height;
    enc_cfg.pixel_reverse = false;

    uint32_t jpeg_size = 0;
    unsigned long start = millis();
    if (jpeg_encoder_process(jpeg_encoder, &enc_cfg, input, (uint32_t)input_size,
                             jpeg_buffer, (uint32_t)jpeg_buffer_size, &jpeg_size) != ESP_OK) {
        return false;
    }
    unsigned long elapsed = millis() - start;

    String filename = nextFilename();
    File file = SD_MMC.open(filename.c_str(), FILE_WRITE);
    if (!file) return false;
    size_t written = file.write(jpeg_buffer, jpeg_size);
    file.close();

    if (written == jpeg_size) {
        Serial.printf("Saved: %s (%lu bytes, %lu ms)\n", filename.c_str(), (unsigned long)jpeg_size, elapsed);
        return true;
    }
    return false;
}

void setup() {
    Serial.begin(115200);
    delay(1000);
    Serial.println("\n=== SYSTEM INIT ===");

    #ifdef BOARD_HAS_PSRAM
    Serial.printf("PSRAM: ENABLED (%u / %u bytes free)\n", 
                  (unsigned int)ESP.getFreePsram(), (unsigned int)ESP.getPsramSize());
    #endif

    if (!initSD() || !initCamera() || !initJPEG()) {
        Serial.println("SYSTEM ERROR: Init failed");
        return;
    }
    Serial.println("SYSTEM READY");
    last_capture = millis() - CAPTURE_INTERVAL_MS;
}

void loop() {
    if (!capture_dev.isOpened() || !capture_dev.isCaptureStarted()) {
        delay(1000);
        return;
    }

    if (millis() - last_capture >= CAPTURE_INTERVAL_MS) {
        last_capture = millis();
        capturePhoto();
        #ifdef BOARD_HAS_PSRAM
        Serial.printf("Free PSRAM: %u bytes\n", (unsigned int)ESP.getFreePsram());
        #endif
    }
    delay(10);
}

Arduino coreは3.3.11を使用。それ以下だとESP_Video.hが見つからない。ビルド設定はデフォルトからPSRAMはEnableにしただけ。もし新しいリビジョンのESP32-P4を使っている場合はChip Variantをv3.00 or Newerにしないとうまく動かないかも。

とりあえずシリアルコンソールを確認してみたらちゃんと動いてそう。適当なタイミングで電源を切ってmicroSDの中身を確認してみるとちゃんとjpgファイルがたくさん生成されていて写真が取れてる。

解像度は800x800で動いてるみたいだけどどうやらドライバーのデフォルト値?どうやって解像度を変更できるのかESP_Video.hの仕様を調べる必要がありそう。

2026年9月5日土曜日

USB WiFiアダプタの使用チップをリストにしてみた。

 最近BuildrootとかをビルドしているとWiFiのドライバとかを組み込む際にどれを組み込めばいいのとか忘れてしまうので品番と内蔵ICのリストを作ってみることに。

あと購入する前に調べた情報も忘れてもう一回調べてたりするので持ってるUSBドングルだけじゃなくて調べたやつもリストにしておこうかなと。


製品名 使用チップ Wi-Fiバージョン 技適番号 ドライバ内蔵 所有
ELECOM WDC-433SU2M2BK RTL8811AU Wi-Fi 5 (ac) 201-150394 × ×
TOKAIZ TWA-003 AIC8800M80 Wi-Fi 6 (ax) 214-250303 ×
TOKAIZ TWA-001 RTL8811CU Wi-Fi 5 (ac) 210-138110 × 関連記事
10Gtek WD-4510AC RTL8821CU Wi-Fi 5 (ac) 214-118694 関連記事
BUFFALO WLI-UC-GNM2 RT3070 Wi-Fi 4 (n) 003WWA110733 × 関連記事
BUFFALO WI-U2-866DM RTL8822BU Wi-Fi 5 (ac) 003-240097 × ×
UGREEN CM845 RTL8851BU Wi-Fi 6 (ax) 201-250245 ×


Linuxの場合はドライバ内蔵してないほうが使いやすい。仮想CD-ROMとして認識するとusb_modeswitchでイジェクト処理とかをしないとWiFiデバイスとして認識しなかったりするし。再起動のときにたまにusb_modeswitchでも切り替わらなかったりするので安定性が…

安価でドライバも結構公開されていて使いやすいのはRealtekなのかなぁ。RTL8811CUなTWA-001はラズパイ3Bにつけて常時稼働してるけどとりあえずは問題なさそうだし。


2026年8月29日土曜日

CN3791を使ってソーラーでLiPoを充電してみた。

 この前改造したLEDセンサーライトをCN3791搭載のソーラーチャージャーで充電できるようにしてみた。

CN3791はワンチップでソーラーからLiPoを充電できるICっぽい。しかもMPPT対応らしい。CN3791のデータシートを見ると充電電流はシャント抵抗で変更できるらしい。この基板はR050ってなってるので50mΩがついてるので2.4Aまで流せるらしい。

今回は500mAhのバッテリーに充電したいので電流を落とさないと…

300mΩにすると400mAぐらいまで充電電流を落とせるのでとりあえずこのぐらいにしてみようかな。

PHコネクタが片側に2つずつ付いてるんだけど、これはただパラでつながってるだけだった。バッテリー側は片方出力として使えるようにしてあるのかな?何にせよ出力の制御とかをやってるわけではない。ソーラも2つパラでつなぐときは逆流とか注意しないとだめかも。

ちなみにMPPTの設定は6Vになっていたのでそのまま使おうと思う。使おうとしていたソーラーが6Vっぽいし。
とりあえず330mΩに変更して熱収縮チューブで絶縁完了。

付属のPHコネクタ付きのケーブル、両端で極性が反対になっていたりするので注意かも。とりあえずカットして使った。

そして出力はこの前改造したソーラーライトに接続して、入力は外に設置してるセンサーライトから分岐して接続。
充電中は赤色のLEDが点灯。そして充電中でもセンサーライトが光るのは便利。
LEDが青く光ったら充電完了。

しかしここで夜になっても分岐元のセンサーライトがつかない…
原因を調べてみるとどうやらCN3791から電源が微妙に逆流して、もともとセンサーライトはソーラーの発電がなくならないとつかないのでCN3791から逆流した電気でまだ昼だと勘違いしているっぽい。
なのでとりあえずCN3971の入力にダイオードを突っ込んで解決した。電圧降下が少ないほうがいいと思ったのでとりあえずは40V 1A程度のショットキーバリアダイオードにしておいた。これでとりあえず漏れ電流も問題なさそうなので大丈夫そう。

2026年8月27日木曜日

USB扇風機のケーブル交換してみた。

 昔ドスパラで安売りしていたときに購入したDEEPCOOL USB FAN M6なんだけど、ケーブルの被覆が剥けて来てしまっていたので交換してみることに。

スイッチ側のほうのケーブルの付け根がこんなことに。まだ銅線まで見えていないのでそのまま使っていたが被覆剥けたらUSBにダメージが入りそうなので危険。(USB側にポリスイッチは入ってるんだろうけど)
USBコネクタ側も同じ感じに。

分解しつつ掃除も。ところどころ錆びてるし。

モータを外すとこんな感じに。意外とスペースが有るぞ…
モータステーの奥行きは20mmぐらいある。
ケーブルグランドを外してみると穴はΦ12で小判型(11mm)になっていた。

スペースがあるのでファンコントロール機能とかつけて真のUSB扇風機として改造も出来そうなんだけどとりあえずケーブル交換が楽なようにTypeC化してみようかなと思ったり。

USB Type Cコネクタを探していると便利そうな丸形のパネルマウント式があった。パネルカットアウト寸法的にはちょっとUSB扇風機の穴が大きいような気がするけど、USB Type Cコネクタ自体が13.8mmということでうまくセンターをあわせて固定すればきれいに収まりそう。6線式ならType C充電器のCC1とCC2の接続をうまくしてやればUSB TypeAコネクタ以外のケーブルでもちゃんと使えそうだし。

というわけで早速手配してみたらレビューにもあった通りCC1とCC2(緑と黄色)には5.1kΩがもともとGNDの間に入っていそう。なので両端がType Cのケーブルで接続しても使えそう。
穴も無加工で取り付けられた。逆に穴は結構スカスカなのでいい感じの角度で固定した。要らないケーブルは短くカットして、黒と赤のケーブルをもともとつながっていたように接続。
これでType C化完了。

ケーブルが少し太くなったからなのか断線気味だったからなのかわからないけど風と音が前より強くなっている。

これでまだまだ使えそうだ。

2026年8月23日日曜日

Anker T8410XにThinginoをインストールしてみた

 というわけで前回Eufy Indoor Cam 2K Pan & Tiltのファームウェアをバックアップできたので、カスタムファームのThinginoに書き換えてみることに。
ThinginoはIngenic社製のSOC搭載のIPカメラ用のオープンソースファームウェアで、メーカー専用のクラウドサービスを使わずにカメラを運用したりできるファームウェア。設定アプリも要らなくなるのでPCだけでしか使わない場合とかも便利かも。

Bootモードが使えることでUSBで簡単に書き換えられるのは良いな。

前回同様にthingino Web Flasherでブートストラップを行ってからDFUモードで接続する。今回はファームウェアを書き込むボタンを押してファームウェアを書き込む。

ボタンを押すとbinファイルをアップロードするように言われるので、Thinginoの公式サイトからT8410X用のファームウェアをダウンロードしてこないといけないんだけど、イメージセンサーがどれかわからない…

ということでとりあえずSC3336のを書き込んでみた。6分ぐらいかかった。Chromeは非アクティブにすると遅くなったりするのでアップデートのウインドウはアクティブ状態にしておいたほうがいいかも。

書き込みが終わるとTHINGINOから始まるSSIDでSoftAPが立ち上がるので、Wifiで接続する。

PCからだと自動で繋がらないのでhttp://172.16.0.1/を開くと設定画面が出てくる。

とりあえずSoftAPで単独で動かして見るためこんな感じに設定してProceed

再起動されて設定したSSIDで起動してくるので接続したら完了。

またブラウザで今度はhttp://100.64.1.1/に接続するとSoftAPで起動してきたカメラの映像が出るはず。とりあえず写ったのでSC3336で間違いなかったのかな?

SSHからもアクセスできるし、プレビュー画面ではカメラも動かせる。

試しにSSHからカメラのレジスタを読んでみる。

root@ing-eufy-t8410x-0358 ~# i2ctransfer -f -y 0 w2@0x30 0x31 0x07 r2
0xcc 0x41

0xcc41ってことはSC3336で間違いないのかな?0xcc1aだとSC3335らしい。

なんかThinginoにしたらカメラが熱い気がする。ということで一旦戻してみることに。

同じ手順で純正ファームウェアを流し込んだら10分ぐらいで書き込めた。Thinginoは16MBだけど純正ファームは32MBなので少し時間がかかる。

そこまで熱くはならないけど純正ファームウェアでもRTSPでストリーミングしておくと結構あったかい。

ファームウェアの違いかもしれないので試しにSC3338版を入れてみた。そんでもってsensor-infoを実行してみる。

root@ing-eufy-t8410x-0358 ~# sensor-info
Probing bus 0... 0/288sinfo: [Warning] cannot claim reset GPIO 18 (a sensor driver probably holds it); probing without the reset dance, use -r -1 to silence
Probing bus 0... 288/288
sc3336 on bus 0 at 0x30 (MCLK 24 MHz)
  also matches: sc3338
  ID: 0x3107=0xCC 0x3108=0x41

なるほど、SC3336とSC3338は同じIDなのね。フレームレートが違うみたいだけど。ちゃんとカメラが写ってるし問題なさそう。sensor-infoを実行すると映像が止まるのでカメラの再起動が必要。

とりあえずSC3338では熱くならずにうまく動いてそうなのでこれで試してみよう。

ちなみに後から気がついたんだけど公式サイトにもっと簡単にBootモード入れられる方法が書いてあった。どうやらBootピンがmicroUSBのIDピンにつながってるようで、OTGケーブルを使えば勝手にBootモードに入るらしいのでUSBオスオスケーブルOTGケーブルでPCに接続すれば非分解でファームウェアをインストールできるらしい。microUSBの端子部分が狭いのであまり大きいコネクタのOTGケーブルだと刺さらないので注意だけど。

Thinginoはだいぶ色々設定できそうなんだけど、まずは遅延を測ってみた。

とりあえずデフォルト設定でWeb設定画面のプレビューで640msぐらいの遅延。
デフォルト設定は2304 x 1296の25fpsになっていたので、Eufyファームウェアの15fpsよりもフレームレートが高く設定できるようになっているっぽい。

ThinginoではRaptor Streaming Systemを入れるとカメラ側でWebRTCが使えたりするっぽいのでさらに遅延がすくなることを期待してRapterを導入してみたい。

2026年8月22日土曜日

Anker T8410Xのファームウェアをバックアップしてみた

 この前購入したEufy Indoor Cam 2K Pan & Tiltなんだけど、とりあえず一通り動作確認が出来たので、分解してカスタムファームウェアを入れてみることに。

分解方法はYoutubeに動画があったのでそれを参考にさせてもらった。

E220

分解方法はまず裏面のネジを外す。ゴム足の端っこに合計4つのネジが隠れているのでこれを外して蓋を引っ張ると外れる。

蓋の中にはインターフェース基板がいるのでここのケーブルを外してから基板を取り外す。

フラットケーブルは少し長いので基板を取り外してから外しても大丈夫かも。

インターフェース基板を取り外すとなんとBootって書いてあるTPが…

Bootって書いてあるTPを見つけたのでここで一旦分解は終了。boot_sel0につながってるTPならここからBootモード入れるじゃん!

早速フラットケーブルだけ接続して電源を入れる。
テスターで測ってみると3.3Vになってるのでそれっぽい。

このBootと近くのGND(USB端子のハウジングがハンダされてるところ)をピンセットでショートさせてUSBに繋いでみる…

デバイスマネージャーで確認してみると"Ã USB Boot Device"ってのが増えてる。カメラをBootモードで起動することに成功したっぽい。ちなみにIngenic T31Xはboot_sel0をGNDに落としながら起動するとSDカード起動をトライして、それでも起動できない場合はUSB bootモードになるっぽい。
VID:A108 PID:C309

カメラ本体まで分解するのは面倒なのでネジ6本外すだけでBootモードに入れるのは便利。ROMライターもいらないし。

Zadigを使ってドライバをインストールしてみる。

Zadigを起動するとÃ USB Boot Deviceが選ばれているので、WinUSBを選んでInstall Driverをしてみる。WinUSBにするとthingino Web Flasherが使用できる。(Chrome系のブラウザのみ対応)
thingino Web Flasherを開いて、Connect Deviceをクリックするとドライバがちゃんとインストールされていればデバイスリストに表示されるはず。
うまく接続できたらブートストラップボタンを押して、DFUモードで起動させる。
DFUモードでデバイスが認識し直されるので次はUSB download gadgetで接続。
ファームウェアを読み出すボタンを押して10分ぐらい待つとバックアップされたbinファイルがダウンロードされる。
バックアップが終わったら普通にUSBを抜いて、またACアダプタに接続してカメラの動作確認を行う。今回はファームウェアを読み込んだだけなので純正ファームウェアでうまく起動した。

とりあえずこれで純正ファームウェアがバックアップできたので、次回はThinginoをインストールしてみようと思う。