Observers API Reference

This page provides detailed documentation for the observers in GUM.

Base Observer

gum.observers.Observer(name: Optional[str] = None)

Bases: ABC

Base class for all observers in the GUM system.

This abstract base class defines the interface for all observers that monitor user behavior. Observers are responsible for collecting data about user interactions and sending updates through an asynchronous queue.

Parameters:

Name	Type	Description	Default
name	Optional[str]	A custom name for the observer. If not provided, the class name will be used.	None

Attributes:

Name	Type	Description
update_queue	Queue	Queue for sending updates to the main GUM system.
_name	str	The name of the observer.
_running	bool	Flag indicating if the observer is currently running.
_task	Optional[Task]	Background task handle for the observer's worker.

Source code in gum/observers/observer.py

def __init__(self, name: Optional[str] = None) -> None:
    self.update_queue = asyncio.Queue()
    self._name = name or self.__class__.__name__

    # running flag + background task handle
    self._running = True
    self._task: asyncio.Task | None = asyncio.create_task(self._worker_wrapper())

Attributes

name: str property

Get the name of the observer.

Returns:

Name	Type	Description
str	str	The observer's name.

update_queue = asyncio.Queue() instance-attribute

Functions

get_update() async

Get the next update from the queue if available.

Returns:

Type	Description
	Optional[Update]: The next update from the queue, or None if the queue is empty.

Source code in gum/observers/observer.py

async def get_update(self):
    """Get the next update from the queue if available.

    Returns:
        Optional[Update]: The next update from the queue, or None if the queue is empty.
    """
    try:
        return self.update_queue.get_nowait()
    except asyncio.QueueEmpty:
        return None

stop() -> None async

Stop the observer and clean up resources.

This method cancels the worker task and drains the update queue.

Source code in gum/observers/observer.py

async def stop(self) -> None:
    """Stop the observer and clean up resources.

    This method cancels the worker task and drains the update queue.
    """
    if self._task and not self._task.done():
        self._task.cancel()
        try:
            await self._task
        except asyncio.CancelledError:
            pass
    # unblock any awaiters
    while not self.update_queue.empty():
        self.update_queue.get_nowait()

Screen Observer

gum.observers.Screen(model_name: str = 'gpt-4o-mini', screenshots_dir: str = '~/.cache/gum/screenshots', skip_when_visible: Optional[str | list[str]] = None, transcription_prompt: Optional[str] = None, summary_prompt: Optional[str] = None, history_k: int = 10, debug: bool = False, api_key: str | None = None, api_base: str | None = None)

Bases: Observer

Observer that captures and analyzes screen content around user interactions.

This observer captures screenshots before and after user interactions (mouse movements, clicks, and scrolls) and uses GPT-4 Vision to analyze the content. It can also take periodic screenshots and skip captures when certain applications are visible.

Parameters:

Name	Type	Description	Default
screenshots_dir	str	Directory to store screenshots. Defaults to "~/.cache/gum/screenshots".	'~/.cache/gum/screenshots'
skip_when_visible	Optional[str | list[str]]	Application names to skip when visible. Defaults to None.	None
transcription_prompt	Optional[str]	Custom prompt for transcribing screenshots. Defaults to None.	None
summary_prompt	Optional[str]	Custom prompt for summarizing screenshots. Defaults to None.	None
model_name	str	GPT model to use for vision analysis. Defaults to "gpt-4o-mini".	'gpt-4o-mini'
history_k	int	Number of recent screenshots to keep in history. Defaults to 10.	10
debug	bool	Enable debug logging. Defaults to False.	False

Attributes:

Name	Type	Description
_CAPTURE_FPS	int	Frames per second for screen capture.
_PERIODIC_SEC	int	Seconds between periodic screenshots.
_DEBOUNCE_SEC	int	Seconds to wait before processing an interaction.
_MON_START	int	Index of first real display in mss.

Initialize the Screen observer.

Parameters:

Name	Type	Description	Default
screenshots_dir	str	Directory to store screenshots. Defaults to "~/.cache/gum/screenshots".	'~/.cache/gum/screenshots'
skip_when_visible	Optional[str | list[str]]	Application names to skip when visible. Defaults to None.	None
transcription_prompt	Optional[str]	Custom prompt for transcribing screenshots. Defaults to None.	None
summary_prompt	Optional[str]	Custom prompt for summarizing screenshots. Defaults to None.	None
model_name	str	GPT model to use for vision analysis. Defaults to "gpt-4o-mini".	'gpt-4o-mini'
history_k	int	Number of recent screenshots to keep in history. Defaults to 10.	10
debug	bool	Enable debug logging. Defaults to False.	False

Source code in gum/observers/screen.py

def __init__(
    self,
    model_name: str = "gpt-4o-mini",
    screenshots_dir: str = "~/.cache/gum/screenshots",
    skip_when_visible: Optional[str | list[str]] = None,
    transcription_prompt: Optional[str] = None,
    summary_prompt: Optional[str] = None,
    history_k: int = 10,
    debug: bool = False,
    api_key: str | None = None,
    api_base: str | None = None,
) -> None:
    """Initialize the Screen observer.

    Args:
        screenshots_dir (str, optional): Directory to store screenshots. Defaults to "~/.cache/gum/screenshots".
        skip_when_visible (Optional[str | list[str]], optional): Application names to skip when visible.
            Defaults to None.
        transcription_prompt (Optional[str], optional): Custom prompt for transcribing screenshots.
            Defaults to None.
        summary_prompt (Optional[str], optional): Custom prompt for summarizing screenshots.
            Defaults to None.
        model_name (str, optional): GPT model to use for vision analysis. Defaults to "gpt-4o-mini".
        history_k (int, optional): Number of recent screenshots to keep in history. Defaults to 10.
        debug (bool, optional): Enable debug logging. Defaults to False.
    """
    self.screens_dir = os.path.abspath(os.path.expanduser(screenshots_dir))
    os.makedirs(self.screens_dir, exist_ok=True)

    self._guard = {skip_when_visible} if isinstance(skip_when_visible, str) else set(skip_when_visible or [])

    self.transcription_prompt = transcription_prompt or TRANSCRIPTION_PROMPT
    self.summary_prompt = summary_prompt or SUMMARY_PROMPT
    self.model_name = model_name

    self.debug = debug

    # state shared with worker
    self._frames: Dict[int, Any] = {}
    self._frame_lock = asyncio.Lock()

    self._history: deque[str] = deque(maxlen=max(0, history_k))
    self._pending_event: Optional[dict] = None
    self._debounce_handle: Optional[asyncio.TimerHandle] = None
    self.client = AsyncOpenAI(
        # try the class, then the env for screen, then the env for gum
        base_url=api_base or os.getenv("SCREEN_LM_API_BASE") or os.getenv("GUM_LM_API_BASE"), 

        # try the class, then the env for screen, then the env for GUM, then none
        api_key=api_key or os.getenv("SCREEN_LM_API_KEY") or os.getenv("GUM_LM_API_KEY") or os.getenv("OPENAI_API_KEY") or "None"
    )

    # call parent
    super().__init__()

Attributes

client = AsyncOpenAI(base_url=api_base or os.getenv('SCREEN_LM_API_BASE') or os.getenv('GUM_LM_API_BASE'), api_key=api_key or os.getenv('SCREEN_LM_API_KEY') or os.getenv('GUM_LM_API_KEY') or os.getenv('OPENAI_API_KEY') or 'None') instance-attribute

debug = debug instance-attribute

model_name = model_name instance-attribute

screens_dir = os.path.abspath(os.path.expanduser(screenshots_dir)) instance-attribute

summary_prompt = summary_prompt or SUMMARY_PROMPT instance-attribute

transcription_prompt = transcription_prompt or TRANSCRIPTION_PROMPT instance-attribute

Functions