1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
|
#!/usr/bin/python3
#
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# The headphones panel's backend: noise cancelling on Sony headphones that
# speak the "v2" MDR protocol (the WH-CH720N here) over RFCOMM. Volume and
# battery are not here: volume is AVRCP through PipeWire, and battery is
# BlueZ's Battery1, both of which the island already reads.
#
# sonyctl MAC status {ok, mode, level, voice}
# sonyctl MAC nc|off set the mode, print the new state
# sonyctl MAC ambient [LEVEL] ambient sound, LEVEL 1-20 (default: kept)
# sonyctl MAC voice on|off ambient sound with focus on voice on or off
#
# Every set keeps the level and focus on voice it does not change.
# sonyctl MAC poweroff turn the headphones off
#
# mode is "nc", "ambient" or "off". Errors print {ok: false, error}.
#
# Frames, as measured on a WH-CH720N: 3e, type, seq, 4-byte big-endian
# length, payload, 1-byte sum of everything after 3e, 3c. 3c/3d/3e inside a
# frame are sent as 3d then the byte & 0xef. Type 0c is a command, 01 an
# ack; each side acks with seq 1 - the received seq, and the next command
# uses the seq of the last ack received. The headphones answer nothing until
# an init (00 00) has been sent on the connection.
#
# The ambient payload subtype is 17 on this model. 15, which Gadgetbridge
# sends for headphones without wind-noise reduction, is acked and never
# answered. The state is 17, 01, on, ambient, focus-on-voice, level.
import errno
import json
import socket
import sys
import time
CMD, ACK = 0x0C, 0x01
# ponytail: the SDP record measured on a WH-CH720N puts the MDR service
# (956c7b26-...) on channel 18. Look it up over SDP if another model differs.
CHANNEL = 18
SPECIAL = (0x3C, 0x3D, 0x3E)
def encode(kind, seq, payload):
body = bytes([kind, seq]) + len(payload).to_bytes(4, "big") + bytes(payload)
body += bytes([sum(body) & 0xFF])
out = bytearray([0x3E])
for b in body:
out += bytes([0x3D, b & 0xEF]) if b in SPECIAL else bytes([b])
return bytes(out) + b"\x3c"
def decode(raw):
"""One frame, without its 3c trailer, to (type, seq, payload) or None."""
if not raw.startswith(b"\x3e"):
return None
body, i = bytearray(), 1
while i < len(raw):
if raw[i] == 0x3D and i + 1 < len(raw):
i += 1
body.append(raw[i] | 0x10)
else:
body.append(raw[i])
i += 1
if len(body) < 7 or sum(body[:-1]) & 0xFF != body[-1]:
return None
n = int.from_bytes(body[2:6], "big")
if n != len(body) - 7:
return None
return body[0], body[1], bytes(body[6:-1])
def state(p):
"""A 67/69 17 payload to {mode, level, voice}."""
mode = "off" if p[3] == 0 else ("ambient" if p[4] else "nc")
return {"mode": mode, "level": p[6], "voice": p[5] == 1}
class Link:
def __init__(self, mac):
# A connect right after the previous run closed fails with EBUSY
# (measured) while the old channel is still going down.
for attempt in range(10):
self.s = socket.socket(socket.AF_BLUETOOTH, socket.SOCK_STREAM, socket.BTPROTO_RFCOMM)
self.s.settimeout(3)
try:
self.s.connect((mac, CHANNEL))
break
except OSError as e:
self.s.close()
if e.errno != errno.EBUSY or attempt == 9:
raise
time.sleep(0.3)
self.seq = 0
self.buf = b""
def frames(self):
while True:
while b"\x3c" in self.buf:
raw, self.buf = self.buf.split(b"\x3c", 1)
f = decode(raw[raw.find(b"\x3e"):]) if b"\x3e" in raw else None
if f:
yield f
chunk = self.s.recv(1024)
if not chunk:
raise ConnectionError("headphones closed the connection")
self.buf += chunk
def request(self, payload, reply):
"""Send a command and return the first command payload starting with
one of the reply codes, acking everything that arrives meanwhile."""
self.s.send(encode(CMD, self.seq, payload))
for kind, seq, p in self.frames():
if kind == ACK:
self.seq = seq
continue
self.s.send(encode(ACK, 1 - seq, b""))
if p and p[0] in reply:
return p
def run(mac, cmd, arg):
link = Link(mac)
try:
link.request([0x00, 0x00], (0x01,))
if cmd == "poweroff":
# Unanswered: the headphones go away instead. The ack is enough.
link.s.send(encode(CMD, link.seq, [0x24, 0x03, 0x01]))
for kind, _, _ in link.frames():
if kind == ACK:
return {"ok": True}
cur = state(link.request([0x66, 0x17], (0x67,)))
if cmd == "status":
return {"ok": True, **cur}
level, voice = cur["level"], int(cur["voice"])
if cmd == "voice":
voice = arg
elif arg is not None:
level = arg
on = 0 if cmd == "off" else 1
amb = 1 if cmd in ("ambient", "voice") else 0
p = link.request([0x68, 0x17, 0x01, on, amb, voice, level], (0x69,))
return {"ok": True, **state(p)}
finally:
link.s.close()
def main(argv):
if len(argv) < 3 or argv[2] not in ("status", "nc", "ambient", "voice", "off", "poweroff"):
return {"ok": False, "error": "usage: sonyctl MAC status|nc|ambient [LEVEL]|voice on|off|off|poweroff"}
arg = None
if argv[2] == "voice":
if len(argv) < 4 or argv[3] not in ("on", "off"):
return {"ok": False, "error": "voice takes on or off"}
arg = int(argv[3] == "on")
if argv[2] == "ambient" and len(argv) > 3:
if not argv[3].isdigit() or not 1 <= int(argv[3]) <= 20:
return {"ok": False, "error": "level must be 1-20"}
arg = int(argv[3])
try:
return run(argv[1], argv[2], arg)
except (OSError, ConnectionError) as e:
return {"ok": False, "error": str(e) or type(e).__name__}
if __name__ == "__main__":
r = main(sys.argv)
print(json.dumps(r))
sys.exit(0 if r["ok"] else 1)
|
