forked from auracaster/bumble_mirror
Rust library cleanup
* Fix error code extraction from Python to Rust * Add documentation for dealing with HCI packets
This commit is contained in:
@@ -69,3 +69,68 @@ To regenerate the assigned number tables based on the Python codebase:
|
||||
```
|
||||
PYTHONPATH=.. cargo run --bin gen-assigned-numbers --features dev-tools
|
||||
```
|
||||
|
||||
## HCI packets
|
||||
|
||||
Sending a command packet from a device is composed to of two major steps.
|
||||
There are more generalized ways of dealing with packets in other scenarios.
|
||||
|
||||
### Construct the command
|
||||
Pick a command from `src/internal/hci/packets.pdl` and construct its associated "builder" struct.
|
||||
|
||||
```rust
|
||||
// The "LE Set Scan Enable" command can be found in the Core Bluetooth Spec.
|
||||
// It can also be found in `packets.pdl` as `packet LeSetScanEnable : Command`
|
||||
fn main() {
|
||||
let device = init_device_as_desired();
|
||||
|
||||
let le_set_scan_enable_command_builder = LeSetScanEnableBuilder {
|
||||
filter_duplicates: Enable::Disabled,
|
||||
le_scan_enable: Enable::Enabled,
|
||||
};
|
||||
}
|
||||
```
|
||||
|
||||
### Send the command and interpret the event response
|
||||
Send the command from an initialized device, and then receive the response.
|
||||
|
||||
```rust
|
||||
fn main() {
|
||||
// ...
|
||||
|
||||
// `check_result` to false to receive the event response even if the controller returns a failure code
|
||||
let event = device.send_command(le_set_scan_enable_command_builder.into(), /*check_result*/ false);
|
||||
// Coerce the event into the expected format. A `Command` should have an associated event response
|
||||
// "<command name>Complete".
|
||||
let le_set_scan_enable_complete_event: LeSetScanEnableComplete = event.try_into().unwrap();
|
||||
}
|
||||
```
|
||||
|
||||
### Generic packet handling
|
||||
At the very least, you should expect to at least know _which_ kind of base packet you are dealing with. Base packets in
|
||||
`packets.pdl` can be identified because they do not extend any other packet. They are easily found with the regex:
|
||||
`^packet [^:]* \{`. For Bluetooth LE (BLE) HCI, one should find some kind of header preceding the packet with the purpose of
|
||||
packet disambiguation. We do some of that disambiguation for H4 BLE packets using the `WithPacketHeader` trait at `internal/hci/mod.rs`.
|
||||
|
||||
Say you've identified a series of bytes that are certainly an `Acl` packet. They can be parsed using the `Acl` struct.
|
||||
```rust
|
||||
fn main() {
|
||||
let bytes = bytes_that_are_certainly_acl();
|
||||
let acl_packet = Acl::parse(bytes).unwrap();
|
||||
}
|
||||
```
|
||||
|
||||
Since you don't yet know what kind of `Acl` packet it is, you need to specialize it and then handle the various
|
||||
potential cases.
|
||||
```rust
|
||||
fn main() {
|
||||
// ...
|
||||
match acl_packet.specialize() {
|
||||
Payload(bytes) => do_something(bytes),
|
||||
None => do_something_else(),
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Some packets may yet further embed other packets, in which case you may need to further specialize until no more
|
||||
specialization is needed.
|
||||
|
||||
@@ -25,7 +25,6 @@ use clap::Parser as _;
|
||||
use pyo3::PyResult;
|
||||
use rand::Rng;
|
||||
use std::path;
|
||||
|
||||
#[pyo3_asyncio::tokio::main]
|
||||
async fn main() -> PyResult<()> {
|
||||
env_logger::builder()
|
||||
|
||||
@@ -28,7 +28,7 @@ use bumble::wrapper::{
|
||||
};
|
||||
use pyo3::{
|
||||
exceptions::PyException,
|
||||
{PyErr, PyResult},
|
||||
FromPyObject, IntoPy, Python, {PyErr, PyResult},
|
||||
};
|
||||
|
||||
#[pyo3_asyncio::tokio::test]
|
||||
@@ -78,6 +78,28 @@ async fn test_hci_roundtrip_success_and_failure() -> PyResult<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[pyo3_asyncio::tokio::test]
|
||||
fn valid_error_code_extraction_succeeds() -> PyResult<()> {
|
||||
let error_code = Python::with_gil(|py| {
|
||||
let python_error_code_success = 0x00_u8.into_py(py);
|
||||
ErrorCode::extract(python_error_code_success.as_ref(py))
|
||||
})?;
|
||||
|
||||
assert_eq!(ErrorCode::Success, error_code);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[pyo3_asyncio::tokio::test]
|
||||
fn invalid_error_code_extraction_fails() -> PyResult<()> {
|
||||
let failed_extraction = Python::with_gil(|py| {
|
||||
let python_invalid_error_code = 0xFE_u8.into_py(py);
|
||||
ErrorCode::extract(python_invalid_error_code.as_ref(py))
|
||||
});
|
||||
|
||||
assert!(failed_extraction.is_err());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn create_local_device(address: Address) -> PyResult<Device> {
|
||||
let link = Link::new_local_link()?;
|
||||
let controller = Controller::new("C1", None, None, Some(link), Some(address.clone())).await?;
|
||||
|
||||
@@ -178,7 +178,11 @@ impl IntoPy<PyObject> for AddressType {
|
||||
|
||||
impl<'source> FromPyObject<'source> for ErrorCode {
|
||||
fn extract(ob: &'source PyAny) -> PyResult<Self> {
|
||||
ob.extract()
|
||||
// Bumble represents error codes simply as a single-byte number (in Rust, u8)
|
||||
let value: u8 = ob.extract()?;
|
||||
ErrorCode::try_from(value).map_err(|b| {
|
||||
PyErr::new::<PyException, _>(format!("Failed to map {b} to an error code"))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user