Gemini Robotics ER - Live API Examples

[thorwebdev]

No description provided.

[gemini-code-assist]

Code Review

This pull request introduces a comprehensive physical agent server and web UI designed to connect Gemini Robotics ER with physical robot embodiments (such as Boston Dynamics Spot and Tinybot) using the Gemini Live API. The implementation includes modular agent configurations, a custom async event bus, real-time audio/video streaming handlers, and pixel-grounded manipulation tools. Key feedback from the review highlights several critical improvements: preventing state leaks in the TTS background task by shielding cleanup from cancellation, ensuring observation loops survive WebSocket reconnection, optimizing CPU-intensive calculations (such as frame differences and peak audio amplitude) using NumPy, avoiding runtime errors in the event bus by iterating over a copy of the handler set, instantly cutting off audio playback on model interruption, and guarding against division-by-zero errors in the audio downsampling processor.

[gemini-code-assist]

In _run_tts_background, the finally block awaits self._decrement_tts_and_maybe_clear(). If the background task is cancelled (which happens frequently when new text arrives or the user interrupts), the await inside the finally block can raise a CancelledError before acquiring the lock or decrementing the counter. This will permanently leak the _pending_tts_count and keep the speaking event set, blocking future heartbeats. Wrap the cleanup call in asyncio.shield to ensure it runs to completion even during task cancellation.

Suggested change

	finally:
	await self._decrement_tts_and_maybe_clear()
	finally:
	await asyncio.shield(self._decrement_tts_and_maybe_clear())

[gemini-code-assist]

The background observation loops (_send_audio, _send_video, _send_text) will permanently terminate if send_message raises any exception (e.g., due to a temporary WebSocket disconnection). Although _reconnect() successfully re-establishes the WebSocket stream, the observation loops are never restarted, leaving the session unable to stream any further audio, video, or text. Consider handling connection errors gracefully within send_message or the observation loops to prevent them from dying.

[gemini-code-assist]

The generator expression sum(abs(left - right) for left, right in zip(previous_pixels, current_pixels)) is executed in pure Python for every frame difference calculation. Since previous_pixels and current_pixels contain thousands of bytes, this is highly inefficient and can block the asyncio event loop. Since numpy is already a project dependency, you can use numpy.frombuffer to perform this calculation in compiled C code, which is orders of magnitude faster.

Suggested change

	difference = sum(
	abs(left - right)
	for left, right in zip(previous_pixels, current_pixels)
	)
	return difference / (len(previous_pixels) * 255.0)
	import numpy as np
	prev_arr = np.frombuffer(previous_pixels, dtype=np.uint8)
	curr_arr = np.frombuffer(current_pixels, dtype=np.uint8)
	difference = np.sum(np.abs(prev_arr.astype(np.int32) - curr_arr.astype(np.int32)))
	return difference / (len(previous_pixels) * 255.0)

[gemini-code-assist]

Iterating directly over handlers (which is a set from self._handlers_by_event_type) during event dispatch is risky. If any handler subscribes or unsubscribes during the dispatch loop, it will raise a RuntimeError: Set size changed during iteration. Iterate over a copy of the set to prevent this.

Suggested change

	for handler in handlers:
	for handler in list(handlers):

[gemini-code-assist]

Calculating the peak audio amplitude using a pure Python generator expression max(abs(s) for s in samples) for every incoming audio chunk is highly CPU-intensive and can block the asyncio event loop. Since numpy is a project dependency, use numpy to perform this calculation efficiently.

        if len(chunk) >= 2:
          import numpy as np
          samples_arr = np.frombuffer(chunk, dtype=np.int16)
          peak = np.max(np.abs(samples_arr))
          if peak > 100:

[gemini-code-assist]

When the model is interrupted, audioFlushPending is set to true, but mediaHandler.stopAudioPlayback() is not called immediately. The currently playing audio buffer will continue to play until the next audio chunk arrives (which might be delayed or never happen if the model remains silent). Call mediaHandler.stopAudioPlayback() immediately to instantly cut off the speech.

Suggested change

	if (msg.type === 'interrupted') {
	audioFlushPending = true;
	if (msg.type === 'interrupted') {
	mediaHandler.stopAudioPlayback();
	audioFlushPending = true;

[gemini-code-assist]

In downsampleBuffer, if count is 0 (which can happen due to rounding at the end of the buffer), dividing accum / count will result in NaN values in the downsampled audio buffer. Add a guard to default to 0 if count is 0.

Suggested change

	let accum = 0, count = 0;
	for (let i = offsetBuffer; i < nextOffsetBuffer && i < buffer.length;
	i++) {
	accum += buffer[i];
	count++;
	}
	result[offsetResult] = accum / count;
	let accum = 0, count = 0;
	for (let i = offsetBuffer; i < nextOffsetBuffer && i < buffer.length;
	i++) {
	accum += buffer[i];
	count++;
	}
	result[offsetResult] = count > 0 ? accum / count : 0;