Awareness

elva.awareness

Module with the awareness component definition.

Classes:

• Awareness –

  Component version of pycrdt's Awareness class.

Attributes:

• AwarenessState –

  The states of the Awareness component.

AwarenessState = create_component_state('AwarenessState') module-attribute

The states of the Awareness component.

Awareness

Bases: Component, Awareness

Component version of pycrdt's Awareness class.

Methods:

• run –

  Hook performing periodic awareness updates.

• cleanup –

  Hook removing the local awareness state.

• observe –

  Add a callback to be run on awareness state changes.

• unobserve –

  Remove a registered callback.

Attributes:

• client_states (dict) –

  The client states.

• states (AwarenessState) –

  The states this component can have.

client_states property

The client states.

states property

The states this component can have.

run() async

Hook performing periodic awareness updates.

Source code in src/elva/awareness.py

async def run(self):
    """
    Hook performing periodic awareness updates.
    """
    await self._start()

cleanup() async

Hook removing the local awareness state.

Source code in src/elva/awareness.py

async def cleanup(self):
    """
    Hook removing the local awareness state.
    """
    self.remove_awareness_states([self.client_id], origin="local")

observe(callback)

Add a callback to be run on awareness state changes.

Parameters:

• callback (Callable[[str, tuple[dict[str, Any], Any]], None]) –

  the function to call on state changes.

Returns:

• str –

  the observer identifier to use for unobserve.

Source code in src/elva/awareness.py

def observe(
    self, callback: Callable[[str, tuple[dict[str, Any], Any]], None]
) -> str:
    """
    Add a callback to be run on awareness state changes.

    Arguments:
        callback: the function to call on state changes.

    Returns:
        the observer identifier to use for [`unobserve`][elva.awareness.Awareness.unobserve].
    """
    observer_id = super().observe(callback)
    self.log.info(f"added observer {observer_id}")

    return observer_id

unobserve(observer_id)

Remove a registered callback.

Parameters:

• observer_id (str) –

  the identifier associated with the callback as given by observe.

Source code in src/elva/awareness.py

def unobserve(self, observer_id: str):
    """
    Remove a registered callback.

    Arguments:
        observer_id: the identifier associated with the callback as given by [`observe`][elva.awareness.Awareness.observe].
    """
    super().unobserve(observer_id)
    self.log.info(f"removed observer {observer_id}")