Examples

This page provides code examples for common tasks using the Zigbee Analyzer.

Basic Capture

Simple packet capture with live output.

use zigbee_drivers::{DriverRegistry, CC2531};
use zigbee_hal::traits::ZigbeeCapture;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Auto-detect device
    let registry = DriverRegistry::new();
    let devices = registry.detect_devices();
    
    if devices.is_empty() {
        eprintln!("No Zigbee devices found!");
        return Ok(());
    }
    
    println!("Using: {}", devices[0].description);
    
    // Create and initialize driver
    let mut driver = CC2531::new()?;
    driver.initialize().await?;
    
    // Set channel (11-26)
    driver.set_channel(15).await?;
    println!("Capturing on channel 15...\n");
    
    // Capture loop
    for i in 1..=10 {
        let packet = driver.capture_packet().await?;
        
        println!("Packet #{}", i);
        println!("  RSSI: {} dBm", packet.rssi);
        println!("  LQI: {}", packet.lqi);
        println!("  Length: {} bytes", packet.data.len());
        println!("  Data: {:02x?}\n", &packet.data[..packet.data.len().min(16)]);
    }
    
    driver.shutdown().await?;
    Ok(())
}

Run:

cargo run --example basic_capture

PCAP Recording

Capture packets and save to a PCAP file for Wireshark.

use zigbee_drivers::{DriverRegistry, PcapWriter};
use zigbee_hal::traits::ZigbeeCapture;
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let args: Vec<String> = std::env::args().collect();
    if args.len() < 2 {
        eprintln!("Usage: {} <output.pcap> [duration_secs]", args[0]);
        return Ok(());
    }
    
    let output_path = &args[1];
    let duration_secs: u64 = args.get(2)
        .and_then(|s| s.parse().ok())
        .unwrap_or(60);
    
    // Setup driver
    let registry = DriverRegistry::new();
    let devices = registry.detect_devices();
    let mut driver = registry.create_driver(&devices[0]).unwrap();
    
    driver.initialize().await?;
    driver.set_channel(11).await?;
    
    // Create PCAP writer
    let mut pcap = PcapWriter::new(output_path)?;
    pcap.write_header()?;
    
    println!("Recording to: {}", output_path);
    println!("Duration: {} seconds\n", duration_secs);
    
    let start = std::time::Instant::now();
    let mut count = 0;
    
    while start.elapsed() < Duration::from_secs(duration_secs) {
        match tokio::time::timeout(
            Duration::from_secs(1),
            driver.capture_packet()
        ).await {
            Ok(Ok(packet)) => {
                pcap.write_packet(&packet)?;
                count += 1;
                
                if count % 100 == 0 {
                    println!("Captured {} packets...", count);
                }
            }
            Ok(Err(e)) => eprintln!("Capture error: {}", e),
            Err(_) => continue, // Timeout
        }
    }
    
    pcap.flush()?;
    println!("\nDone! Captured {} packets", count);
    println!("Open with: wireshark {}", output_path);
    
    Ok(())
}

Run:

cargo run --example pcap_record -- output.pcap 60
wireshark output.pcap

Channel Scanning

Scan all Zigbee channels to find active networks.

use zigbee_drivers::DriverRegistry;
use zigbee_hal::traits::ZigbeeCapture;
use std::collections::HashMap;
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let registry = DriverRegistry::new();
    let devices = registry.detect_devices();
    let mut driver = registry.create_driver(&devices[0]).unwrap();
    
    driver.initialize().await?;
    
    println!("Scanning Zigbee channels...\n");
    
    let mut channel_stats: HashMap<u8, usize> = HashMap::new();
    
    // Scan each channel for 5 seconds
    for channel in 11..=26 {
        println!("Scanning channel {}...", channel);
        driver.set_channel(channel).await?;
        
        let mut count = 0;
        let start = std::time::Instant::now();
        
        while start.elapsed() < Duration::from_secs(5) {
            if let Ok(Ok(_packet)) = tokio::time::timeout(
                Duration::from_millis(100),
                driver.capture_packet()
            ).await {
                count += 1;
            }
        }
        
        channel_stats.insert(channel, count);
        println!("  → {} packets\n", count);
    }
    
    // Display results
    println!("Scan Results:");
    println!("=============");
    
    let mut channels: Vec<_> = channel_stats.iter().collect();
    channels.sort_by_key(|(_, &count)| std::cmp::Reverse(count));
    
    for (channel, count) in channels {
        let bar = "█".repeat(*count / 10);
        println!("Ch {}: {:5} packets {}", channel, count, bar);
    }
    
    driver.shutdown().await?;
    Ok(())
}

Run:

cargo run --example channel_scan

Network Topology Analysis

Build a network topology from captured traffic.

use zigbee_core::packet::ParsedPacket;
use zigbee_analysis::topology::NetworkTopology;
use zigbee_drivers::DriverRegistry;
use zigbee_hal::traits::ZigbeeCapture;
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let registry = DriverRegistry::new();
    let devices = registry.detect_devices();
    let mut driver = registry.create_driver(&devices[0]).unwrap();
    
    driver.initialize().await?;
    driver.set_channel(15).await?;
    
    println!("Analyzing network topology...");
    println!("Capturing for 60 seconds...\n");
    
    let mut topology = NetworkTopology::new();
    let start = std::time::Instant::now();
    
    while start.elapsed() < Duration::from_secs(60) {
        if let Ok(packet) = driver.capture_packet().await {
            // Parse packet
            if let Ok(parsed) = packet.parse() {
                // Update topology
                topology.process_packet(&parsed);
            }
        }
    }
    
    // Display results
    println!("\nNetwork Topology:");
    println!("=================");
    println!("Coordinator: {:?}", topology.coordinator());
    println!("Routers: {} devices", topology.routers().len());
    println!("End Devices: {} devices", topology.end_devices().len());
    println!("Total Links: {}", topology.links().len());
    
    println!("\nDevices:");
    for device in topology.all_devices() {
        println!("  0x{:04x} - {} (LQI: {})", 
            device.short_addr,
            device.device_type,
            device.average_lqi
        );
    }
    
    // Export to Graphviz
    let dot = topology.to_graphviz();
    std::fs::write("network.dot", dot)?;
    println!("\nTopology saved to network.dot");
    println!("Visualize with: dot -Tpng network.dot -o network.png");
    
    driver.shutdown().await?;
    Ok(())
}

Run:

cargo run --example topology_analysis
dot -Tpng network.dot -o network.png

Packet Filtering

Filter packets by type, address, or other criteria.

use zigbee_core::packet::{RawPacket, FrameType};
use zigbee_drivers::DriverRegistry;
use zigbee_hal::traits::ZigbeeCapture;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let registry = DriverRegistry::new();
    let devices = registry.detect_devices();
    let mut driver = registry.create_driver(&devices[0]).unwrap();
    
    driver.initialize().await?;
    driver.set_channel(15).await?;
    
    println!("Filtering for data frames from 0x1234...\n");
    
    let target_addr = 0x1234u16;
    let mut count = 0;
    
    loop {
        let packet = driver.capture_packet().await?;
        
        // Parse packet
        if let Ok(parsed) = packet.parse() {
            // Filter by frame type
            if parsed.mac.frame_control.frame_type != FrameType::Data {
                continue;
            }
            
            // Filter by source address
            if let MacAddress::Short(addr) = parsed.mac.src_addr {
                if addr != target_addr {
                    continue;
                }
            } else {
                continue;
            }
            
            // This packet matches our filter!
            count += 1;
            println!("Match #{}: RSSI {} dBm", count, packet.rssi);
            
            if let Some(nwk) = parsed.network {
                println!("  NWK Dst: 0x{:04x}", nwk.dest_addr);
            }
            
            if let Some(aps) = parsed.aps {
                if let Some(cluster) = aps.cluster_id {
                    println!("  Cluster: 0x{:04x}", cluster);
                }
            }
            println!();
        }
    }
}

Custom Analysis

Implement custom packet analysis logic.

use zigbee_core::packet::ParsedPacket;
use zigbee_drivers::DriverRegistry;
use zigbee_hal::traits::ZigbeeCapture;
use std::collections::HashMap;

struct PacketStats {
    total: usize,
    by_type: HashMap<String, usize>,
    avg_rssi: f32,
    devices_seen: HashSet<u16>,
}

impl PacketStats {
    fn new() -> Self {
        Self {
            total: 0,
            by_type: HashMap::new(),
            avg_rssi: 0.0,
            devices_seen: HashSet::new(),
        }
    }
    
    fn process(&mut self, packet: &ParsedPacket, rssi: i8) {
        self.total += 1;
        
        // Update RSSI average
        let n = self.total as f32;
        self.avg_rssi = (self.avg_rssi * (n - 1.0) + rssi as f32) / n;
        
        // Count by frame type
        let frame_type = format!("{:?}", packet.mac.frame_control.frame_type);
        *self.by_type.entry(frame_type).or_insert(0) += 1;
        
        // Track devices
        if let MacAddress::Short(addr) = packet.mac.src_addr {
            self.devices_seen.insert(addr);
        }
    }
    
    fn display(&self) {
        println!("\nPacket Statistics:");
        println!("==================");
        println!("Total packets: {}", self.total);
        println!("Average RSSI: {:.1} dBm", self.avg_rssi);
        println!("Unique devices: {}", self.devices_seen.len());
        println!("\nBy Frame Type:");
        
        for (frame_type, count) in &self.by_type {
            let pct = (*count as f32 / self.total as f32) * 100.0;
            println!("  {}: {} ({:.1}%)", frame_type, count, pct);
        }
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let registry = DriverRegistry::new();
    let devices = registry.detect_devices();
    let mut driver = registry.create_driver(&devices[0]).unwrap();
    
    driver.initialize().await?;
    driver.set_channel(15).await?;
    
    let mut stats = PacketStats::new();
    
    println!("Collecting statistics for 60 seconds...\n");
    
    let start = std::time::Instant::now();
    while start.elapsed() < std::time::Duration::from_secs(60) {
        if let Ok(packet) = driver.capture_packet().await {
            if let Ok(parsed) = packet.parse() {
                stats.process(&parsed, packet.rssi);
                
                if stats.total % 100 == 0 {
                    print!("\rPackets: {}...", stats.total);
                    std::io::Write::flush(&mut std::io::stdout())?;
                }
            }
        }
    }
    
    stats.display();
    
    driver.shutdown().await?;
    Ok(())
}

Using the API

Integrate Zigbee Analyzer into your own application.

Add Dependencies

[dependencies]
zigbee-core = { path = "../zigbee-core" }
zigbee-hal = { path = "../zigbee-hal" }
zigbee-drivers = { path = "../zigbee-drivers" }
zigbee-analysis = { path = "../zigbee-analysis" }
tokio = { version = "1", features = ["full"] }

Simple Integration

use zigbee_drivers::DriverRegistry;
use zigbee_hal::traits::ZigbeeCapture;
use tokio::sync::mpsc;

struct ZigbeeMonitor {
    driver: Box<dyn ZigbeeCapture>,
    channel: u8,
}

impl ZigbeeMonitor {
    async fn new(channel: u8) -> Result<Self, Box<dyn std::error::Error>> {
        let registry = DriverRegistry::new();
        let devices = registry.detect_devices();
        
        if devices.is_empty() {
            return Err("No devices found".into());
        }
        
        let mut driver = registry.create_driver(&devices[0])
            .ok_or("Failed to create driver")?;
        
        driver.initialize().await?;
        driver.set_channel(channel).await?;
        
        Ok(Self { driver, channel })
    }
    
    async fn start(mut self) -> mpsc::Receiver<ParsedPacket> {
        let (tx, rx) = mpsc::channel(100);
        
        tokio::spawn(async move {
            loop {
                match self.driver.capture_packet().await {
                    Ok(packet) => {
                        if let Ok(parsed) = packet.parse() {
                            if tx.send(parsed).await.is_err() {
                                break; // Receiver dropped
                            }
                        }
                    }
                    Err(e) => {
                        eprintln!("Capture error: {}", e);
                        break;
                    }
                }
            }
        });
        
        rx
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let monitor = ZigbeeMonitor::new(15).await?;
    let mut rx = monitor.start().await;
    
    // Process packets in your application
    while let Some(packet) = rx.recv().await {
        // Do something with packet
        println!("Got packet from 0x{:04x}", packet.mac.src_addr);
    }
    
    Ok(())
}

Advanced Examples

Multi-Channel Monitoring

use tokio::task::JoinSet;

async fn monitor_channel(channel: u8) {
    // Each task monitors a different channel
    let mut driver = create_driver_for_channel(channel).await.unwrap();
    
    loop {
        if let Ok(packet) = driver.capture_packet().await {
            println!("Ch {}: packet from {:?}", channel, packet.mac.src_addr);
        }
    }
}

#[tokio::main]
async fn main() {
    let mut tasks = JoinSet::new();
    
    // Monitor channels 11, 15, 20, 25 simultaneously
    for channel in [11, 15, 20, 25] {
        tasks.spawn(monitor_channel(channel));
    }
    
    // Wait for all tasks
    while let Some(result) = tasks.join_next().await {
        if let Err(e) = result {
            eprintln!("Task error: {}", e);
        }
    }
}

Real-Time Decryption

use zigbee_crypto::{SecurityMaterial, ZigbeeKey};

async fn decrypt_traffic() -> Result<(), Box<dyn std::error::Error>> {
    let mut driver = setup_driver().await?;
    
    // Load network key (from configuration, user input, etc.)
    let network_key = ZigbeeKey::NetworkKey([
        0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08,
        0x09, 0x0A, 0x0B, 0x0C, 0x0D, 0x0E, 0x0F, 0x10,
    ]);
    
    let mut security = SecurityMaterial::new();
    security.add_network_key(network_key);
    
    loop {
        let packet = driver.capture_packet().await?;
        let parsed = packet.parse()?;
        
        if parsed.is_encrypted() {
            // Attempt decryption
            match security.decrypt_packet(&parsed) {
                Ok(decrypted) => {
                    println!("Decrypted: {:02x?}", decrypted);
                }
                Err(e) => {
                    println!("Decryption failed: {}", e);
                }
            }
        }
    }
}

Build Your Own Tool

Template for a custom Zigbee analysis tool:

use clap::Parser;
use zigbee_drivers::DriverRegistry;
use zigbee_hal::traits::ZigbeeCapture;

#[derive(Parser)]
#[clap(name = "my-zigbee-tool")]
#[clap(about = "Custom Zigbee analysis tool")]
struct Args {
    /// Channel to monitor (11-26)
    #[clap(short, long, default_value = "15")]
    channel: u8,
    
    /// Output file
    #[clap(short, long)]
    output: Option<String>,
    
    /// Verbose output
    #[clap(short, long)]
    verbose: bool,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let args = Args::parse();
    
    // Setup logging
    if args.verbose {
        tracing_subscriber::fmt::init();
    }
    
    // Initialize driver
    let registry = DriverRegistry::new();
    let devices = registry.detect_devices();
    let mut driver = registry.create_driver(&devices[0]).unwrap();
    
    driver.initialize().await?;
    driver.set_channel(args.channel).await?;
    
    println!("My Zigbee Tool - Channel {}", args.channel);
    
    // Your custom logic here
    loop {
        let packet = driver.capture_packet().await?;
        // Process packet...
    }
}

See Also:

API Reference - Detailed API documentation
Architecture - Understanding the codebase
Protocol Documentation - Protocol details
Driver Development - Creating custom drivers

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Examples

Examples

Table of Contents

Basic Capture

PCAP Recording

Channel Scanning

Network Topology Analysis

Packet Filtering

Custom Analysis

Using the API

Add Dependencies

Simple Integration

Advanced Examples

Multi-Channel Monitoring

Real-Time Decryption

Build Your Own Tool

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Clone this wiki locally