From 76c75fb42e1bd0457fdf2c70ba762bb5f75c698c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9s=20Antonio=20Riveros?= Date: Sun, 30 Aug 2026 23:36:48 -0700 Subject: [PATCH 1/4] Add QC Ultra 2 Earbuds support --- NOTES.md | 28 ++++ README.md | 13 +- cpp/src/bmap.h | 16 ++- cpp/src/catalog.h | 2 +- cpp/src/connection.h | 30 ++++- cpp/src/device.h | 37 ++++++ cpp/src/devices.h | 11 ++ cpp/src/main.cpp | 13 +- cpp/src/protocol.h | 12 +- cpp/tests/test_catalog.cpp | 2 +- cpp/tests/test_connection.cpp | 95 +++++++++++++ cpp/tests/test_parsers.cpp | 48 +++++++ cpp/tests/test_protocol.cpp | 2 + docs/architecture.md | 32 ++++- fixtures/README.md | 3 +- .../qc-ultra2-earbuds/battery-status.hex | 1 + python/pybmap/__init__.py | 17 ++- python/pybmap/catalog.py | 2 +- python/pybmap/cli.py | 7 + python/pybmap/connection.py | 66 +++++++-- python/pybmap/devices/__init__.py | 3 + python/pybmap/devices/parsers.py | 24 ++++ python/pybmap/devices/qc_ultra2_earbuds.py | 34 +++++ python/pybmap/protocol.py | 6 + python/pybmap/types.py | 9 ++ python/tests/test_catalog.py | 2 +- python/tests/test_channel_probe.py | 21 ++- python/tests/test_cli.py | 56 ++++++++ python/tests/test_connection.py | 74 ++++++++++- python/tests/test_device_parsers.py | 34 +++++ python/tests/test_protocol.py | 6 + python/tests/test_qc_ultra2.py | 9 +- rust/src/catalog.rs | 4 +- rust/src/connection.rs | 125 +++++++++++++++++- rust/src/device.rs | 100 ++++++++++++++ rust/src/devices.rs | 38 ++++++ rust/src/lib.rs | 30 ++++- rust/src/main.rs | 11 +- rust/src/protocol.rs | 8 +- 39 files changed, 973 insertions(+), 58 deletions(-) create mode 100644 fixtures/packets/qc-ultra2-earbuds/battery-status.hex create mode 100644 python/pybmap/devices/qc_ultra2_earbuds.py create mode 100644 python/tests/test_cli.py diff --git a/NOTES.md b/NOTES.md index 8b6d59c..ffc76ff 100644 --- a/NOTES.md +++ b/NOTES.md @@ -21,6 +21,34 @@ the app uses to change modes in real time. - Custom name: "Fargo" - Product ID: 0x4082, Variant: 0x01 +## QC Ultra 2 Earbuds — EDITH + +- Product: Bose QuietComfort Ultra Earbuds (2nd Gen) +- Codename: `edith` +- Product ID: `0x4062` +- Platform: OTG-QCC-384 +- Transport: Bluetooth SPP over RFCOMM channel 2 +- Feature layout: shared with QC Ultra 2 headphones +- Hardware validation: physical `edith` device; battery components, status, + listening modes, and CNC behavior verified + +### Component Battery Response + +EDITH returns multiple four-byte battery records from `[2.2]`. Each record is +`[level, reserved, reserved, component_id]`: + +| Component ID | Component | Example | +|--------------|-----------|---------| +| `0x01` | Right | `3cffff01` = 60% | +| `0x02` | Left | `3cffff02` = 60% | +| `0x03` | Case | `50ffff03` = 80% | +| `0x04` | Combined buds | `3cffff04` = 60%, used for generic Battery | + +Records must be identified by component ID, not payload position. The generic +`Battery` value uses ID `0x04`; the CLI separately displays IDs `0x01`, `0x02`, +and `0x03`. BMAP `[2.5]` tracks whether both earbuds are seated and did not +change across observed charger transitions, so case charging is not inferred. + ## BMAP Protocol - Version: 1.1.0 - Transport: Bluetooth SPP over RFCOMM channel 2 diff --git a/README.md b/README.md index 5bf116b..9b4cd90 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ [![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE) [![Release](https://img.shields.io/github/v/release/aaronsb/bosectl)](https://github.com/aaronsb/bosectl/releases/latest) -[![Devices](https://img.shields.io/badge/Devices-3_supported_·_38_known-green)](docs/architecture.md#device-catalog) +[![Devices](https://img.shields.io/badge/Devices-8_supported_·_38_known-green)](docs/architecture.md#device-catalog) [![Python 3](https://img.shields.io/badge/Python-3-3572A5.svg)](python/) [![Rust](https://img.shields.io/badge/Rust-1.70+-DEA584.svg)](rust/) [![C++17](https://img.shields.io/badge/C++-17-f34b7d.svg)](cpp/) @@ -26,6 +26,7 @@ connection to the headphones. | Device | NC Control | EQ | Spatial | Profiles | Buttons | Status | | --------------------------- | ------------------------------------- | ------ | -------------- | --------------------- | ---------------------- | ------------------ | | **QC Ultra Headphones 2** | CNC 0-10 slider | 3-band | room/head | 7 custom slots | Shortcut remap | Verified | +| **QC Ultra Earbuds 2** | CNC 0-10 (verified) | inherited | modes verified | inherited | inherited | Battery/status/modes/CNC verified (`edith`) | | **QuietComfort Headphones** | CNC 0-10 + Wind Block via ModeConfig | — | field observed | 2 user slots observed | — | Verified (`prince`) | | **QuietComfort 35 / 35 II** | ANR off/high/wind/low | — | — | — | Action remap (VPA/ANC) | Verified | | **QuietComfort Earbuds** | CNC 0-10 via direct SETGET | 3-band | — | 4 fixed modes | Remap | Verified (`lando`) | @@ -97,6 +98,10 @@ bosectl buttons set ANC # Remap programmable button bosectl quiet # Switch to Quiet mode ``` +When `BMAP_MAC` (or Python CLI alias `BOSE_MAC`) is set, also set +`BMAP_DEVICE` to the matching config key. Device type is auto-detected only +when the MAC address is auto-detected. + ### Device Catalog API ```python @@ -114,8 +119,8 @@ pybmap.modalias(0x4082) # "bluetooth:v05A7p4082d0000" # Check support status pybmap.is_supported(0x4082) # True — has tested config pybmap.is_supported(0x4075) # True — QuietComfort Headphones (prince) -pybmap.is_supported(0x4039) # False — QC45, recognized but untested -pybmap.supported_devices() # [wolfcastle, baywolf, edith, prince, wolverine] +pybmap.is_supported(0x4039) # True — QC45 uses an inferred config +pybmap.supported_devices() # eight devices; see docs/architecture.md pybmap.known_devices() # full catalog ``` @@ -241,7 +246,7 @@ Full protocol reference: **[NOTES.md](NOTES.md)** and ```bash make test # All tests (121 Python, 63 Rust, 54 C++) make artifacts # Build + strip + SHA256SUMS in dist/ -make release VERSION=v0.2.0 # Test → build → gh release create +make release VERSION=v0.4.0 # Test → build → gh release create make clean # Remove all build artifacts ``` diff --git a/cpp/src/bmap.h b/cpp/src/bmap.h index dd8facf..800e47e 100644 --- a/cpp/src/bmap.h +++ b/cpp/src/bmap.h @@ -19,6 +19,16 @@ inline constexpr uint8_t FALLBACK_CHANNELS[] = {2, 8, 9}; namespace detail { +inline void validate_device_override( + const std::string& mac_override, + const std::string& device_type_override) +{ + if (!mac_override.empty() && device_type_override.empty()) { + throw std::invalid_argument( + "device_type is required when mac is specified"); + } +} + inline void send_init(Transport& transport, const DeviceConfig& config) { if (config.init_packet) { auto pkt = bmap_packet(config.init_packet->fblock, @@ -81,11 +91,12 @@ inline std::unique_ptr open_transport(const std::string& mac, } // namespace detail -/// Connect to a BMAP device, auto-detecting if mac/device_type are empty. +/// Connect to a BMAP device. Device type is resolved only during MAC discovery. inline std::unique_ptr connect( const std::string& mac_override = "", const std::string& device_type_override = "") { + detail::validate_device_override(mac_override, device_type_override); std::string mac = mac_override; std::string device_type = device_type_override; @@ -100,9 +111,6 @@ inline std::unique_ptr connect( device_type = detected->second; } } - if (device_type.empty()) { - device_type = "qc_ultra2"; - } auto config = get_device(device_type); if (!config) { diff --git a/cpp/src/catalog.h b/cpp/src/catalog.h index d3ee188..835ac30 100644 --- a/cpp/src/catalog.h +++ b/cpp/src/catalog.h @@ -53,7 +53,7 @@ inline const std::vector& catalog() { {0x404C, "celine_ii", "Frames (2nd Gen)", Category::Earbuds, nullptr}, {0x4060, "olivia", "Frames Tempo", Category::Earbuds, nullptr}, {0x4061, "vedder", "Frames", Category::Earbuds, nullptr}, - {0x4062, "edith", "QuietComfort Ultra Earbuds (2nd Gen)", Category::Earbuds, "qc_ultra2"}, + {0x4062, "edith", "QuietComfort Ultra Earbuds (2nd Gen)", Category::Earbuds, "qc_ultra2_earbuds"}, {0x4064, "smalls", "QuietComfort Earbuds II", Category::Earbuds, nullptr}, {0x4068, "serena", "Ultra Open Earbuds", Category::Earbuds, "ultra_open"}, {0x4072, "scotty", "QuietComfort Ultra Earbuds", Category::Earbuds, nullptr}, diff --git a/cpp/src/connection.h b/cpp/src/connection.h index dadd391..904907a 100644 --- a/cpp/src/connection.h +++ b/cpp/src/connection.h @@ -4,6 +4,7 @@ #include #include #include +#include #include #include @@ -25,7 +26,28 @@ class BmapConnection { // ── Read Operations ───────────────────────────────────────────────────── - uint8_t battery() { return parse_battery(get(require(config_.battery, "battery"))); } + uint8_t battery() { + return battery_status().aggregate; + } + BatteryStatus battery_status() { + auto payload = get(require(config_.battery, "battery")); + if (config_.battery_aggregate_id) { + auto readings = parse_battery_readings(payload); + for (const auto& reading : readings) { + if (reading.component_id == *config_.battery_aggregate_id) + return {reading.level, std::move(readings)}; + } + throw std::runtime_error( + "Battery response missing aggregate component " + + std::to_string(*config_.battery_aggregate_id)); + } + if (payload.empty()) + throw std::runtime_error("Empty battery response"); + return {parse_battery(payload), {}}; + } + std::vector battery_readings() { + return battery_status().readings; + } std::string firmware() { return parse_firmware(get(require(config_.firmware, "firmware"))); } std::string name() { return parse_product_name(get(require(config_.product_name, "product_name"))); } std::pair cnc() { return parse_cnc(get(require(config_.cnc, "cnc"))); } @@ -90,9 +112,11 @@ class BmapConnection { [&]{ return cnc(); }, {0, 10}); auto [prom_on, prom_lang] = safe_call>( [&]{ return prompts(); }, {false, ""}); + auto battery_state = battery_status(); DeviceStatus s; - s.battery = battery(); + s.battery = battery_state.aggregate; + s.battery_readings = std::move(battery_state.readings); s.mode = mode_name; s.mode_idx = idx; s.cnc_level = cnc_cur; @@ -293,7 +317,7 @@ class BmapConnection { auto pkt = bmap_packet(addr.fblock, addr.func, Operator::Get); auto data = transport_->send_recv(pkt); auto resp = parse_response(data); - if (!resp) throw std::runtime_error("Empty response"); + if (!resp) throw std::runtime_error("Invalid or empty response"); check_error(*resp); return resp->payload; } diff --git a/cpp/src/device.h b/cpp/src/device.h index dad22b2..e3d71e8 100644 --- a/cpp/src/device.h +++ b/cpp/src/device.h @@ -6,7 +6,9 @@ #include #include #include +#include #include +#include #include #include "protocol.h" @@ -59,6 +61,16 @@ struct ButtonMapping { std::string action_name; }; +struct BatteryReading { + uint8_t component_id; + uint8_t level; +}; + +struct BatteryStatus { + uint8_t aggregate; + std::vector readings; +}; + struct AudioSource { std::string source_type; std::string source_mac; // empty if not bluetooth @@ -66,6 +78,7 @@ struct AudioSource { struct DeviceStatus { uint8_t battery; + std::vector battery_readings; std::string mode; uint8_t mode_idx; uint8_t cnc_level; @@ -91,6 +104,10 @@ struct DeviceConfig { /// Init packet required before device responds (QC35 needs GET [0.1]). std::optional init_packet; std::optional battery; + /// Component IDs and display labels for multi-cell battery responses. + std::vector> battery_components; + /// Component ID used for the generic aggregate battery value. + std::optional battery_aggregate_id; std::optional firmware; std::optional product_name; std::optional voice_prompts; @@ -128,6 +145,26 @@ inline uint8_t parse_battery(const std::vector& p) { return p.empty() ? 0 : p[0]; } +/// Parse four-byte component battery records from [2.2]. +inline std::vector parse_battery_readings(const std::vector& p) { + if (p.size() % 4 != 0) { + throw std::invalid_argument("Malformed battery response"); + } + std::vector readings; + std::vector seen_components; + for (size_t i = 0; i + 3 < p.size(); i += 4) { + for (auto component_id : seen_components) { + if (component_id == p[i + 3]) { + throw std::invalid_argument( + "Duplicate battery component " + std::to_string(p[i + 3])); + } + } + seen_components.push_back(p[i + 3]); + if (p[i] <= 100) readings.push_back({p[i + 3], p[i]}); + } + return readings; +} + inline std::string parse_firmware(const std::vector& p) { return {p.begin(), p.end()}; } diff --git a/cpp/src/devices.h b/cpp/src/devices.h index b9f23eb..add9fb5 100644 --- a/cpp/src/devices.h +++ b/cpp/src/devices.h @@ -41,6 +41,16 @@ inline DeviceConfig qc_ultra2() { return c; } +/// Bose QuietComfort Ultra Earbuds 2 -- edith, product ID 0x4062. +/// Shares the QC Ultra 2 protocol layout and reports separate battery cells. +inline DeviceConfig qc_ultra2_earbuds() { + auto c = qc_ultra2(); + c.info = {"Bose QuietComfort Ultra Earbuds (2nd Gen)", "edith", "OTG-QCC-384"}; + c.battery_components = {{1, "Right"}, {2, "Left"}, {3, "Case"}}; + c.battery_aggregate_id = 4; + return c; +} + /// Bose QuietComfort Headphones -- prince, product ID 0x4075. /// Verified against firmware 1.0.6-80+f5f219b. RFCOMM channel 8. inline DeviceConfig qc_prince() { @@ -187,6 +197,7 @@ inline DeviceConfig ultra_open() { inline std::optional get_device(const std::string& name) { if (name == "qc_ultra2") return qc_ultra2(); + if (name == "qc_ultra2_earbuds") return qc_ultra2_earbuds(); if (name == "qc35") return qc35(); if (name == "qc_prince") return qc_prince(); if (name == "qc_earbuds") return qc_earbuds(); diff --git a/cpp/src/main.cpp b/cpp/src/main.cpp index 4411cf7..078f21a 100644 --- a/cpp/src/main.cpp +++ b/cpp/src/main.cpp @@ -1,6 +1,7 @@ // bmapctl — Minimal CLI for controlling BMAP devices (C++ version). #include +#include #include #include @@ -30,7 +31,8 @@ static void usage() { << " off Power off\n\n" << "Environment:\n" << " BMAP_MAC=XX:XX:XX:XX:XX:XX Device MAC\n" - << " BMAP_DEVICE=qc_ultra2|qc_prince|qc35 Device type\n"; + << " BMAP_DEVICE=qc_ultra2|qc_ultra2_earbuds|qc_prince|qc35|" + "qc_earbuds|qc45|ultra_open\n"; } static bool is_on(const std::string& s) { @@ -63,6 +65,15 @@ int main(int argc, char** argv) { auto s = dev.status(); std::cout << " Model " << dev.config().info.name << "\n"; std::cout << " Battery " << (int)s.battery << "%\n"; + for (const auto& [id, label] : dev.config().battery_components) { + for (const auto& reading : s.battery_readings) { + if (id == reading.component_id) { + std::cout << " " << std::left << std::setw(12) << label + << std::right << (int)reading.level << "%\n"; + break; + } + } + } if (!s.mode.empty()) std::cout << " Mode " << s.mode << "\n"; if (dev.has_feature("anr")) { diff --git a/cpp/src/protocol.h b/cpp/src/protocol.h index cae02af..af7d236 100644 --- a/cpp/src/protocol.h +++ b/cpp/src/protocol.h @@ -90,13 +90,15 @@ inline std::vector bmap_packet(uint8_t fblock, uint8_t func, inline std::optional parse_response(const std::vector& data) { if (data.size() < 4) return std::nullopt; + auto raw_op = data[2] & 0x0F; + if (raw_op > static_cast(Operator::Processing)) return std::nullopt; BmapResponse resp; resp.fblock = data[0]; resp.func = data[1]; - resp.op = static_cast(data[2] & 0x0F); + resp.op = static_cast(raw_op); uint8_t length = data[3]; - size_t end = std::min(4 + length, data.size()); - resp.payload.assign(data.begin() + 4, data.begin() + end); + if (data.size() < 4 + length) return std::nullopt; + resp.payload.assign(data.begin() + 4, data.begin() + 4 + length); return resp; } @@ -107,7 +109,9 @@ inline std::vector parse_all_responses(const std::vector& BmapResponse resp; resp.fblock = data[pos]; resp.func = data[pos + 1]; - resp.op = static_cast(data[pos + 2] & 0x0F); + auto raw_op = data[pos + 2] & 0x0F; + if (raw_op > static_cast(Operator::Processing)) break; + resp.op = static_cast(raw_op); uint8_t length = data[pos + 3]; if (pos + 4 + length > data.size()) break; resp.payload.assign(data.begin() + pos + 4, data.begin() + pos + 4 + length); diff --git a/cpp/tests/test_catalog.cpp b/cpp/tests/test_catalog.cpp index ae43446..333b4c8 100644 --- a/cpp/tests/test_catalog.cpp +++ b/cpp/tests/test_catalog.cpp @@ -32,7 +32,7 @@ TEST(catalog_lookup_qc_ultra2_earbuds) { ASSERT_TRUE(dev != nullptr); ASSERT_EQ(std::string(dev->codename), std::string("edith")); ASSERT_TRUE(dev->config != nullptr); - ASSERT_EQ(std::string(dev->config), std::string("qc_ultra2")); + ASSERT_EQ(std::string(dev->config), std::string("qc_ultra2_earbuds")); } TEST(catalog_lookup_qc_prince) { diff --git a/cpp/tests/test_connection.cpp b/cpp/tests/test_connection.cpp index 80e30b5..9d2ece9 100644 --- a/cpp/tests/test_connection.cpp +++ b/cpp/tests/test_connection.cpp @@ -7,6 +7,7 @@ #include "../src/connection.h" #include "../src/devices.h" +#include "../src/bmap.h" using namespace bmap; @@ -52,6 +53,91 @@ static std::unique_ptr mock_qc_ultra2() { } TEST(battery) { ASSERT_EQ(mock_qc_ultra2()->battery(), 80); } + +TEST(battery_empty_response) { + auto raw = new MockTransport(); + raw->add(2, 2, 0x03, {}); + BmapConnection dev(std::unique_ptr(raw), qc_ultra2()); + bool threw = false; + try { dev.battery(); } + catch (const std::runtime_error& error) { + threw = std::string(error.what()).find("Empty battery response") != std::string::npos; + } + ASSERT_TRUE(threw); +} + +TEST(battery_invalid_response) { + auto raw = new MockTransport(); + raw->responses[{2, 2}] = {2, 2, 0x08, 0}; + BmapConnection dev(std::unique_ptr(raw), qc_ultra2()); + bool threw = false; + try { dev.battery(); } + catch (const std::runtime_error& error) { + threw = std::string(error.what()).find("Invalid or empty response") != std::string::npos; + } + ASSERT_TRUE(threw); +} + +TEST(battery_readings) { + auto raw = new MockTransport(); + raw->add(2, 2, 0x03, { + 0x50,0xff,0xff,0x03, 0x3c,0xff,0xff,0x01, + 0x3c,0xff,0xff,0x02, 0x46,0xff,0xff,0x04, + }); + BmapConnection dev(std::unique_ptr(raw), qc_ultra2_earbuds()); + auto battery = dev.battery_status(); + ASSERT_EQ(battery.readings.size(), 4u); + ASSERT_EQ(battery.readings[0].component_id, 3); + ASSERT_EQ(battery.readings[0].level, 80); + ASSERT_EQ(battery.readings[3].component_id, 4); + ASSERT_EQ(dev.config().battery_components.size(), 3u); + ASSERT_EQ(battery.aggregate, 70); + ASSERT_EQ(raw->sent.size(), 1u); +} + +TEST(battery_missing_aggregate_component) { + auto raw = new MockTransport(); + raw->add(2, 2, 0x03, { + 0x3c,0xff,0xff,0x01, 0x3c,0xff,0xff,0x02, + 0x50,0xff,0xff,0x03, + }); + BmapConnection dev(std::unique_ptr(raw), qc_ultra2_earbuds()); + bool threw = false; + try { dev.battery(); } + catch (const std::runtime_error& error) { + threw = std::string(error.what()).find("aggregate component 4") != std::string::npos; + } + ASSERT_TRUE(threw); +} + +TEST(status_uses_one_battery_response) { + auto raw = new MockTransport(); + raw->add(2, 2, 0x03, { + 0x50,0xff,0xff,0x03, 0x3c,0xff,0xff,0x01, + 0x46,0xff,0xff,0x04, 0x3c,0xff,0xff,0x02, + }); + BmapConnection dev(std::unique_ptr(raw), qc_ultra2_earbuds()); + auto status = dev.status(); + ASSERT_EQ(status.battery, 70); + ASSERT_EQ(status.battery_readings.size(), 4u); + size_t battery_requests = 0; + for (const auto& packet : raw->sent) { + if (packet.size() >= 2 && packet[0] == 2 && packet[1] == 2) battery_requests++; + } + ASSERT_EQ(battery_requests, 1u); +} + +TEST(qc_ultra2_earbuds_config) { + auto config = qc_ultra2_earbuds(); + ASSERT_EQ(config.info.codename, "edith"); + ASSERT_EQ(config.info.name, "Bose QuietComfort Ultra Earbuds (2nd Gen)"); + ASSERT_EQ(config.battery_components.size(), 3u); + ASSERT_EQ(config.battery_components[0].second, "Right"); + ASSERT_EQ(config.battery_components[1].second, "Left"); + ASSERT_EQ(config.battery_components[2].second, "Case"); + ASSERT_EQ(*config.battery_aggregate_id, 4); + ASSERT_EQ(config.preset_modes.size(), 4u); +} TEST(firmware) { ASSERT_EQ(mock_qc_ultra2()->firmware(), "8.2.20+g34cf029"); } TEST(device_name) { ASSERT_EQ(mock_qc_ultra2()->name(), "Fargo"); } @@ -93,6 +179,15 @@ TEST(has_feature_battery) { ASSERT_TRUE(mock_qc_ultra2()->has_feature("battery") TEST(has_feature_eq) { ASSERT_TRUE(mock_qc_ultra2()->has_feature("eq")); } TEST(has_feature_missing) { ASSERT_FALSE(mock_qc_ultra2()->has_feature("nonexistent")); } +TEST(explicit_mac_requires_device_type) { + bool threw = false; + try { bmap::detail::validate_device_override("00:11:22:33:44:55", ""); } + catch (const std::invalid_argument& error) { + threw = std::string(error.what()).find("device_type is required") != std::string::npos; + } + ASSERT_TRUE(threw); +} + TEST(qc35_no_eq) { auto t = std::make_unique(); BmapConnection dev(std::move(t), qc35()); diff --git a/cpp/tests/test_parsers.cpp b/cpp/tests/test_parsers.cpp index 59c87a9..dd1e3c2 100644 --- a/cpp/tests/test_parsers.cpp +++ b/cpp/tests/test_parsers.cpp @@ -1,9 +1,29 @@ // Tests for shared device parsers using real captured data. #include "test_common.h" +#include +#include +#include +#include #include "../src/device.h" using namespace bmap; +static std::vector decode_hex_fixture(const std::string& relative_path) { + auto path = std::filesystem::path(__FILE__).parent_path() / relative_path; + std::ifstream input(path); + if (!input) throw std::runtime_error("Could not open fixture: " + path.string()); + std::string hex((std::istreambuf_iterator(input)), {}); + hex.erase(std::remove_if(hex.begin(), hex.end(), [](unsigned char c) { + return std::isspace(c); + }), hex.end()); + if (hex.size() % 2 != 0) throw std::runtime_error("Fixture contains incomplete hex byte"); + std::vector bytes; + for (size_t i = 0; i < hex.size(); i += 2) { + bytes.push_back(static_cast(std::stoul(hex.substr(i, 2), nullptr, 16))); + } + return bytes; +} + TEST(parse_battery_from_capture) { ASSERT_EQ(parse_battery({0x50, 0xff, 0xff, 0x00}), 0x50); } @@ -12,6 +32,34 @@ TEST(parse_battery_empty) { ASSERT_EQ(parse_battery({}), 0); } +TEST(parse_battery_readings) { + auto readings = parse_battery_readings(decode_hex_fixture( + "../../fixtures/packets/qc-ultra2-earbuds/battery-status.hex")); + ASSERT_EQ(readings.size(), 4u); + ASSERT_EQ(readings[0].component_id, 1); + ASSERT_EQ(readings[0].level, 60); + ASSERT_EQ(readings[3].component_id, 3); + ASSERT_EQ(readings[3].level, 80); + auto shuffled = parse_battery_readings({ + 0x50,0xff,0xff,0x03, 0x46,0xff,0xff,0x04, + 0x32,0xff,0xff,0x09, 0x3c,0xff,0xff,0x01, + 0x3c,0xff,0xff,0x02, + }); + ASSERT_EQ(shuffled.size(), 5u); + ASSERT_EQ(shuffled[0].component_id, 3); + ASSERT_EQ(shuffled[1].component_id, 4); + ASSERT_EQ(shuffled[2].component_id, 9); + bool threw = false; + try { parse_battery_readings({0x50, 0xff}); } + catch (const std::invalid_argument&) { threw = true; } + ASSERT_TRUE(threw); + threw = false; + try { parse_battery_readings({ + 0x3c,0xff,0xff,0x01, 0x50,0xff,0xff,0x01}); } + catch (const std::invalid_argument&) { threw = true; } + ASSERT_TRUE(threw); +} + TEST(parse_firmware_from_capture) { std::vector p = {'8','.','2','.','2','0','+','g','3','4','c','f','0','2','9'}; ASSERT_EQ(parse_firmware(p), "8.2.20+g34cf029"); diff --git a/cpp/tests/test_protocol.cpp b/cpp/tests/test_protocol.cpp index e78227e..c82e8aa 100644 --- a/cpp/tests/test_protocol.cpp +++ b/cpp/tests/test_protocol.cpp @@ -34,6 +34,8 @@ TEST(parse_response_basic) { TEST(parse_response_too_short) { auto resp = parse_response({1, 2}); ASSERT_FALSE(resp.has_value()); + ASSERT_FALSE(parse_response({2, 2, 0x03, 4, 80, 0xff}).has_value()); + ASSERT_FALSE(parse_response({2, 2, 0x08, 0}).has_value()); } TEST(parse_all_responses_two) { diff --git a/docs/architecture.md b/docs/architecture.md index cf94ea3..9d0d21e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -129,6 +129,7 @@ graph LR FEAT[Feature Map
name → addr + parser + builder] MODES[Preset Modes
name → index] SLOTS[Editable Slots
profile indices] + CELLS[Battery Metadata
component ID → label
aggregate component ID] end ``` @@ -144,6 +145,11 @@ Each feature entry maps a name to its protocol address and codec functions: **Device quirks are expressed as config differences, not code branches:** +Multi-component battery devices declare display labels and an aggregate +component ID in config. For `edith`, IDs 1/2/3 are Right/Left/Case and ID 4 +is the combined earbud level. Unknown IDs are preserved in the snapshot but +only configured components are rendered. + | Property | QC Ultra 2 | QuietComfort Headphones (`prince`) | QC35 | |----------|-----------|-------------------------------------|------| | RFCOMM channel | 2 | 8 | 8 | @@ -185,9 +191,14 @@ sequenceDiagram **Read pattern** (GET): ``` -battery() → _get("battery") → lookup addr → send GET → parse response → apply parser +battery_status() → one GET → aggregate level + component ID/level pairs +battery() / status() → reuse the parsed battery status ``` +`BatteryStatus` keeps the aggregate and all component readings from that one +response. `DeviceStatus.battery_readings` carries the same snapshot to CLIs, +which render known components in config order rather than packet order. + **Write pattern** (SETGET): ``` set_anr("high") → lookup addr + builder → build payload → send SETGET → check for errors @@ -207,6 +218,7 @@ feature map, not inherited. **Parsers** decode response payloads: ``` parse_battery([0x50, 0xff, 0xff, 0x00]) → 80 +parse_battery_readings([0x3c,0xff,0xff,0x01, ...]) → [(component=1, level=60), ...] parse_cnc([0x0b, 0x07, 0x03]) → (current=7, max=10) parse_anr([0x01, 0x0b]) → "high" parse_buttons([0x10, 0x04, 0x01, 0x07]) → ButtonMapping{Action, single_press, VPA} @@ -258,6 +270,8 @@ sequenceDiagram 3. Modalias product ID maps to a known device type Connected devices are preferred over paired-but-disconnected. +When a caller supplies a MAC explicitly, it must also supply the device config +key; the type can only be detected while discovering the MAC. ## Error Handling @@ -317,7 +331,7 @@ graph TD CAT[Device Catalog
known BMAP devices] --> SUP[Supported
config ≠ None] CAT --> UNSUP[Recognized but Unsupported
config = None] SUP --> DISC[Discovery
PID → config lookup] - SUP --> CFG[Device Configs
qc_ultra2, qc_prince, qc35] + SUP --> CFG[Device Configs
qc_ultra2, qc_ultra2_earbuds, qc_prince, qc35] DISC --> CONN[connect()] CFG --> CONN @@ -343,9 +357,10 @@ Each catalog entry carries: ``` lookup_device(0x4082) → BoseDevice{wolverine, "QuietComfort Ultra Headphones (2nd Gen)", config="qc_ultra2"} +lookup_device(0x4062) → BoseDevice{edith, "QuietComfort Ultra Earbuds (2nd Gen)", config="qc_ultra2_earbuds"} lookup_device(0x4075) → BoseDevice{prince, "QuietComfort Headphones", config="qc_prince"} is_supported(0x4024) → False (NCH 700: recognized, no config yet) -supported_devices() → [wolfcastle, baywolf, edith, prince, wolverine] +supported_devices() → [wolfcastle, baywolf, duran, prince, wolverine, lando, edith, serena] known_devices() → full catalog usb_ids(0x4082) → (0x05A7, 0x4082) modalias(0x4082) → "bluetooth:v05A7p4082d0000" @@ -368,7 +383,7 @@ a default config since they don't have a tested implementation yet. |-----|----------|---------|--------| | `0x400C` | wolfcastle | QuietComfort 35 | `qc35` | | `0x4020` | baywolf | QuietComfort 35 II | `qc35` | -| `0x4062` | edith | QuietComfort Ultra Earbuds (2nd Gen) | `qc_ultra2` | +| `0x4062` | edith | QuietComfort Ultra Earbuds (2nd Gen) | `qc_ultra2_earbuds` | | `0x4075` | prince | QuietComfort Headphones | `qc_prince` | | `0x402F` | lando | QuietComfort Earbuds | `qc_earbuds` | | `0x4039` | duran | QuietComfort 45 | `qc45` (inferred, untested) | @@ -414,6 +429,7 @@ pybmap/ ├── protocol.py # Packet codec (bmap_packet, parse_response) ├── transport.py # RfcommTransport (AF_BLUETOOTH socket) ├── connection.py # BmapConnection class +├── cli.py # bosectl command-line interface ├── discovery.py # find_bmap_device() via bluetoothctl ├── constants.py # Operators, error codes, button/action/language tables ├── types.py # NamedTuples (BmapResponse, ModeConfig, ButtonMapping, etc.) @@ -422,7 +438,11 @@ pybmap/ ├── __init__.py # Device registry (DEVICES dict, get_device()) ├── parsers.py # Shared parser/builder functions ├── qc_ultra2.py # QC Ultra 2 config (module-level constants) + ├── qc_ultra2_earbuds.py # QC Ultra 2 Earbuds config (shared layout) ├── qc_prince.py # QuietComfort Headphones / prince config + ├── qc45.py # QC45 config + ├── qc_earbuds.py # QuietComfort Earbuds config + ├── ultra_open.py # Ultra Open Earbuds partial config └── qc35.py # QC35 config (module-level constants) ``` @@ -459,7 +479,7 @@ rust/src/ ├── connection.rs # BmapConnection generic struct ├── discovery.rs # find_bmap_device() via bluetoothctl ├── device.rs # DeviceConfig struct, all parsers/builders -├── devices.rs # qc_ultra2() / qc35() factory functions +├── devices.rs # qc_ultra2() / qc_ultra2_earbuds() / qc35() ├── error.rs # BmapError enum, BmapResult type alias └── main.rs # CLI binary (bmapctl) ``` @@ -514,7 +534,7 @@ cpp/src/ ├── transport.cpp # RfcommTransport (BlueZ sockets) ├── connection.h # BmapConnection class (header-only) ├── device.h # DeviceConfig, all parsers/builders (header-only) -├── devices.h # qc_ultra2() / qc35() inline functions +├── devices.h # qc_ultra2() / qc_ultra2_earbuds() / qc35() ├── discovery.h # find_bmap_device() declaration ├── discovery.cpp # Discovery implementation └── main.cpp # CLI binary (bmapctl) diff --git a/fixtures/README.md b/fixtures/README.md index c105598..cb7c5ec 100644 --- a/fixtures/README.md +++ b/fixtures/README.md @@ -5,7 +5,8 @@ Shared across Python, Rust, and C++ test suites. ## Structure -- `packets//` — Raw captured request/response pairs (JSON files from bmap-capture.py) +- `packets//` — Raw captured request/response pairs and payloads + (`.json` snapshots or plain hexadecimal `.hex` files) - `expected/` — Expected parse results for unit tests (JSON) ## Packet format diff --git a/fixtures/packets/qc-ultra2-earbuds/battery-status.hex b/fixtures/packets/qc-ultra2-earbuds/battery-status.hex new file mode 100644 index 0000000..b634f9a --- /dev/null +++ b/fixtures/packets/qc-ultra2-earbuds/battery-status.hex @@ -0,0 +1 @@ +3cffff013cffff023cffff0450ffff03 diff --git a/python/pybmap/__init__.py b/python/pybmap/__init__.py index 07ca412..206d496 100644 --- a/python/pybmap/__init__.py +++ b/python/pybmap/__init__.py @@ -27,11 +27,14 @@ BmapError, BmapConnectionError, BmapAuthError, BmapDeviceError, BmapTimeoutError, BmapNotFoundError, ) -from .types import DeviceStatus, ModeConfig, EqBand, ButtonMapping, BmapResponse +from .types import ( + BatteryReading, BatteryStatus, BmapResponse, ButtonMapping, DeviceStatus, + EqBand, ModeConfig, +) from .protocol import bmap_packet, parse_response, parse_all_responses from .constants import OP_STATUS -__version__ = "0.1.0" +__version__ = "0.4.0" def connect(mac=None, device_type=None): @@ -40,7 +43,7 @@ def connect(mac=None, device_type=None): Args: mac: Bluetooth MAC address. Auto-detected if None. device_type: Device type string (e.g. "qc_ultra2", "qc35"). - Defaults to "qc_ultra2" if not specified. + Required when mac is specified; auto-detected otherwise. Returns: BmapConnection context manager. @@ -49,6 +52,9 @@ def connect(mac=None, device_type=None): BmapNotFoundError: If no device is found. BmapConnectionError: If the connection fails. """ + mac = mac or None + device_type = device_type or None + if mac is None: detected_mac, detected_type = find_bmap_device() if detected_mac is None: @@ -59,9 +65,8 @@ def connect(mac=None, device_type=None): mac = detected_mac if device_type is None: device_type = detected_type - - if device_type is None: - device_type = "qc_ultra2" + elif device_type is None: + raise BmapError("device_type is required when mac is specified") device = get_device(device_type) channel = getattr(device, "RFCOMM_CHANNEL", 2) diff --git a/python/pybmap/catalog.py b/python/pybmap/catalog.py index e1b4c49..b29fa3d 100644 --- a/python/pybmap/catalog.py +++ b/python/pybmap/catalog.py @@ -59,7 +59,7 @@ 0x404C: BoseDevice(0x404C, "celine_ii", "Frames (2nd Gen)", "earbuds", None), 0x4060: BoseDevice(0x4060, "olivia", "Frames Tempo", "earbuds", None), 0x4061: BoseDevice(0x4061, "vedder", "Frames", "earbuds", None), - 0x4062: BoseDevice(0x4062, "edith", "QuietComfort Ultra Earbuds (2nd Gen)", "earbuds", "qc_ultra2"), + 0x4062: BoseDevice(0x4062, "edith", "QuietComfort Ultra Earbuds (2nd Gen)", "earbuds", "qc_ultra2_earbuds"), 0x4064: BoseDevice(0x4064, "smalls", "QuietComfort Earbuds II", "earbuds", None), 0x4068: BoseDevice(0x4068, "serena", "Ultra Open Earbuds", "earbuds", "ultra_open"), 0x4072: BoseDevice(0x4072, "scotty", "QuietComfort Ultra Earbuds", "earbuds", None), diff --git a/python/pybmap/cli.py b/python/pybmap/cli.py index bad938d..3e6f154 100644 --- a/python/pybmap/cli.py +++ b/python/pybmap/cli.py @@ -63,6 +63,13 @@ def cmd_status(dev): row("Model", dev.device_info.get("name", "Unknown"), C_MAGENTA) batt_color = C_GREEN if s.battery > 30 else C_YELLOW if s.battery > 10 else C_RED row("Battery", "%d%%" % s.battery, batt_color) + component_names = dev.battery_components + readings = {reading.component_id: reading.level for reading in s.battery_readings} + for component_id, label in component_names.items(): + if component_id in readings: + level = readings[component_id] + color = C_GREEN if level > 30 else C_YELLOW if level > 10 else C_RED + row(label, "%d%%" % level, color) if s.mode: row("Mode", s.mode, C_CYAN) diff --git a/python/pybmap/connection.py b/python/pybmap/connection.py index 7ee51d4..f309381 100644 --- a/python/pybmap/connection.py +++ b/python/pybmap/connection.py @@ -18,7 +18,8 @@ ) from .protocol import bmap_packet, parse_response, parse_all_responses, fmt_response from .errors import BmapError, BmapAuthError, BmapDeviceError -from .types import DeviceStatus, AudioSettings +from .types import AudioSettings, BatteryStatus, DeviceStatus +from .devices import parsers # The device name field is 32 bytes on every BMAP device seen so far. @@ -53,20 +54,26 @@ def _feature(self, name): ) return features[name] - def _get(self, feature_name): - """Send a GET request and return the parsed payload.""" + def _get_payload(self, feature_name): + """Send a GET request and return its raw payload.""" feat = self._feature(feature_name) fblock, func = feat["addr"] resp = self._transport.send_recv(bmap_packet(fblock, func, OP_GET)) parsed = parse_response(resp) if parsed is None: - return None + raise BmapDeviceError("Invalid or empty response") if parsed.op == OP_ERROR: self._raise_error(parsed) + return parsed.payload + + def _get(self, feature_name): + """Send a GET request and return its parsed payload.""" + feat = self._feature(feature_name) + payload = self._get_payload(feature_name) parser = feat.get("parser") if parser: - return parser(parsed.payload) - return parsed.payload + return parser(payload) + return payload def _setget(self, feature_name, payload): """Send a SETGET request and return the parsed response.""" @@ -126,7 +133,38 @@ def _raise_error(self, parsed): def battery(self): """Battery percentage (int).""" - return self._get("battery") + return self.battery_status().aggregate + + def battery_status(self): + """Aggregate and component levels from one battery response.""" + feature = self._feature("battery") + payload = self._get_payload("battery") + + aggregate_id = self.battery_aggregate_id + if aggregate_id is None: + parser = feature.get("parser", parsers.parse_battery) + aggregate = parser(payload) + if aggregate is None: + raise BmapDeviceError("Empty battery response") + return BatteryStatus(aggregate=aggregate, readings=[]) + + try: + readings = parsers.parse_battery_readings(payload) + except ValueError as error: + raise BmapDeviceError(str(error)) from error + aggregate = next( + (reading.level for reading in readings + if reading.component_id == aggregate_id), + None, + ) + if aggregate is None: + raise BmapDeviceError( + "Battery response missing aggregate component %d" % aggregate_id) + return BatteryStatus(aggregate=aggregate, readings=readings) + + def battery_readings(self): + """Component battery readings, when the device reports multiple cells.""" + return self.battery_status().readings def firmware(self): """Firmware version string.""" @@ -261,9 +299,11 @@ def status(self): current_name = self._mode_name_from_idx(current_idx) if current_idx is not None else "" cnc_cur, cnc_max = self._safe_read(self.cnc, (0, 10)) prompts_on, prompts_lang = self._safe_read(self.prompts, (False, "")) + battery = self.battery_status() return DeviceStatus( - battery=self.battery(), + battery=battery.aggregate, + battery_readings=battery.readings, mode=current_name, mode_idx=current_idx, cnc_level=cnc_cur, cnc_max=cnc_max, @@ -552,6 +592,16 @@ def device_info(self): """Device identification dict.""" return self._device.DEVICE_INFO + @property + def battery_components(self): + """Known component ID to label mappings for battery readings.""" + return getattr(self._device, "BATTERY_COMPONENTS", {}) + + @property + def battery_aggregate_id(self): + """Component ID used for the generic battery level, when applicable.""" + return getattr(self._device, "BATTERY_AGGREGATE_ID", None) + @property def preset_modes(self): """Preset mode definitions dict (name -> {idx, description}).""" diff --git a/python/pybmap/devices/__init__.py b/python/pybmap/devices/__init__.py index 86fb886..49c3cd2 100644 --- a/python/pybmap/devices/__init__.py +++ b/python/pybmap/devices/__init__.py @@ -1,6 +1,7 @@ """Device registry for BMAP-capable devices.""" from . import qc_ultra2 +from . import qc_ultra2_earbuds from . import qc35 from . import qc_prince from . import qc_earbuds @@ -10,6 +11,7 @@ # Registry of supported devices keyed by type string. DEVICES = { "qc_ultra2": qc_ultra2, + "qc_ultra2_earbuds": qc_ultra2_earbuds, "qc35": qc35, "qc_prince": qc_prince, "qc_earbuds": qc_earbuds, @@ -20,6 +22,7 @@ # Product ID -> device type (for auto-detection after connecting). PRODUCT_IDS = { 0x4082: "qc_ultra2", + 0x4062: "qc_ultra2_earbuds", 0x4075: "qc_prince", 0x402F: "qc_earbuds", 0x4039: "qc45", diff --git a/python/pybmap/devices/parsers.py b/python/pybmap/devices/parsers.py index 96ef98d..291c6d8 100644 --- a/python/pybmap/devices/parsers.py +++ b/python/pybmap/devices/parsers.py @@ -23,6 +23,30 @@ def parse_battery(payload): return None +def parse_battery_readings(payload): + """Parse four-byte component battery records. + + Earbud responses observed on BMAP [2.2] use the layout + ``[level, reserved, reserved, component_id]``. + """ + from ..types import BatteryReading + + if len(payload) % 4: + raise ValueError("Malformed battery response") + + readings = [] + seen_components = set() + for i in range(0, len(payload) - 3, 4): + level = payload[i] + component_id = payload[i + 3] + if component_id in seen_components: + raise ValueError("Duplicate battery component %d" % component_id) + seen_components.add(component_id) + if level <= 100: + readings.append(BatteryReading(component_id, level)) + return readings + + def parse_firmware(payload): """Parse firmware version string.""" return payload.decode("ascii", errors="replace") diff --git a/python/pybmap/devices/qc_ultra2_earbuds.py b/python/pybmap/devices/qc_ultra2_earbuds.py new file mode 100644 index 0000000..6285ba7 --- /dev/null +++ b/python/pybmap/devices/qc_ultra2_earbuds.py @@ -0,0 +1,34 @@ +"""Bose QuietComfort Ultra Earbuds (2nd Gen) device configuration. + +The earbuds use the same tested BMAP feature layout as QC Ultra 2 headphones, +but report separate battery records for the left bud, right bud, and case. +""" + +from .qc_ultra2 import ( + DEVICE_INFO as _HEADPHONE_INFO, + EDITABLE_SLOTS, + FEATURES as _HEADPHONE_FEATURES, + MODE_BY_IDX, + PRESET_MODES, + RFCOMM_CHANNEL, + STATUS_OFFSETS, +) + + +DEVICE_INFO = dict(_HEADPHONE_INFO) +DEVICE_INFO.update({ + "name": "Bose QuietComfort Ultra Earbuds (2nd Gen)", + "codename": "edith", + "product_id": 0x4062, + "category": "earbuds", +}) + +# Keep the shared feature layout. Case charging is intentionally not exposed: +# BMAP [2.5] changes with earbud seating, but stayed the same across observed +# charger transitions, so it is not a reliable charging boolean. +FEATURES = dict(_HEADPHONE_FEATURES) + +# Product-specific IDs: 1=right, 2=left, 3=case, 4=combined buds. +# ID 4 is the combined earbud reading and is intentionally not displayed. +BATTERY_COMPONENTS = {1: "Right", 2: "Left", 3: "Case"} +BATTERY_AGGREGATE_ID = 4 diff --git a/python/pybmap/protocol.py b/python/pybmap/protocol.py index 78dc577..827165f 100644 --- a/python/pybmap/protocol.py +++ b/python/pybmap/protocol.py @@ -39,7 +39,11 @@ def parse_response(data): fblock = data[0] func = data[1] op = data[2] & 0x0F + if op not in OP_NAMES: + return None length = data[3] + if len(data) < 4 + length: + return None payload = data[4:4 + length] return BmapResponse(fblock, func, op, payload) @@ -60,6 +64,8 @@ def parse_all_responses(data): fblock = data[pos] func = data[pos + 1] op = data[pos + 2] & 0x0F + if op not in OP_NAMES: + break length = data[pos + 3] if pos + 4 + length > len(data): break # Truncated packet diff --git a/python/pybmap/types.py b/python/pybmap/types.py index fd40c04..442a33a 100644 --- a/python/pybmap/types.py +++ b/python/pybmap/types.py @@ -25,6 +25,12 @@ # A single EQ band reading. EqBand = namedtuple("EqBand", ["band_id", "name", "min_val", "max_val", "current"]) +# A single component battery reading from a multi-battery device. +BatteryReading = namedtuple("BatteryReading", ["component_id", "level"]) + +# Aggregate and component readings parsed from one battery response. +BatteryStatus = namedtuple("BatteryStatus", ["aggregate", "readings"]) + # Complete device status snapshot. DeviceStatus = namedtuple("DeviceStatus", [ "battery", @@ -41,7 +47,10 @@ "auto_answer", "prompts_enabled", "prompts_language", + "battery_readings", # List of BatteryReading from the same response ]) +# Preserve positional construction of the pre-snapshot status shape. +DeviceStatus.__new__.__defaults__ = ((),) # Current audio settings (CNC, spatial, wind, ANC) from [31.10]. AudioSettings = namedtuple("AudioSettings", [ diff --git a/python/tests/test_catalog.py b/python/tests/test_catalog.py index 424d4ac..ef20fa7 100644 --- a/python/tests/test_catalog.py +++ b/python/tests/test_catalog.py @@ -28,7 +28,7 @@ def test_qc35_original(self): def test_qc_ultra2_earbuds(self): dev = lookup_device(0x4062) assert dev.codename == "edith" - assert dev.config == "qc_ultra2" + assert dev.config == "qc_ultra2_earbuds" def test_quietcomfort_headphones_prince(self): dev = lookup_device(0x4075) diff --git a/python/tests/test_channel_probe.py b/python/tests/test_channel_probe.py index 71c1b2a..4507f96 100644 --- a/python/tests/test_channel_probe.py +++ b/python/tests/test_channel_probe.py @@ -4,7 +4,7 @@ import pybmap from pybmap.constants import OP_STATUS, OP_PROCESSING, OP_ERROR -from pybmap.errors import BmapConnectionError, BmapTimeoutError, BmapDeviceError +from pybmap.errors import BmapConnectionError, BmapTimeoutError, BmapDeviceError, BmapError from pybmap.devices import qc_prince from tests.test_connection import MockTransport @@ -95,6 +95,25 @@ def test_fallback_order_skips_duplicate_of_configured(patch_transport): assert f.attempts == [2, 8, 9] +@pytest.mark.parametrize("device_type", [None, ""]) +def test_explicit_mac_requires_device_type(patch_transport, device_type): + f = patch_transport({2: "bmap"}) + with pytest.raises(BmapError, match="device_type is required"): + pybmap.connect(mac="00:11:22:33:44:55", device_type=device_type) + assert f.attempts == [] + + +def test_empty_mac_uses_discovery(patch_transport, monkeypatch): + f = patch_transport({2: "bmap"}) + monkeypatch.setattr( + pybmap, "find_bmap_device", + lambda: ("00:11:22:33:44:55", "qc_ultra2"), + ) + dev = pybmap.connect(mac="") + dev.close() + assert f.attempts == [2] + + class TestModeSwitchAck: def _dev(self, op): t = MockTransport() diff --git a/python/tests/test_cli.py b/python/tests/test_cli.py new file mode 100644 index 0000000..cf9015d --- /dev/null +++ b/python/tests/test_cli.py @@ -0,0 +1,56 @@ +"""Tests for user-visible CLI output.""" + +from pybmap.cli import cmd_status +from pybmap.types import BatteryReading, DeviceStatus + + +class StatusDevice: + device_info = {"name": "Bose QuietComfort Ultra Earbuds (2nd Gen)"} + battery_components = {1: "Right", 2: "Left", 3: "Case"} + + def status(self): + return DeviceStatus( + battery=70, + battery_readings=[ + BatteryReading(3, 80), + BatteryReading(4, 70), + BatteryReading(2, 60), + BatteryReading(1, 50), + ], + mode="quiet", + mode_idx=0, + cnc_level=0, + cnc_max=10, + eq=[], + name="edith", + firmware="1.0.0", + sidetone="off", + multipoint=False, + auto_pause=True, + auto_answer=False, + prompts_enabled=False, + prompts_language="US English", + ) + + def has_feature(self, _name): + return False + + +def test_status_orders_known_components_and_hides_combined(capsys): + cmd_status(StatusDevice()) + output = capsys.readouterr().out + + assert output.index("Right") < output.index("Left") < output.index("Case") + assert "Right 50%" in output + assert "Left 60%" in output + assert "Case 80%" in output + assert "Combined" not in output + + +def test_device_status_preserves_old_positional_shape(): + status = DeviceStatus( + 80, "quiet", 0, 7, 10, [], "Device", "1.0.0", "off", + False, True, False, True, "English", + ) + assert status.mode == "quiet" + assert status.battery_readings == () diff --git a/python/tests/test_connection.py b/python/tests/test_connection.py index c22b3f1..56503f0 100644 --- a/python/tests/test_connection.py +++ b/python/tests/test_connection.py @@ -1,11 +1,13 @@ """Tests for BmapConnection using a mock transport.""" +from types import SimpleNamespace + import pytest from pybmap.connection import BmapConnection from pybmap.protocol import bmap_packet from pybmap.constants import OP_GET, OP_SETGET, OP_STATUS, OP_RESULT, OP_ERROR from pybmap.errors import BmapError, BmapAuthError, BmapDeviceError -from pybmap.devices import qc_ultra2, qc_prince +from pybmap.devices import qc_ultra2, qc_ultra2_earbuds, qc_prince class MockTransport: @@ -58,6 +60,75 @@ class TestReadOperations: def test_battery(self, mock_dev): assert mock_dev.battery() == 80 + def test_battery_rejects_empty_response(self): + transport = MockTransport() + transport.add_response(2, 2, OP_STATUS, b"") + dev = BmapConnection(transport, qc_ultra2) + with pytest.raises(BmapDeviceError, match="Empty battery response"): + dev.battery() + + @pytest.mark.parametrize("response", [ + bytes([2, 2, OP_STATUS, 4, 80, 0xff]), + bytes([2, 2, 0x08, 0]), + ]) + def test_battery_rejects_invalid_frame(self, response): + transport = MockTransport() + transport.responses[(2, 2)] = response + dev = BmapConnection(transport, qc_ultra2) + with pytest.raises(BmapDeviceError, match="Invalid or empty response"): + dev.battery() + + def test_battery_uses_configured_parser(self): + transport = MockTransport() + transport.add_response(2, 2, OP_STATUS, bytes([80])) + device = SimpleNamespace( + DEVICE_INFO={"name": "Custom"}, + FEATURES={ + "battery": { + "addr": (2, 2), + "parser": lambda payload: payload[0] - 1, + }, + }, + ) + assert BmapConnection(transport, device).battery() == 79 + + def test_battery_readings(self): + transport = MockTransport() + transport.add_response(2, 2, OP_STATUS, + bytes.fromhex("50ffff033cffff013cffff0246ffff04")) + dev = BmapConnection(transport, qc_ultra2_earbuds) + readings = dev.battery_readings() + assert [(r.component_id, r.level) for r in readings] == [ + (3, 80), (1, 60), (2, 60), (4, 70) + ] + + def test_battery_uses_combined_earbud_record(self): + transport = MockTransport() + transport.add_response(2, 2, OP_STATUS, + bytes.fromhex("46ffff0450ffff023cffff0140ffff03")) + dev = BmapConnection(transport, qc_ultra2_earbuds) + assert dev.battery() == 70 + + def test_battery_rejects_missing_aggregate_component(self): + transport = MockTransport() + transport.add_response(2, 2, OP_STATUS, + bytes.fromhex("3cffff0150ffff0240ffff03")) + dev = BmapConnection(transport, qc_ultra2_earbuds) + with pytest.raises(BmapDeviceError, match="aggregate component 4"): + dev.battery() + + def test_status_uses_one_battery_response(self): + transport = MockTransport() + transport.add_response(2, 2, OP_STATUS, + bytes.fromhex("50ffff033cffff0146ffff043cffff02")) + dev = BmapConnection(transport, qc_ultra2_earbuds) + status = dev.status() + assert status.battery == 70 + assert [(r.component_id, r.level) for r in status.battery_readings] == [ + (3, 80), (1, 60), (4, 70), (2, 60) + ] + assert sum(packet[:2] == bytes([2, 2]) for packet in transport.sent) == 1 + def test_firmware(self, mock_dev): assert mock_dev.firmware() == "8.2.20+g34cf029" @@ -111,6 +182,7 @@ class TestStatus: def test_returns_full_status(self, mock_dev): s = mock_dev.status() assert s.battery == 80 + assert s.battery_readings == [] assert s.mode == "quiet" assert s.cnc_level == 7 assert s.cnc_max == 10 diff --git a/python/tests/test_device_parsers.py b/python/tests/test_device_parsers.py index 57700bf..8b22262 100644 --- a/python/tests/test_device_parsers.py +++ b/python/tests/test_device_parsers.py @@ -1,7 +1,12 @@ """Tests for device-specific parsers using real captured data.""" +from pathlib import Path + +import pytest + from pybmap.devices.parsers import ( parse_battery, parse_firmware, parse_product_name, parse_cnc, + parse_battery_readings, parse_eq, parse_buttons, parse_multipoint, parse_bool, parse_sidetone, parse_voice_prompts, parse_mode_config_48, parse_mode_config_47, @@ -12,6 +17,11 @@ from pybmap.types import ModeConfig, EqBand, ButtonMapping +EDITH_BATTERY_CAPTURE = Path( + __file__ +).parents[2] / "fixtures/packets/qc-ultra2-earbuds/battery-status.hex" + + # ── Test data from real QC Ultra 2 captures ────────────────────────────────── # These hex values come from fixtures/packets/qc-ultra-2/ capture files. @@ -25,6 +35,30 @@ def test_empty(self): assert parse_battery(bytes()) is None +class TestParseBatteryReadings: + def test_multi_component_capture(self): + payload = bytes.fromhex(EDITH_BATTERY_CAPTURE.read_text().strip()) + readings = parse_battery_readings(payload) + assert [(r.component_id, r.level) for r in readings] == [ + (1, 60), (2, 60), (4, 60), (3, 80) + ] + + def test_rejects_incomplete_record(self): + with pytest.raises(ValueError, match="Malformed battery response"): + parse_battery_readings(bytes.fromhex("50ffff")) + + def test_rejects_duplicate_component(self): + with pytest.raises(ValueError, match="Duplicate battery component 1"): + parse_battery_readings(bytes.fromhex("3cffff0150ffff01")) + + def test_preserves_shuffled_component_ids(self): + payload = bytes.fromhex("50ffff0332ffff0946ffff043cffff013cffff02") + readings = parse_battery_readings(payload) + assert [(r.component_id, r.level) for r in readings] == [ + (3, 80), (9, 50), (4, 70), (1, 60), (2, 60) + ] + + class TestParseFirmware: def test_from_capture(self): # "0.5": "382e322e32302b6733346366303239" diff --git a/python/tests/test_protocol.py b/python/tests/test_protocol.py index 95988ad..2c6bd96 100644 --- a/python/tests/test_protocol.py +++ b/python/tests/test_protocol.py @@ -60,6 +60,12 @@ def test_too_short(self): assert parse_response(bytes([1, 2, 3])) is None assert parse_response(bytes()) is None + def test_truncated_payload(self): + assert parse_response(bytes([2, 2, OP_STATUS, 4, 80, 0xff])) is None + + def test_invalid_operator(self): + assert parse_response(bytes([2, 2, 0x08, 0])) is None + def test_status_with_payload(self): payload = bytes([0x50, 0xff, 0xff, 0x00]) # Battery STATUS data = bytes([2, 2, 0x03, len(payload)]) + payload diff --git a/python/tests/test_qc_ultra2.py b/python/tests/test_qc_ultra2.py index a5c8de2..9c325a0 100644 --- a/python/tests/test_qc_ultra2.py +++ b/python/tests/test_qc_ultra2.py @@ -1,12 +1,15 @@ """Tests for QC Ultra 2 device configuration.""" -from pybmap.devices import qc_ultra2, qc_prince, DEVICES, get_device +from pybmap.devices import qc_ultra2, qc_ultra2_earbuds, qc_prince, DEVICES, get_device class TestDeviceRegistry: def test_qc_ultra2_registered(self): assert "qc_ultra2" in DEVICES + def test_qc_ultra2_earbuds_registered(self): + assert "qc_ultra2_earbuds" in DEVICES + def test_qc35_registered(self): assert "qc35" in DEVICES @@ -31,6 +34,10 @@ def test_has_device_info(self): assert qc_ultra2.DEVICE_INFO["name"] == "Bose QC Ultra Headphones 2" assert qc_ultra2.DEVICE_INFO["product_id"] == 0x4082 + def test_qc_ultra2_earbuds_config(self): + assert qc_ultra2_earbuds.DEVICE_INFO["name"] == "Bose QuietComfort Ultra Earbuds (2nd Gen)" + assert qc_ultra2_earbuds.BATTERY_COMPONENTS == {1: "Right", 2: "Left", 3: "Case"} + def test_has_all_features(self): expected = [ "battery", "firmware", "product_name", "voice_prompts", diff --git a/rust/src/catalog.rs b/rust/src/catalog.rs index d0fd75f..e930c82 100644 --- a/rust/src/catalog.rs +++ b/rust/src/catalog.rs @@ -54,7 +54,7 @@ pub const CATALOG: &[BoseDevice] = &[ BoseDevice { product_id: 0x404C, codename: "celine_ii", name: "Frames (2nd Gen)", category: Category::Earbuds, config: None }, BoseDevice { product_id: 0x4060, codename: "olivia", name: "Frames Tempo", category: Category::Earbuds, config: None }, BoseDevice { product_id: 0x4061, codename: "vedder", name: "Frames", category: Category::Earbuds, config: None }, - BoseDevice { product_id: 0x4062, codename: "edith", name: "QuietComfort Ultra Earbuds (2nd Gen)", category: Category::Earbuds, config: Some("qc_ultra2") }, + BoseDevice { product_id: 0x4062, codename: "edith", name: "QuietComfort Ultra Earbuds (2nd Gen)", category: Category::Earbuds, config: Some("qc_ultra2_earbuds") }, BoseDevice { product_id: 0x4064, codename: "smalls", name: "QuietComfort Earbuds II", category: Category::Earbuds, config: None }, BoseDevice { product_id: 0x4068, codename: "serena", name: "Ultra Open Earbuds", category: Category::Earbuds, config: Some("ultra_open") }, BoseDevice { product_id: 0x4072, codename: "scotty", name: "QuietComfort Ultra Earbuds", category: Category::Earbuds, config: None }, @@ -130,7 +130,7 @@ mod tests { fn test_lookup_qc_ultra2_earbuds() { let dev = lookup_device(0x4062).unwrap(); assert_eq!(dev.codename, "edith"); - assert_eq!(dev.config, Some("qc_ultra2")); + assert_eq!(dev.config, Some("qc_ultra2_earbuds")); } #[test] diff --git a/rust/src/connection.rs b/rust/src/connection.rs index 08ce3e3..91e940f 100644 --- a/rust/src/connection.rs +++ b/rust/src/connection.rs @@ -37,8 +37,10 @@ impl BmapConnection { fn get(&self, addr: Addr) -> BmapResult> { let pkt = bmap_packet(addr.0, addr.1, Operator::Get, &[]); let data = self.transport.send_recv(&pkt)?; - let resp = parse_response(&data) - .ok_or_else(|| BmapError::Timeout("Empty response".into()))?; + let resp = parse_response(&data).ok_or_else(|| BmapError::Device { + message: "Invalid or empty response".into(), + code: 0, + })?; self.check_error(&resp)?; Ok(resp.payload) } @@ -86,11 +88,49 @@ impl BmapConnection { /// Battery percentage. pub fn battery(&self) -> BmapResult { + Ok(self.battery_status()?.aggregate) + } + + /// Aggregate and component levels from one battery response. + pub fn battery_status(&self) -> BmapResult { let addr = self.addr(self.config.battery)?; let payload = self.get(addr)?; - parse_battery(&payload).ok_or_else(|| BmapError::Device { - message: "Empty battery response".into(), code: 0, - }) + if let Some(component_id) = self.config.battery_aggregate_id { + let readings = + parse_battery_readings(&payload).map_err(|message| BmapError::Device { + message: message.into(), + code: 0, + })?; + let aggregate = readings + .iter() + .find(|reading| reading.component_id == component_id) + .map(|reading| reading.level) + .ok_or_else(|| BmapError::Device { + message: format!( + "Battery response missing aggregate component {}", + component_id + ), + code: 0, + })?; + Ok(BatteryStatus { + aggregate, + readings, + }) + } else { + let aggregate = parse_battery(&payload).ok_or_else(|| BmapError::Device { + message: "Empty battery response".into(), + code: 0, + })?; + Ok(BatteryStatus { + aggregate, + readings: Vec::new(), + }) + } + } + + /// Component battery readings, when the device reports multiple cells. + pub fn battery_readings(&self) -> BmapResult> { + Ok(self.battery_status()?.readings) } /// Firmware version string. @@ -219,9 +259,11 @@ impl BmapConnection { }; let (cnc_level, cnc_max) = self.cnc().unwrap_or((0, 10)); let (prompts_enabled, prompts_language) = self.prompts().unwrap_or((false, "Unknown")); + let battery = self.battery_status()?; Ok(DeviceStatus { - battery: self.battery()?, + battery: battery.aggregate, + battery_readings: battery.readings, mode: current_name, mode_idx: current_idx, cnc_level, @@ -674,6 +716,76 @@ mod tests { assert_eq!(mock_qc_ultra2().battery().unwrap(), 80); } + #[test] + fn test_battery_rejects_empty_response() { + let mut t = MockTransport::new(); + t.add(2, 2, 0x03, &[]); + let dev = BmapConnection::new(t, devices::qc_ultra2()); + assert!(matches!(dev.battery(), Err(BmapError::Device { message, .. }) + if message.contains("Empty battery response"))); + } + + #[test] + fn test_battery_rejects_invalid_frame() { + let mut t = MockTransport::new(); + t.responses.insert((2, 2), vec![2, 2, 0x08, 0]); + let dev = BmapConnection::new(t, devices::qc_ultra2()); + assert!(matches!(dev.battery(), Err(BmapError::Device { message, .. }) + if message.contains("Invalid or empty response"))); + } + + #[test] + fn test_battery_readings() { + let mut t = MockTransport::new(); + t.add(2, 2, 0x03, &[ + 0x50,0xff,0xff,0x03, 0x3c,0xff,0xff,0x01, + 0x3c,0xff,0xff,0x02, 0x46,0xff,0xff,0x04, + ]); + let dev = BmapConnection::new(t, devices::qc_ultra2_earbuds()); + let battery = dev.battery_status().unwrap(); + assert_eq!(battery.readings, vec![ + BatteryReading { component_id: 3, level: 80 }, + BatteryReading { component_id: 1, level: 60 }, + BatteryReading { component_id: 2, level: 60 }, + BatteryReading { component_id: 4, level: 70 }, + ]); + assert_eq!(battery.aggregate, 70); + assert_eq!(dev.transport.sent.borrow().len(), 1); + } + + #[test] + fn test_battery_rejects_missing_aggregate_component() { + let mut t = MockTransport::new(); + t.add(2, 2, 0x03, &[ + 0x3c,0xff,0xff,0x01, 0x3c,0xff,0xff,0x02, + 0x50,0xff,0xff,0x03, + ]); + let dev = BmapConnection::new(t, devices::qc_ultra2_earbuds()); + assert!(matches!(dev.battery(), Err(BmapError::Device { message, .. }) + if message.contains("aggregate component 4"))); + } + + #[test] + fn test_status_uses_one_battery_response() { + let mut t = MockTransport::new(); + t.add(2, 2, 0x03, &[ + 0x50,0xff,0xff,0x03, 0x3c,0xff,0xff,0x01, + 0x46,0xff,0xff,0x04, 0x3c,0xff,0xff,0x02, + ]); + let dev = BmapConnection::new(t, devices::qc_ultra2_earbuds()); + let status = dev.status().unwrap(); + assert_eq!(status.battery, 70); + assert_eq!(status.battery_readings.len(), 4); + let battery_requests = dev + .transport + .sent + .borrow() + .iter() + .filter(|packet| packet.starts_with(&[2, 2])) + .count(); + assert_eq!(battery_requests, 1); + } + #[test] fn test_firmware() { assert_eq!(mock_qc_ultra2().firmware().unwrap(), "8.2.20+g34cf029"); @@ -738,6 +850,7 @@ mod tests { fn test_status() { let s = mock_qc_ultra2().status().unwrap(); assert_eq!(s.battery, 80); + assert!(s.battery_readings.is_empty()); assert_eq!(s.mode, "quiet"); assert_eq!(s.cnc_level, 7); assert_eq!(s.cnc_max, 10); diff --git a/rust/src/device.rs b/rust/src/device.rs index 451fd85..5230bed 100644 --- a/rust/src/device.rs +++ b/rust/src/device.rs @@ -60,6 +60,20 @@ pub struct ButtonMapping { pub action_name: &'static str, } +/// A component battery reading from a device with multiple cells. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BatteryReading { + pub component_id: u8, + pub level: u8, +} + +/// Aggregate and component levels parsed from one battery response. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BatteryStatus { + pub aggregate: u8, + pub readings: Vec, +} + /// Audio source information. #[derive(Debug, Clone)] pub struct AudioSource { @@ -71,6 +85,7 @@ pub struct AudioSource { #[derive(Debug, Clone)] pub struct DeviceStatus { pub battery: u8, + pub battery_readings: Vec, pub mode: String, pub mode_idx: u8, pub cnc_level: u8, @@ -94,6 +109,10 @@ pub struct DeviceConfig { /// Init packet required before device responds (Some((fblock, func)) for QC35). pub init_packet: Option, pub battery: Option, + /// Component IDs and display labels for multi-cell battery responses. + pub battery_components: &'static [(u8, &'static str)], + /// Component ID used for the generic aggregate battery value. + pub battery_aggregate_id: Option, pub firmware: Option, pub product_name: Option, pub voice_prompts: Option, @@ -135,6 +154,29 @@ pub fn parse_battery(payload: &[u8]) -> Option { payload.first().copied() } +/// Parse four-byte component battery records from [2.2]. +pub fn parse_battery_readings(payload: &[u8]) -> Result, &'static str> { + if payload.len() % 4 != 0 { + return Err("Malformed battery response"); + } + let mut readings = Vec::new(); + let mut seen_components = Vec::new(); + for record in payload.chunks_exact(4) { + let component_id = record[3]; + if seen_components.contains(&component_id) { + return Err("Duplicate battery component"); + } + seen_components.push(component_id); + if record[0] <= 100 { + readings.push(BatteryReading { + component_id, + level: record[0], + }); + } + } + Ok(readings) +} + pub fn parse_firmware(payload: &[u8]) -> String { String::from_utf8_lossy(payload).into_owned() } @@ -587,12 +629,70 @@ pub fn build_mode_config_39( mod tests { use super::*; + fn decode_hex(text: &str) -> Vec { + let text = text.trim(); + (0..text.len()) + .step_by(2) + .map(|i| u8::from_str_radix(&text[i..i + 2], 16).unwrap()) + .collect() + } + #[test] fn test_parse_battery() { assert_eq!(parse_battery(&[0x50, 0xff, 0xff, 0x00]), Some(0x50)); assert_eq!(parse_battery(&[]), None); } + #[test] + fn test_parse_battery_readings() { + let payload = decode_hex(include_str!( + "../../fixtures/packets/qc-ultra2-earbuds/battery-status.hex" + )); + assert_eq!( + parse_battery_readings(&payload).unwrap(), + vec![ + BatteryReading { + component_id: 1, + level: 60 + }, + BatteryReading { + component_id: 2, + level: 60 + }, + BatteryReading { + component_id: 4, + level: 60 + }, + BatteryReading { + component_id: 3, + level: 80 + }, + ] + ); + assert_eq!( + parse_battery_readings(&[ + 0x50,0xff,0xff,0x03, 0x46,0xff,0xff,0x04, + 0x32,0xff,0xff,0x09, 0x3c,0xff,0xff,0x01, + 0x3c,0xff,0xff,0x02, + ]) + .unwrap() + .iter() + .map(|reading| reading.component_id) + .collect::>(), + vec![3, 4, 9, 1, 2], + ); + assert_eq!( + parse_battery_readings(&[0x50, 0xff]), + Err("Malformed battery response"), + ); + assert_eq!( + parse_battery_readings(&[ + 0x3c, 0xff, 0xff, 0x01, 0x50, 0xff, 0xff, 0x01, + ]), + Err("Duplicate battery component"), + ); + } + #[test] fn test_parse_firmware() { let payload = b"8.2.20+g34cf029"; diff --git a/rust/src/devices.rs b/rust/src/devices.rs index e36ac21..5265cb1 100644 --- a/rust/src/devices.rs +++ b/rust/src/devices.rs @@ -17,6 +17,8 @@ pub fn qc_ultra2() -> DeviceConfig { rfcomm_channel: 2, init_packet: None, battery: Some(Addr(2, 2)), + battery_components: &[], + battery_aggregate_id: None, firmware: Some(Addr(0, 5)), product_name: Some(Addr(1, 2)), voice_prompts: Some(Addr(1, 3)), @@ -51,6 +53,20 @@ pub fn qc_ultra2() -> DeviceConfig { } } +/// Bose QuietComfort Ultra Earbuds 2 configuration -- edith, product ID 0x4062. +/// Shares the QC Ultra 2 protocol layout and reports separate battery cells. +pub fn qc_ultra2_earbuds() -> DeviceConfig { + let mut c = qc_ultra2(); + c.info = DeviceInfo { + name: "Bose QuietComfort Ultra Earbuds (2nd Gen)", + codename: "edith", + platform: "OTG-QCC-384", + }; + c.battery_components = &[(1, "Right"), (2, "Left"), (3, "Case")]; + c.battery_aggregate_id = Some(4); + c +} + /// Bose QuietComfort Headphones configuration -- prince, product ID 0x4075. /// Verified against firmware 1.0.6-80+f5f219b. /// BMAP over RFCOMM channel 8 with 47-byte ModeConfig STATUS responses. @@ -64,6 +80,8 @@ pub fn qc_prince() -> DeviceConfig { rfcomm_channel: 8, init_packet: None, battery: Some(Addr(2, 2)), + battery_components: &[], + battery_aggregate_id: None, firmware: Some(Addr(0, 5)), product_name: Some(Addr(1, 2)), voice_prompts: Some(Addr(1, 3)), @@ -110,6 +128,8 @@ pub fn qc35() -> DeviceConfig { rfcomm_channel: 8, init_packet: Some(Addr(0, 1)), // GET [0.1] required before QC35 responds battery: Some(Addr(2, 2)), + battery_components: &[], + battery_aggregate_id: None, firmware: Some(Addr(0, 5)), product_name: Some(Addr(1, 2)), voice_prompts: Some(Addr(1, 3)), @@ -157,6 +177,8 @@ pub fn qc_earbuds() -> DeviceConfig { rfcomm_channel: 8, init_packet: None, battery: Some(Addr(2, 2)), + battery_components: &[], + battery_aggregate_id: None, firmware: Some(Addr(0, 5)), product_name: Some(Addr(1, 2)), voice_prompts: Some(Addr(1, 3)), @@ -202,6 +224,8 @@ pub fn qc45() -> DeviceConfig { rfcomm_channel: 8, init_packet: Some(Addr(0, 1)), battery: Some(Addr(2, 2)), + battery_components: &[], + battery_aggregate_id: None, firmware: Some(Addr(0, 5)), product_name: Some(Addr(1, 2)), voice_prompts: Some(Addr(1, 3)), @@ -248,6 +272,8 @@ pub fn ultra_open() -> DeviceConfig { rfcomm_channel: 2, init_packet: None, battery: Some(Addr(2, 2)), + battery_components: &[], + battery_aggregate_id: None, firmware: Some(Addr(0, 5)), product_name: Some(Addr(1, 2)), voice_prompts: Some(Addr(1, 3)), @@ -280,6 +306,7 @@ pub fn ultra_open() -> DeviceConfig { pub fn get_device(name: &str) -> Option { match name { "qc_ultra2" => Some(qc_ultra2()), + "qc_ultra2_earbuds" => Some(qc_ultra2_earbuds()), "qc35" => Some(qc35()), "qc_prince" => Some(qc_prince()), "qc_earbuds" => Some(qc_earbuds()), @@ -326,6 +353,7 @@ mod tests { #[test] fn test_get_device() { assert!(get_device("qc_ultra2").is_some()); + assert!(get_device("qc_ultra2_earbuds").is_some()); assert!(get_device("qc35").is_some()); assert!(get_device("qc_prince").is_some()); assert!(get_device("qc_earbuds").is_some()); @@ -334,6 +362,16 @@ mod tests { assert!(get_device("nonexistent").is_none()); } + #[test] + fn test_qc_ultra2_earbuds_identity_and_battery_components() { + let dev = qc_ultra2_earbuds(); + assert_eq!(dev.info.codename, "edith"); + assert_eq!(dev.battery_components, &[(1, "Right"), (2, "Left"), (3, "Case")]); + assert_eq!(dev.battery_aggregate_id, Some(4)); + assert_eq!(dev.preset_modes.len(), 4); + assert!(dev.audio_settings.is_some()); + } + #[test] fn test_qc_earbuds_direct_cnc_no_profile_editing() { let dev = qc_earbuds(); diff --git a/rust/src/lib.rs b/rust/src/lib.rs index 42dc162..143e4c4 100644 --- a/rust/src/lib.rs +++ b/rust/src/lib.rs @@ -22,17 +22,27 @@ pub mod catalog; pub use connection::BmapConnection; pub use transport::Transport; -pub use device::{DeviceConfig, DeviceStatus, ModeConfig, EqBand, ButtonMapping}; +pub use device::{ + BatteryReading, BatteryStatus, ButtonMapping, DeviceConfig, DeviceStatus, + EqBand, ModeConfig, +}; pub use error::{BmapError, BmapResult}; pub use protocol::{Operator, BmapResponse}; /// Connect to a BMAP device over Bluetooth RFCOMM. /// /// - `mac`: Bluetooth MAC address. Auto-detected if None. -/// - `device_type`: Device type string. Auto-detected if None. +/// - `device_type`: Device type string. Auto-detected only when `mac` is None. pub fn connect(mac: Option<&str>, device_type: Option<&str>) -> BmapResult> { + let mac = mac.filter(|value| !value.is_empty()); + let device_type = device_type.filter(|value| !value.is_empty()); let (mac, resolved_type) = match mac { - Some(m) => (m.to_string(), device_type.unwrap_or("qc_ultra2").to_string()), + Some(m) => { + let dtype = device_type.ok_or_else(|| BmapError::InvalidArg( + "device_type is required when mac is specified".into() + ))?; + (m.to_string(), dtype.to_string()) + } None => { let (detected_mac, detected_type) = discovery::find_bmap_device() .ok_or_else(|| BmapError::NotFound( @@ -50,6 +60,20 @@ pub fn connect(mac: Option<&str>, device_type: Option<&str>) -> BmapResult) -> Result<(), Bm println!(" Model {}", dev.config().info.name); println!(" Battery {}%", s.battery); + for (component_id, label) in dev.config().battery_components { + if let Some(reading) = s + .battery_readings + .iter() + .find(|reading| reading.component_id == *component_id) + { + println!(" {:12} {}%", label, reading.level); + } + } if !s.mode.is_empty() { println!(" Mode {}", s.mode); } @@ -326,5 +335,5 @@ fn usage() { println!(); println!("Environment:"); println!(" BMAP_MAC=XX:XX:XX:XX:XX:XX Device MAC (auto-detected if unset)"); - println!(" BMAP_DEVICE=qc_ultra2|qc_prince|qc35 Device type"); + println!(" BMAP_DEVICE=qc_ultra2|qc_ultra2_earbuds|qc_prince|qc35|qc_earbuds|qc45|ultra_open"); } diff --git a/rust/src/protocol.rs b/rust/src/protocol.rs index 3cefe4e..3433a80 100644 --- a/rust/src/protocol.rs +++ b/rust/src/protocol.rs @@ -113,8 +113,10 @@ pub fn parse_response(data: &[u8]) -> Option { let func = data[1]; let op = Operator::from_u8(data[2])?; let length = data[3] as usize; - let end = std::cmp::min(4 + length, data.len()); - let payload = data[4..end].to_vec(); + if data.len() < 4 + length { + return None; + } + let payload = data[4..4 + length].to_vec(); Some(BmapResponse { fblock, func, op, payload }) } @@ -178,6 +180,8 @@ mod tests { #[test] fn test_parse_response_too_short() { assert!(parse_response(&[1, 2]).is_none()); + assert!(parse_response(&[2, 2, 0x03, 4, 80, 0xff]).is_none()); + assert!(parse_response(&[2, 2, 0x08, 0]).is_none()); } #[test] From 7da70565fd7f11c3c8a50a25dbe35353caf0fee1 Mon Sep 17 00:00:00 2001 From: Aaron Bockelie Date: Mon, 28 Sep 2026 11:58:44 -0500 Subject: [PATCH 2/4] Fall back to bud battery levels when the aggregate is missing - battery_status: when the combined-buds record (0x04) is absent or 0xFF, use the lowest valid reading from the configured aggregate sources (edith: right/left); raise only when none are valid - status(): battery is now a soft-failure field like the other optional reads (battery 0, no readings) instead of failing the whole snapshot - Rust duplicate-component error now names the component id, matching Python and C++ - README: revert out-of-scope badge, QC45 catalog and release example edits - tests in Python, Rust and C++ using the edith fixture with 0x04 set to 0xFF and with 0x04 removed --- NOTES.md | 3 +- README.md | 8 +- cpp/src/connection.h | 13 ++- cpp/src/device.h | 2 + cpp/src/devices.h | 1 + cpp/tests/test_common.h | 24 ++++++ cpp/tests/test_connection.cpp | 73 ++++++++++++++++- cpp/tests/test_parsers.cpp | 20 ----- docs/architecture.md | 9 +- python/pybmap/connection.py | 15 +++- python/pybmap/devices/qc_ultra2_earbuds.py | 2 + python/tests/test_connection.py | 53 +++++++++++- rust/src/connection.rs | 95 +++++++++++++++++++++- rust/src/device.rs | 12 +-- rust/src/devices.rs | 8 ++ 15 files changed, 295 insertions(+), 43 deletions(-) diff --git a/NOTES.md b/NOTES.md index ffc76ff..9e1ebec 100644 --- a/NOTES.md +++ b/NOTES.md @@ -45,7 +45,8 @@ EDITH returns multiple four-byte battery records from `[2.2]`. Each record is | `0x04` | Combined buds | `3cffff04` = 60%, used for generic Battery | Records must be identified by component ID, not payload position. The generic -`Battery` value uses ID `0x04`; the CLI separately displays IDs `0x01`, `0x02`, +`Battery` value uses ID `0x04` (falling back to the lower of `0x01`/`0x02` +when `0x04` is absent or `0xFF`); the CLI separately displays IDs `0x01`, `0x02`, and `0x03`. BMAP `[2.5]` tracks whether both earbuds are seated and did not change across observed charger transitions, so case charging is not inferred. diff --git a/README.md b/README.md index aa2b34c..c4c0d9f 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ [![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE) [![Release](https://img.shields.io/github/v/release/aaronsb/bosectl)](https://github.com/aaronsb/bosectl/releases/latest) -[![Devices](https://img.shields.io/badge/Devices-8_supported_·_38_known-green)](docs/architecture.md#device-catalog) +[![Devices](https://img.shields.io/badge/Devices-3_supported_·_38_known-green)](docs/architecture.md#device-catalog) [![Python 3](https://img.shields.io/badge/Python-3-3572A5.svg)](python/) [![Rust](https://img.shields.io/badge/Rust-1.70+-DEA584.svg)](rust/) [![C++17](https://img.shields.io/badge/C++-17-f34b7d.svg)](cpp/) @@ -119,8 +119,8 @@ pybmap.modalias(0x4082) # "bluetooth:v05A7p4082d0000" # Check support status pybmap.is_supported(0x4082) # True — has tested config pybmap.is_supported(0x4075) # True — QuietComfort Headphones (prince) -pybmap.is_supported(0x4039) # True — QC45 uses an inferred config -pybmap.supported_devices() # eight devices; see docs/architecture.md +pybmap.is_supported(0x4039) # False — QC45, recognized but untested +pybmap.supported_devices() # [wolfcastle, baywolf, edith, prince, wolverine] pybmap.known_devices() # full catalog ``` @@ -253,7 +253,7 @@ Full protocol reference: **[NOTES.md](NOTES.md)** and ```bash make test # All tests (121 Python, 63 Rust, 54 C++) make artifacts # Build + strip + SHA256SUMS in dist/ -make release VERSION=v0.4.0 # Test → build → gh release create +make release VERSION=v0.2.0 # Test → build → gh release create make clean # Remove all build artifacts ``` diff --git a/cpp/src/connection.h b/cpp/src/connection.h index 904907a..08c5ab8 100644 --- a/cpp/src/connection.h +++ b/cpp/src/connection.h @@ -37,6 +37,16 @@ class BmapConnection { if (reading.component_id == *config_.battery_aggregate_id) return {reading.level, std::move(readings)}; } + // Aggregate missing or 0xFF: fall back to the lowest bud reading. + std::optional lowest; + for (const auto& reading : readings) { + const auto& sources = config_.battery_aggregate_sources; + if (std::find(sources.begin(), sources.end(), reading.component_id) == + sources.end()) + continue; + if (!lowest || reading.level < *lowest) lowest = reading.level; + } + if (lowest) return {*lowest, std::move(readings)}; throw std::runtime_error( "Battery response missing aggregate component " + std::to_string(*config_.battery_aggregate_id)); @@ -112,7 +122,8 @@ class BmapConnection { [&]{ return cnc(); }, {0, 10}); auto [prom_on, prom_lang] = safe_call>( [&]{ return prompts(); }, {false, ""}); - auto battery_state = battery_status(); + auto battery_state = safe_call( + [&]{ return battery_status(); }, {0, {}}); DeviceStatus s; s.battery = battery_state.aggregate; diff --git a/cpp/src/device.h b/cpp/src/device.h index e3d71e8..e9cb0ee 100644 --- a/cpp/src/device.h +++ b/cpp/src/device.h @@ -108,6 +108,8 @@ struct DeviceConfig { std::vector> battery_components; /// Component ID used for the generic aggregate battery value. std::optional battery_aggregate_id; + /// Component IDs whose lowest level stands in for a missing aggregate. + std::vector battery_aggregate_sources; std::optional firmware; std::optional product_name; std::optional voice_prompts; diff --git a/cpp/src/devices.h b/cpp/src/devices.h index 6dc3305..5dd71ad 100644 --- a/cpp/src/devices.h +++ b/cpp/src/devices.h @@ -48,6 +48,7 @@ inline DeviceConfig qc_ultra2_earbuds() { c.info = {"Bose QuietComfort Ultra Earbuds (2nd Gen)", "edith", "OTG-QCC-384"}; c.battery_components = {{1, "Right"}, {2, "Left"}, {3, "Case"}}; c.battery_aggregate_id = 4; + c.battery_aggregate_sources = {1, 2}; return c; } diff --git a/cpp/tests/test_common.h b/cpp/tests/test_common.h index e881b9a..55bd048 100644 --- a/cpp/tests/test_common.h +++ b/cpp/tests/test_common.h @@ -1,9 +1,16 @@ // Minimal test framework — no external deps. #pragma once +#include +#include +#include #include #include +#include +#include #include +#include +#include #include #include @@ -45,3 +52,20 @@ struct TestRegistrar { } while(0) #define ASSERT_FALSE(x) ASSERT_TRUE(!(x)) + +/// Read a hex fixture relative to the tests directory. +inline std::vector decode_hex_fixture(const std::string& relative_path) { + auto path = std::filesystem::path(__FILE__).parent_path() / relative_path; + std::ifstream input(path); + if (!input) throw std::runtime_error("Could not open fixture: " + path.string()); + std::string hex((std::istreambuf_iterator(input)), {}); + hex.erase(std::remove_if(hex.begin(), hex.end(), [](unsigned char c) { + return std::isspace(c); + }), hex.end()); + if (hex.size() % 2 != 0) throw std::runtime_error("Fixture contains incomplete hex byte"); + std::vector bytes; + for (size_t i = 0; i < hex.size(); i += 2) { + bytes.push_back(static_cast(std::stoul(hex.substr(i, 2), nullptr, 16))); + } + return bytes; +} diff --git a/cpp/tests/test_connection.cpp b/cpp/tests/test_connection.cpp index fb5700b..643a9b0 100644 --- a/cpp/tests/test_connection.cpp +++ b/cpp/tests/test_connection.cpp @@ -95,11 +95,22 @@ TEST(battery_readings) { ASSERT_EQ(raw->sent.size(), 1u); } -TEST(battery_missing_aggregate_component) { +TEST(battery_falls_back_to_lowest_bud_without_aggregate) { auto raw = new MockTransport(); raw->add(2, 2, 0x03, { - 0x3c,0xff,0xff,0x01, 0x3c,0xff,0xff,0x02, - 0x50,0xff,0xff,0x03, + 0x3c,0xff,0xff,0x01, 0x50,0xff,0xff,0x02, + 0x28,0xff,0xff,0x03, + }); + BmapConnection dev(std::unique_ptr(raw), qc_ultra2_earbuds()); + // Case (3) is lower but is not a bud; right bud (1) is the lowest. + ASSERT_EQ(dev.battery(), 60); +} + +TEST(battery_rejects_response_without_valid_buds) { + auto raw = new MockTransport(); + raw->add(2, 2, 0x03, { + 0xff,0xff,0xff,0x01, 0xff,0xff,0xff,0x02, + 0xff,0xff,0xff,0x04, 0x28,0xff,0xff,0x03, }); BmapConnection dev(std::unique_ptr(raw), qc_ultra2_earbuds()); bool threw = false; @@ -110,6 +121,61 @@ TEST(battery_missing_aggregate_component) { ASSERT_TRUE(threw); } +static std::vector earbuds_battery_fixture() { + return decode_hex_fixture("../../fixtures/packets/qc-ultra2-earbuds/battery-status.hex"); +} + +static void assert_fixture_bud_readings(const std::vector& readings) { + ASSERT_EQ(readings.size(), 3u); + ASSERT_EQ(readings[0].component_id, 1); + ASSERT_EQ(readings[0].level, 60); + ASSERT_EQ(readings[1].component_id, 2); + ASSERT_EQ(readings[1].level, 60); + ASSERT_EQ(readings[2].component_id, 3); + ASSERT_EQ(readings[2].level, 80); +} + +TEST(status_falls_back_when_fixture_aggregate_is_invalid) { + auto payload = earbuds_battery_fixture(); + for (size_t i = 0; i + 3 < payload.size(); i += 4) { + if (payload[i + 3] == 4) payload[i] = 0xff; + } + auto raw = new MockTransport(); + raw->add(2, 2, 0x03, payload); + BmapConnection dev(std::unique_ptr(raw), qc_ultra2_earbuds()); + auto status = dev.status(); + ASSERT_EQ(status.battery, 60); + assert_fixture_bud_readings(status.battery_readings); +} + +TEST(status_falls_back_when_fixture_aggregate_is_absent) { + auto fixture = earbuds_battery_fixture(); + std::vector payload; + for (size_t i = 0; i + 3 < fixture.size(); i += 4) { + if (fixture[i + 3] != 4) payload.insert(payload.end(), &fixture[i], &fixture[i] + 4); + } + auto raw = new MockTransport(); + raw->add(2, 2, 0x03, payload); + BmapConnection dev(std::unique_ptr(raw), qc_ultra2_earbuds()); + auto status = dev.status(); + ASSERT_EQ(status.battery, 60); + assert_fixture_bud_readings(status.battery_readings); +} + +TEST(status_tolerates_battery_without_valid_readings) { + auto raw = new MockTransport(); + raw->add(2, 2, 0x03, { + 0xff,0xff,0xff,0x01, 0xff,0xff,0xff,0x02, + 0xff,0xff,0xff,0x04, 0xff,0xff,0xff,0x03, + }); + raw->add(31, 3, 0x03, {0x01}); + BmapConnection dev(std::unique_ptr(raw), qc_ultra2_earbuds()); + auto status = dev.status(); + ASSERT_EQ(status.battery, 0); + ASSERT_TRUE(status.battery_readings.empty()); + ASSERT_EQ(status.mode, "aware"); +} + TEST(status_uses_one_battery_response) { auto raw = new MockTransport(); raw->add(2, 2, 0x03, { @@ -136,6 +202,7 @@ TEST(qc_ultra2_earbuds_config) { ASSERT_EQ(config.battery_components[1].second, "Left"); ASSERT_EQ(config.battery_components[2].second, "Case"); ASSERT_EQ(*config.battery_aggregate_id, 4); + ASSERT_EQ(config.battery_aggregate_sources, (std::vector{1, 2})); ASSERT_EQ(config.preset_modes.size(), 4u); } TEST(firmware) { ASSERT_EQ(mock_qc_ultra2()->firmware(), "8.2.20+g34cf029"); } diff --git a/cpp/tests/test_parsers.cpp b/cpp/tests/test_parsers.cpp index dd1e3c2..373df8c 100644 --- a/cpp/tests/test_parsers.cpp +++ b/cpp/tests/test_parsers.cpp @@ -1,29 +1,9 @@ // Tests for shared device parsers using real captured data. #include "test_common.h" -#include -#include -#include -#include #include "../src/device.h" using namespace bmap; -static std::vector decode_hex_fixture(const std::string& relative_path) { - auto path = std::filesystem::path(__FILE__).parent_path() / relative_path; - std::ifstream input(path); - if (!input) throw std::runtime_error("Could not open fixture: " + path.string()); - std::string hex((std::istreambuf_iterator(input)), {}); - hex.erase(std::remove_if(hex.begin(), hex.end(), [](unsigned char c) { - return std::isspace(c); - }), hex.end()); - if (hex.size() % 2 != 0) throw std::runtime_error("Fixture contains incomplete hex byte"); - std::vector bytes; - for (size_t i = 0; i < hex.size(); i += 2) { - bytes.push_back(static_cast(std::stoul(hex.substr(i, 2), nullptr, 16))); - } - return bytes; -} - TEST(parse_battery_from_capture) { ASSERT_EQ(parse_battery({0x50, 0xff, 0xff, 0x00}), 0x50); } diff --git a/docs/architecture.md b/docs/architecture.md index b6f3c92..b97af7f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -147,8 +147,10 @@ Each feature entry maps a name to its protocol address and codec functions: Multi-component battery devices declare display labels and an aggregate component ID in config. For `edith`, IDs 1/2/3 are Right/Left/Case and ID 4 -is the combined earbud level. Unknown IDs are preserved in the snapshot but -only configured components are rendered. +is the combined earbud level. If ID 4 is absent or reports 0xFF, the lowest +valid reading from the configured aggregate sources (the buds, IDs 1/2) is +used instead. Unknown IDs are preserved in the snapshot but only configured +components are rendered. | Property | QC Ultra 2 | QuietComfort Headphones (`prince`) | QC35 | |----------|-----------|-------------------------------------|------| @@ -198,6 +200,9 @@ battery() / status() → reuse the parsed battery status `BatteryStatus` keeps the aggregate and all component readings from that one response. `DeviceStatus.battery_readings` carries the same snapshot to CLIs, which render known components in config order rather than packet order. +`status()` treats battery like its other optional fields: a failed or +unusable battery read yields `battery = 0` with no readings instead of +failing the whole snapshot. **Write pattern** (SETGET): ``` diff --git a/python/pybmap/connection.py b/python/pybmap/connection.py index f309381..43d5545 100644 --- a/python/pybmap/connection.py +++ b/python/pybmap/connection.py @@ -157,6 +157,14 @@ def battery_status(self): if reading.component_id == aggregate_id), None, ) + if aggregate is None: + # Aggregate missing or 0xFF: fall back to the lowest bud reading. + sources = self.battery_aggregate_sources + aggregate = min( + (reading.level for reading in readings + if reading.component_id in sources), + default=None, + ) if aggregate is None: raise BmapDeviceError( "Battery response missing aggregate component %d" % aggregate_id) @@ -299,7 +307,7 @@ def status(self): current_name = self._mode_name_from_idx(current_idx) if current_idx is not None else "" cnc_cur, cnc_max = self._safe_read(self.cnc, (0, 10)) prompts_on, prompts_lang = self._safe_read(self.prompts, (False, "")) - battery = self.battery_status() + battery = self._safe_read(self.battery_status, BatteryStatus(0, [])) return DeviceStatus( battery=battery.aggregate, @@ -602,6 +610,11 @@ def battery_aggregate_id(self): """Component ID used for the generic battery level, when applicable.""" return getattr(self._device, "BATTERY_AGGREGATE_ID", None) + @property + def battery_aggregate_sources(self): + """Component IDs whose lowest level stands in for a missing aggregate.""" + return getattr(self._device, "BATTERY_AGGREGATE_SOURCES", ()) + @property def preset_modes(self): """Preset mode definitions dict (name -> {idx, description}).""" diff --git a/python/pybmap/devices/qc_ultra2_earbuds.py b/python/pybmap/devices/qc_ultra2_earbuds.py index 6285ba7..00c32c0 100644 --- a/python/pybmap/devices/qc_ultra2_earbuds.py +++ b/python/pybmap/devices/qc_ultra2_earbuds.py @@ -32,3 +32,5 @@ # ID 4 is the combined earbud reading and is intentionally not displayed. BATTERY_COMPONENTS = {1: "Right", 2: "Left", 3: "Case"} BATTERY_AGGREGATE_ID = 4 +# When ID 4 is absent or 0xFF, report the lowest bud level instead. +BATTERY_AGGREGATE_SOURCES = (1, 2) diff --git a/python/tests/test_connection.py b/python/tests/test_connection.py index 56503f0..cb327da 100644 --- a/python/tests/test_connection.py +++ b/python/tests/test_connection.py @@ -1,5 +1,6 @@ """Tests for BmapConnection using a mock transport.""" +from pathlib import Path from types import SimpleNamespace import pytest @@ -10,6 +11,12 @@ from pybmap.devices import qc_ultra2, qc_ultra2_earbuds, qc_prince +EARBUDS_BATTERY_FIXTURE = bytes.fromhex( + (Path(__file__).parents[2] + / "fixtures/packets/qc-ultra2-earbuds/battery-status.hex").read_text().strip() +) + + class MockTransport: """Fake RFCOMM transport that returns canned responses.""" @@ -109,14 +116,56 @@ def test_battery_uses_combined_earbud_record(self): dev = BmapConnection(transport, qc_ultra2_earbuds) assert dev.battery() == 70 - def test_battery_rejects_missing_aggregate_component(self): + def test_battery_falls_back_to_lowest_bud_without_aggregate(self): transport = MockTransport() transport.add_response(2, 2, OP_STATUS, - bytes.fromhex("3cffff0150ffff0240ffff03")) + bytes.fromhex("3cffff0150ffff0228ffff03")) + dev = BmapConnection(transport, qc_ultra2_earbuds) + # Case (3) is lower but is not a bud; right bud (1) is the lowest. + assert dev.battery() == 60 + + def test_battery_rejects_response_without_valid_buds(self): + transport = MockTransport() + transport.add_response(2, 2, OP_STATUS, + bytes.fromhex("ffffff01ffffff02ffffff0428ffff03")) dev = BmapConnection(transport, qc_ultra2_earbuds) with pytest.raises(BmapDeviceError, match="aggregate component 4"): dev.battery() + def test_status_falls_back_when_fixture_aggregate_is_invalid(self): + records = [EARBUDS_BATTERY_FIXTURE[i:i + 4] + for i in range(0, len(EARBUDS_BATTERY_FIXTURE), 4)] + invalid = b"".join(b"\xff" + r[1:] if r[3] == 4 else r for r in records) + transport = MockTransport() + transport.add_response(2, 2, OP_STATUS, invalid) + status = BmapConnection(transport, qc_ultra2_earbuds).status() + assert status.battery == 60 + assert [(r.component_id, r.level) for r in status.battery_readings] == [ + (1, 60), (2, 60), (3, 80) + ] + + def test_status_falls_back_when_fixture_aggregate_is_absent(self): + records = [EARBUDS_BATTERY_FIXTURE[i:i + 4] + for i in range(0, len(EARBUDS_BATTERY_FIXTURE), 4)] + absent = b"".join(r for r in records if r[3] != 4) + transport = MockTransport() + transport.add_response(2, 2, OP_STATUS, absent) + status = BmapConnection(transport, qc_ultra2_earbuds).status() + assert status.battery == 60 + assert [(r.component_id, r.level) for r in status.battery_readings] == [ + (1, 60), (2, 60), (3, 80) + ] + + def test_status_tolerates_battery_without_valid_readings(self): + transport = MockTransport() + transport.add_response(2, 2, OP_STATUS, + bytes.fromhex("ffffff01ffffff02ffffff04ffffff03")) + transport.add_response(31, 3, OP_STATUS, bytes([0x01])) + status = BmapConnection(transport, qc_ultra2_earbuds).status() + assert status.battery == 0 + assert status.battery_readings == [] + assert status.mode == "aware" + def test_status_uses_one_battery_response(self): transport = MockTransport() transport.add_response(2, 2, OP_STATUS, diff --git a/rust/src/connection.rs b/rust/src/connection.rs index 91e940f..11f155c 100644 --- a/rust/src/connection.rs +++ b/rust/src/connection.rs @@ -101,10 +101,19 @@ impl BmapConnection { message: message.into(), code: 0, })?; + let sources = self.config.battery_aggregate_sources; let aggregate = readings .iter() .find(|reading| reading.component_id == component_id) .map(|reading| reading.level) + // Aggregate missing or 0xFF: fall back to the lowest bud reading. + .or_else(|| { + readings + .iter() + .filter(|reading| sources.contains(&reading.component_id)) + .map(|reading| reading.level) + .min() + }) .ok_or_else(|| BmapError::Device { message: format!( "Battery response missing aggregate component {}", @@ -259,7 +268,10 @@ impl BmapConnection { }; let (cnc_level, cnc_max) = self.cnc().unwrap_or((0, 10)); let (prompts_enabled, prompts_language) = self.prompts().unwrap_or((false, "Unknown")); - let battery = self.battery_status()?; + let battery = self.battery_status().unwrap_or(BatteryStatus { + aggregate: 0, + readings: Vec::new(), + }); Ok(DeviceStatus { battery: battery.aggregate, @@ -753,18 +765,93 @@ mod tests { assert_eq!(dev.transport.sent.borrow().len(), 1); } + fn earbuds_battery_fixture() -> Vec { + let text = include_str!( + "../../fixtures/packets/qc-ultra2-earbuds/battery-status.hex" + ) + .trim(); + (0..text.len()) + .step_by(2) + .map(|i| u8::from_str_radix(&text[i..i + 2], 16).unwrap()) + .collect() + } + + fn fixture_bud_readings() -> Vec { + vec![ + BatteryReading { component_id: 1, level: 60 }, + BatteryReading { component_id: 2, level: 60 }, + BatteryReading { component_id: 3, level: 80 }, + ] + } + + #[test] + fn test_battery_falls_back_to_lowest_bud_without_aggregate() { + let mut t = MockTransport::new(); + t.add(2, 2, 0x03, &[ + 0x3c,0xff,0xff,0x01, 0x50,0xff,0xff,0x02, + 0x28,0xff,0xff,0x03, + ]); + let dev = BmapConnection::new(t, devices::qc_ultra2_earbuds()); + // Case (3) is lower but is not a bud; right bud (1) is the lowest. + assert_eq!(dev.battery().unwrap(), 60); + } + #[test] - fn test_battery_rejects_missing_aggregate_component() { + fn test_battery_rejects_response_without_valid_buds() { let mut t = MockTransport::new(); t.add(2, 2, 0x03, &[ - 0x3c,0xff,0xff,0x01, 0x3c,0xff,0xff,0x02, - 0x50,0xff,0xff,0x03, + 0xff,0xff,0xff,0x01, 0xff,0xff,0xff,0x02, + 0xff,0xff,0xff,0x04, 0x28,0xff,0xff,0x03, ]); let dev = BmapConnection::new(t, devices::qc_ultra2_earbuds()); assert!(matches!(dev.battery(), Err(BmapError::Device { message, .. }) if message.contains("aggregate component 4"))); } + #[test] + fn test_status_falls_back_when_fixture_aggregate_is_invalid() { + let mut payload = earbuds_battery_fixture(); + for record in payload.chunks_exact_mut(4) { + if record[3] == 4 { + record[0] = 0xff; + } + } + let mut t = MockTransport::new(); + t.add(2, 2, 0x03, &payload); + let status = BmapConnection::new(t, devices::qc_ultra2_earbuds()).status().unwrap(); + assert_eq!(status.battery, 60); + assert_eq!(status.battery_readings, fixture_bud_readings()); + } + + #[test] + fn test_status_falls_back_when_fixture_aggregate_is_absent() { + let payload: Vec = earbuds_battery_fixture() + .chunks_exact(4) + .filter(|record| record[3] != 4) + .flatten() + .copied() + .collect(); + let mut t = MockTransport::new(); + t.add(2, 2, 0x03, &payload); + let status = BmapConnection::new(t, devices::qc_ultra2_earbuds()).status().unwrap(); + assert_eq!(status.battery, 60); + assert_eq!(status.battery_readings, fixture_bud_readings()); + } + + #[test] + fn test_status_tolerates_battery_without_valid_readings() { + let mut t = MockTransport::new(); + t.add(2, 2, 0x03, &[ + 0xff,0xff,0xff,0x01, 0xff,0xff,0xff,0x02, + 0xff,0xff,0xff,0x04, 0xff,0xff,0xff,0x03, + ]); + t.add(31, 3, 0x03, &[0x01]); + let status = BmapConnection::new(t, devices::qc_ultra2_earbuds()).status().unwrap(); + assert_eq!(status.battery, 0); + assert!(status.battery_readings.is_empty()); + assert_eq!(status.mode, "aware"); + } + #[test] fn test_status_uses_one_battery_response() { let mut t = MockTransport::new(); diff --git a/rust/src/device.rs b/rust/src/device.rs index 5230bed..0860b4f 100644 --- a/rust/src/device.rs +++ b/rust/src/device.rs @@ -113,6 +113,8 @@ pub struct DeviceConfig { pub battery_components: &'static [(u8, &'static str)], /// Component ID used for the generic aggregate battery value. pub battery_aggregate_id: Option, + /// Component IDs whose lowest level stands in for a missing aggregate. + pub battery_aggregate_sources: &'static [u8], pub firmware: Option, pub product_name: Option, pub voice_prompts: Option, @@ -155,16 +157,16 @@ pub fn parse_battery(payload: &[u8]) -> Option { } /// Parse four-byte component battery records from [2.2]. -pub fn parse_battery_readings(payload: &[u8]) -> Result, &'static str> { +pub fn parse_battery_readings(payload: &[u8]) -> Result, String> { if payload.len() % 4 != 0 { - return Err("Malformed battery response"); + return Err("Malformed battery response".into()); } let mut readings = Vec::new(); let mut seen_components = Vec::new(); for record in payload.chunks_exact(4) { let component_id = record[3]; if seen_components.contains(&component_id) { - return Err("Duplicate battery component"); + return Err(format!("Duplicate battery component {}", component_id)); } seen_components.push(component_id); if record[0] <= 100 { @@ -683,13 +685,13 @@ mod tests { ); assert_eq!( parse_battery_readings(&[0x50, 0xff]), - Err("Malformed battery response"), + Err("Malformed battery response".to_string()), ); assert_eq!( parse_battery_readings(&[ 0x3c, 0xff, 0xff, 0x01, 0x50, 0xff, 0xff, 0x01, ]), - Err("Duplicate battery component"), + Err("Duplicate battery component 1".to_string()), ); } diff --git a/rust/src/devices.rs b/rust/src/devices.rs index b77aa27..6724a99 100644 --- a/rust/src/devices.rs +++ b/rust/src/devices.rs @@ -19,6 +19,7 @@ pub fn qc_ultra2() -> DeviceConfig { battery: Some(Addr(2, 2)), battery_components: &[], battery_aggregate_id: None, + battery_aggregate_sources: &[], firmware: Some(Addr(0, 5)), product_name: Some(Addr(1, 2)), voice_prompts: Some(Addr(1, 3)), @@ -64,6 +65,7 @@ pub fn qc_ultra2_earbuds() -> DeviceConfig { }; c.battery_components = &[(1, "Right"), (2, "Left"), (3, "Case")]; c.battery_aggregate_id = Some(4); + c.battery_aggregate_sources = &[1, 2]; c } @@ -82,6 +84,7 @@ pub fn qc_prince() -> DeviceConfig { battery: Some(Addr(2, 2)), battery_components: &[], battery_aggregate_id: None, + battery_aggregate_sources: &[], firmware: Some(Addr(0, 5)), product_name: Some(Addr(1, 2)), voice_prompts: Some(Addr(1, 3)), @@ -130,6 +133,7 @@ pub fn qc35() -> DeviceConfig { battery: Some(Addr(2, 2)), battery_components: &[], battery_aggregate_id: None, + battery_aggregate_sources: &[], firmware: Some(Addr(0, 5)), product_name: Some(Addr(1, 2)), voice_prompts: Some(Addr(1, 3)), @@ -179,6 +183,7 @@ pub fn qc_earbuds() -> DeviceConfig { battery: Some(Addr(2, 2)), battery_components: &[], battery_aggregate_id: None, + battery_aggregate_sources: &[], firmware: Some(Addr(0, 5)), product_name: Some(Addr(1, 2)), voice_prompts: Some(Addr(1, 3)), @@ -226,6 +231,7 @@ pub fn qc45() -> DeviceConfig { battery: Some(Addr(2, 2)), battery_components: &[], battery_aggregate_id: None, + battery_aggregate_sources: &[], firmware: Some(Addr(0, 5)), product_name: Some(Addr(1, 2)), voice_prompts: Some(Addr(1, 3)), @@ -274,6 +280,7 @@ pub fn ultra_open() -> DeviceConfig { battery: Some(Addr(2, 2)), battery_components: &[], battery_aggregate_id: None, + battery_aggregate_sources: &[], firmware: Some(Addr(0, 5)), product_name: Some(Addr(1, 2)), voice_prompts: Some(Addr(1, 3)), @@ -369,6 +376,7 @@ mod tests { assert_eq!(dev.info.codename, "edith"); assert_eq!(dev.battery_components, &[(1, "Right"), (2, "Left"), (3, "Case")]); assert_eq!(dev.battery_aggregate_id, Some(4)); + assert_eq!(dev.battery_aggregate_sources, &[1, 2]); assert_eq!(dev.preset_modes.len(), 4); assert!(dev.audio_settings.is_some()); } From 2cc2dcfc0a5f938846300cb256aa11c7288760e0 Mon Sep 17 00:00:00 2001 From: Aaron Bockelie Date: Mon, 28 Sep 2026 11:59:56 -0500 Subject: [PATCH 3/4] Keep battery a hard failure in status() A failed battery read reported as 0 is indistinguishable from a measured 0%, and bosectl-qt publishes that value to the desktop via BlueZ's battery provider. status() calls battery_status() directly again, as on main; the bud fallback alone covers a missing or 0xFF aggregate. --- cpp/src/connection.h | 3 +-- cpp/tests/test_connection.cpp | 13 ++++++++----- docs/architecture.md | 3 --- python/pybmap/connection.py | 2 +- python/tests/test_connection.py | 10 +++++----- rust/src/connection.rs | 15 ++++++--------- 6 files changed, 21 insertions(+), 25 deletions(-) diff --git a/cpp/src/connection.h b/cpp/src/connection.h index 08c5ab8..9736c93 100644 --- a/cpp/src/connection.h +++ b/cpp/src/connection.h @@ -122,8 +122,7 @@ class BmapConnection { [&]{ return cnc(); }, {0, 10}); auto [prom_on, prom_lang] = safe_call>( [&]{ return prompts(); }, {false, ""}); - auto battery_state = safe_call( - [&]{ return battery_status(); }, {0, {}}); + auto battery_state = battery_status(); DeviceStatus s; s.battery = battery_state.aggregate; diff --git a/cpp/tests/test_connection.cpp b/cpp/tests/test_connection.cpp index 643a9b0..a63ec5b 100644 --- a/cpp/tests/test_connection.cpp +++ b/cpp/tests/test_connection.cpp @@ -162,7 +162,8 @@ TEST(status_falls_back_when_fixture_aggregate_is_absent) { assert_fixture_bud_readings(status.battery_readings); } -TEST(status_tolerates_battery_without_valid_readings) { +TEST(status_rejects_battery_without_valid_readings) { + // A failed read must not surface as a measured 0%. auto raw = new MockTransport(); raw->add(2, 2, 0x03, { 0xff,0xff,0xff,0x01, 0xff,0xff,0xff,0x02, @@ -170,10 +171,12 @@ TEST(status_tolerates_battery_without_valid_readings) { }); raw->add(31, 3, 0x03, {0x01}); BmapConnection dev(std::unique_ptr(raw), qc_ultra2_earbuds()); - auto status = dev.status(); - ASSERT_EQ(status.battery, 0); - ASSERT_TRUE(status.battery_readings.empty()); - ASSERT_EQ(status.mode, "aware"); + bool threw = false; + try { dev.status(); } + catch (const std::runtime_error& error) { + threw = std::string(error.what()).find("aggregate component 4") != std::string::npos; + } + ASSERT_TRUE(threw); } TEST(status_uses_one_battery_response) { diff --git a/docs/architecture.md b/docs/architecture.md index b97af7f..cd0922b 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -200,9 +200,6 @@ battery() / status() → reuse the parsed battery status `BatteryStatus` keeps the aggregate and all component readings from that one response. `DeviceStatus.battery_readings` carries the same snapshot to CLIs, which render known components in config order rather than packet order. -`status()` treats battery like its other optional fields: a failed or -unusable battery read yields `battery = 0` with no readings instead of -failing the whole snapshot. **Write pattern** (SETGET): ``` diff --git a/python/pybmap/connection.py b/python/pybmap/connection.py index 43d5545..60b6071 100644 --- a/python/pybmap/connection.py +++ b/python/pybmap/connection.py @@ -307,7 +307,7 @@ def status(self): current_name = self._mode_name_from_idx(current_idx) if current_idx is not None else "" cnc_cur, cnc_max = self._safe_read(self.cnc, (0, 10)) prompts_on, prompts_lang = self._safe_read(self.prompts, (False, "")) - battery = self._safe_read(self.battery_status, BatteryStatus(0, [])) + battery = self.battery_status() return DeviceStatus( battery=battery.aggregate, diff --git a/python/tests/test_connection.py b/python/tests/test_connection.py index cb327da..fa726ad 100644 --- a/python/tests/test_connection.py +++ b/python/tests/test_connection.py @@ -156,15 +156,15 @@ def test_status_falls_back_when_fixture_aggregate_is_absent(self): (1, 60), (2, 60), (3, 80) ] - def test_status_tolerates_battery_without_valid_readings(self): + def test_status_rejects_battery_without_valid_readings(self): + # A failed read must not surface as a measured 0%. transport = MockTransport() transport.add_response(2, 2, OP_STATUS, bytes.fromhex("ffffff01ffffff02ffffff04ffffff03")) transport.add_response(31, 3, OP_STATUS, bytes([0x01])) - status = BmapConnection(transport, qc_ultra2_earbuds).status() - assert status.battery == 0 - assert status.battery_readings == [] - assert status.mode == "aware" + dev = BmapConnection(transport, qc_ultra2_earbuds) + with pytest.raises(BmapDeviceError, match="aggregate component 4"): + dev.status() def test_status_uses_one_battery_response(self): transport = MockTransport() diff --git a/rust/src/connection.rs b/rust/src/connection.rs index 11f155c..5af5b54 100644 --- a/rust/src/connection.rs +++ b/rust/src/connection.rs @@ -268,10 +268,7 @@ impl BmapConnection { }; let (cnc_level, cnc_max) = self.cnc().unwrap_or((0, 10)); let (prompts_enabled, prompts_language) = self.prompts().unwrap_or((false, "Unknown")); - let battery = self.battery_status().unwrap_or(BatteryStatus { - aggregate: 0, - readings: Vec::new(), - }); + let battery = self.battery_status()?; Ok(DeviceStatus { battery: battery.aggregate, @@ -839,17 +836,17 @@ mod tests { } #[test] - fn test_status_tolerates_battery_without_valid_readings() { + fn test_status_rejects_battery_without_valid_readings() { + // A failed read must not surface as a measured 0%. let mut t = MockTransport::new(); t.add(2, 2, 0x03, &[ 0xff,0xff,0xff,0x01, 0xff,0xff,0xff,0x02, 0xff,0xff,0xff,0x04, 0xff,0xff,0xff,0x03, ]); t.add(31, 3, 0x03, &[0x01]); - let status = BmapConnection::new(t, devices::qc_ultra2_earbuds()).status().unwrap(); - assert_eq!(status.battery, 0); - assert!(status.battery_readings.is_empty()); - assert_eq!(status.mode, "aware"); + let dev = BmapConnection::new(t, devices::qc_ultra2_earbuds()); + assert!(matches!(dev.status(), Err(BmapError::Device { message, .. }) + if message.contains("aggregate component 4"))); } #[test] From 36a51ffcdf8769c18bdd34962b9f08eaa492c239 Mon Sep 17 00:00:00 2001 From: Aaron Bockelie Date: Mon, 28 Sep 2026 12:05:50 -0500 Subject: [PATCH 4/4] Skip the Bluetooth hint when connect() arguments are invalid A MAC without a device type is a setup mistake, not a Bluetooth problem, so the CLIs no longer follow it with "Is Bluetooth on?". Python gains BmapInvalidArgError to mirror Rust's InvalidArg; C++ matches std::invalid_argument. Also test that parse_all_responses stops at an unknown operator, and call the real C++ connect() with a MAC and no device type (the test binary now links libbluetooth like bmapctl). --- cpp/CMakeLists.txt | 7 ++++++- cpp/src/bmap.h | 7 +++++++ cpp/src/main.cpp | 4 ++-- cpp/tests/test_connection.cpp | 12 ++++++++++++ cpp/tests/test_protocol.cpp | 11 +++++++++++ python/pybmap/__init__.py | 4 ++-- python/pybmap/cli.py | 6 ++++-- python/pybmap/errors.py | 4 ++++ python/tests/test_channel_probe.py | 7 +++++-- python/tests/test_cli.py | 20 ++++++++++++++++++++ python/tests/test_protocol.py | 8 ++++++++ rust/src/main.rs | 25 ++++++++++++++++++++++++- rust/src/protocol.rs | 10 ++++++++++ 13 files changed, 115 insertions(+), 10 deletions(-) diff --git a/cpp/CMakeLists.txt b/cpp/CMakeLists.txt index 6ad8c33..bca6cfc 100644 --- a/cpp/CMakeLists.txt +++ b/cpp/CMakeLists.txt @@ -43,5 +43,10 @@ add_executable(bmap_tests tests/test_catalog.cpp tests/test_transport.cpp ) -target_link_libraries(bmap_tests PRIVATE bmap) +# Linked like bmapctl so tests can call the real connect(); no BT at runtime. +if(APPLE) + target_link_libraries(bmap_tests PRIVATE bmap) +else() + target_link_libraries(bmap_tests PRIVATE bmap bluetooth) +endif() add_test(NAME bmap_tests COMMAND bmap_tests) diff --git a/cpp/src/bmap.h b/cpp/src/bmap.h index 800e47e..7c3f361 100644 --- a/cpp/src/bmap.h +++ b/cpp/src/bmap.h @@ -91,6 +91,13 @@ inline std::unique_ptr open_transport(const std::string& mac, } // namespace detail +/// Follow-up hint for a failed connect(), or nullptr when the failure is a +/// caller setup mistake (std::invalid_argument) rather than a Bluetooth issue. +inline const char* connection_hint(const std::exception& error) { + if (dynamic_cast(&error)) return nullptr; + return "Is Bluetooth on? Are the headphones paired and connected?"; +} + /// Connect to a BMAP device. Device type is resolved only during MAC discovery. inline std::unique_ptr connect( const std::string& mac_override = "", diff --git a/cpp/src/main.cpp b/cpp/src/main.cpp index 078f21a..070bd70 100644 --- a/cpp/src/main.cpp +++ b/cpp/src/main.cpp @@ -54,8 +54,8 @@ int main(int argc, char** argv) { try { devptr = connect(mac_str, dev_str); } catch (const std::exception& e) { - std::cerr << "Connection failed: " << e.what() << "\n" - << "Is Bluetooth on? Are the headphones paired and connected?\n"; + std::cerr << "Connection failed: " << e.what() << "\n"; + if (auto hint = connection_hint(e)) std::cerr << hint << "\n"; return 1; } auto& dev = *devptr; diff --git a/cpp/tests/test_connection.cpp b/cpp/tests/test_connection.cpp index a63ec5b..7f55523 100644 --- a/cpp/tests/test_connection.cpp +++ b/cpp/tests/test_connection.cpp @@ -258,6 +258,18 @@ TEST(explicit_mac_requires_device_type) { ASSERT_TRUE(threw); } +TEST(connect_with_mac_requires_device_type) { + // Real connect(): validation fails before any transport is opened. + bool threw = false; + try { bmap::connect("00:11:22:33:44:55", ""); } + catch (const std::invalid_argument& error) { + threw = std::string(error.what()).find("device_type is required") != std::string::npos; + ASSERT_TRUE(bmap::connection_hint(error) == nullptr); + } + ASSERT_TRUE(threw); + ASSERT_TRUE(bmap::connection_hint(std::runtime_error("no device")) != nullptr); +} + TEST(qc35_no_eq) { auto t = std::make_unique(); BmapConnection dev(std::move(t), qc35()); diff --git a/cpp/tests/test_protocol.cpp b/cpp/tests/test_protocol.cpp index c82e8aa..84dd514 100644 --- a/cpp/tests/test_protocol.cpp +++ b/cpp/tests/test_protocol.cpp @@ -46,6 +46,17 @@ TEST(parse_all_responses_two) { ASSERT_EQ(responses[1].func, 3); } +TEST(parse_all_stops_at_unknown_operator) { + std::vector data = { + 31, 6, 0x03, 2, 0xAA, 0xBB, + 31, 3, 0x08, 1, 0x00, + 31, 3, 0x06, 1, 0x00, + }; + auto responses = parse_all_responses(data); + ASSERT_EQ(responses.size(), 1u); + ASSERT_EQ(responses[0].func, 6); +} + TEST(parse_all_truncated) { std::vector data = {31, 3, 0x06, 10, 0x00, 0x01}; auto responses = parse_all_responses(data); diff --git a/python/pybmap/__init__.py b/python/pybmap/__init__.py index a4bf9bb..1e09f27 100644 --- a/python/pybmap/__init__.py +++ b/python/pybmap/__init__.py @@ -25,7 +25,7 @@ ) from .errors import ( BmapError, BmapConnectionError, BmapAuthError, - BmapDeviceError, BmapTimeoutError, BmapNotFoundError, + BmapDeviceError, BmapTimeoutError, BmapNotFoundError, BmapInvalidArgError, ) from .types import ( BatteryReading, BatteryStatus, BmapResponse, ButtonMapping, DeviceStatus, @@ -66,7 +66,7 @@ def connect(mac=None, device_type=None): if device_type is None: device_type = detected_type elif device_type is None: - raise BmapError("device_type is required when mac is specified") + raise BmapInvalidArgError("device_type is required when mac is specified") device = get_device(device_type) channel = getattr(device, "RFCOMM_CHANNEL", 2) diff --git a/python/pybmap/cli.py b/python/pybmap/cli.py index 3e6f154..212bdaf 100644 --- a/python/pybmap/cli.py +++ b/python/pybmap/cli.py @@ -6,7 +6,7 @@ import pybmap from pybmap.constants import SPATIAL_NAMES, SIDETONE_NAMES, VOICE_LANGUAGES -from pybmap.errors import BmapError, BmapConnectionError +from pybmap.errors import BmapError, BmapConnectionError, BmapInvalidArgError from pybmap.protocol import fmt_response # ── ANSI Colors ────────────────────────────────────────────────────────────── @@ -330,7 +330,9 @@ def main(): dev = pybmap.connect(mac=mac, device_type=device_type) except BmapError as e: print("%sConnection failed:%s %s" % (C_RED, C_RESET, e), file=sys.stderr) - print("%sIs Bluetooth on? Are the headphones paired and connected?%s" % (C_DIM, C_RESET), file=sys.stderr) + # A setup mistake is not a Bluetooth problem; skip the pairing hint. + if not isinstance(e, BmapInvalidArgError): + print("%sIs Bluetooth on? Are the headphones paired and connected?%s" % (C_DIM, C_RESET), file=sys.stderr) sys.exit(1) preset_names = set(dev.preset_modes.keys()) diff --git a/python/pybmap/errors.py b/python/pybmap/errors.py index 505fc0b..ae349a3 100644 --- a/python/pybmap/errors.py +++ b/python/pybmap/errors.py @@ -21,6 +21,10 @@ def __init__(self, message, error_code=None): self.error_code = error_code +class BmapInvalidArgError(BmapError): + """Caller supplied invalid or incomplete arguments.""" + + class BmapTimeoutError(BmapError): """Device did not respond in time.""" diff --git a/python/tests/test_channel_probe.py b/python/tests/test_channel_probe.py index 4507f96..4c2afed 100644 --- a/python/tests/test_channel_probe.py +++ b/python/tests/test_channel_probe.py @@ -4,7 +4,10 @@ import pybmap from pybmap.constants import OP_STATUS, OP_PROCESSING, OP_ERROR -from pybmap.errors import BmapConnectionError, BmapTimeoutError, BmapDeviceError, BmapError +from pybmap.errors import ( + BmapConnectionError, BmapTimeoutError, BmapDeviceError, + BmapInvalidArgError, +) from pybmap.devices import qc_prince from tests.test_connection import MockTransport @@ -98,7 +101,7 @@ def test_fallback_order_skips_duplicate_of_configured(patch_transport): @pytest.mark.parametrize("device_type", [None, ""]) def test_explicit_mac_requires_device_type(patch_transport, device_type): f = patch_transport({2: "bmap"}) - with pytest.raises(BmapError, match="device_type is required"): + with pytest.raises(BmapInvalidArgError, match="device_type is required"): pybmap.connect(mac="00:11:22:33:44:55", device_type=device_type) assert f.attempts == [] diff --git a/python/tests/test_cli.py b/python/tests/test_cli.py index cf9015d..0d38f63 100644 --- a/python/tests/test_cli.py +++ b/python/tests/test_cli.py @@ -1,5 +1,8 @@ """Tests for user-visible CLI output.""" +import pytest + +from pybmap import cli from pybmap.cli import cmd_status from pybmap.types import BatteryReading, DeviceStatus @@ -54,3 +57,20 @@ def test_device_status_preserves_old_positional_shape(): ) assert status.mode == "quiet" assert status.battery_readings == () + + +@pytest.mark.parametrize("device_env", [None, ""]) +def test_mac_without_device_type_skips_bluetooth_hint(monkeypatch, capsys, device_env): + monkeypatch.setattr(cli.sys, "argv", ["bosectl", "status"]) + monkeypatch.setenv("BMAP_MAC", "00:11:22:33:44:55") + monkeypatch.delenv("BOSE_MAC", raising=False) + if device_env is None: + monkeypatch.delenv("BMAP_DEVICE", raising=False) + else: + monkeypatch.setenv("BMAP_DEVICE", device_env) + with pytest.raises(SystemExit) as exit_info: + cli.main() + assert exit_info.value.code == 1 + err = capsys.readouterr().err + assert "device_type is required" in err + assert "Is Bluetooth on?" not in err diff --git a/python/tests/test_protocol.py b/python/tests/test_protocol.py index 2c6bd96..edc004e 100644 --- a/python/tests/test_protocol.py +++ b/python/tests/test_protocol.py @@ -99,6 +99,14 @@ def test_concatenated_packets(self): assert responses[1].func == 3 assert responses[1].payload == bytes([0x00]) + def test_stops_at_unknown_operator(self): + pkt1 = bytes([31, 6, 0x03, 2, 0xAA, 0xBB]) + unknown = bytes([31, 3, 0x08, 1, 0x00]) + pkt3 = bytes([31, 3, 0x06, 1, 0x00]) + responses = parse_all_responses(pkt1 + unknown + pkt3) + assert len(responses) == 1 + assert responses[0].func == 6 + def test_empty_data(self): assert parse_all_responses(bytes()) == [] assert parse_all_responses(bytes([1, 2, 3])) == [] diff --git a/rust/src/main.rs b/rust/src/main.rs index 4ee8f3a..52c7c28 100644 --- a/rust/src/main.rs +++ b/rust/src/main.rs @@ -27,7 +27,9 @@ fn main() { Ok(d) => d, Err(e) => { eprintln!("Connection failed: {}", e); - eprintln!("Is Bluetooth on? Are the headphones paired and connected?"); + if let Some(hint) = connection_hint(&e) { + eprintln!("{}", hint); + } process::exit(1); } }; @@ -254,6 +256,14 @@ fn main() { } } +/// Follow-up hint for a failed connect; setup mistakes are not Bluetooth problems. +fn connection_hint(e: &BmapError) -> Option<&'static str> { + match e { + BmapError::InvalidArg(_) => None, + _ => Some("Is Bluetooth on? Are the headphones paired and connected?"), + } +} + fn err_exit(e: &BmapError) { eprintln!("Error: {}", e); process::exit(1); @@ -337,3 +347,16 @@ fn usage() { println!(" BMAP_MAC=XX:XX:XX:XX:XX:XX Device MAC (auto-detected if unset)"); println!(" BMAP_DEVICE=qc_ultra2|qc_ultra2_earbuds|qc_prince|qc35|qc_earbuds|qc45|ultra_open"); } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn missing_device_type_skips_bluetooth_hint() { + let err = connect(Some("00:11:22:33:44:55"), None).err().unwrap(); + assert!(matches!(err, BmapError::InvalidArg(_))); + assert_eq!(connection_hint(&err), None); + assert!(connection_hint(&BmapError::NotFound("none".into())).is_some()); + } +} diff --git a/rust/src/protocol.rs b/rust/src/protocol.rs index 3433a80..ad0e49a 100644 --- a/rust/src/protocol.rs +++ b/rust/src/protocol.rs @@ -194,6 +194,16 @@ mod tests { assert_eq!(responses[1].func, 3); } + #[test] + fn test_parse_all_stops_at_unknown_operator() { + let mut data = vec![31, 6, 0x03, 2, 0xAA, 0xBB]; + data.extend_from_slice(&[31, 3, 0x08, 1, 0x00]); + data.extend_from_slice(&[31, 3, 0x06, 1, 0x00]); + let responses = parse_all_responses(&data); + assert_eq!(responses.len(), 1); + assert_eq!(responses[0].func, 6); + } + #[test] fn test_parse_all_truncated() { // Length says 10 bytes but only 2 available