App

elva.apps.editor.app

App definition.

Classes:

• UI –

  User interface.

Attributes:

• LANGUAGES –

  Supported languages.

LANGUAGES = {'py': 'python', 'md': 'markdown', 'sh': 'bash', 'js': 'javascript', 'rs': 'rust', 'yml': 'yaml'} module-attribute

Supported languages.

UI(config)

Bases: App

User interface.

Parameters:

• config (dict) –

  mapping of configuration parameters to their values.

Methods:

• on_provider_exception –

  Wrapper method around the provider exception handler

• on_awareness_update –

  Wrapper method around the

• wait_for_component_state –

  Wait for a component to set a specific state.

• on_mount –

  Hook called on mounting the app.

• on_unmount –

  Hook called on unmounting the app.

• compose –

  Hook arranging child widgets.

• action_save –

  Action performed on triggering the save key binding.

• get_and_set_file_paths –

  Get and set the data or render file paths after the input prompt.

• action_render –

  Action performed on triggering the render key binding.

• action_toggle_dashboard –

  Action performed on triggering the toggle_dashboard key binding.

• push_client_states –

  Method pushing the client states to the active dashboard.

• push_config –

  Method pushing the configuration mapping to the active dashboard.

Attributes:

• CSS_PATH –

  The path to the default CSS.

• SCREENS –

  The installed screens.

• BINDINGS –

  Key bindings for actions of the app.

• language (str) –

  The language the text document is written in.

Source code in src/elva/apps/editor/app.py

def __init__(self, config: dict):
    """
    Arguments:
        config: mapping of configuration parameters to their values.
    """
    self.config = c = config

    ansi_color = c.get("ansi_color", False)
    super().__init__(ansi_color=ansi_color)

    # document structure
    self.ydoc = Doc()
    self.ytext = Text()
    self.ydoc["ytext"] = self.ytext

    self.components = list()

    if c.get("host") is not None:
        self.provider = WebsocketProvider(
            self.ydoc,
            c["identifier"],
            c["host"],
            port=c.get("port"),
            safe=c.get("safe", True),
            on_exception=self.on_provider_exception,
        )

        data = {}
        if c.get("name") is not None:
            data = {"user": {"name": c["name"]}}

        self.provider.awareness.set_local_state(data)

        self.components.append(self.provider)

    if c.get("file") is not None:
        self.store = SQLiteStore(
            self.ydoc,
            c["identifier"],
            c["file"],
        )
        self.components.append(self.store)

    if c.get("render") is not None:
        self.renderer = TextRenderer(
            self.ytext,
            c["render"],
            c.get("auto_save", False),
            c.get("timeout", 300),
        )
        self.components.append(self.renderer)

    self._language = c.get("language")

CSS_PATH = 'style.tcss' class-attribute instance-attribute

The path to the default CSS.

SCREENS = {'dashboard': Dashboard, 'input': InputScreen} class-attribute instance-attribute

The installed screens.

BINDINGS = [Binding('ctrl+b', 'toggle_dashboard', 'Toggle the dashboard'), Binding('ctrl+s', 'save', 'Save to data file'), Binding('ctrl+r', 'render', 'Render to file')] class-attribute instance-attribute

Key bindings for actions of the app.

language property

The language the text document is written in.

on_provider_exception(exc, config)

Wrapper method around the provider exception handler _on_provider_exception.

Parameters:

• exc (WebSocketException) –

  the exception raised by the provider.

• config (dict) –

  the configuration stored in the provider.

Source code in src/elva/apps/editor/app.py

def on_provider_exception(self, exc: WebSocketException, config: dict):
    """
    Wrapper method around the provider exception handler
    [`_on_provider_exception`][elva.apps.editor.app.UI._on_provider_exception].

    Arguments:
        exc: the exception raised by the provider.
        config: the configuration stored in the provider.
    """
    self.run_worker(self._on_provider_exception(exc, config))

_on_provider_exception(exc, config) async

Handler for exceptions raised by the provider.

It exits the app after displaying the error message to the user.

Parameters:

• exc (WebSocketException) –

  the exception raised by the provider.

• config (dict) –

  the configuration stored in the provider.

Source code in src/elva/apps/editor/app.py

async def _on_provider_exception(self, exc: WebSocketException, config: dict):
    """
    Handler for exceptions raised by the provider.

    It exits the app after displaying the error message to the user.

    Arguments:
        exc: the exception raised by the provider.
        config: the configuration stored in the provider.
    """
    await self.provider.stop()

    if type(exc) is InvalidStatus:
        response = exc.response
        exc = f"HTTP {response.status_code}: {response.reason_phrase}"

    await self.push_screen_wait(ErrorScreen(exc))
    self.exit(return_code=1)

on_awareness_update(topic, data)

Wrapper method around the _on_awareness_update callback.

Parameters:

• topic (Literal['update', 'change']) –

  the topic under which the changes are published.

• data (tuple[dict, Any]) –

  manipulation actions taken as well as the origin of the changes.

Source code in src/elva/apps/editor/app.py

def on_awareness_update(
    self, topic: Literal["update", "change"], data: tuple[dict, Any]
):
    """
    Wrapper method around the
    [`_on_awareness_update`][elva.apps.editor.app.UI._on_awareness_update]
    callback.

    Arguments:
        topic: the topic under which the changes are published.
        data: manipulation actions taken as well as the origin of the changes.
    """
    if topic == "change":
        self.run_worker(self._on_awareness_update(topic, data))

_on_awareness_update(topic, data) async

Hook called on a change in the awareness states.

It pushes client states to the dashboard and removes offline client IDs from the future.

Parameters:

• topic (Literal['update', 'change']) –

  the topic under which the changes are published.

• data (tuple[dict, Any]) –

  manipulation actions taken as well as the origin of the changes.

Source code in src/elva/apps/editor/app.py

async def _on_awareness_update(
    self, topic: Literal["update", "change"], data: tuple[dict, Any]
):
    """
    Hook called on a change in the awareness states.

    It pushes client states to the dashboard and removes offline client IDs from the future.

    Arguments:
        topic: the topic under which the changes are published.
        data: manipulation actions taken as well as the origin of the changes.
    """
    if self.screen == self.get_screen("dashboard"):
        self.push_client_states()

wait_for_component_state(component, state) async

Wait for a component to set a specific state.

Parameters:

• component (Component) –

  the component of interest.

• state (ComponentState) –

  the awaited state.

Source code in src/elva/apps/editor/app.py

async def wait_for_component_state(
    self, component: Component, state: ComponentState
):
    """
    Wait for a component to set a specific state.

    Arguments:
        component: the component of interest.
        state: the awaited state.
    """
    sub = component.subscribe()
    while state != component.state:
        await sub.receive()
    component.unsubscribe(sub)

on_mount() async

Hook called on mounting the app.

Source code in src/elva/apps/editor/app.py

async def on_mount(self):
    """
    Hook called on mounting the app.
    """
    if hasattr(self, "provider"):
        self.subscription = self.provider.awareness.observe(
            self.on_awareness_update
        )

    # load text from rendered file
    c = self.config

    text = ""

    render_file_path = c.get("render")
    if render_file_path is not None and render_file_path.exists():
        # we found some content on disk;
        # now check whether this has precedence over the data file
        data_file_path = c.get("file")
        if data_file_path is None or not data_file_path.exists():
            # there is no data file on disk associated with this
            # file name; we load the content
            with render_file_path.open(mode="r") as fd:
                text = fd.read()

    # wait for components to run
    for comp in self.components:
        self.run_worker(comp.start())
        await self.wait_for_component_state(
            comp, comp.states.ACTIVE | comp.states.RUNNING
        )

    # now add the text to save updates to disk and send them over wire
    if text:
        ytextarea = self.query_one(YTextArea)
        ytextarea.load_text(text)

on_unmount() async

Hook called on unmounting the app.

Source code in src/elva/apps/editor/app.py

async def on_unmount(self):
    """
    Hook called on unmounting the app.
    """
    if hasattr(self, "subscription"):
        self.provider.awareness.unobserve(self.subscription)
        del self.subscription

    for comp in self.components:
        await self.wait_for_component_state(comp, comp.states.NONE)

compose()

Hook arranging child widgets.

Source code in src/elva/apps/editor/app.py

def compose(self):
    """
    Hook arranging child widgets.
    """
    yield YTextArea(
        self.ytext,
        tab_behavior="indent",
        show_line_numbers=True,
        id="editor",
        language=self.language,
    )

action_save() async

Action performed on triggering the save key binding.

Source code in src/elva/apps/editor/app.py

async def action_save(self):
    """
    Action performed on triggering the `save` key binding.
    """
    if self.config.get("file") is None:
        self.run_worker(self.get_and_set_file_paths())

get_and_set_file_paths(data_file=True) async

Get and set the data or render file paths after the input prompt.

Parameters:

• data_file (bool, default: True ) –

  flag whether to add a data file path to the config.

Source code in src/elva/apps/editor/app.py

async def get_and_set_file_paths(self, data_file: bool = True):
    """
    Get and set the data or render file paths after the input prompt.

    Arguments:
        data_file: flag whether to add a data file path to the config.
    """
    name = await self.push_screen_wait("input")

    if not name:
        return

    path = Path(name)

    data_file_path = get_data_file_path(path)
    if data_file:
        self.config["file"] = data_file_path
        self.store = SQLiteStore(
            self.ydoc, self.config["identifier"], data_file_path
        )
        self.components.append(self.store)
        self.run_worker(self.store.start())

    if self.config.get("render") is None:
        render_file_path = get_render_file_path(data_file_path)
        self.config["render"] = render_file_path

        self.renderer = TextRenderer(
            self.ytext, render_file_path, self.config.get("auto_save", False)
        )
        self.components.append(self.renderer)
        self.run_worker(self.renderer.start())

    if self.screen == self.get_screen("dashboard"):
        self.push_config()

action_render() async

Action performed on triggering the render key binding.

Source code in src/elva/apps/editor/app.py

async def action_render(self):
    """
    Action performed on triggering the `render` key binding.
    """
    if self.config.get("render") is None:
        self.run_worker(self.get_and_set_file_paths(data_file=False))
    else:
        await self.renderer.write()

action_toggle_dashboard() async

Action performed on triggering the toggle_dashboard key binding.

Source code in src/elva/apps/editor/app.py

async def action_toggle_dashboard(self):
    """
    Action performed on triggering the `toggle_dashboard` key binding.
    """
    if self.screen == self.get_screen("dashboard"):
        self.pop_screen()
    else:
        await self.push_screen("dashboard")
        self.push_client_states()
        self.push_config()

push_client_states()

Method pushing the client states to the active dashboard.

Source code in src/elva/apps/editor/app.py

def push_client_states(self):
    """
    Method pushing the client states to the active dashboard.
    """
    if hasattr(self, "provider"):
        client_states = self.provider.awareness.client_states.copy()
        client_id = self.provider.awareness.client_id
        if client_id not in client_states:
            return
        states = list()
        states.append((client_id, client_states.pop(client_id)))
        states.extend(list(client_states.items()))
        states = tuple(states)

        awareness_view = self.screen.query_one(AwarenessView)
        awareness_view.states = states

push_config()

Method pushing the configuration mapping to the active dashboard.

Source code in src/elva/apps/editor/app.py

def push_config(self):
    """
    Method pushing the configuration mapping to the active dashboard.
    """
    config = tuple(self.config.items())

    config_view = self.screen.query_one(ConfigView)
    config_view.config = config