controlling acquisition rate

[DiTo97]

it would be nice if the streamer had a way to control the acquisition rate.

any plans for that? might be duplicate of #6.

[BBBBBBob]

@DiTo97 Hi, I am also interested in this issue, have you found any solutions?

[younghyopark]

Hi @DiTo97 @BBBBBBob ,

With the recently updated app, I'm getting

Results:
Duration: 10.00 seconds
Total Messages Received: 1391
Unique Messages Received: 904
Average Message Rate: 139.04 Hz
Average Unique Data Rate: 90.36 Hz

which the 90Hz cap, I belive, is coming from ARKit's hand tracking acquisition rate limitation rather than the gRPC side. I'm not sure if there's any way to increase the acquisition rate of ARKit hand tracking.

---

I've used this script:

import time
import argparse
from avp_stream import VisionProStreamer


class BenchmarkStreamer(VisionProStreamer):
    def __init__(self, *args, **kwargs):
        self.message_count = 0
        self.unique_count = 0
        self.last_position = None
        self.measuring = False
        super().__init__(*args, **kwargs)
        
    def _process_hand_update(self, hand_update, source="grpc"):
        super()._process_hand_update(hand_update, source)
        
        if not self.measuring:
            return
            
        self.message_count += 1
        
        current_pos = (
            hand_update.right_hand.wristMatrix.m03,
            hand_update.right_hand.wristMatrix.m13,
            hand_update.right_hand.wristMatrix.m23
        )
        
        if self.last_position != current_pos:
            self.unique_count += 1
            self.last_position = current_pos


def main():
    parser = argparse.ArgumentParser(description="Benchmark Hand Tracking Data Rate")
    parser.add_argument("--ip", type=str, required=True, help="Vision Pro IP address")
    parser.add_argument("--duration", type=float, default=10.0, help="Duration of the benchmark in seconds")
    args = parser.parse_args()

    print(f"Connecting to Vision Pro at {args.ip}...")
    streamer = BenchmarkStreamer(ip=args.ip, record=False)
    
        
    print(f"Collecting data for {args.duration} seconds...")
    streamer.measuring = True
    start_time = time.time()
    time.sleep(args.duration)
    streamer.measuring = False
    duration = time.time() - start_time
    
    print(f"\nResults:")
    print(f"Duration: {duration:.2f} seconds")
    print(f"Total Messages: {streamer.message_count}")
    print(f"Unique Messages: {streamer.unique_count}")
    print(f"Message Rate: {streamer.message_count / duration:.2f} Hz")
    print(f"Unique Data Rate: {streamer.unique_count / duration:.2f} Hz")

if __name__ == "__main__":
    main()