BP1 Pro: BLE accepts the same command bytes as RFCOMM (+ undocumented Game Mode opcodes)

[Zyxilon]

BP1 Pro: BLE accepts the same command bytes as RFCOMM (+ undocumented Game Mode opcodes)

Hey, great repo, really appreciated having clean, documented UUIDs and opcodes to test against. Wanted to share results from independently confirming and extending the BP1 Pro protocol.

Background

I'd already reverse-engineered the BP1 Pro's RFCOMM/SPP control channel via an HCI snoop log pulled via Android bug report while using the official app: that's where the actual command bytes (ANC modes, game mode) came from. I'd separately done static analysis of the Android APK (HeadPhoneDataResolveManager, HeadPhoneCrcUtil, etc. in com.base.module_common), which gave me the frame structure (length field, CRC16, opcode position) but not the specific per-command payloads; those I pulled from the live snoop capture instead. That gave me a working SPP script:

EARBUDS_MAC = "XX:XX:XX:XX:XX:XX"
RFCOMM_CHANNEL = 1
COMMANDS = {

"anc_off":      bytes.fromhex("ba3400ff"),

"anc_on":       bytes.fromhex("ba340165"),

"transparency": bytes.fromhex("ba3402ff"),

"game_mode_on":  bytes.fromhex("ba2401"),

"game_mode_off": bytes.fromhex("ba2400"),

}
sock = socket.socket(socket.AF_BLUETOOTH, socket.SOCK_STREAM, socket.BTPROTO_RFCOMM)

sock.connect((EARBUDS_MAC, RFCOMM_CHANNEL))

sock.send(COMMANDS["anc_on"])

This RFCOMM path was already fully working on its own before I found this repo. After finding it and seeing gatt_uuids() / bp1_pro.rs, I wanted to check whether BLE needed its own separate byte format, or whether it would accept the same payloads. Tested the exact same command bytes, unmodified, over BLE GATT using bleak, against your already-verified UUIDs:

WRITE_UUID  = "ee684b1a-1e9b-ed3e-ee55-f894667e92ac"
NOTIFY_UUID = "654b749c-e37f-ae1f-ebab-40ca133e3690"
await client.write_gatt_char(WRITE_UUID, bytes.fromhex("ba340165"), response=True)

Result: BLE accepts the same bytes as RFCOMM, full session log below

RFCOMM/SPP control was already confirmed working separately (see anc_send.py below). What's new here: the same command bytes, unmodified, also work when written to the BLE GATT write characteristic instead — same payload, different transport. Ran a full cycle over BLE (ANC off → on → transparency → off → game mode on → off), 4s pause between each command:

[+] Connected, subscribed to notify characteristic.
    [NOTIFY] aa0246004601
[*] Sending: ANC OFF -> ba3400ff

[NOTIFY] aa274600

[NOTIFY] aa3000

[NOTIFY] aa3401
[*] Sending: ANC ON -> ba340165

[NOTIFY] aa3401
[*] Sending: Transparency -> ba3402ff

[NOTIFY] aa3401
[*] Sending: ANC OFF -> ba3400ff

[NOTIFY] aa3401
[*] Sending: Game Mode ON -> ba2401

[NOTIFY] aa2401

[NOTIFY] aa2301
[*] Sending: Game Mode OFF -> ba2400

[NOTIFY] aa2401

[NOTIFY] aa2300

ANC physically toggled on the earbuds (audibly confirmed) for every command. So, the command/opcode layer (BA [opcode] [payload] outgoing, AA [opcode] [payload] incoming) is shared across SPP and BLE for this model, meaning it uses the same protocol and two transports, not two separate byte formats. This should mean BlueZ/Linux BLE support doesn't need any separate protocol RE from the RFCOMM side.

Finding 1: ANC ack behavior differs from what's documented in bp1_pro.rs

Your decoder currently expects distinct acks per mode:

0x32 => Ok(DeviceEvent::AncModeUpdate(AncMode::Transparency)),
0x33 => Ok(DeviceEvent::AncModeUpdate(AncMode::Anc)),

with 0x34 reserved for ANC-off-only, and 0x30 as a separate "periodic keepalive, not a state update."

On my unit, every ANC command, including off, on, and transparency, produced the identical ack AA 34 01, never AA 32 or AA 33. Only one ack opcode was returned with the same payload byte, regardless of which mode was actually set. So either:

• this is a firmware revision difference between units, or
• 0x34 01 is actually a generic "ANC command applied" ack rather than mode-specific, and the 0x32/0x33 mapping in the current code was extrapolated rather than empirically confirmed for every mode

Worth flagging in case others see the same, as it might be safer for the decoder to not assume 0x32/0x33 will appear at all. Instead, it could track ANC state from the outgoing command rather than relying on a mode-specific incoming ack, if this turns out to be common.

Finding 2: Game/Low-Latency Mode, not currently modeled at all

AncMode and DeviceEvent have no game-mode variant, and opcodes 0x23/0x24 don't appear anywhere in bp1_pro.rs. This was found via the same HCI snoop capture and confirmed live over both SPP and BLE:

Action	Outgoing	Ack(s)
Game Mode ON	BA 24 01	AA 24 01, then AA 23 01
Game Mode OFF	BA 24 00	AA 24 01, then AA 23 00

The pattern looks consistent and clean: the 0x24 ack is a generic "command received" ack (always payload 01 regardless of on/off, which mirrors the ANC behavior in Finding 1 and interestingly confirms the flat ack isn't a bug but might just be this device's convention), and 0x23 is the actual state-confirmation opcode, with payload 01/00 for on/off.

It looks like a straightforward addition to AncMode/DeviceEvent, or it may deserve its own DeviceEvent::GameModeUpdate variant rather than folding into AncMode, since it appears to be an independent toggle, not a mutually-exclusive ANC state.

This may directly unblock crates/baseus-transport/src/win/rfcomm.rs

Noticed RfcommTransport already exists in the repo, using the standard SPP service ID (0x1101) via RfcommServiceId::SerialPort() — looks like the transport layer was scaffolded ahead of the protocol RE, per the "Phase 0 RE" comment near the top of that file. The command bytes confirmed in this issue should be directly usable through RfcommTransport::send() as-is, since they're confirmed working over a standard SPP/RFCOMM channel 1 connection — the same service ID this code already targets. Happy to help, this might just serve as confirmation that the existing transport is ready to use.

Test scripts (for reproduction)

anc_send.py (RFCOMM / SPP)

#!/usr/bin/env python3
"""
anc_send.py - send commands to Baseus BP1 Pro ANC earbuds over RFCOMM.
Transport: classic Bluetooth RFCOMM, Channel 1 ("Serial Port" / SPP)

"""
import socket

import sys

import time

import argparse
EARBUDS_MAC = "XX:XX:XX:XX:XX:XX"  # Classic BT / SPP address (distinct from BLE address)

RFCOMM_CHANNEL = 1  # confirmed "Serial Port" channel from SABM handshake
COMMANDS = {

"on":           ("ANC ON",       bytes.fromhex("ba340165")),

"off":          ("ANC OFF",      bytes.fromhex("ba3400ff")),

"transparency": ("Transparency", bytes.fromhex("ba3402ff")),

"gameoff":      ("Gameoff",      bytes.fromhex("ba2400")),

"gameon":       ("Gameon",       bytes.fromhex("ba2401")),

}
MAX_ATTEMPTS = 4

RETRY_DELAY_SECONDS = 1.5
def log(verbose: bool, message: str) -> None:

if verbose:

print(message)
def send_command(command_bytes: bytes, verbose: bool) -> int:

last_error = None
for attempt in range(1, MAX_ATTEMPTS + 1):
    log(verbose, f"[*] Attempt {attempt}/{MAX_ATTEMPTS}...")

    try:
        sock = socket.socket(socket.AF_BLUETOOTH, socket.SOCK_STREAM, socket.BTPROTO_RFCOMM)
    except OSError as e:
        log(verbose, f"[!] Failed to create socket: {e}")
        return 3

    try:
        sock.settimeout(5)
        sock.connect((EARBUDS_MAC, RFCOMM_CHANNEL))
    except OSError as e:
        last_error = e
        log(verbose, f"[!] connect() failed: {e}")
        sock.close()
        if attempt < MAX_ATTEMPTS:
            time.sleep(RETRY_DELAY_SECONDS)
        continue

    log(verbose, "[+] Connected.")

    try:
        sock.send(command_bytes)
    except OSError as e:
        log(verbose, f"[!] send() failed: {e}")
        sock.close()
        return 5

    sock.close()
    log(verbose, "[+] Command sent successfully.")
    return 0

log(verbose, f"[!] Giving up after {MAX_ATTEMPTS} attempts. Last error: {last_error}")
return 4

def main() -> int:

parser = argparse.ArgumentParser(description="Send commands to Baseus BP1 Pro ANC earbuds")

parser.add_argument("mode", choices=COMMANDS.keys(), help="mode to set")

parser.add_argument("-v", "--verbose", action="store_true")

args = parser.parse_args()
command_name, command_bytes = COMMANDS[args.mode]
log(args.verbose, f"[*] {command_name} -> {command_bytes.hex()}")

return send_command(command_bytes, args.verbose)

if name == "main":

sys.exit(main())

Usage: python3 anc_send.py <on|off|transparency|gameon|gameoff> [--verbose]

Note: if the phone's Baseus app is already connected, connect() may fail with ECONNREFUSED. The earbuds' Bluetooth chip has only one RFCOMM slot, and the phone claims it on connecting — not just while audio happens to be streaming. This is not fixable with retries once the phone holds the slot — but a connection opened before the phone connects stays alive afterward, even with the phone connected too. See anc_daemon.py below for a persistent-connection approach that grabs the slot early and sidesteps this entirely.

anc_daemon.py (persistent RFCOMM connection — works around the ECONNREFUSED limitation)

Opens one RFCOMM connection when the earbuds connect and holds it open indefinitely, so commands sent later don't need a fresh connect() (which is what fails if phone audio is active at that moment). Watches bluetoothctl for connect/disconnect events and exposes a Unix socket for a lightweight client to send commands through.

#!/usr/bin/env python3
"""
anc_daemon.py - persistent background daemon for Baseus Bass BP1 Pro ANC control.
WHY THIS EXISTS:

A fresh RFCOMM connect() to the earbuds gets refused once the phone's

Baseus app has already established its own connection - the earbuds'

Bluetooth chip only has one RFCOMM slot, and the phone claims it

immediately on connecting, not just while audio happens to be playing.

But a connection opened BEFORE the phone claims that slot keeps working

fine afterward, even with the phone connected. So instead of opening a

new connection on every command (what anc_send.py does), this daemon

opens ONE connection as early as possible - ideally before the phone

connects at all - and keeps it alive indefinitely, ready to send

commands instantly whenever asked. This sidesteps the

connection-refused problem entirely, as long as the daemon grabs the

slot first.
ARCHITECTURE:

|

v

[this daemon]

|

persistent RFCOMM

|

v

[earbuds]
The daemon does two things concurrently:

1. Watches bluetoothctl's monitor output for the earbuds connecting/

disconnecting, and opens/closes its persistent RFCOMM socket to

match.

2. Listens on a Unix domain socket at SOCKET_PATH for single-word

commands ("on", "off", "transparency", "status"), and writes a

single-line response back ("OK", or an error description).
KNOWN LIMITATION: if the daemon's RFCOMM connection is not currently

open and the phone's Baseus app connects first, the daemon will NOT be

able to open one until the phone disconnects and releases the slot.

This is a hardware/firmware limit on the earbuds (one RFCOMM slot

total), not a bug in this script. The daemon will keep retrying

periodically in the background regardless.
Run with sudo (raw Bluetooth sockets need elevated privileges):

sudo python3 anc_daemon.py
Stop with Ctrl+C.

"""
import socket

import subprocess

import threading

import time

import os

import re

import sys

import signal
EARBUDS_MAC = "XX:XX:XX:XX:XX:XX"

RFCOMM_CHANNEL = 1

SOCKET_PATH = "/tmp/anc_daemon.sock"

RECONNECT_RETRY_SECONDS = 5  # how often to retry opening RFCOMM if not connected
COMMANDS = {

"on":           ("ANC ON",       bytes.fromhex("ba340165")),

"off":          ("ANC OFF",      bytes.fromhex("ba3400ff")),

"transparency": ("Transparency", bytes.fromhex("ba3402ff")),

}
CONNECTED_PATTERN = re.compile(

rf"Device {re.escape(EARBUDS_MAC)} Connected: yes", re.IGNORECASE

)

DISCONNECTED_PATTERN = re.compile(

rf"Device {re.escape(EARBUDS_MAC)} Connected: no", re.IGNORECASE

)
class AncDaemon:

def init(self):

self.rfcomm_sock = None

self.rfcomm_lock = threading.Lock()

self.earbuds_connected = False

self.running = True
# ---- RFCOMM connection management ----

def open_rfcomm(self) -> bool:
    """Try to open the persistent RFCOMM connection. Returns True on success."""
    with self.rfcomm_lock:
        if self.rfcomm_sock is not None:
            return True  # already open

        try:
            sock = socket.socket(socket.AF_BLUETOOTH, socket.SOCK_STREAM, socket.BTPROTO_RFCOMM)
            sock.settimeout(5)
            sock.connect((EARBUDS_MAC, RFCOMM_CHANNEL))
        except OSError as e:
            print(f"[!] RFCOMM connect failed: {e}")
            return False

        self.rfcomm_sock = sock
        print("[+] RFCOMM connection established and held open.")
        return True

def close_rfcomm(self):
    with self.rfcomm_lock:
        if self.rfcomm_sock is not None:
            try:
                self.rfcomm_sock.close()
            except OSError:
                pass
            self.rfcomm_sock = None
            print("[*] RFCOMM connection closed.")

def send_command(self, mode: str) -> str:
    """Send a command over the persistent connection. Returns a result string."""
    if mode not in COMMANDS:
        return f"ERROR unknown mode '{mode}'"

    name, payload = COMMANDS[mode]

    with self.rfcomm_lock:
        if self.rfcomm_sock is None:
            return "ERROR not connected (earbuds disconnected, or phone already holds the RFCOMM slot)"
        try:
            self.rfcomm_sock.send(payload)
            return f"OK {name}"
        except OSError as e:
            # the persistent connection itself died - drop it so the
            # reconnect loop will try to re-establish it
            print(f"[!] send() failed on persistent connection: {e}")
            try:
                self.rfcomm_sock.close()
            except OSError:
                pass
            self.rfcomm_sock = None
            return f"ERROR send failed, connection dropped: {e}"

# ---- bluetoothctl monitor thread ----

def bluetoothctl_monitor_loop(self):
    proc = subprocess.Popen(
        ["bluetoothctl"],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        bufsize=1,
    )
    try:
        for line in proc.stdout:
            if not self.running:
                break
            if CONNECTED_PATTERN.search(line):
                print("[*] Earbuds connected.")
                self.earbuds_connected = True
                self.open_rfcomm()
            elif DISCONNECTED_PATTERN.search(line):
                print("[*] Earbuds disconnected.")
                self.earbuds_connected = False
                self.close_rfcomm()
    finally:
        proc.terminate()

# ---- background reconnect loop ----

def reconnect_loop(self):
    while self.running:
        time.sleep(RECONNECT_RETRY_SECONDS)
        if self.earbuds_connected and self.rfcomm_sock is None:
            print("[*] Retrying RFCOMM connection...")
            self.open_rfcomm()

# ---- Unix socket command server ----

def socket_server_loop(self):
    if os.path.exists(SOCKET_PATH):
        os.remove(SOCKET_PATH)

    server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    server.bind(SOCKET_PATH)
    os.chmod(SOCKET_PATH, 0o666)  # allow non-root widget process to connect
    server.listen(5)
    server.settimeout(1)  # so we can check self.running periodically

    print(f"[*] Listening for commands on {SOCKET_PATH}")

    while self.running:
        try:
            conn, _ = server.accept()
        except socket.timeout:
            continue

        try:
            data = conn.recv(64).decode("utf-8", errors="replace").strip().lower()
            if data == "status":
                reply = "CONNECTED" if self.rfcomm_sock is not None else "DISCONNECTED"
            else:
                reply = self.send_command(data)
            conn.sendall(reply.encode("utf-8"))
        except OSError as e:
            print(f"[!] Socket server error: {e}")
        finally:
            conn.close()

    server.close()
    if os.path.exists(SOCKET_PATH):
        os.remove(SOCKET_PATH)

def shutdown(self, *_):
    print("\n[*] Shutting down...")
    self.running = False
    self.close_rfcomm()

def main():

daemon = AncDaemon()

signal.signal(signal.SIGINT, daemon.shutdown)

signal.signal(signal.SIGTERM, daemon.shutdown)
threads = [
    threading.Thread(target=daemon.bluetoothctl_monitor_loop, daemon=True),
    threading.Thread(target=daemon.reconnect_loop, daemon=True),
]
for t in threads:
    t.start()

print("[*] anc_daemon running. Waiting for earbuds connection...")
print("    (if already connected, disconnect/reconnect to trigger detection)")

# socket server runs on the main thread so Ctrl+C / signals work cleanly
daemon.socket_server_loop()

if name == "main":

main()

Run with sudo python3 anc_daemon.py (raw Bluetooth sockets need elevated privileges).

anc_client.py (lightweight client for the daemon)

#!/usr/bin/env python3
"""
anc_client.py - send a command to the running anc_daemon.py via its Unix
socket. This is the lightweight client a UI widget can call instead of
talking Bluetooth directly.
Usage:

python3 anc_client.py on

python3 anc_client.py off

python3 anc_client.py transparency

python3 anc_client.py status
Exit codes:

0 = daemon responded with OK / CONNECTED

1 = usage error

2 = could not reach daemon (not running, or socket missing)

3 = daemon responded with an error (e.g. not connected to earbuds)
No sudo needed - this just talks to the daemon's Unix socket, it doesn't

touch Bluetooth directly. (The daemon itself needs sudo.)

"""
import socket

import sys
SOCKET_PATH = "/tmp/anc_daemon.sock"

VALID_COMMANDS = {"on", "off", "transparency", "status"}
def main():

if len(sys.argv) != 2 or sys.argv[1] not in VALID_COMMANDS:

print(f"Usage: python3 {sys.argv[0]} <{'|'.join(sorted(VALID_COMMANDS))}>")

return 1
command = sys.argv[1]

try:
    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    sock.settimeout(5)
    sock.connect(SOCKET_PATH)
except OSError as e:
    print(f"ERROR could not reach daemon: {e}")
    print("Is anc_daemon.py running? (sudo python3 anc_daemon.py)")
    return 2

try:
    sock.sendall(command.encode("utf-8"))
    response = sock.recv(256).decode("utf-8", errors="replace")
except OSError as e:
    print(f"ERROR communicating with daemon: {e}")
    return 2
finally:
    sock.close()

print(response)
if response.startswith("OK") or response in ("CONNECTED", "DISCONNECTED"):
    return 0
return 3

if name == "main":

sys.exit(main())

ble_full_test.py (BLE GATT)

#!/usr/bin/env python3
"""
ble_full_test.py - cycle through known Baseus BP1 Pro commands over BLE GATT,
confirming each against the SPP-derived byte payloads.
Logs every notification so request/response can be correlated.

"""
import asyncio

from bleak import BleakClient
ADDRESS = "XX:XX:XX:XX:XX:XX"  # BLE address (distinct from the Classic BT/SPP address)

WRITE_UUID  = "ee684b1a-1e9b-ed3e-ee55-f894667e92ac"

NOTIFY_UUID = "654b749c-e37f-ae1f-ebab-40ca133e3690"
COMMANDS = [

("ANC OFF",       bytes.fromhex("ba3400ff")),

("ANC ON",        bytes.fromhex("ba340165")),

("Transparency",  bytes.fromhex("ba3402ff")),

("ANC OFF",       bytes.fromhex("ba3400ff")),

("Game Mode ON",  bytes.fromhex("ba2401")),

("Game Mode OFF", bytes.fromhex("ba2400")),

]
PAUSE_BETWEEN_COMMANDS = 4  # seconds
def handle_notify(sender, data):

print(f"    [NOTIFY] {data.hex()}")
async def main():

async with BleakClient(ADDRESS) as client:

await client.start_notify(NOTIFY_UUID, handle_notify)

print("[+] Connected, subscribed to notify characteristic.\n")

await asyncio.sleep(1)
    for label, payload in COMMANDS:
        print(f"[*] Sending: {label} -> {payload.hex()}")
        await client.write_gatt_char(WRITE_UUID, payload, response=True)
        await asyncio.sleep(PAUSE_BETWEEN_COMMANDS)
        print()

    print("[+] Sequence complete.")

if name == "main":

asyncio.run(main())

Requires pip install bleak. Tested on Linux Mint via BlueZ.

Other notes

• Connected via direct BLE address XX:XX:XX:XX:XX:XX, which is distinct from the Classic BT/SPP address on the same unit. This is worth noting for anyone confused why the two addresses differ on a single physical device.
• Also confirmed battery (AA 02 ...) and case (AA 27 ...) notifications match your documented format exactly, both observed unprompted shortly after connecting.

[elaxptr]

Hi! ive been quite busy lately, i will look into this later, thanks for your effort!!

[elaxptr]

Finally got a chance to sit down with this properly — this is a fantastic writeup, thank you. Independent confirmation on a second unit is exactly what this project needed, and both findings checked out against the code. Everything below has landed on main (2fab41a).

Finding 1 — flat AA 34 01 ack: confirmed this was a real problem. The decoder already tracked ANC state from the outgoing command, but it trusted the ack payload to distinguish Off (00) from active (non-zero) — on your firmware that would have flipped the UI back to "ANC on" right after switching ANC off. Ack handling now goes through Bp1ProAnc::resolve_anc_ack(): a zero payload always means Off, and a non-zero payload only confirms the last commanded mode rather than being read as a mode value — safe on both firmware variants. For reference, my unit did echo distinct AA 30/AA 32/AA 33 frames in nRF Connect captures, so this looks like genuine firmware variance; both behaviors are now documented side by side in docs/protocol/bp1-pro-anc.md.

Finding 2 — game mode: added end to end. BA 24 [00|01] command, AA 23 decoded as the state confirmation (DeviceEvent::GameModeUpdate), AA 24 treated as the flat "command received" ack and ignored, plus a Low Latency toggle in the UI. Modeled as an independent toggle rather than an AncMode variant, as you suggested.

Transport notes: the protocol doc no longer claims SPP is OTA-only, and now records that the device accepts bare BLE-style frames (BA op payload, no BA BA + CRC framing) over RFCOMM channel 1 — the official app frames its RFCOMM traffic, but the device is lenient. Great news for BlueZ/Linux support, and it means the scaffolded RfcommTransport can reuse the exact same command bytes, as you predicted. The single-RFCOMM-slot caveat you documented is noted too.

One loose end from your session log: your first ANC OFF elicited an AA 30 00 but the second didn't, so 0x30 stays classified as a keepalive for now. If you ever spot a pattern there — or see AA 32/AA 33 at all on your unit, or figure out what 0x92 does (it looks gesture-related in my captures) — I'd love to hear about it.

Thanks again — this saved a lot of RE time, and I'll credit this issue in the next release's changelog.