Function measure_streaming_frequency

avp_stream/sim_latency_test.py:314–366

Measure the actual streaming frequency achieved. Args: streamer: VisionProStreamer instance model: MuJoCo model data: MuJoCo data duration: Test duration in seconds target_hz: Target update rate in Hz Returns: Dictionary with

(streamer: VisionProStreamer, 
                                 model, data,
                                 duration: float = 5.0,
                                 target_hz: float = 120.0)

def measure_streaming_frequency(streamer: VisionProStreamer, 
                                 model, data,
                                 duration: float = 5.0,
                                 target_hz: float = 120.0) -> dict:
    """
    Measure the actual streaming frequency achieved.
    
    Args:
        streamer: VisionProStreamer instance
        model: MuJoCo model
        data: MuJoCo data
        duration: Test duration in seconds
        target_hz: Target update rate in Hz
        
    Returns:
        Dictionary with frequency statistics
    """
    update_count = 0
    update_times = []
    last_time = None
    
    target_dt = 1.0 / target_hz
    start_time = time.perf_counter()
    
    while time.perf_counter() - start_time < duration:
        mujoco.mj_step(model, data)
        streamer.update_sim()
        update_count += 1
        
        now = time.perf_counter()
        if last_time is not None:
            update_times.append(now - last_time)
        last_time = now
        
        # Sleep to maintain target rate
        elapsed = time.perf_counter() - start_time
        expected_updates = int(elapsed * target_hz)
        if update_count > expected_updates:
            sleep_time = (update_count / target_hz) - elapsed
            if sleep_time > 0:
                time.sleep(sleep_time)
    
    total_time = time.perf_counter() - start_time
    actual_hz = update_count / total_time
    
    return {
        "target_hz": target_hz,
        "actual_hz": actual_hz,
        "update_count": update_count,
        "duration_s": total_time,
        "mean_dt_ms": float(np.mean(update_times)) * 1000 if update_times else 0,
        "std_dt_ms": float(np.std(update_times)) * 1000 if update_times else 0,
    }


def run_frequency_sweep(args):